SpringBoot使用SSE实现服务端实时推送的完整方案
作者：(farerboy)
引言
之前做个需求，服务端要往浏览器推消息，我头脑一热决定用 WebSocket，结果客户端只需要收，根本不需要发。后面换成 SSE，代码量少了一大截，还不用处理协议升级和断线重连。借着这篇文章，把 Spring Boot 里用 SSE 做实时推送的完整过程捋一遍，顺便把集群广播的坑也填了。
SSE 是个啥，凭啥比 WebSocket 省事
SSE（Server-Sent Events）是 HTML5 的标准特性，本质就是服务端往客户端推送一段文本流。客户端发一个普通的 GET 请求，服务端不返回常规 JSON，而是返回 Content-Type: text/event-stream 的响应，连接保持打开，服务端可以持续往里写数据。
它和 WebSocket 最大的区别在方向：SSE 是单向的（服务端 → 客户端），WebSocket 是双向的。如果业务上客户端基本不往后发数据，用 WebSocket 等于杀鸡用了牛刀，还得自己操心握手、心跳、断线重连这些破事。SSE 自带自动重连，浏览器原生支持，连第三方库都不用引。
SSE 的消息格式也不复杂，每个字段一行，消息之间空一行：
id: 1
event: stock
data: {"symbol": "AAPL", "price": 175.32}
: heartbeat
这里 event 不是必需，默认 message；id 用于断线续传；retry 能控制重连间隔；: 开头是注释，主要拿来当心跳，客户端直接忽略。
客户端用 EventSource 监听，写起来比 WebSocket 简单得多：
const source = new EventSource('/api/sse/connect');
source.addEventListener('stock', (event) => {
const data = JSON.parse(event.data);
console.log(data);
});
source.onerror = (error) => {
// 断了会自动重连，这里主要打日志
};
Spring Boot 里写个 SSE 端点
Spring MVC 从 4.2 开始支持 SseEmitter，这就是为 SSE 准备的异步响应类型。新建一个接口：
@RestController
@RequestMapping("/api/sse")
public class SseController {
private final SseSessionRegistry registry;
public SseController(SseSessionRegistry registry) {
this.registry = registry;
}
@GetMapping(value = "/connect", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter connect() {
SseEmitter emitter = new SseEmitter(0L); // 0L 就是不超时
String sessionId = UUID.randomUUID().toString();
registry.add(sessionId, emitter);
emitter.onCompletion(() -> registry.remove(sessionId));
emitter.onTimeout(() -> registry.remove(sessionId));
emitter.onError((ex) -> registry.remove(sessionId));
try {
emitter.send(SseEmitter.event()
.name("connected")
.data("session=" + sessionId));
} catch (IOException e) {
emitter.completeWithError(e);
}
return emitter;
}
}
SseEmitter.event() 可以设置事件名和数据。produces 必须写 text/event-stream，不然浏览器不认。
后端往指定连接推消息，就把事件发到对应的 SseEmitter：
SseEmitter emitter = registry.get(sessionId);
if (emitter != null) {
try {
emitter.send(SseEmitter.event()
.name("message")
.data(notificationPayload));
} catch (IOException e) {
// 记录日志，从注册表移除连接
}
}
连接注册表：别用 static 裸奔
SSE 连接是挂着不走的，服务端得记着每个 SseEmitter，不然推消息时找不到人。用 ConcurrentHashMap 存 sessionId 和 emitter：
@Component
public class SseSessionRegistry {
private final Map<String, SseEmitter> sessions = new ConcurrentHashMap<>();
public void add(String sessionId, SseEmitter emitter) {
sessions.put(sessionId, emitter);
}
public void remove(String sessionId) {
sessions.remove(sessionId);
}
public SseEmitter get(String sessionId) {
return sessions.get(sessionId);
}
public Collection<SseEmitter> getAll() {
return sessions.values();
}
public int size() {
return sessions.size();
}
}
虽然 SseEmitter 注册了 onCompletion，但客户端直接拔网线、浏览器崩了这些情况，回调不一定触发，Map 里就会积累一堆“僵尸”连接。所以得开个定时任务扫一扫。简单点的办法是发个注释消息试试，失败了就踢掉：
@Component
public class SseSessionScheduler {
private final SseSessionRegistry registry;
public SseSessionScheduler(SseSessionRegistry registry) {
this.registry = registry;
}
@Scheduled(fixedRate = 30000)
public void purgeExpiredConnections() {
for (SseEmitter emitter : registry.getAll()) {
try {
emitter.send(SseEmitter.event().comment("ping"));
} catch (Exception e) {
registry.removeByEmitter(emitter);
emitter.complete();
}
}
}
}
这里最好在 SseSessionRegistry 里加个 removeByEmitter，直接通过引用来删：
public void removeByEmitter(SseEmitter emitter) {
sessions.values().removeIf(value -> value == emitter);
}
记得在启动类上加上 @EnableScheduling。
心跳保活，不然连接会被 Nginx 掐断
SSE 连接如果长时间没数据，Nginx 或防火墙会认为是死连接，直接给你断开。所以得定期发心跳。上面定时任务里的 comment("ping") 就是心跳，浏览器收到后直接忽略，但这条数据能保持 TCP 连接活跃。
SseEmitter 有默认超时时间，大概 30 秒（取决于容器）。如果不希望超时，可以在构造时传 0L，但这么做有风险：如果客户端断线了，服务端也不知道，连接和线程会一直被占用。所以要么设一个较大的超时时间，要么靠心跳和清理任务配合。我一般设 SseEmitter emitter = new SseEmitter(60_000L);，然后心跳间隔是 15 秒，超时时间设成心跳的 4 倍左右，万一客户端真挂了，两端都能及时释放。
心跳和清理可以放在同一个定时任务里，没必要写两个，省得增加复杂度：
@Scheduled(fixedRate = 15000)
public void heartbeatAndClean() {
for (SseEmitter emitter : registry.getAll()) {
try {
emitter.send(SseEmitter.event().comment("heartbeat"));
} catch (Exception e) {
registry.removeByEmitter(emitter);
emitter.complete();
}
}
}
断线重连和续传：Last-Event-ID 是个好协议
EventSource 自带重连，断线后会自动重新发起请求。如果之前的响应里带过 id 字段，重连请求头里就会自动带上 Last-Event-ID。服务端读到这个值，就知道该从哪条消息继续发。
比如你有张事件表，每条事件有个递增 ID，那控制器可以这样写：
@GetMapping(value = "/connect", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter connect(@RequestHeader(value = "Last-Event-ID", required = false) String lastEventId) {
SseEmitter emitter = new SseEmitter(60_000L);
String sessionId = registry.add(emitter);
if (lastEventId != null) {
List<EventMessage> missed = messageService.fetchAfterId(Long.parseLong(lastEventId));
for (EventMessage msg : missed) {
try {
emitter.send(SseEmitter.event()
.name("message")
.id(String.valueOf(msg.getId()))
.data(msg.getPayload()));
} catch (Exception e) {
break;
}
}
}
// 然后继续发实时事件
return emitter;
}
有个地方要注意：如果你手动重连，比如弄个退避策略，那浏览器不会自动带上 Last-Event-ID，你只能自己放在 URL 参数里带过去。要是用默认的重连机制，它虽然会带这个头，但没法自定义 header，所以鉴权一般只能靠 Cookie 或者 URL token。Cookie 在浏览器重连时会自动带上，算是比较省心的方案。
集群部署：用 Redis Pub/Sub 做广播
生产环境不会一个实例扛所有，客户端分散连到不同机器。问题来了：A 实例的某个请求产生了一条消息，要推给所有客户端，但客户端在 B 实例上的连接，A 实例根本不知道。这时候就得引入中间件做广播。
用 Redis Pub/Sub 是最简单的方案，所有实例订阅同一个频道，谁发布消息谁就能被所有人收到：
先加依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-data-redis</artifactId>
</dependency>配置 Redis 监听容器：
@Configuration
public class RedisPubSubConfig {
@Bean
public RedisMessageListenerContainer redisContainer(RedisConnectionFactory factory,
MessageListenerAdapter listenerAdapter) {
RedisMessageListenerContainer container = new RedisMessageListenerContainer();
container.setConnectionFactory(factory);
container.addMessageListener(listenerAdapter, new PatternTopic("sse-notification"));
return container;
}
@Bean
public MessageListenerAdapter listenerAdapter(SseBroadcastListener listener) {
return new MessageListenerAdapter(listener, "onMessage");
}
}
注意这里有个大坑：MessageListenerAdapter 的第二个参数是方法名，我在这里指定了 onMessage，那么监听器的方法签名必须是 onMessage(String message, byte[] pattern)——第一个参数是消息内容（取决于序列化器），第二个参数是订阅的 pattern，不是消息体！很多人在这里把第二个参数当消息内容转 JSON，结果半天查不出问题。正确的实现像这样：
@Component
public class SseBroadcastListener {
private final SseSessionRegistry registry;
public SseBroadcastListener(SseSessionRegistry registry) {
this.registry = registry;
}
public void onMessage(String message, byte[] pattern) {
NotificationDTO dto = JSON.parseObject(message, NotificationDTO.class);
for (SseEmitter emitter : registry.getAll()) {
try {
emitter.send(SseEmitter.event().name("notice").data(dto));
} catch (Exception e) {
// 推送失败，等清理任务踢掉
}
}
}
}
这里如果 RedisTemplate 默认序列化的是 String，那第一个参数就是 JSON 字符串。要是你配了 Jackson 序列化器，那第一个参数可能直接就是个对象，那更省事，但注意类型别搞错。
发布消息的地方：
redisTemplate.convertAndSend("sse-notification", JSON.toJSONString(dto));
再提醒一下，Redis Pub/Sub 是即发即弃的，发的时候如果某个实例宕机了，那它错过的消息就丢了。如果是重要业务，建议用 Redis Streams 或者 RabbitMQ 的持久化队列，别拿 Pub/Sub 当消息队列用。
传二进制文件？Base64 是唯一选择
SSE 是文本流，传不了原生二进制，但是我们可以把二进制编码成 Base64 塞进 data 字段。服务端：
byte[] fileBytes = Files.readAllBytes(path);
String base64Data = Base64.getEncoder().encodeToString(fileBytes);
emitter.send(SseEmitter.event()
.name("file")
.id(String.valueOf(fileId))
.data("{\"filename\":\"report.pdf\",\"content\":\"" + base64Data + "\"}"));
客户端拿到后解码：
source.addEventListener('file', (event) => {
const response = JSON.parse(event.data);
const binaryString = atob(response.content);
const bytes = new Uint8Array(binaryString.length);
for (let i = 0; i < binaryString.length; i++) {
bytes[i] = binaryString.charCodeAt(i);
}
const blob = new Blob([bytes], { type: 'application/pdf' });
// 触发下载
});
但是 Base64 会让数据体积膨胀 33%，要是传个 100MB 的文件，光编码就浪费不少带宽，而且长连接也容易卡死。真传大文件，老老实实用 HTTP 分块下载或对象存储的直传链接。SSE 适合小消息高频推送，比如行情、通知、AI 增量输出。
生产环境你还得注意这些
用 Nginx 做反向代理时，SSE 默认会被缓冲，导致消息攒到一定量才一起发给客户端，实时性直接废掉。需要在 location 里关闭缓冲：
location /api/sse/ {
proxy_buffering off;
proxy_cache off;
proxy_read_timeout 3600s;
proxy_http_version 1.1;
proxy_set_header Connection "";
}
proxy_buffering off 保证每发一条数据，Nginx 就立刻转给客户端；proxy_read_timeout 设长一点，不然空闲时间超过默认 60 秒，Nginx 就把连接掐了。
连接多了以后，线程资源是个大问题。Tomcat 默认每个连接占一个线程，SSE 连接又长期占用，所以线程池最大值要调高，比如：
server.tomcat.threads.max=400 server.tomcat.threads.min-spare=50
如果你担心 C10K 级别，那就得换 Spring WebFlux，非阻塞模型能撑更多连接，但复杂度也上去了，大部分场景 MVC 异步够用。
SSE 在 HTTP/1.1 下会占用一个连接，而浏览器对同一域名并发连接数通常是 6 个，如果页面同时开了好几个 SSE 连接，其他 HTTP 请求就得排队。这是 SSE 的天然限制，不是 Spring 的问题，设计时留意一下。
安全方面，SSE 没法像 WebSocket 那样自定义升级握手时带 header，所以认证主要靠 Cookie 或 URL 参数。Cookie 默认跨子域不太好使，需要注意；URL 参数会落在 Nginx access log 里，token 也会泄露，最好放 Cookie 并且配 HTTPS。CORS 也要配置，避免别的站点利用你的 SSE 端点做点坏事。
最后，线上一定要监控连接状态：当前活跃连接数、推送速率、断开率、注册表大小。这些指标可以直接通过 Actuator 暴露出来，或者自己写个 @RestController 统计一下，集群里再汇总，方便排查问题。
总结一下
SSE 不是新技术，但很多人一聊实时推送就想到 WebSocket。如果你的场景是服务端单向推送，用 SSE 绝对够用，代码量少一半，还不容易踩坑。选型这事，单向选 SSE，双向选 WebSocket，核心看需求，别图“高级”就上重武器。
以上就是SpringBoot使用SSE实现服务端实时推送的完整方案的详细内容，更多关于SpringBoot SSE服务端实时推送的资料请关注脚本之家其它相关文章！