之前做个需求服务端要往浏览器推消息我头脑一热决定用 WebSocket结果客户端只需要收根本不需要发。后面换成 SSE代码量少了一大截还不用处理协议升级和断线重连。借着这篇文章把 Spring Boot 里用 SSE 做实时推送的完整过程捋一遍顺便把集群广播的坑也填了。SSE 是个啥凭啥比 WebSocket 省事SSEServer-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不是必需默认messageid用于断线续传retry能控制重连间隔:开头是注释主要拿来当心跳客户端直接忽略。客户端用EventSource监听写起来比 WebSocket 简单得多constsourcenewEventSource(/api/sse/connect);source.addEventListener(stock,(event){constdataJSON.parse(event.data);console.log(data);});source.onerror(error){// 断了会自动重连这里主要打日志};Spring Boot 里写个 SSE 端点Spring MVC 从 4.2 开始支持SseEmitter这就是为 SSE 准备的异步响应类型。新建一个接口RestControllerRequestMapping(/api/sse)publicclassSseController{privatefinalSseSessionRegistryregistry;publicSseController(SseSessionRegistryregistry){this.registryregistry;}GetMapping(value/connect,producesMediaType.TEXT_EVENT_STREAM_VALUE)publicSseEmitterconnect(){SseEmitteremitternewSseEmitter(0L);// 0L 就是不超时StringsessionIdUUID.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(sessionsessionId));}catch(IOExceptione){emitter.completeWithError(e);}returnemitter;}}SseEmitter.event()可以设置事件名和数据。produces必须写text/event-stream不然浏览器不认。后端往指定连接推消息就把事件发到对应的SseEmitterSseEmitteremitterregistry.get(sessionId);if(emitter!null){try{emitter.send(SseEmitter.event().name(message).data(notificationPayload));}catch(IOExceptione){// 记录日志从注册表移除连接}}连接注册表别用 static 裸奔SSE 连接是挂着不走的服务端得记着每个SseEmitter不然推消息时找不到人。用ConcurrentHashMap存 sessionId 和 emitterComponentpublicclassSseSessionRegistry{privatefinalMapString,SseEmittersessionsnewConcurrentHashMap();publicvoidadd(StringsessionId,SseEmitteremitter){sessions.put(sessionId,emitter);}publicvoidremove(StringsessionId){sessions.remove(sessionId);}publicSseEmitterget(StringsessionId){returnsessions.get(sessionId);}publicCollectionSseEmittergetAll(){returnsessions.values();}publicintsize(){returnsessions.size();}}虽然SseEmitter注册了onCompletion但客户端直接拔网线、浏览器崩了这些情况回调不一定触发Map 里就会积累一堆“僵尸”连接。所以得开个定时任务扫一扫。简单点的办法是发个注释消息试试失败了就踢掉ComponentpublicclassSseSessionScheduler{privatefinalSseSessionRegistryregistry;publicSseSessionScheduler(SseSessionRegistryregistry){this.registryregistry;}Scheduled(fixedRate30000)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-valueemitter);}记得在启动类上加上EnableScheduling。心跳保活不然连接会被 Nginx 掐断SSE 连接如果长时间没数据Nginx 或防火墙会认为是死连接直接给你断开。所以得定期发心跳。上面定时任务里的comment(ping)就是心跳浏览器收到后直接忽略但这条数据能保持 TCP 连接活跃。SseEmitter有默认超时时间大概 30 秒取决于容器。如果不希望超时可以在构造时传0L但这么做有风险如果客户端断线了服务端也不知道连接和线程会一直被占用。所以要么设一个较大的超时时间要么靠心跳和清理任务配合。我一般设SseEmitter emitter new SseEmitter(60_000L);然后心跳间隔是 15 秒超时时间设成心跳的 4 倍左右万一客户端真挂了两端都能及时释放。心跳和清理可以放在同一个定时任务里没必要写两个省得增加复杂度Scheduled(fixedRate15000)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,producesMediaType.TEXT_EVENT_STREAM_VALUE)publicSseEmitterconnect(RequestHeader(valueLast-Event-ID,requiredfalse)StringlastEventId){SseEmitteremitternewSseEmitter(60_000L);StringsessionIdregistry.add(emitter);if(lastEventId!null){ListEventMessagemissedmessageService.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 是最简单的方案所有实例订阅同一个频道谁发布消息谁就能被所有人收到publishpublishsubscribesubscribepushpush应用实例 1Redis Pub/Sub应用实例 2客户端 1客户端 2先加依赖dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-data-redis/artifactId/dependency配置 Redis 监听容器ConfigurationpublicclassRedisPubSubConfig{BeanpublicRedisMessageListenerContainerredisContainer(RedisConnectionFactoryfactory,MessageListenerAdapterlistenerAdapter){RedisMessageListenerContainercontainernewRedisMessageListenerContainer();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.registryregistry;}publicvoidonMessage(Stringmessage,byte[]pattern){NotificationDTOdtoJSON.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[]fileBytesFiles.readAllBytes(path);Stringbase64DataBase64.getEncoder().encodeToString(fileBytes);emitter.send(SseEmitter.event().name(file).id(String.valueOf(fileId)).data({\filename\:\report.pdf\,\content\:\base64Data\}));客户端拿到后解码source.addEventListener(file,(event){constresponseJSON.parse(event.data);constbinaryStringatob(response.content);constbytesnewUint8Array(binaryString.length);for(leti0;ibinaryString.length;i){bytes[i]binaryString.charCodeAt(i);}constblobnewBlob([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.max400 server.tomcat.threads.min-spare50如果你担心 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核心看需求别图“高级”就上重武器。 福利时间如果你正在备战面试或者想要学习其他知识给大家推荐一个宝藏知识库作者整理了一些列 Java 程序员需要掌握的核心知识有需要的自取不谢。知识库地址https://farerboy.com/