SSE服务器推送事件原理深度解析与实战应用

一、什么是SSE

SSE（Server-Sent Events） 是一种基于HTTP的服务器推送技术，允许服务器向客户端单向发送事件流。它是HTML5规范的一部分，专为服务器到客户端的单向实时通信而设计。

1.1 SSE的核心特性

• 单向通信：仅支持服务器向客户端推送数据
• 基于HTTP：使用标准HTTP协议，无需额外的连接建立
• 自动重连：浏览器会自动处理断线重连
• 文本格式：传输的数据必须是文本格式
• 事件模型：支持事件类型和ID，便于客户端处理

二、SSE工作原理

2.1 连接建立流程

服务器客户端服务器客户端#mermaid-svg-XqZ19Gdk79LGThdG{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-XqZ19Gdk79LGThdG .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-XqZ19Gdk79LGThdG .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-XqZ19Gdk79LGThdG .error-icon{fill:#552222;}#mermaid-svg-XqZ19Gdk79LGThdG .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-XqZ19Gdk79LGThdG .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-XqZ19Gdk79LGThdG .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-XqZ19Gdk79LGThdG .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-XqZ19Gdk79LGThdG .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-XqZ19Gdk79LGThdG .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-XqZ19Gdk79LGThdG .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-XqZ19Gdk79LGThdG .marker{fill:#333333;stroke:#333333;}#mermaid-svg-XqZ19Gdk79LGThdG .marker.cross{stroke:#333333;}#mermaid-svg-XqZ19Gdk79LGThdG svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-XqZ19Gdk79LGThdG p{margin:0;}#mermaid-svg-XqZ19Gdk79LGThdG .actor{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-XqZ19Gdk79LGThdG text.actor>tspan{fill:black;stroke:none;}#mermaid-svg-XqZ19Gdk79LGThdG .actor-line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-XqZ19Gdk79LGThdG .innerArc{stroke-width:1.5;stroke-dasharray:none;}#mermaid-svg-XqZ19Gdk79LGThdG .messageLine0{stroke-width:1.5;stroke-dasharray:none;stroke:#333;}#mermaid-svg-XqZ19Gdk79LGThdG .messageLine1{stroke-width:1.5;stroke-dasharray:2,2;stroke:#333;}#mermaid-svg-XqZ19Gdk79LGThdG #arrowhead path{fill:#333;stroke:#333;}#mermaid-svg-XqZ19Gdk79LGThdG .sequenceNumber{fill:white;}#mermaid-svg-XqZ19Gdk79LGThdG #sequencenumber{fill:#333;}#mermaid-svg-XqZ19Gdk79LGThdG #crosshead path{fill:#333;stroke:#333;}#mermaid-svg-XqZ19Gdk79LGThdG .messageText{fill:#333;stroke:none;}#mermaid-svg-XqZ19Gdk79LGThdG .labelBox{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-XqZ19Gdk79LGThdG .labelText,#mermaid-svg-XqZ19Gdk79LGThdG .labelText>tspan{fill:black;stroke:none;}#mermaid-svg-XqZ19Gdk79LGThdG .loopText,#mermaid-svg-XqZ19Gdk79LGThdG .loopText>tspan{fill:black;stroke:none;}#mermaid-svg-XqZ19Gdk79LGThdG .loopLine{stroke-width:2px;stroke-dasharray:2,2;stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);}#mermaid-svg-XqZ19Gdk79LGThdG .note{stroke:#aaaa33;fill:#fff5ad;}#mermaid-svg-XqZ19Gdk79LGThdG .noteText,#mermaid-svg-XqZ19Gdk79LGThdG .noteText>tspan{fill:black;stroke:none;}#mermaid-svg-XqZ19Gdk79LGThdG .activation0{fill:#f4f4f4;stroke:#666;}#mermaid-svg-XqZ19Gdk79LGThdG .activation1{fill:#f4f4f4;stroke:#666;}#mermaid-svg-XqZ19Gdk79LGThdG .activation2{fill:#f4f4f4;stroke:#666;}#mermaid-svg-XqZ19Gdk79LGThdG .actorPopupMenu{position:absolute;}#mermaid-svg-XqZ19Gdk79LGThdG .actorPopupMenuPanel{position:absolute;fill:#ECECFF;box-shadow:0px 8px 16px 0px rgba(0,0,0,0.2);filter:drop-shadow(3px 5px 2px rgb(0 0 0 / 0.4));}#mermaid-svg-XqZ19Gdk79LGThdG .actor-man line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;}#mermaid-svg-XqZ19Gdk79LGThdG .actor-man circle,#mermaid-svg-XqZ19Gdk79LGThdG line{stroke:hsl(259.6261682243, 59.7765363128%, 87.9019607843%);fill:#ECECFF;stroke-width:2px;}#mermaid-svg-XqZ19Gdk79LGThdG :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}保持连接开放持续传输…发送HTTP请求Accept: text/event-stream返回200 OKContent-Type: text/event-streamdata: 消息1\\n\\ndata: 消息2\\n\\ndata: 消息3\\n\\n连接断开自动重连请求Last-Event-ID: xxx继续发送后续消息

2.2 SSE数据格式

SSE的传输格式非常简单，每条消息由一个或多个字段组成，以换行符分隔：

字段名: 值\\n

常用字段：

• data: 消息内容（必需）
• event: 事件类型（可选）
• id: 事件ID（可选）
• retry: 重连间隔时间（可选，单位毫秒）

示例：

id: 1
event: message
data: {"content": "Hello SSE"}

id: 2
event: notification
data: {"title": "新消息", "count": 5}

data: 简单文本消息

注意： 每条消息必须以两个换行符 \\n\\n 结束。

2.3 消息类型处理

#mermaid-svg-HJDL3gbvbU1eetIe{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-HJDL3gbvbU1eetIe .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-HJDL3gbvbU1eetIe .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-HJDL3gbvbU1eetIe .error-icon{fill:#552222;}#mermaid-svg-HJDL3gbvbU1eetIe .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-HJDL3gbvbU1eetIe .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-HJDL3gbvbU1eetIe .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-HJDL3gbvbU1eetIe .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-HJDL3gbvbU1eetIe .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-HJDL3gbvbU1eetIe .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-HJDL3gbvbU1eetIe .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-HJDL3gbvbU1eetIe .marker{fill:#333333;stroke:#333333;}#mermaid-svg-HJDL3gbvbU1eetIe .marker.cross{stroke:#333333;}#mermaid-svg-HJDL3gbvbU1eetIe svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-HJDL3gbvbU1eetIe p{margin:0;}#mermaid-svg-HJDL3gbvbU1eetIe .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-HJDL3gbvbU1eetIe .cluster-label text{fill:#333;}#mermaid-svg-HJDL3gbvbU1eetIe .cluster-label span{color:#333;}#mermaid-svg-HJDL3gbvbU1eetIe .cluster-label span p{background-color:transparent;}#mermaid-svg-HJDL3gbvbU1eetIe .label text,#mermaid-svg-HJDL3gbvbU1eetIe span{fill:#333;color:#333;}#mermaid-svg-HJDL3gbvbU1eetIe .node rect,#mermaid-svg-HJDL3gbvbU1eetIe .node circle,#mermaid-svg-HJDL3gbvbU1eetIe .node ellipse,#mermaid-svg-HJDL3gbvbU1eetIe .node polygon,#mermaid-svg-HJDL3gbvbU1eetIe .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-HJDL3gbvbU1eetIe .rough-node .label text,#mermaid-svg-HJDL3gbvbU1eetIe .node .label text,#mermaid-svg-HJDL3gbvbU1eetIe .image-shape .label,#mermaid-svg-HJDL3gbvbU1eetIe .icon-shape .label{text-anchor:middle;}#mermaid-svg-HJDL3gbvbU1eetIe .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-HJDL3gbvbU1eetIe .rough-node .label,#mermaid-svg-HJDL3gbvbU1eetIe .node .label,#mermaid-svg-HJDL3gbvbU1eetIe .image-shape .label,#mermaid-svg-HJDL3gbvbU1eetIe .icon-shape .label{text-align:center;}#mermaid-svg-HJDL3gbvbU1eetIe .node.clickable{cursor:pointer;}#mermaid-svg-HJDL3gbvbU1eetIe .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-HJDL3gbvbU1eetIe .arrowheadPath{fill:#333333;}#mermaid-svg-HJDL3gbvbU1eetIe .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-HJDL3gbvbU1eetIe .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-HJDL3gbvbU1eetIe .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-HJDL3gbvbU1eetIe .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-HJDL3gbvbU1eetIe .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-HJDL3gbvbU1eetIe .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-HJDL3gbvbU1eetIe .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-HJDL3gbvbU1eetIe .cluster text{fill:#333;}#mermaid-svg-HJDL3gbvbU1eetIe .cluster span{color:#333;}#mermaid-svg-HJDL3gbvbU1eetIe div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-HJDL3gbvbU1eetIe .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-HJDL3gbvbU1eetIe rect.text{fill:none;stroke-width:0;}#mermaid-svg-HJDL3gbvbU1eetIe .icon-shape,#mermaid-svg-HJDL3gbvbU1eetIe .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-HJDL3gbvbU1eetIe .icon-shape p,#mermaid-svg-HJDL3gbvbU1eetIe .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-HJDL3gbvbU1eetIe .icon-shape rect,#mermaid-svg-HJDL3gbvbU1eetIe .image-shape rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-HJDL3gbvbU1eetIe .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-HJDL3gbvbU1eetIe .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-HJDL3gbvbU1eetIe :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

是

否

是

否

SSE消息到达

有event字段?

触发对应事件监听器

触发默认message事件

客户端处理特定事件

有id字段?

保存Last-Event-ID

不更新ID

断线重连时使用此ID

三、SSE与WebSocket对比

3.1 选择建议

选择SSE的场景：

• 服务器单向推送数据
• 需要自动重连机制
• 推送的是文本数据
• 实时通知、数据更新

选择WebSocket的场景：

• 需要双向通信
• 需要传输二进制数据
• 低延迟要求极高的实时交互
• 聊天、协作编辑、在线游戏

四、SSE与传统轮询对比

#mermaid-svg-s64Wp2AR2W2VYNQ3{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .error-icon{fill:#552222;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .marker{fill:#333333;stroke:#333333;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .marker.cross{stroke:#333333;}#mermaid-svg-s64Wp2AR2W2VYNQ3 svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-s64Wp2AR2W2VYNQ3 p{margin:0;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .cluster-label text{fill:#333;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .cluster-label span{color:#333;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .cluster-label span p{background-color:transparent;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .label text,#mermaid-svg-s64Wp2AR2W2VYNQ3 span{fill:#333;color:#333;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .node rect,#mermaid-svg-s64Wp2AR2W2VYNQ3 .node circle,#mermaid-svg-s64Wp2AR2W2VYNQ3 .node ellipse,#mermaid-svg-s64Wp2AR2W2VYNQ3 .node polygon,#mermaid-svg-s64Wp2AR2W2VYNQ3 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .rough-node .label text,#mermaid-svg-s64Wp2AR2W2VYNQ3 .node .label text,#mermaid-svg-s64Wp2AR2W2VYNQ3 .image-shape .label,#mermaid-svg-s64Wp2AR2W2VYNQ3 .icon-shape .label{text-anchor:middle;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .rough-node .label,#mermaid-svg-s64Wp2AR2W2VYNQ3 .node .label,#mermaid-svg-s64Wp2AR2W2VYNQ3 .image-shape .label,#mermaid-svg-s64Wp2AR2W2VYNQ3 .icon-shape .label{text-align:center;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .node.clickable{cursor:pointer;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .arrowheadPath{fill:#333333;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-s64Wp2AR2W2VYNQ3 .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-s64Wp2AR2W2VYNQ3 .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-s64Wp2AR2W2VYNQ3 .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .cluster text{fill:#333;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .cluster span{color:#333;}#mermaid-svg-s64Wp2AR2W2VYNQ3 div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-s64Wp2AR2W2VYNQ3 rect.text{fill:none;stroke-width:0;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .icon-shape,#mermaid-svg-s64Wp2AR2W2VYNQ3 .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .icon-shape p,#mermaid-svg-s64Wp2AR2W2VYNQ3 .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .icon-shape rect,#mermaid-svg-s64Wp2AR2W2VYNQ3 .image-shape rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-s64Wp2AR2W2VYNQ3 .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-s64Wp2AR2W2VYNQ3 .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-s64Wp2AR2W2VYNQ3 :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

SSE

一次请求

持续推送

客户端

服务器

传统轮询

定时请求

返回数据

客户端

服务器

轮询的问题：

• 频繁请求消耗服务器资源
• 数据延迟（取决于轮询间隔）
• 大量无效请求（无新数据时）

SSE的优势：

• 一次连接，持续推送
• 实时性高
• 资源消耗低

五、Spring Boot集成SSE

5.1 基础实现

创建SSE控制器：

import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

@RestController
@RequestMapping("/sse")
public class SseController {

private final ExecutorService executor = Executors.newCachedThreadPool();

@GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter streamEvents() {
SseEmitter emitter = new SseEmitter(60_000L); // 超时时间60秒

executor.execute(() -> {
try {
for (int i = 0; i < 10; i++) {
// 模拟业务处理
Thread.sleep(1000);

// 发送消息
emitter.send(SseEmitter.event()
.id(String.valueOf(i))
.name("message")
.data("消息内容: " + i)
.reconnectTime(5000L)); // 重连间隔5秒

System.out.println("发送消息: " + i);
}

// 发送完成事件
emitter.complete();
} catch (IOException | InterruptedException e) {
emitter.completeWithError(e);
}
});

return emitter;
}
}

5.2 完整的SSE服务实现

SSE服务类：

import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;

@Service
public class SseService {

// 存储所有活跃的SSE连接
private final ConcurrentHashMap<String, SseEmitter> emitters = new ConcurrentHashMap<>();

// 存储用户ID到连接ID的映射
private final ConcurrentHashMap<String, String> userConnections = new ConcurrentHashMap<>();

// 存储订阅关系（用户ID -> 订阅的频道）
private final ConcurrentHashMap<String, CopyOnWriteArraySet<String>> subscriptions = new ConcurrentHashMap<>();

/**
* 创建SSE连接
*/
public SseEmitter createConnection(String userId) {
String connectionId = generateConnectionId();

// 设置超时时间为30分钟
SseEmitter emitter = new SseEmitter(30 * 60 * 1000L);

// 连接关闭时的处理
emitter.onCompletion(() -> {
System.out.println("连接完成: " + connectionId);
removeConnection(connectionId);
});

emitter.onTimeout(() -> {
System.out.println("连接超时: " + connectionId);
removeConnection(connectionId);
});

emitter.onError((ex) -> {
System.err.println("连接错误: " + connectionId);
ex.printStackTrace();
removeConnection(connectionId);
});

// 存储连接
emitters.put(connectionId, emitter);
userConnections.put(userId, connectionId);

// 发送连接成功消息
try {
emitter.send(SseEmitter.event()
.name("connected")
.data("连接建立成功")
.id(connectionId));
} catch (IOException e) {
removeConnection(connectionId);
throw new RuntimeException("建立连接失败", e);
}

return emitter;
}

/**
* 向指定用户发送消息
*/
public boolean sendToUser(String userId, String eventName, Object data) {
String connectionId = userConnections.get(userId);
if (connectionId == null) {
return false;
}

return sendToConnection(connectionId, eventName, data);
}

/**
* 向指定连接发送消息
*/
public boolean sendToConnection(String connectionId, String eventName, Object data) {
SseEmitter emitter = emitters.get(connectionId);
if (emitter == null) {
return false;
}

try {
emitter.send(SseEmitter.event()
.name(eventName)
.data(data)
.id(String.valueOf(System.currentTimeMillis())));
return true;
} catch (IOException e) {
System.err.println("发送消息失败: " + connectionId);
removeConnection(connectionId);
return false;
}
}

/**
* 向所有连接广播消息
*/
public void broadcast(String eventName, Object data) {
emitters.forEach((connectionId, emitter) -> {
try {
emitter.send(SseEmitter.event()
.name(eventName)
.data(data)
.id(String.valueOf(System.currentTimeMillis())));
} catch (IOException e) {
System.err.println("广播消息失败: " + connectionId);
removeConnection(connectionId);
}
});
}

/**
* 移除连接
*/
private void removeConnection(String connectionId) {
SseEmitter emitter = emitters.remove(connectionId);
if (emitter != null) {
emitter.complete();
}

// 移除用户映射
userConnections.entrySet().removeIf(entry -> entry.getValue().equals(connectionId));

System.out.println("连接已移除: " + connectionId + ", 当前连接数: " + emitters.size());
}

/**
* 获取当前连接数
*/
public int getConnectionCount() {
return emitters.size();
}

/**
* 生成连接ID
*/
private String generateConnectionId() {
return java.util.UUID.randomUUID().toString();
}
}

控制器使用服务：

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.MediaType;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

@RestController
@RequestMapping("/api/sse")
public class SseController {

@Autowired
private SseService sseService;

/**
* 建立SSE连接
*/
@GetMapping(value = "/connect", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter connect(@RequestParam String userId) {
System.out.println("用户连接: " + userId);
return sseService.createConnection(userId);
}

/**
* 向指定用户发送消息（测试用）
*/
@PostMapping("/send")
public String sendToUser(
@RequestParam String userId,
@RequestParam(defaultValue = "message") String eventName,
@RequestBody Object data) {
boolean success = sseService.sendToUser(userId, eventName, data);
return success ? "发送成功" : "发送失败（用户未连接）";
}

/**
* 广播消息（测试用）
*/
@PostMapping("/broadcast")
public String broadcast(
@RequestParam(defaultValue = "message") String eventName,
@RequestBody Object data) {
sseService.broadcast(eventName, data);
return "广播成功";
}

/**
* 获取在线连接数
*/
@GetMapping("/count")
public int getConnectionCount() {
return sseService.getConnectionCount();
}
}

5.3 客户端代码示例

JavaScript客户端：

// 建立SSE连接
const userId = 'user123';
const eventSource = new EventSource(`http://localhost:8080/api/sse/connect?userId=${userId}`);

// 监听连接成功事件
eventSource.addEventListener('connected', (event) => {
console.log('连接成功:', event.data);
console.log('连接ID:', event.lastEventId);
});

// 监听自定义消息事件
eventSource.addEventListener('message', (event) => {
console.log('收到消息:', event.data);
const data = JSON.parse(event.data);
// 处理消息…
});

// 监听通知事件
eventSource.addEventListener('notification', (event) => {
console.log('收到通知:', event.data);
const notification = JSON.parse(event.data);
// 显示通知…
});

// 监听错误
eventSource.onerror = (error) => {
console.error('SSE连接错误:', error);
// 浏览器会自动尝试重连
};

// 关闭连接
// eventSource.close();

带重试和心跳的客户端：

class SseClient {
constructor(url, options = {}) {
this.url = url;
this.options = {
reconnectInterval: 5000,
maxReconnectAttempts: 10,
heartbeatInterval: 30000,
…options
};
this.reconnectAttempts = 0;
this.eventSource = null;
this.heartbeatTimer = null;
this.isConnected = false;
}

connect() {
this.eventSource = new EventSource(this.url);

this.eventSource.onopen = () => {
console.log('SSE连接已建立');
this.isConnected = true;
this.reconnectAttempts = 0;
this.startHeartbeat();
};

this.eventSource.onerror = (error) => {
console.error('SSE连接错误:', error);
this.isConnected = false;
this.stopHeartbeat();

if (this.reconnectAttempts < this.options.maxReconnectAttempts) {
this.reconnectAttempts++;
console.log(`尝试重连 (${this.reconnectAttempts}/${this.options.maxReconnectAttempts})…`);
setTimeout(() => this.connect(), this.options.reconnectInterval);
} else {
console.error('达到最大重连次数，停止重连');
}
};

return this.eventSource;
}

addEventListener(event, callback) {
if (this.eventSource) {
this.eventSource.addEventListener(event, callback);
}
}

startHeartbeat() {
this.heartbeatTimer = setInterval(() => {
if (this.isConnected) {
console.log('心跳检测');
}
}, this.options.heartbeatInterval);
}

stopHeartbeat() {
if (this.heartbeatTimer) {
clearInterval(this.heartbeatTimer);
this.heartbeatTimer = null;
}
}

close() {
this.stopHeartbeat();
if (this.eventSource) {
this.eventSource.close();
this.eventSource = null;
}
this.isConnected = false;
}
}

// 使用示例
const sseClient = new SseClient('http://localhost:8080/api/sse/connect?userId=user123', {
reconnectInterval: 3000,
maxReconnectAttempts: 5,
heartbeatInterval: 60000
});

sseClient.connect();

sseClient.addEventListener('message', (event) => {
console.log('收到消息:', event.data);
});

sseClient.addEventListener('notification', (event) => {
console.log('收到通知:', event.data);
});

// 关闭连接
// sseClient.close();

六、实际应用场景

6.1 实时消息通知

#mermaid-svg-b1TpQYOgGV9xnVI2{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;fill:#333;}@keyframes edge-animation-frame{from{stroke-dashoffset:0;}}@keyframes dash{to{stroke-dashoffset:0;}}#mermaid-svg-b1TpQYOgGV9xnVI2 .edge-animation-slow{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 50s linear infinite;stroke-linecap:round;}#mermaid-svg-b1TpQYOgGV9xnVI2 .edge-animation-fast{stroke-dasharray:9,5!important;stroke-dashoffset:900;animation:dash 20s linear infinite;stroke-linecap:round;}#mermaid-svg-b1TpQYOgGV9xnVI2 .error-icon{fill:#552222;}#mermaid-svg-b1TpQYOgGV9xnVI2 .error-text{fill:#552222;stroke:#552222;}#mermaid-svg-b1TpQYOgGV9xnVI2 .edge-thickness-normal{stroke-width:1px;}#mermaid-svg-b1TpQYOgGV9xnVI2 .edge-thickness-thick{stroke-width:3.5px;}#mermaid-svg-b1TpQYOgGV9xnVI2 .edge-pattern-solid{stroke-dasharray:0;}#mermaid-svg-b1TpQYOgGV9xnVI2 .edge-thickness-invisible{stroke-width:0;fill:none;}#mermaid-svg-b1TpQYOgGV9xnVI2 .edge-pattern-dashed{stroke-dasharray:3;}#mermaid-svg-b1TpQYOgGV9xnVI2 .edge-pattern-dotted{stroke-dasharray:2;}#mermaid-svg-b1TpQYOgGV9xnVI2 .marker{fill:#333333;stroke:#333333;}#mermaid-svg-b1TpQYOgGV9xnVI2 .marker.cross{stroke:#333333;}#mermaid-svg-b1TpQYOgGV9xnVI2 svg{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:16px;}#mermaid-svg-b1TpQYOgGV9xnVI2 p{margin:0;}#mermaid-svg-b1TpQYOgGV9xnVI2 .label{font-family:\”trebuchet ms\”,verdana,arial,sans-serif;color:#333;}#mermaid-svg-b1TpQYOgGV9xnVI2 .cluster-label text{fill:#333;}#mermaid-svg-b1TpQYOgGV9xnVI2 .cluster-label span{color:#333;}#mermaid-svg-b1TpQYOgGV9xnVI2 .cluster-label span p{background-color:transparent;}#mermaid-svg-b1TpQYOgGV9xnVI2 .label text,#mermaid-svg-b1TpQYOgGV9xnVI2 span{fill:#333;color:#333;}#mermaid-svg-b1TpQYOgGV9xnVI2 .node rect,#mermaid-svg-b1TpQYOgGV9xnVI2 .node circle,#mermaid-svg-b1TpQYOgGV9xnVI2 .node ellipse,#mermaid-svg-b1TpQYOgGV9xnVI2 .node polygon,#mermaid-svg-b1TpQYOgGV9xnVI2 .node path{fill:#ECECFF;stroke:#9370DB;stroke-width:1px;}#mermaid-svg-b1TpQYOgGV9xnVI2 .rough-node .label text,#mermaid-svg-b1TpQYOgGV9xnVI2 .node .label text,#mermaid-svg-b1TpQYOgGV9xnVI2 .image-shape .label,#mermaid-svg-b1TpQYOgGV9xnVI2 .icon-shape .label{text-anchor:middle;}#mermaid-svg-b1TpQYOgGV9xnVI2 .node .katex path{fill:#000;stroke:#000;stroke-width:1px;}#mermaid-svg-b1TpQYOgGV9xnVI2 .rough-node .label,#mermaid-svg-b1TpQYOgGV9xnVI2 .node .label,#mermaid-svg-b1TpQYOgGV9xnVI2 .image-shape .label,#mermaid-svg-b1TpQYOgGV9xnVI2 .icon-shape .label{text-align:center;}#mermaid-svg-b1TpQYOgGV9xnVI2 .node.clickable{cursor:pointer;}#mermaid-svg-b1TpQYOgGV9xnVI2 .root .anchor path{fill:#333333!important;stroke-width:0;stroke:#333333;}#mermaid-svg-b1TpQYOgGV9xnVI2 .arrowheadPath{fill:#333333;}#mermaid-svg-b1TpQYOgGV9xnVI2 .edgePath .path{stroke:#333333;stroke-width:2.0px;}#mermaid-svg-b1TpQYOgGV9xnVI2 .flowchart-link{stroke:#333333;fill:none;}#mermaid-svg-b1TpQYOgGV9xnVI2 .edgeLabel{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-b1TpQYOgGV9xnVI2 .edgeLabel p{background-color:rgba(232,232,232, 0.8);}#mermaid-svg-b1TpQYOgGV9xnVI2 .edgeLabel rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-b1TpQYOgGV9xnVI2 .labelBkg{background-color:rgba(232, 232, 232, 0.5);}#mermaid-svg-b1TpQYOgGV9xnVI2 .cluster rect{fill:#ffffde;stroke:#aaaa33;stroke-width:1px;}#mermaid-svg-b1TpQYOgGV9xnVI2 .cluster text{fill:#333;}#mermaid-svg-b1TpQYOgGV9xnVI2 .cluster span{color:#333;}#mermaid-svg-b1TpQYOgGV9xnVI2 div.mermaidTooltip{position:absolute;text-align:center;max-width:200px;padding:2px;font-family:\”trebuchet ms\”,verdana,arial,sans-serif;font-size:12px;background:hsl(80, 100%, 96.2745098039%);border:1px solid #aaaa33;border-radius:2px;pointer-events:none;z-index:100;}#mermaid-svg-b1TpQYOgGV9xnVI2 .flowchartTitleText{text-anchor:middle;font-size:18px;fill:#333;}#mermaid-svg-b1TpQYOgGV9xnVI2 rect.text{fill:none;stroke-width:0;}#mermaid-svg-b1TpQYOgGV9xnVI2 .icon-shape,#mermaid-svg-b1TpQYOgGV9xnVI2 .image-shape{background-color:rgba(232,232,232, 0.8);text-align:center;}#mermaid-svg-b1TpQYOgGV9xnVI2 .icon-shape p,#mermaid-svg-b1TpQYOgGV9xnVI2 .image-shape p{background-color:rgba(232,232,232, 0.8);padding:2px;}#mermaid-svg-b1TpQYOgGV9xnVI2 .icon-shape rect,#mermaid-svg-b1TpQYOgGV9xnVI2 .image-shape rect{opacity:0.5;background-color:rgba(232,232,232, 0.8);fill:rgba(232,232,232, 0.8);}#mermaid-svg-b1TpQYOgGV9xnVI2 .label-icon{display:inline-block;height:1em;overflow:visible;vertical-align:-0.125em;}#mermaid-svg-b1TpQYOgGV9xnVI2 .node .label-icon path{fill:currentColor;stroke:revert;stroke-width:revert;}#mermaid-svg-b1TpQYOgGV9xnVI2 :root{–mermaid-font-family:\”trebuchet ms\”,verdana,arial,sans-serif;}

建立SSE连接

实时推送

订单状态变更

新评论

系统通知

显示通知

用户

前端页面

通知服务

订单服务

评论服务

系统服务

实现示例：

@Service
public class NotificationService {

@Autowired
private SseService sseService;

/**
* 发送订单状态变更通知
*/
public void sendOrderNotification(String userId, Order order) {
Notification notification = Notification.builder()
.type("order")
.title("订单状态更新")
.message("您的订单 " + order.getOrderNo() + " 状态已变更为: " + order.getStatus())
.data(order)
.timestamp(LocalDateTime.now())
.build();

sseService.sendToUser(userId, "notification", notification);
}

/**
* 发送评论通知
*/
public void sendCommentNotification(String userId, Comment comment) {
Notification notification = Notification.builder()
.type("comment")
.title("新评论")
.message("用户 " + comment.getAuthor() + " 评论了你的文章")
.data(comment)
.timestamp(LocalDateTime.now())
.build();

sseService.sendToUser(userId, "notification", notification);
}
}

6.2 实时数据大屏

@RestController
@RequestMapping("/api/dashboard")
public class DashboardController {

@Autowired
private SseService sseService;

@Autowired
private DataService dataService;

/**
* 实时数据推送
*/
@GetMapping(value = "/realtime", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter realtimeData() {
String userId = "dashboard_" + System.currentTimeMillis();
SseEmitter emitter = sseService.createConnection(userId);

ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor();
scheduler.scheduleAtFixedRate(() -> {
try {
// 获取实时数据
DashboardData data = dataService.getRealtimeData();

// 推送数据
sseService.sendToUser(userId, "dashboard", data);
} catch (Exception e) {
e.printStackTrace();
}
}, 0, 1, TimeUnit.SECONDS); // 每秒推送一次

emitter.onCompletion(() -> scheduler.shutdown());
emitter.onTimeout(() -> scheduler.shutdown());

return emitter;
}
}

6.3 实时日志监控

@Service
public class LogMonitorService {

@Autowired
private SseService sseService;

private final CopyOnWriteArraySet<String> subscribers = new CopyOnWriteArraySet<>();

/**
* 订阅日志
*/
public SseEmitter subscribeLogs(String userId) {
subscribers.add(userId);
return sseService.createConnection(userId);
}

/**
* 推送日志
*/
public void pushLog(LogEntry logEntry) {
if (!subscribers.isEmpty()) {
subscribers.forEach(userId -> {
sseService.sendToUser(userId, "log", logEntry);
});
}
}

/**
* 取消订阅
*/
public void unsubscribe(String userId) {
subscribers.remove(userId);
}
}

6.4 进度条推送

@RestController
@RequestMapping("/api/task")
public class TaskController {

@Autowired
private SseService sseService;

/**
* 执行长任务并推送进度
*/
@GetMapping(value = "/execute", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter executeTask(@RequestParam String userId, @RequestParam String taskId) {
SseEmitter emitter = sseService.createConnection(userId);

ExecutorService executor = Executors.newSingleThreadExecutor();
executor.execute(() -> {
try {
// 推送任务开始
sseService.sendToUser(userId, "progress", Progress.builder()
.taskId(taskId)
.stage("初始化")
.percentage(0)
.message("任务开始")
.build());

// 模拟任务执行
for (int i = 0; i <= 100; i += 10) {
Thread.sleep(1000);

sseService.sendToUser(userId, "progress", Progress.builder()
.taskId(taskId)
.stage("处理中")
.percentage(i)
.message("正在处理… " + i + "%")
.build());
}

// 任务完成
sseService.sendToUser(userId, "progress", Progress.builder()
.taskId(taskId)
.stage("完成")
.percentage(100)
.message("任务完成")
.build());

sseService.sendToUser(userId, "complete", TaskResult.builder()
.taskId(taskId)
.success(true)
.message("任务执行成功")
.build());

emitter.complete();
} catch (Exception e) {
sseService.sendToUser(userId, "error", TaskResult.builder()
.taskId(taskId)
.success(false)
.message("任务执行失败: " + e.getMessage())
.build());
emitter.completeWithError(e);
} finally {
executor.shutdown();
}
});

return emitter;
}
}

七、生产环境最佳实践

7.1 连接管理

@Service
public class SseConnectionManager {

private final ConcurrentHashMap<String, SseEmitter> emitters = new ConcurrentHashMap<>();
private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);

@PostConstruct
public void init() {
// 定期清理过期连接
scheduler.scheduleAtFixedRate(this::cleanExpiredConnections, 5, 5, TimeUnit.MINUTES);
}

private void cleanExpiredConnections() {
emitters.forEach((id, emitter) -> {
try {
// 发送心跳检测
emitter.send(SseEmitter.event().name("heartbeat").data("ping"));
} catch (IOException e) {
// 发送失败，移除连接
removeConnection(id);
}
});
}

@PreDestroy
public void destroy() {
scheduler.shutdown();
emitters.forEach((id, emitter) -> emitter.complete());
emitters.clear();
}
}

7.2 消息队列集成

@Service
public class SseMessageQueueService {

@Autowired
private SseService sseService;

@Autowired
private RabbitTemplate rabbitTemplate;

/**
* 发送消息到队列
*/
public void sendMessageToQueue(String userId, String eventName, Object data) {
SseMessage message = SseMessage.builder()
.userId(userId)
.eventName(eventName)
.data(data)
.timestamp(LocalDateTime.now())
.build();

rabbitTemplate.convertAndSend("sse.exchange", "sse.routing", message);
}

/**
* 从队列消费消息并推送
*/
@RabbitListener(queues = "sse.queue")
public void handleSseMessage(SseMessage message) {
sseService.sendToUser(message.getUserId(), message.getEventName(), message.getData());
}
}

7.3 分布式环境支持

@Service
public class DistributedSseService {

@Autowired
private RedisTemplate<String, Object> redisTemplate;

@Autowired
private SseService sseService;

/**
* 发布消息到Redis
*/
public void publishMessage(String channel, String eventName, Object data) {
SseMessage message = SseMessage.builder()
.eventName(eventName)
.data(data)
.timestamp(LocalDateTime.now())
.build();

redisTemplate.convertAndSend(channel, message);
}

/**
* 订阅Redis消息
*/
@PostConstruct
public void subscribeChannels() {
redisTemplate.getConnectionFactory()
.getConnection()
.subscribe((message, pattern) -> {
try {
SseMessage sseMessage = JsonUtils.fromJson(
new String(message.getBody()),
SseMessage.class
);

// 广播给当前服务器的所有连接
sseService.broadcast(sseMessage.getEventName(), sseMessage.getData());
} catch (Exception e) {
e.printStackTrace();
}
}, "sse.channel.*");
}
}

八、性能优化建议

8.1 连接数限制

@Configuration
public class SseConfig {

@Bean
public SseService sseService() {
return new SseService() {
private static final int MAX_CONNECTIONS = 10000;
private final AtomicInteger connectionCount = new AtomicInteger(0);

@Override
public SseEmitter createConnection(String userId) {
if (connectionCount.get() >= MAX_CONNECTIONS) {
throw new RuntimeException("连接数已达上限");
}
connectionCount.incrementAndGet();
return super.createConnection(userId);
}

@Override
public void removeConnection(String connectionId) {
super.removeConnection(connectionId);
connectionCount.decrementAndGet();
}
};
}
}

8.2 消息压缩

@RestController
@RequestMapping("/api/sse")
public class CompressedSseController {

@GetMapping(value = "/compressed", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter streamCompressedEvents(@RequestParam String userId) {
SseEmitter emitter = new SseEmitter(60_000L);

ExecutorService executor = Executors.newSingleThreadExecutor();
executor.execute(() -> {
try {
for (int i = 0; i < 100; i++) {
// 原始数据
Map<String, Object> data = new HashMap<>();
data.put("id", i);
data.put("timestamp", System.currentTimeMillis());
data.put("content", "这是一条测试消息");

// 序列化为JSON
String json = JsonUtils.toJson(data);

// 使用gzip压缩
ByteArrayOutputStream bos = new ByteArrayOutputStream();
try (GZIPOutputStream gzip = new GZIPOutputStream(bos)) {
gzip.write(json.getBytes(StandardCharsets.UTF_8));
}

// 发送压缩后的数据
emitter.send(SseEmitter.event()
.name("message")
.data(Base64.getEncoder().encodeToString(bos.toByteArray()))
.comment("gzip"));

Thread.sleep(1000);
}
emitter.complete();
} catch (Exception e) {
emitter.completeWithError(e);
} finally {
executor.shutdown();
}
});

return emitter;
}
}

九、常见问题与解决方案

9.1 连接超时问题

问题： SSE连接在一段时间后自动断开

解决方案：

// 1. 设置合理的超时时间
SseEmitter emitter = new SseEmitter(30 * 60 * 1000L); // 30分钟

// 2. 定期发送心跳保持连接
scheduledExecutor.scheduleAtFixedRate(() -> {
try {
emitter.send(SseEmitter.event().name("heartbeat").data("ping"));
} catch (IOException e) {
emitter.completeWithError(e);
}
}, 0, 30, TimeUnit.SECONDS); // 每30秒发送一次心跳

9.2 消息丢失问题

问题： 客户端断线重连后丢失部分消息

解决方案：

@Service
public class ReliableSseService {

// 存储消息历史
private final ConcurrentHashMap<String, LinkedList<SseMessage>> messageHistory = new ConcurrentHashMap<>();

private static final int MAX_HISTORY_SIZE = 100;

/**
* 发送消息并保存历史
*/
public void sendWithHistory(String userId, String eventName, Object data) {
SseMessage message = SseMessage.builder()
.id(generateMessageId())
.eventName(eventName)
.data(data)
.timestamp(LocalDateTime.now())
.build();

// 保存到历史
messageHistory.computeIfAbsent(userId, k -> new LinkedList<>())
.addLast(message);

// 限制历史大小
if (messageHistory.get(userId).size() > MAX_HISTORY_SIZE) {
messageHistory.get(userId).removeFirst();
}

// 发送消息
sseService.sendToUser(userId, eventName, data);
}

/**
* 重连时发送历史消息
*/
public void sendHistoryOnReconnect(String userId, String lastEventId) {
LinkedList<SseMessage> history = messageHistory.get(userId);
if (history != null) {
history.stream()
.filter(msg -> msg.getId().compareTo(lastEventId) > 0)
.forEach(msg -> sseService.sendToUser(userId, msg.getEventName(), msg.getData()));
}
}
}

9.3 Nginx代理配置

问题： 通过Nginx代理后SSE连接不稳定

解决方案：

location /sse {
proxy_pass http://backend;
proxy_http_version 1.1;
proxy_set_header Connection "";
proxy_set_header X-Accel-Buffering no; # 禁用缓冲
proxy_cache off; # 禁用缓存
proxy_read_timeout 3600s; # 增加超时时间
proxy_send_timeout 3600s;
chunked_transfer_encoding on;
}

十、总结

SSE作为一种轻量级的服务器推送技术，在以下场景中具有明显优势：

✅ 适用场景：

• 实时消息通知
• 数据大屏实时更新
• 进度条推送
• 日志监控
• 股票/价格实时更新
• 新闻推送

❌ 不适用场景：

• 需要双向通信的场景（使用WebSocket）
• 需要传输二进制数据的场景（使用WebSocket）
• 需要低延迟高频交互的场景（使用WebSocket）

核心优势：

• 实现简单，基于标准HTTP
• 自动重连机制
• 服务器资源消耗低
• 浏览器原生支持

注意事项：

• 注意连接超时和心跳保持
• 生产环境需要考虑分布式支持
• 合理控制连接数和消息频率
• 做好错误处理和重连机制

SSE是实时通信领域的重要工具，掌握其原理和实现方式，能够帮助我们构建更加实时和高效的Web应用。