1. 先搞清楚SSE能做什么和WebSocket到底怎么选很多人一听到“实时推送”就想到WebSocket然后陷入前后端联调的深坑里。实际上有一类很常见的业务场景——服务端往客户端单向推数据用SSEServer-Sent Events服务器发送事件反而更省事。SSE是HTML5标准里自带的能力基于HTTP协议实现客户端通过EventSource接口订阅服务端通过普通的HTTP连接持续输出数据流。它解决的核心问题很简单服务端有了新数据主动把数据推给浏览器或客户端而客户端不需要轮询。举个例子一台服务器上有N个任务在跑前端要实时看到每个任务的进度百分比、日志输出、状态变更。用WebSocket当然能做但你需要处理握手协议、消息帧格式、双向通信而SSE完全不需要这些它就是一次普通的HTTP请求服务端不关连接源源不断地把数据写出去就行。客户端拿到的是text/event-stream格式的数据流用浏览器的EventSource对象一两行代码就能接住。这里先说适合用SSE的场景服务端状态变化通知、实时日志流、任务进度上报、消息中心弹窗、股票行情或监控指标的大屏展示。共同特点是方向单一基本不需要客户端往服务端推消息。如果有强双向交互需求——比如聊天、多人协作编辑器、实时对战那再去考虑WebSocket。还有一个非常现实的考量SSE走的就是普通HTTP/HTTPS协议不需要额外的协议解析和端口对防火墙、反向代理、负载均衡这些基础设施非常友好而WebSocket在复杂网络环境里经常要配额外的超时时间、代理缓冲策略。很多项目里SSE就是那个让你省掉一大堆事的方案。2. 动手前先搞懂SSE的核心协议细节2.1 事件流到底长什么样SSE的数据格式看起来平淡无奇但它定义了一套自己的文本协议MIME类型是text/event-stream字符集UTF-8。服务端推送的每一条消息可以包含这些字段data:数据内容可以多行客户端收到的是一整块文本event:事件类型不写就默认是messageid:消息ID客户端会自动记录断线重连时通过Last-Event-ID头传给服务端retry:重连时间毫秒告诉客户端如果断开了隔多久再来连消息之间用空行分隔。一个最简单的推送片段是这样data: 第一条消息 data: 第二条消息 id: 2 event: status data: {progress: 78}客户端在浏览器里对应的事件监听是默认的message事件、直接用addEventListener(status)监听自定义事件。这个格式设计的好处是通过标准化的文本就能实现事件路由、断点续传和自动重连不需要任何第三方库。2.2 为什么SSE能“一条连接推到底”本质上这就是一个长驻的HTTP响应。服务端收到GET请求后把响应头设置好然后体内容以分块传输chunked的方式持续往客户端写文本。客户端不关心你是一次性写完还是慢慢写只要连接不断数据就一直流过来。关键响应头有这么几个每一个都有讲究响应头值作用Content-Typetext/event-stream;charsetUTF-8告诉客户端这是SSE流Cache-Controlno-cache禁止缓存否则数据被代理或浏览器缓存起来就收不到实时推送了Connectionkeep-alive保持长连接X-Accel-Bufferingno如果前面挂了Nginx要关掉缓冲否则数据会攒在Nginx里不往下发第三项在Spring Boot里一般不手动设置容器会自动处理。第四项看起来不起眼但很多生产事故就出在这里——Nginx默认开启了proxy bufferingSSE数据会卡在缓冲区里攒够4KB或一段时间才flush一次前端看到的效果就是“数据一跳一跳地出现”。针对这类问题我后面会在常见问题章节里专门展开讲。3. Spring Boot接入SSE的完整实操3.1 用Spring MVC的SseEmitter打通第一版从Spring Framework 4.2开始就原生支持了SseEmitter这个异步控制器返回值。接口的核心足够简单几分钟就能跑通。下面是一个基础示例RestController RequestMapping(/api/sse) public class SseController { /** * 建立SSE连接客户端连接后服务端每2秒推送一次当前时间 */ GetMapping(value /clock, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter clock() { // 1小时不操作自动断开 SseEmitter emitter new SseEmitter(3600_000L); Executors.newSingleThreadScheduledExecutor().scheduleAtFixedRate(() - { try { emitter.send(SseEmitter.event() .name(clock) .data(当前时间: LocalDateTime.now())); } catch (IOException e) { emitter.completeWithError(e); } }, 0, 2, TimeUnit.SECONDS); return emitter; } }关键就三件事设定超时时间、拿到SseEmitter返回给Spring MVC、在有数据的时候调用emitter.send()。注意send()方法里如果抛了IOException说明客户端已经断开这时候要completeWithError或者直接complete()把线程释放掉。3.2 线程池别乱用这是第一道坑SseEmitter本质上有两个线程参与工作一个是Tomcat容器的工作线程负责把HTTP请求挂起另一个是你自己线程池里的线程负责定时或业务触发去调send()。如果图省事直接在控制器里new线程池每来一个连接就创建一个线程客户端一多线程数和句柄数会直接被打满。我的做法是准备一个全局的调度线程池专门负责往各个emitter发数据Configuration public class SseThreadPoolConfig { Bean(sseScheduler) public ScheduledExecutorService sseScheduler() { return Executors.newScheduledThreadPool(8, r - { Thread t new Thread(r, sse-push-thread); t.setDaemon(true); return t; }); } }使用的时候注入这个Bean不要在每次请求的时候Executors.newSingleThreadScheduledExecutor()。8个线程够用吗这里有个很实际的计算思路如果你的每条推送只发送一小段文本一个线程一秒钟能做几千次send()瓶颈通常不在线程数而在容器的IO和网络带宽。8个线程支持几百个并发连接没大问题。要再省事也可以直接用TaskScheduler。提示SseEmitter超时设置为0的话表示永不超时。但我仍然建议设置一个合理的上限比如1小时避免异常情况下连接永远挂着不释放。超时后Spring容器会自动清理节省资源。3.3 把连接池管理起来实现主动推送上一节同步推送的写法适合固定频率推送但真实业务多半是“某个事件发生了我要立刻推给相关客户端”。这就得把所有活跃的SseEmitter统一管起来。我习惯用一个简单的事件总线管理器Component public class SsePushManager { private final ConcurrentHashMapString, SseEmitter emitterMap new ConcurrentHashMap(); /** * 加入连接池 */ public SseEmitter connect(String clientId) { SseEmitter emitter new SseEmitter(3600_000L); // 注册回调完成、超时、报错都要从map里移除 emitter.onCompletion(() - emitterMap.remove(clientId)); emitter.onTimeout(() - emitterMap.remove(clientId)); emitter.onError(e - emitterMap.remove(clientId)); emitterMap.put(clientId, emitter); return emitter; } /** * 往指定客户端推送 */ public void sendToClient(String clientId, Object data) { SseEmitter emitter emitterMap.get(clientId); if (emitter ! null) { try { emitter.send(SseEmitter.event().data(data)); } catch (IOException e) { emitterMap.remove(clientId); } } } /** * 广播给所有客户端 */ public void broadcast(Object data) { emitterMap.forEach((id, emitter) - { try { emitter.send(SseEmitter.event().data(data)); } catch (IOException e) { emitterMap.remove(id); } }); } }控制层就薄薄一层RestController RequestMapping(/api/sse) public class SubscribeController { Resource private SsePushManager pushManager; GetMapping(/subscribe/{clientId}) public SseEmitter subscribe(PathVariable String clientId) { return pushManager.connect(clientId); } PostMapping(/publish) public void publish(RequestBody MapString, Object payload) { pushManager.broadcast(payload); } }这样前端只要先subscribe拿到一个长连接后端任何一个入口——接口、定时任务、MQ消费者——调到pushManager.broadcast()就能把消息实时推出去。这个模式基本覆盖了我遇到过的九成需求。4. 一个完整的实例任务进度实时上报4.1 整体需求与前端设计上面的框架如果能跑通我们来做一个更接近生产环境的Demo前端发起一个耗时任务后端异步执行执行过程中把每一步的进度推给前端。假设场景是批量处理1000条数据前端要看每个阶段的处理条数和耗时。我直接用EventSource接数据用fetch去触发任务启动。前端页面核心代码HTML 原生JS!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleSSE批量任务进度/title /head body h2批量任务进度/h2 button idstartBtn启动任务/button ul idprogressList/ul script let eventSource null; document.getElementById(startBtn).addEventListener(click, async () { // 先触发后端任务 const resp await fetch(/api/task/start, { method: POST }); const body await resp.json(); if (body.taskId) { connectSse(body.taskId); } }); function connectSse(taskId) { if (eventSource) { eventSource.close(); } // 带上taskId建立连接 eventSource new EventSource(/api/task/progress/ taskId); // 默认事件这里后端没写event名所以走message eventSource.onmessage function(event) { const data JSON.parse(event.data); const li document.createElement(li); li.textContent 进度: data.percent %, 当前阶段: data.stage , 已处理: data.processed / data.total; document.getElementById(progressList).prepend(li); }; // 监听特殊结束事件也可以使用message事件数据判断这里演示自定义事件 eventSource.addEventListener(finish, function(event) { const data JSON.parse(event.data); alert(任务完成总耗时: data.cost ms); eventSource.close(); }); eventSource.onerror function() { // EventSource自带重连不用手动处理 console.log(连接异常等待自动重连...); }; } /script /body /htmlEventSource有两个值得记住的特性断线会自动重连可以在构造函数里带URL参数。我们的taskId直接拼在URL里就行不需要额外设置Header这一点比WebSocket省心。4.2 后端任务执行 SSE推送落地后端流程拆分清楚任务启动接口返回taskId任务执行接口拿着taskId注册好SseEmitter然后在一个异步线程里模拟执行步骤每次处理一批数据就推一条进度消息。RestController RequestMapping(/api/task) public class TaskController { Resource private SsePushManager pushManager; Resource private ScheduledExecutorService sseScheduler; private final AtomicInteger taskIdGen new AtomicInteger(0); private final MapString, SseEmitter activeTasks new ConcurrentHashMap(); PostMapping(/start) public MapString, String startTask() { String taskId task- taskIdGen.incrementAndGet(); // 立即注册SSE连接但前端还没连上没关系EventSource连接很快 SseEmitter emitter pushManager.connect(taskId); activeTasks.put(taskId, emitter); // 异步执行任务 sseScheduler.execute(() - runTask(taskId)); return Map.of(taskId, taskId); } /** * 前端订阅 */ GetMapping(value /progress/{taskId}, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter subscribe(PathVariable String taskId) { SseEmitter emitter activeTasks.get(taskId); if (emitter null) { SseEmitter e new SseEmitter(); e.complete(); return e; } return emitter; } private void runTask(String taskId) { int total 100; int batchSize 10; long start System.currentTimeMillis(); try { for (int i 0; ; i batchSize) { // 模拟耗时处理 Thread.sleep(500); int processed Math.min(i batchSize, total); MapString, Object data new HashMap(); data.put(percent, processed * 100 / total); data.put(stage, processStage(processed, total)); data.put(processed, processed); data.put(total, total); pushManager.sendToClient(taskId, data); if (processed total) { break; } } // 发送完成事件 MapString, Object finish new HashMap(); finish.put(cost, System.currentTimeMillis() - start); pushManager.sendEvent(taskId, finish, finish); } catch (IOException | InterruptedException e) { // 客户端断开任务照常执行只是推送失败 Thread.currentThread().interrupt(); } finally { activeTasks.remove(taskId); } } private String processStage(int processed, int total) { if (processed total * 0.3) return 开始阶段; if (processed total * 0.7) return 中间阶段; return 收尾阶段; } }这个Demo里有几个细节值得展开任务启动时pushManager.connect(taskId)和前端订阅是分开的两个请求。connect()创建了emitter放进map里前端再new EventSource()连过来时Spring MVC会找到这个SseEmitter对象作为Response返回。所以即使前端晚了一秒连上也没事消息不会丢——因为还没开始跑任务呢。发送完成时用了一个sendEvent方法这个方法我建议封装成支持event字段。上一条sendToClient发默认消息完成通知用自定义事件名方便前端区分。Thread.sleep要放在异步线程里绝对不能在Tomcat工作线程里做否则会占用连接线程导致其他请求等待。我在实战中还习惯把activeTasks和pushManager里的map合并成一套直接用SsePushManager来完成订阅和发送避免两个map数据不一致。上面代码为了解释清楚分开了两个容器。真正落地时建议把subscribe接口直接委托给pushManager.connect保证同一个clientId对应同一个emitter。5. 不可忽视的几个进阶处理5.1 心跳保活防止连接“假死”SSE连接虽然挂着但网络链路中很多设备路由器、负载均衡、云平台的网关会对长时间空闲的连接做超时清理。比如某云负载均衡默认空闲超时60秒你的连接60秒没数据就被咔嚓断了。心跳就是每隔一段时间推一个注释行或空数据告诉链路“我还活着”。由于只是占位信息客户端完全感知不到。用Spring Boot实现心跳非常简单就在调度线程里定时往emitter里写event: ping或直接发注释行sseScheduler.scheduleAtFixedRate(() - { activeEmitters.forEach((id, emitter) - { try { emitter.send(SseEmitter.event().name(ping).comment(heartbeat)); } catch (IOException e) { // 连接已断开清理 } }); }, 30, 30, TimeUnit.SECONDS);实测下来30秒一次比较稳妥对流量消耗可以忽略不计。5.2 断线重连和消息去重要一起考虑浏览器EventSource内置了自动重连断线后默认重连间隔大概是1~3秒可以服务端通过retry:字段控制。重连后如果服务端还有未消费的数据完全可以通过上一节讲过的id和Last-Event-ID头实现增量续传。实现思路客户端每次收到带id的消息都记录下来重连时请求头里自动携带Last-Event-ID服务端判断这个值之后的数据继续推。在Spring Boot里读取这个请求头GetMapping(/progress/{taskId}) public SseEmitter subscribe(PathVariable String taskId, RequestHeader(value Last-Event-ID, required false) String lastEventId) { if (lastEventId ! null) { // 从lastEventId之后继续推送实现断线续传 } // ... }这套“消息序列号 客户端ack”的模式做起来容易但绝大多数业务用不上。日常开发更常见的是短时重连——业务数据不追求完全无丢失重启一下重新拉全量就行。我的建议是实时性要求高且允许少量重建缓存、重新拉数据的场景直接让前端close()后再建立新连接必须保证不丢数据的金融级场景再去考虑队列、序列号和窗口确认。5.3 鉴权怎么处理SSE的接口说到底是个GET请求鉴权方式比WebSocket简单得多。常用的三种Cookie鉴权浏览器自动带上和普通页面一致Token拼在URL上new EventSource(/api/sse?tokenxxx)注意Token会出现在访问日志里适合临时连接Header鉴权通过EventSource无法自定义Header但可以在服务端进行拦截校验。如果必须用Header一个变通方案是先请求一个鉴权接口拿到一次性连接凭证再建SSE连接个人建议用Cookie或一次性短期Token省心可靠。对于内网或B端场景Token拼URL是最常见的因为浏览器的EventSource天生不支持自定义请求头。6. 常见问题排查与避坑实录6.1 前端一直是重连状态但收不到任何数据九成原因是反向代理缓冲。在Nginx配置里对SSE接口的location加上location /api/sse/ { proxy_pass http://backend; proxy_set_header Connection ; proxy_http_version 1.1; proxy_buffering off; proxy_cache off; chunked_transfer_encoding on; proxy_read_timeout 3600s; }核心就是proxy_buffering off让数据来一条转一条。如果后端是TomcatConnection: keep-alive会有些兼容性问题把proxy_set_header Connection 配置上然后让Nginx自己管理和后端的长连接Nginx默认会复用上游连接。proxy_read_timeout这个很关键它表示后端在多久没发新数据时Nginx认为连接超时。之前一个项目就是因为这个参数默认60秒SSE推送间隔一长就被Nginx切断。6.2 SseEmitter发送时报IOException但前端看起来还连着这通常是网络设备断开的“滞后感知”。TCP连接一端断掉另一端往往要等下次写数据时才感知到。所以send()抛异常时别犹豫直接emitter.completeWithError()并从map中移除。心跳可以加速这种断线发现。6.3 开启SSE接口后系统报“不能处理异步请求”Spring Boot默认是支持异步请求的。如果自己写了一层Filter要确保Filter里做完了逻辑就chain.doFilter()不要吞掉请求更不要用request.getInputStream()去读SSE的请求体因为SSE是GET请求正常没体可以读。曾经遇到一个同事在权限Filter里尝试读取request body做统一鉴权GET请求读到EOF以后把流关了Tomcat就认为请求不完整后续SSE写数据直接失败。排查半天才发现是这个原因。6.4 连接数暴涨疑似“连接泄漏”SseEmitter的三个回调千万别漏emitter.onCompletion(() - activeTasks.remove(taskId)); emitter.onTimeout(() - activeTasks.remove(taskId)); emitter.onError((e) - activeTasks.remove(taskId));前端如果直接关掉页面服务端短时间内可能感知不到超时回调和错误回调就是兜底回收机制。我建议再做一层冗余清理用调度任务扫描map里上一次心跳时间超过N分钟的emitter直接complete()。6.5 WebSocket和SSE混用时的选择恐惧再补充一个决策工具消息频率极高每秒几十条以上且需要对客户端发消息的场景用WebSocket频率一般、单向推送为主选SSE。SSE还自带HTTP压缩文本数据用gzip压下去带宽可以减少70%以上。WebSocket要做压缩就得自己定制扩展。7. 生产环境级别的SSE封装建议如果项目里SSE用得多我推荐把最外层封装成一个消息推送服务统一屏蔽底层细节。参考设计如下public interface RealtimePusher { SseEmitter subscribe(String topic, String userId); void publish(String topic, Object message); }内部用ConcurrentHashMapString, CopyOnWriteArrayListSseEmitter维护topic和emitter的关系。为什么用CopyOnWriteArrayList因为广播场景是读多写少遍历比改动频繁得多写锁开销小读的时候直接拿快照遍历并发安全。发布消息时可以对消息做简单的toString序列化或者直接用Jackson统一序列化为JSON字符串。注意不要序列化Java对象后直接emitter.send()因为SseEmitter.send()对非String对象也是走JSON序列化但有时Date、LocalDateTime的格式会不一致建议手动统一。最后再分享一个运营层面的注意点SSE连接在服务端是有状态的应用多实例部署时必须考虑连接路由问题。比如A实例维护了一堆连接请求打到B实例就找不到对应emitter。解决办法引入消息中间件做广播每个实例订阅同一个topic收到消息后推给本地维护的连接或者让客户端在订阅时带上实例标识再用负载均衡器做一致性哈希路由。如果你只是单机应用或Demo上面这些都可以忽略但如果是生产环境多实例一定要在架构设计时想清楚。实际操作里比较稳的一条路先不做复杂路由用Redis Pub/Sub或者消息中间件做广播每个实例把自己本地的emitter都注册到一个全局连接管理上消费者收到消息后再走本地推送。等连接数真正大到需要分级时再做更精细的定向路由。这个方案我已经在几个项目里用了一年多稳定度不错而且SSE的维护成本是真的低。当年调WebSocket心跳、重连、断线补包的记忆还历历在目换到SSE之后大部分问题都被浏览器和HTTP规范消解掉了。