Spring Boot 集成 SSE，别啥都上 WebSocket-平芜编程栈

之前做个需求，服务端要往浏览器推消息，我头脑一热决定用 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 简单得多：

constsource=newEventSource('/api/sse/connect');source.addEventListener('stock',(event)=>{constdata=JSON.parse(event.data);console.log(data);});source.onerror=(error)=>{// 断了会自动重连，这里主要打日志};

Spring Boot 里写个 SSE 端点

Spring MVC 从 4.2 开始支持SseEmitter，这就是为 SSE 准备的异步响应类型。新建一个接口：

@RestController@RequestMapping("/api/sse")publicclassSseController{privatefinalSseSessionRegistryregistry;publicSseController(SseSessionRegistryregistry){this.registry=registry;}@GetMapping(value="/connect",produces=MediaType.TEXT_EVENT_STREAM_VALUE)publicSseEmitterconnect(){SseEmitteremitter=newSseEmitter(0L);// 0L 就是不超时StringsessionId=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(IOExceptione){emitter.completeWithError(e);}returnemitter;}}

SseEmitter.event()可以设置事件名和数据。produces必须写text/event-stream，不然浏览器不认。

后端往指定连接推消息，就把事件发到对应的SseEmitter：

SseEmitteremitter=registry.get(sessionId);if(emitter!=null){try{emitter.send(SseEmitter.event().name("message").data(notificationPayload));}catch(IOExceptione){// 记录日志，从注册表移除连接}}

连接注册表：别用 static 裸奔

SSE 连接是挂着不走的，服务端得记着每个SseEmitter，不然推消息时找不到人。用ConcurrentHashMap存 sessionId 和 emitter：

@ComponentpublicclassSseSessionRegistry{privatefinalMap<String,SseEmitter>sessions=newConcurrentHashMap<>();publicvoidadd(StringsessionId,SseEmitteremitter){sessions.put(sessionId,emitter);}publicvoidremove(StringsessionId){sessions.remove(sessionId);}publicSseEmitterget(StringsessionId){returnsessions.get(sessionId);}publicCollection<SseEmitter>getAll(){returnsessions.values();}publicintsize(){returnsessions.size();}}

虽然SseEmitter注册了onCompletion，但客户端直接拔网线、浏览器崩了这些情况，回调不一定触发，Map 里就会积累一堆“僵尸”连接。所以得开个定时任务扫一扫。简单点的办法是发个注释消息试试，失败了就踢掉：

@ComponentpublicclassSseSessionScheduler{privatefinalSseSessionRegistryregistry;publicSseSessionScheduler(SseSessionRegistryregistry){this.registry=registry;}@Scheduled(fixedRate=30000)publicvoidpurgeExpiredConnections(){for(SseEmitteremitter:registry.getAll()){try{emitter.send(SseEmitter.event().comment("ping"));}catch(Exceptione){registry.removeByEmitter(emitter);emitter.complete();}}}}

这里最好在SseSessionRegistry里加个removeByEmitter，直接通过引用来删：

publicvoidremoveByEmitter(SseEmitteremitter){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)publicvoidheartbeatAndClean(){for(SseEmitteremitter:registry.getAll()){try{emitter.send(SseEmitter.event().comment("heartbeat"));}catch(Exceptione){registry.removeByEmitter(emitter);emitter.complete();}}}

断线重连和续传：Last-Event-ID 是个好协议

EventSource自带重连，断线后会自动重新发起请求。如果之前的响应里带过id字段，重连请求头里就会自动带上Last-Event-ID。服务端读到这个值，就知道该从哪条消息继续发。

比如你有张事件表，每条事件有个递增 ID，那控制器可以这样写：

@GetMapping(value="/connect",produces=MediaType.TEXT_EVENT_STREAM_VALUE)publicSseEmitterconnect(@RequestHeader(value="Last-Event-ID",required=false)StringlastEventId){SseEmitteremitter=newSseEmitter(60_000L);StringsessionId=registry.add(emitter);if(lastEventId!=null){List<EventMessage>missed=messageService.fetchAfterId(Long.parseLong(lastEventId));for(EventMessagemsg:missed){try{emitter.send(SseEmitter.event().name("message").id(String.valueOf(msg.getId())).data(msg.getPayload()));}catch(Exceptione){break;}}}// 然后继续发实时事件returnemitter;}

有个地方要注意：如果你手动重连，比如弄个退避策略，那浏览器不会自动带上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 监听容器：

@ConfigurationpublicclassRedisPubSubConfig{@BeanpublicRedisMessageListenerContainerredisContainer(RedisConnectionFactoryfactory,MessageListenerAdapterlistenerAdapter){RedisMessageListenerContainercontainer=newRedisMessageListenerContainer();container.setConnectionFactory(factory);container.addMessageListener(listenerAdapter,newPatternTopic("sse-notification"));returncontainer;}@BeanpublicMessageListenerAdapterlistenerAdapter(SseBroadcastListenerlistener){returnnewMessageListenerAdapter(listener,"onMessage");}}

注意这里有个大坑：MessageListenerAdapter的第二个参数是方法名，我在这里指定了onMessage，那么监听器的方法签名必须是onMessage(String message, byte[] pattern)——第一个参数是消息内容（取决于序列化器），第二个参数是订阅的 pattern，不是消息体！很多人在这里把第二个参数当消息内容转 JSON，结果半天查不出问题。正确的实现像这样：

@ComponentpublicclassSseBroadcastListener{privatefinalSseSessionRegistryregistry;publicSseBroadcastListener(SseSessionRegistryregistry){this.registry=registry;}publicvoidonMessage(Stringmessage,byte[]pattern){NotificationDTOdto=JSON.parseObject(message,NotificationDTO.class);for(SseEmitteremitter:registry.getAll()){try{emitter.send(SseEmitter.event().name("notice").data(dto));}catch(Exceptione){// 推送失败，等清理任务踢掉}}}}

这里如果 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);Stringbase64Data=Base64.getEncoder().encodeToString(fileBytes);emitter.send(SseEmitter.event().name("file").id(String.valueOf(fileId)).data("{\"filename\":\"report.pdf\",\"content\":\""+base64Data+"\"}"));

客户端拿到后解码：

source.addEventListener('file',(event)=>{constresponse=JSON.parse(event.data);constbinaryString=atob(response.content);constbytes=newUint8Array(binaryString.length);for(leti=0;i<binaryString.length;i++){bytes[i]=binaryString.charCodeAt(i);}constblob=newBlob([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统计一下，集群里再汇总，方便排查问题。