1. 项目缘起从“等待”到“流淌”的体验升级最近在做一个后台管理系统的功能需求是用户在前端点击一个“数据导出”按钮后端需要处理一个比较耗时的任务比如生成一份包含几十万条记录的报告。传统的做法是前端发起一个HTTP请求然后就开始“转圈圈”用户只能干等着直到后端处理完所有数据一次性打包成一个文件返回前端才能下载。这个过程短则十几秒长则几分钟用户界面完全卡死体验非常糟糕。更头疼的是如果网络不稳定在最后时刻请求超时了那之前所有的等待和计算都白费了用户还得重来一次。这种“批处理-等待-返回”的模式在需要即时反馈或处理流式数据的场景下显得力不从心。于是我开始寻找一种能让数据“流淌”起来的技术让后端可以一边处理一边就把已经完成的部分推送给前端让用户能实时看到进度甚至先看到部分结果。这就是我选择用SpringBoot搭建SSEServer-Sent Events服务端来实现流式响应的初衷。SSE不是什么新技术它其实是HTML5规范的一部分但正因为其简单、轻量且基于标准的HTTP协议在需要服务器向客户端单向推送数据的场景下比如实时日志、进度通知、新闻推送、股票价格更新等它往往比WebSocket更合适因为后者是为双向通信设计的架构和实现上都更重。简单来说这次的目标就是告别“黑盒”等待实现一个“透明”的、可感知进度的数据流服务。下面我就把自己从零搭建、调试到优化这个SpringBoot SSE服务端的完整过程包括其中的关键决策、踩过的坑和总结的经验毫无保留地分享出来。2. SSE协议核心理解“长连接”与“事件流”在动手写代码之前我们必须先搞清楚SSE到底是什么以及它和普通HTTP请求、WebSocket的区别。这决定了我们后续的代码结构和配置思路。SSE的本质是在客户端和服务器之间建立一条长时间的、单向的HTTP连接。请注意这两个关键词“长时间”和“单向”。普通的HTTP请求是“一问一答”客户端问完服务器答完连接立即关闭。而SSE连接一旦建立就会一直保持打开状态直到服务器主动关闭或发生网络错误。在这个持久的连接上服务器可以随时、多次地向客户端发送数据这就是“单向”的数据流。数据是如何组织的呢SSE规定了一种非常简单的文本格式。服务器推送的每条消息由若干行field: value组成并以一个空行\n\n作为消息的结束分隔符。其中最重要的字段有三个data:消息的数据内容。一行或多行都可以最终客户端接收时会用换行符连接起来。event:事件类型。这是一个字符串标识符客户端可以根据不同的事件类型来绑定不同的处理函数。id:消息ID。用于实现断线重连机制。如果连接意外中断客户端重新连接时可以通过HTTP头Last-Event-ID告诉服务器“我从哪个ID之后的消息开始要”服务器就可以只发送遗漏的消息。一个典型的SSE响应体看起来是这样的data: 这是第一条消息的第一行 data: 这是第一条消息的第二行 event: update data: {progress: 50, status: 处理中} id: 100 event: complete data: 任务处理完成客户端通常是浏览器的EventSourceAPI会解析这个流每当遇到一个空行就触发一次消息事件并把data字段的内容、event字段的类型传递给我们的JavaScript回调函数。那么它和WebSocket的主要区别在哪WebSocket是真正的全双工协议连接建立后客户端和服务器可以随时互相发送消息适合聊天、游戏、协同编辑等强交互场景。而SSE是服务器向客户端的单向推送客户端只能接收。但SSE的优势在于简单基于HTTP/HTTPS无需额外的协议升级握手几乎不需要处理兼容性问题。自动重连EventSource内置了断线重连机制。天然支持现代浏览器都原生支持后端实现也相对简单。对于我们的“进度通知”、“流式日志”这类场景SSE的简单和高效是巨大的优势。理解了这些我们就能明白在SpringBoot中实现SSE核心就是如何保持一个HTTP连接不立即关闭并持续地向这个连接的输出流中写入符合SSE格式的数据。3. SpringBoot中的SSE实现方案选型Spring框架提供了多种处理异步和流式响应的方法对于SSE我们主要有两种主流选择使用SseEmitter或者使用ResponseBodyEmitter。这里需要做一个清晰的区分和选型。方案一使用SseEmitter这是Spring专门为SSE设计的一个类位于org.springframework.web.servlet.mvc.method.annotation包下。它本质上是对ResponseBodyEmitter的一个封装帮我们自动处理了SSE格式的细节比如自动添加data:前缀和结尾的空行。它的API非常直观SseEmitter emitter new SseEmitter(); emitter.send(这是一条消息); // 自动包装为 data: 这是一条消息\n\n emitter.send(SseEmitter.event().name(update).data(进度50%)); emitter.complete(); // 发送结束信号SseEmitter最大的好处是省心。你不需要关心格式只需要关注业务数据和事件。它内部还维护了超时管理和完成/错误回调。对于快速实现一个标准的SSE端点它是首选。方案二使用ResponseBodyEmitter这是一个更底层的抽象它代表一个异步的响应体允许你手动控制向响应输出流中写入任何内容。如果你要实现的不是严格的SSE或者需要混合输出其他内容或者需要对输出格式有绝对的控制权那么可以用它。ResponseBodyEmitter emitter new ResponseBodyEmitter(); emitter.send(data: 自定义格式\n\n, MediaType.TEXT_EVENT_STREAM);使用它来实现SSE就需要自己拼接data:、event:这些前缀和空行。为什么我选择SseEmitter对于绝大多数“服务器向浏览器推送事件”的场景SseEmitter都是更合适的选择。理由如下语义清晰类名SseEmitter直接表明了用途代码可读性高。格式保障避免了手动拼接字符串可能带来的格式错误比如忘了加空行导致客户端无法解析。功能集成它内置了超时处理、完成和错误事件的回调注册这些机制在异步场景下非常重要。社区共识这是Spring社区推荐和普遍使用的方式相关资料和解决方案更丰富。因此我们的项目将围绕SseEmitter来构建。接下来的重点就是如何在一个SpringBoot应用中正确地创建、管理并最终通过SseEmitter对象将数据流式地推送给客户端。4. 实战构建从控制器到业务逻辑的全链路实现现在我们进入具体的代码实现环节。我会按照从外到内、从接口到业务的顺序把每个环节的关键代码和设计思路讲清楚。4.1 控制器层定义SSE端点与连接管理首先我们需要一个Spring MVC的控制器来暴露一个SSE的连接入口。RestController RequestMapping(/api/sse) Slf4j public class SseController { // 用于保存每个客户端的SseEmitterkey可以为用户ID或会话ID private static final MapString, SseEmitter emitterMap new ConcurrentHashMap(); /** * 客户端连接SSE的端点 * param clientId 客户端标识可以从请求参数或Header中传递 * return SseEmitter */ GetMapping(path /connect, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter connect(RequestParam String clientId) { // 设置连接超时时间0表示永不超时但通常建议设置一个较长的值如30分钟LONG_TIMEOUT long timeout 30 * 60 * 1000L; // 30分钟 SseEmitter emitter new SseEmitter(timeout); // 将新的emitter存入Map emitterMap.put(clientId, emitter); // 设置连接完成和超时的回调用于资源清理 emitter.onCompletion(() - { log.info(SSE连接完成clientId: {}, clientId); emitterMap.remove(clientId); }); emitter.onTimeout(() - { log.warn(SSE连接超时clientId: {}, clientId); emitter.completeWithError(new RuntimeException(连接超时)); }); emitter.onError((ex) - { log.error(SSE连接发生错误clientId: {}, clientId, ex); emitterMap.remove(clientId); }); // 可选发送一条连接成功的初始消息 try { emitter.send(SseEmitter.event() .name(connect) .data(SSE连接已建立clientId: clientId) .reconnectTime(5000L)); // 建议客户端5秒后重连 } catch (IOException e) { log.error(发送初始连接消息失败, e); } log.info(新的SSE客户端连接clientId: {}, clientId); return emitter; } /** * 提供一个内部方法供业务服务调用向指定客户端发送消息 */ public static void sendMessage(String clientId, String eventName, Object data) { SseEmitter emitter emitterMap.get(clientId); if (emitter ! null) { try { emitter.send(SseEmitter.event().name(eventName).data(data)); } catch (IOException e) { log.error(向客户端 {} 发送消息失败事件类型: {}, clientId, eventName, e); // 发送失败通常意味着连接已中断移除失效的emitter emitterMap.remove(clientId); } } else { log.warn(客户端 {} 的SSE连接不存在或已关闭, clientId); } } }关键点解析produces MediaType.TEXT_EVENT_STREAM_VALUE这是最重要的注解属性。它告诉Spring这个接口的响应内容类型是text/event-stream这是SSE协议规定的MIME类型。浏览器EventSource对象会识别这个类型。连接管理Map我们使用一个静态的ConcurrentHashMap来管理所有在线的SseEmitter对象。Key通常使用能唯一标识客户端的ID比如用户ID或前端生成的UUID。这里用ConcurrentHashMap是为了线程安全。超时设置SseEmitter构造函数可以传入超时时间。如果不设置会使用Spring MVC的默认异步请求超时时间。对于长连接建议显式设置一个较长的值如30分钟。设置为0代表不超时但要小心资源泄漏。回调函数onCompletion、onTimeout、onError这三个回调是资源清理的黄金位置。无论连接是正常结束、超时还是出错都必须在这里将对应的emitter从Map中移除防止内存泄漏。静态发送方法sendMessage是一个静态工具方法这样任何业务服务如Service注解的类都可以方便地调用SseController.sendMessage(clientId, eventName, data)来推送消息实现了业务逻辑与推送机制的松耦合。4.2 业务服务层模拟耗时任务与进度推送控制器搭建好了接下来我们需要一个模拟的业务服务它执行一个耗时任务并分阶段向客户端推送进度。Service Slf4j public class TaskService { Async(taskExecutor) // 指定使用异步线程池执行 public void executeLongRunningTask(String clientId, String taskId) { log.info(开始执行耗时任务taskId: {}, clientId: {}, taskId, clientId); try { // 阶段1任务开始 SseController.sendMessage(clientId, status, Map.of(taskId, taskId, progress, 0, message, 任务开始初始化...)); Thread.sleep(2000); // 模拟初始化耗时 SseController.sendMessage(clientId, status, Map.of(taskId, taskId, progress, 20, message, 初始化完成开始处理数据...)); // 阶段2数据处理 int totalItems 100; for (int i 1; i totalItems; i) { Thread.sleep(50); // 模拟处理每条数据的耗时 int progress 20 (i * 60 / totalItems); // 进度从20%到80% SseController.sendMessage(clientId, status, Map.of(taskId, taskId, progress, progress, message, 正在处理第 i 条数据...)); } Thread.sleep(1000); // 模拟最终处理耗时 SseController.sendMessage(clientId, status, Map.of(taskId, taskId, progress, 95, message, 数据整合中...)); // 阶段3任务完成 Thread.sleep(500); SseController.sendMessage(clientId, complete, Map.of(taskId, taskId, progress, 100, message, 任务执行成功, downloadUrl, /api/download/ taskId)); log.info(耗时任务执行完毕taskId: {}, taskId); } catch (InterruptedException e) { Thread.currentThread().interrupt(); SseController.sendMessage(clientId, error, Map.of(taskId, taskId, message, 任务被中断)); log.error(任务被中断taskId: {}, taskId, e); } catch (Exception e) { SseController.sendMessage(clientId, error, Map.of(taskId, taskId, message, 任务执行失败: e.getMessage())); log.error(任务执行失败taskId: {}, taskId, e); } } }关键点解析Async异步执行这是核心耗时的任务绝对不能在SSE的连接线程即connect接口的线程中执行。否则你会阻塞这个连接线程导致它无法及时响应其他请求甚至无法发送后续的SSE消息。Async注解将方法提交到Spring的异步线程池中执行立即返回从而释放了HTTP连接线程。进度计算与推送在任务的各个关键节点通过调用SseController.sendMessage推送不同事件类型status,complete,error的消息。消息内容通常用JSON格式便于前端解析。进度百分比需要根据业务逻辑合理计算。异常处理务必在异步任务中捕获所有异常并通过SSE通道将错误信息推送给客户端。如果异常未被捕获任务会静默失败用户将收不到任何反馈。4.3 配置层启用异步与线程池调优要让Async生效并保证系统稳定我们必须进行配置。Configuration EnableAsync // 启用Spring的异步执行能力 public class AsyncConfig { Bean(taskExecutor) public TaskExecutor taskExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); // 核心线程数即使空闲也保留的线程数 executor.setCorePoolSize(5); // 最大线程数队列满后能创建的最大线程数 executor.setMaxPoolSize(20); // 队列容量核心线程满后新任务进入队列等待 executor.setQueueCapacity(100); // 线程名前缀 executor.setThreadNamePrefix(sse-task-); // 拒绝策略当线程池和队列都满时新任务的处理策略 // CallerRunsPolicy: 由调用者线程这里是Tomcat的HTTP线程自己执行这是一种简单的降级 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 非核心线程空闲存活时间秒 executor.setKeepAliveSeconds(60); executor.initialize(); return executor; } }关键点解析EnableAsync这个注解必须加它是开启异步功能的开关。线程池参数这是性能与稳定的关键。不能使用默认的SimpleAsyncTaskExecutor它为每个任务新建线程否则高并发下线程数会爆炸。CorePoolSize和MaxPoolSize需要根据你的服务器资源和任务特性来设定。对于IO密集型如我们的模拟任务可以设大一些。QueueCapacity是缓冲队列。设置太小容易触发拒绝策略设置太大会消耗内存并增加延迟。拒绝策略这里用了CallerRunsPolicy当池和队列满时任务会在调用者线程即Tomcat工作线程中运行。这保证了任务不会被丢弃但会影响到HTTP线程处理新请求的能力。另一种常见策略是AbortPolicy直接抛出异常你需要根据业务重要性来选择。线程命名给线程设置清晰的前缀在排查问题如用jstack看线程堆栈时非常有用。4.4 前端页面使用EventSource接收事件后端准备好了我们还需要一个简单的前端页面来测试。创建一个index.html。!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleSSE流式响应测试/title /head body h2SSE流式任务进度演示/h2 button onclickconnectSSE()连接SSE/button button onclickstartTask()开始模拟任务/button button onclickdisconnectSSE()断开连接/button br/br/ div连接状态: span idstatus未连接/span/div div任务进度: progress idprogress value0 max100/progress span idprogressText0%/span/div div最新消息: span idmessage-/span/div div ideventLog styleborder:1px solid #ccc; height:300px; overflow-y:scroll; padding:10px; margin-top:20px; strong事件日志/strongbr/ /div script let eventSource null; const clientId user_ Math.random().toString(36).substr(2, 9); // 生成一个随机客户端ID function connectSSE() { if (eventSource eventSource.readyState ! EventSource.CLOSED) { logEvent(SSE连接已存在); return; } // 连接后端SSE端点带上clientId参数 const url http://localhost:8080/api/sse/connect?clientId${clientId}; eventSource new EventSource(url); eventSource.onopen function(event) { document.getElementById(status).textContent 已连接; logEvent(SSE连接已建立。); }; // 监听通用消息未指定event字段的消息 eventSource.onmessage function(event) { logEvent(收到消息: ${event.data}); }; // 监听特定事件类型的消息 eventSource.addEventListener(connect, function(event) { logEvent([connect事件] ${event.data}); }); eventSource.addEventListener(status, function(event) { const data JSON.parse(event.data); document.getElementById(progress).value data.progress; document.getElementById(progressText).textContent data.progress %; document.getElementById(message).textContent data.message; logEvent([status事件] 进度: ${data.progress}%, 信息: ${data.message}); }); eventSource.addEventListener(complete, function(event) { const data JSON.parse(event.data); document.getElementById(message).textContent data.message; logEvent([complete事件] ${data.message} 下载地址: ${data.downloadUrl}); // 可以在这里触发文件下载等操作 }); eventSource.addEventListener(error, function(event) { const data JSON.parse(event.data); document.getElementById(message).textContent 错误: data.message; logEvent([error事件] ${data.message}); }); eventSource.onerror function(event) { document.getElementById(status).textContent 连接错误; logEvent(SSE连接发生错误或已关闭。); // EventSource会自动尝试重连 }; } function startTask() { if (!eventSource || eventSource.readyState ! EventSource.OPEN) { alert(请先连接SSE); return; } const taskId task_ Date.now(); // 发起一个普通的HTTP请求来启动后台任务 fetch(/api/task/start?clientId${clientId}taskId${taskId}, { method: POST }).then(response { if (response.ok) { logEvent(任务 ${taskId} 已开始执行。); } else { logEvent(启动任务失败: ${response.status}); } }).catch(err { logEvent(启动任务请求失败: ${err}); }); } function disconnectSSE() { if (eventSource) { eventSource.close(); document.getElementById(status).textContent 已断开; logEvent(SSE连接已手动关闭。); eventSource null; } } function logEvent(msg) { const logDiv document.getElementById(eventLog); logDiv.innerHTML [${new Date().toLocaleTimeString()}] ${msg}br/; logDiv.scrollTop logDiv.scrollHeight; // 自动滚动到底部 } /script /body /html同时需要在后端增加一个触发任务的接口RestController RequestMapping(/api/task) public class TaskController { Autowired private TaskService taskService; PostMapping(/start) public ResponseEntityString startTask(RequestParam String clientId, RequestParam String taskId) { taskService.executeLongRunningTask(clientId, taskId); return ResponseEntity.ok(Task started: taskId); } }前端关键点解析EventSource对象这是浏览器原生API用于创建SSE连接。传入的URL就是我们的/api/sse/connect端点。事件监听onmessage监听所有未指定event字段的消息。addEventListener监听特定event类型如status,complete的消息。这是我们业务推送的主要方式。onopen和onerror监听连接状态。连接管理前端需要维护eventSource对象并在页面卸载时或用户主动操作时调用close()方法以通知服务器清理资源。虽然服务器端有超时清理但主动关闭是更好的实践。启动任务注意启动任务是通过另一个普通的HTTP接口/api/task/start触发的而不是通过SSE连接。SSE连接只用于接收服务器推送的消息。这是一个清晰的职责分离。5. 部署与生产环境下的关键考量把代码跑起来只是第一步。要真正在生产环境使用SSE有几个绕不开的坎必须跨过去。5.1 连接数限制与服务器优化一个SSE连接就是一个长期的HTTP连接。像Tomcat这样的Servlet容器其对并发连接数是有限制的受限于最大线程数maxThreads和连接器配置。默认配置可能只能处理一两百个并发连接。优化建议调整Servlet容器配置以SpringBoot内嵌Tomcat为例在application.yml中server: tomcat: max-connections: 10000 # 最大连接数 max-threads: 200 # 最大工作线程数 min-spare-threads: 10 # 最小空闲线程数增加max-connections和max-threads可以支持更多并发SSE连接。但要注意线程是昂贵的资源线程数过多会导致大量的上下文切换反而降低性能。SSE连接在等待消息期间线程实际上是被挂起的在异步模式下所以对线程的占用不像同步请求那么严重但依然需要合理评估。考虑使用Netty或Undertow对于需要维持大量长连接的场景如消息推送平台可以考虑将SpringBoot的默认容器从Tomcat切换到Netty或Undertow。它们在处理高并发、非阻塞IO方面有更好的设计。SpringBoot WebFlux响应式编程模型默认使用Netty天生适合这种流式、异步的场景。使用反向代理在生产环境中应用前面通常会有Nginx这样的反向代理。你需要配置Nginx支持代理SSE流。location /api/sse/ { proxy_pass http://backend-server; proxy_set_header Connection ; proxy_http_version 1.1; # 必须使用HTTP/1.1 chunked_transfer_encoding off; # 对于某些代理可能需要关闭分块传输编码 proxy_buffering off; # 关键关闭代理缓冲让数据立即转发 proxy_cache off; # 关闭缓存 proxy_read_timeout 3600s; # 设置一个很长的读超时时间 }proxy_buffering off;这一行至关重要。如果Nginx开启了缓冲它会尝试接收完整个后端响应再转发给客户端这就破坏了SSE的“流式”特性客户端会等到所有数据缓冲完才一次性收到。5.2 心跳机制与连接保活网络环境复杂中间可能经过网关、代理、防火墙。这些中间设备为了节省资源可能会关闭长时间没有数据交互的空闲连接。为了解决这个问题我们需要实现心跳机制。心跳就是服务器定期比如每30秒向客户端发送一条没有业务含义的消息例如只包含一个冒号的注释行:\n\n目的只有一个告诉网络中间件和客户端“这个连接还活着”。在SpringBoot中我们可以用一个后台定时任务来实现Component Slf4j public class SseHeartbeatTask { Scheduled(fixedDelay 30000) // 每30秒执行一次 public void sendHeartbeat() { SseController.getEmitterMap().forEach((clientId, emitter) - { if (emitter ! null) { try { // 发送一个注释作为心跳客户端EventSource会忽略它 emitter.send(SseEmitter.event().comment(heartbeat)); } catch (IOException e) { log.debug(发送心跳到客户端 {} 失败连接可能已断开, clientId); // 发送失败可以从Map中移除或者由回调函数处理 } } }); } }同时需要在启动类或配置类上加上EnableScheduling来启用定时任务。这样即使后端长时间没有业务消息推送连接也能保持活跃。5.3 客户端重连与消息可靠性浏览器端的EventSource对象内置了断线重连机制。当连接意外关闭时它会自动尝试重新连接。但是这里有两个问题重连间隔默认的重连时间可能不理想。我们可以在服务器端发送消息时通过retry字段来建议客户端重连的等待时间毫秒。emitter.send(SseEmitter.event().data(Hello).reconnectTime(5000L)); // 建议5秒后重连消息丢失与重复如果连接在服务器发送消息后、客户端接收前中断这条消息就丢失了。更复杂的场景下需要实现消息ID和断点续传。原理是服务器发送每条消息时都带一个递增的id。客户端断线重连时会在请求头中带上最后一次收到的消息IDLast-Event-ID。服务器收到后可以从这个ID之后开始发送消息。// 服务器端发送带ID的消息 String messageId generateNextId(); emitter.send(SseEmitter.event().id(messageId).data(Important Data)); // 在connect接口中可以读取Last-Event-ID头 GetMapping(/connect) public SseEmitter connect(RequestParam String clientId, RequestHeader(value Last-Event-ID, required false) String lastEventId) { // 如果lastEventId不为空可以查询并发送遗漏的消息... }实现完整的消息可靠性保障如确保至少一次、恰好一次送达会引入很大的复杂度通常需要引入消息队列如RabbitMQ, Kafka来持久化消息这超出了基础SSE的范畴。对于进度通知这类允许少量丢失的场景简单的重连机制通常已足够。6. 常见问题排查与性能调优心得在实际开发和压测过程中我遇到了不少典型问题这里总结一下排查思路和优化点。问题一客户端收不到消息或者消息延迟很久才一次性收到。排查步骤检查响应头首先用浏览器开发者工具或curl -i查看SSE接口的响应头确认Content-Type是text/event-stream。检查代理缓冲这是最常见的原因。如果你用了Nginx务必确认配置了proxy_buffering off;。其他代理如Apache, HAProxy也有类似配置。检查服务器端刷新虽然SseEmitter.send()方法内部会处理输出但在某些极端情况下确保在发送关键消息后调用emitter.flush()可以强制刷新缓冲区。检查客户端代码确认前端EventSource的事件监听器绑定正确没有JS错误。问题二连接数上去后服务器负载很高甚至出现OOM内存溢出。排查与优化监控连接数在SseController中记录emitterMap的size()或者通过JMX、Actuator端点监控。严格管理生命周期确保onCompletion、onTimeout、onError回调中一定将emitter从Map中移除。这是防止内存泄漏的生命线。合理设置超时不要设置0无限超时。根据业务场景设置一个合理的超时时间如30分钟让不活跃的连接能被自动清理。优化消息体积SSE消息是文本格式对于复杂数据使用紧凑的JSON避免发送冗余信息。可以考虑对消息进行压缩虽然SSE本身不支持但可以在应用层对data字段的JSON字符串进行gzip后再Base64但会增加客户端复杂度需权衡。评估线程池回顾AsyncConfig中的线程池配置。如果任务都是IO等待型的可以适当调大maxPoolSize和queueCapacity。使用监控工具如VisualVM, Prometheus观察线程池的活动线程数、队列大小避免任务堆积。问题三在分布式部署多台应用服务器时消息无法推送到正确的客户端。问题根源我们的emitterMap是存储在单个应用实例的内存中的。如果用户A连接到了服务器1而触发任务的请求被负载均衡到了服务器2那么服务器2上的TaskService无法找到服务器1内存中的那个emitter推送就会失败。解决方案这就需要引入外部存储来共享连接状态。常见的方案有Redis Pub/Sub每个应用实例订阅一个以clientId命名的频道。当需要向某个客户端推送消息时向对应的Redis频道发布消息。持有该客户端连接的应用实例收到消息后再通过本地的emitter发送出去。这种方式实现相对简单。消息队列如RabbitMQ为每个客户端创建一个队列原理类似。WebSocket集群解决方案如果系统已经使用了Spring的WebSocket且配置了STOMP代理中继如RabbitMQ可以借鉴其思路。但对于纯SSE使用Redis是更轻量的选择。实现分布式SSE会显著增加系统的复杂度因此需要根据实际业务规模和架构需求来决定是否必要。对于中小型应用通过负载均衡器的“会话保持”Session Affinity功能将同一用户的请求尽量路由到同一台后端服务器可以在一定程度上缓解这个问题但这并非高可用架构。7. 进阶思考SSE与WebSocket、HTTP/2 Server Push的对比选型在项目后期我们可能会思考SSE是不是所有场景下的最优解这里简单对比一下其他流式/推送技术。SSE vs WebSocketSSE优势协议简单基于HTTP自动重连浏览器原生支持与现有HTTP基础设施认证、缓存、代理兼容性好。WebSocket优势真正的全双工延迟极低适合高频、双向交互场景如在线游戏、实时协作编辑。选型建议服务器向客户端的单向数据流如通知、日志、进度首选SSE需要客户端频繁向服务器发送数据的双向交互场景选WebSocket。SSE vs HTTP/2 Server PushHTTP/2 Server Push 允许服务器主动向客户端推送资源如CSS, JS文件但它是在单个连接上多路复用的主要目的是优化页面加载而不是用于应用程序数据的实时推送。它缺乏SSE那种“事件流”的语义和客户端API。两者解决的问题域不同SSE在应用数据推送方面更成熟、更专用。SSE vs 长轮询Long Polling长轮询是“伪实时”它需要客户端不断发起新请求。SSE建立一次连接即可持续接收在连接管理、服务器压力和实时性上都优于长轮询。最终技术选型没有银弹。SSE以其简洁、高效和良好的浏览器兼容性在服务器推送领域占据着独特而重要的位置。这次用SpringBoot实现SSE服务端的经历让我深刻体会到将异步处理、连接管理和资源清理这些细节处理好就能将一个看似简单的技术点变成提升用户体验的利器。
1. 项目缘起从“等待”到“流淌”的体验升级最近在做一个后台管理系统的功能需求是用户在前端点击一个“数据导出”按钮后端需要处理一个比较耗时的任务比如生成一份包含几十万条记录的报告。传统的做法是前端发起一个HTTP请求然后就开始“转圈圈”用户只能干等着直到后端处理完所有数据一次性打包成一个文件返回前端才能下载。这个过程短则十几秒长则几分钟用户界面完全卡死体验非常糟糕。更头疼的是如果网络不稳定在最后时刻请求超时了那之前所有的等待和计算都白费了用户还得重来一次。这种“批处理-等待-返回”的模式在需要即时反馈或处理流式数据的场景下显得力不从心。于是我开始寻找一种能让数据“流淌”起来的技术让后端可以一边处理一边就把已经完成的部分推送给前端让用户能实时看到进度甚至先看到部分结果。这就是我选择用SpringBoot搭建SSEServer-Sent Events服务端来实现流式响应的初衷。SSE不是什么新技术它其实是HTML5规范的一部分但正因为其简单、轻量且基于标准的HTTP协议在需要服务器向客户端单向推送数据的场景下比如实时日志、进度通知、新闻推送、股票价格更新等它往往比WebSocket更合适因为后者是为双向通信设计的架构和实现上都更重。简单来说这次的目标就是告别“黑盒”等待实现一个“透明”的、可感知进度的数据流服务。下面我就把自己从零搭建、调试到优化这个SpringBoot SSE服务端的完整过程包括其中的关键决策、踩过的坑和总结的经验毫无保留地分享出来。2. SSE协议核心理解“长连接”与“事件流”在动手写代码之前我们必须先搞清楚SSE到底是什么以及它和普通HTTP请求、WebSocket的区别。这决定了我们后续的代码结构和配置思路。SSE的本质是在客户端和服务器之间建立一条长时间的、单向的HTTP连接。请注意这两个关键词“长时间”和“单向”。普通的HTTP请求是“一问一答”客户端问完服务器答完连接立即关闭。而SSE连接一旦建立就会一直保持打开状态直到服务器主动关闭或发生网络错误。在这个持久的连接上服务器可以随时、多次地向客户端发送数据这就是“单向”的数据流。数据是如何组织的呢SSE规定了一种非常简单的文本格式。服务器推送的每条消息由若干行field: value组成并以一个空行\n\n作为消息的结束分隔符。其中最重要的字段有三个data:消息的数据内容。一行或多行都可以最终客户端接收时会用换行符连接起来。event:事件类型。这是一个字符串标识符客户端可以根据不同的事件类型来绑定不同的处理函数。id:消息ID。用于实现断线重连机制。如果连接意外中断客户端重新连接时可以通过HTTP头Last-Event-ID告诉服务器“我从哪个ID之后的消息开始要”服务器就可以只发送遗漏的消息。一个典型的SSE响应体看起来是这样的data: 这是第一条消息的第一行 data: 这是第一条消息的第二行 event: update data: {progress: 50, status: 处理中} id: 100 event: complete data: 任务处理完成客户端通常是浏览器的EventSourceAPI会解析这个流每当遇到一个空行就触发一次消息事件并把data字段的内容、event字段的类型传递给我们的JavaScript回调函数。那么它和WebSocket的主要区别在哪WebSocket是真正的全双工协议连接建立后客户端和服务器可以随时互相发送消息适合聊天、游戏、协同编辑等强交互场景。而SSE是服务器向客户端的单向推送客户端只能接收。但SSE的优势在于简单基于HTTP/HTTPS无需额外的协议升级握手几乎不需要处理兼容性问题。自动重连EventSource内置了断线重连机制。天然支持现代浏览器都原生支持后端实现也相对简单。对于我们的“进度通知”、“流式日志”这类场景SSE的简单和高效是巨大的优势。理解了这些我们就能明白在SpringBoot中实现SSE核心就是如何保持一个HTTP连接不立即关闭并持续地向这个连接的输出流中写入符合SSE格式的数据。3. SpringBoot中的SSE实现方案选型Spring框架提供了多种处理异步和流式响应的方法对于SSE我们主要有两种主流选择使用SseEmitter或者使用ResponseBodyEmitter。这里需要做一个清晰的区分和选型。方案一使用SseEmitter这是Spring专门为SSE设计的一个类位于org.springframework.web.servlet.mvc.method.annotation包下。它本质上是对ResponseBodyEmitter的一个封装帮我们自动处理了SSE格式的细节比如自动添加data:前缀和结尾的空行。它的API非常直观SseEmitter emitter new SseEmitter(); emitter.send(这是一条消息); // 自动包装为 data: 这是一条消息\n\n emitter.send(SseEmitter.event().name(update).data(进度50%)); emitter.complete(); // 发送结束信号SseEmitter最大的好处是省心。你不需要关心格式只需要关注业务数据和事件。它内部还维护了超时管理和完成/错误回调。对于快速实现一个标准的SSE端点它是首选。方案二使用ResponseBodyEmitter这是一个更底层的抽象它代表一个异步的响应体允许你手动控制向响应输出流中写入任何内容。如果你要实现的不是严格的SSE或者需要混合输出其他内容或者需要对输出格式有绝对的控制权那么可以用它。ResponseBodyEmitter emitter new ResponseBodyEmitter(); emitter.send(data: 自定义格式\n\n, MediaType.TEXT_EVENT_STREAM);使用它来实现SSE就需要自己拼接data:、event:这些前缀和空行。为什么我选择SseEmitter对于绝大多数“服务器向浏览器推送事件”的场景SseEmitter都是更合适的选择。理由如下语义清晰类名SseEmitter直接表明了用途代码可读性高。格式保障避免了手动拼接字符串可能带来的格式错误比如忘了加空行导致客户端无法解析。功能集成它内置了超时处理、完成和错误事件的回调注册这些机制在异步场景下非常重要。社区共识这是Spring社区推荐和普遍使用的方式相关资料和解决方案更丰富。因此我们的项目将围绕SseEmitter来构建。接下来的重点就是如何在一个SpringBoot应用中正确地创建、管理并最终通过SseEmitter对象将数据流式地推送给客户端。4. 实战构建从控制器到业务逻辑的全链路实现现在我们进入具体的代码实现环节。我会按照从外到内、从接口到业务的顺序把每个环节的关键代码和设计思路讲清楚。4.1 控制器层定义SSE端点与连接管理首先我们需要一个Spring MVC的控制器来暴露一个SSE的连接入口。RestController RequestMapping(/api/sse) Slf4j public class SseController { // 用于保存每个客户端的SseEmitterkey可以为用户ID或会话ID private static final MapString, SseEmitter emitterMap new ConcurrentHashMap(); /** * 客户端连接SSE的端点 * param clientId 客户端标识可以从请求参数或Header中传递 * return SseEmitter */ GetMapping(path /connect, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter connect(RequestParam String clientId) { // 设置连接超时时间0表示永不超时但通常建议设置一个较长的值如30分钟LONG_TIMEOUT long timeout 30 * 60 * 1000L; // 30分钟 SseEmitter emitter new SseEmitter(timeout); // 将新的emitter存入Map emitterMap.put(clientId, emitter); // 设置连接完成和超时的回调用于资源清理 emitter.onCompletion(() - { log.info(SSE连接完成clientId: {}, clientId); emitterMap.remove(clientId); }); emitter.onTimeout(() - { log.warn(SSE连接超时clientId: {}, clientId); emitter.completeWithError(new RuntimeException(连接超时)); }); emitter.onError((ex) - { log.error(SSE连接发生错误clientId: {}, clientId, ex); emitterMap.remove(clientId); }); // 可选发送一条连接成功的初始消息 try { emitter.send(SseEmitter.event() .name(connect) .data(SSE连接已建立clientId: clientId) .reconnectTime(5000L)); // 建议客户端5秒后重连 } catch (IOException e) { log.error(发送初始连接消息失败, e); } log.info(新的SSE客户端连接clientId: {}, clientId); return emitter; } /** * 提供一个内部方法供业务服务调用向指定客户端发送消息 */ public static void sendMessage(String clientId, String eventName, Object data) { SseEmitter emitter emitterMap.get(clientId); if (emitter ! null) { try { emitter.send(SseEmitter.event().name(eventName).data(data)); } catch (IOException e) { log.error(向客户端 {} 发送消息失败事件类型: {}, clientId, eventName, e); // 发送失败通常意味着连接已中断移除失效的emitter emitterMap.remove(clientId); } } else { log.warn(客户端 {} 的SSE连接不存在或已关闭, clientId); } } }关键点解析produces MediaType.TEXT_EVENT_STREAM_VALUE这是最重要的注解属性。它告诉Spring这个接口的响应内容类型是text/event-stream这是SSE协议规定的MIME类型。浏览器EventSource对象会识别这个类型。连接管理Map我们使用一个静态的ConcurrentHashMap来管理所有在线的SseEmitter对象。Key通常使用能唯一标识客户端的ID比如用户ID或前端生成的UUID。这里用ConcurrentHashMap是为了线程安全。超时设置SseEmitter构造函数可以传入超时时间。如果不设置会使用Spring MVC的默认异步请求超时时间。对于长连接建议显式设置一个较长的值如30分钟。设置为0代表不超时但要小心资源泄漏。回调函数onCompletion、onTimeout、onError这三个回调是资源清理的黄金位置。无论连接是正常结束、超时还是出错都必须在这里将对应的emitter从Map中移除防止内存泄漏。静态发送方法sendMessage是一个静态工具方法这样任何业务服务如Service注解的类都可以方便地调用SseController.sendMessage(clientId, eventName, data)来推送消息实现了业务逻辑与推送机制的松耦合。4.2 业务服务层模拟耗时任务与进度推送控制器搭建好了接下来我们需要一个模拟的业务服务它执行一个耗时任务并分阶段向客户端推送进度。Service Slf4j public class TaskService { Async(taskExecutor) // 指定使用异步线程池执行 public void executeLongRunningTask(String clientId, String taskId) { log.info(开始执行耗时任务taskId: {}, clientId: {}, taskId, clientId); try { // 阶段1任务开始 SseController.sendMessage(clientId, status, Map.of(taskId, taskId, progress, 0, message, 任务开始初始化...)); Thread.sleep(2000); // 模拟初始化耗时 SseController.sendMessage(clientId, status, Map.of(taskId, taskId, progress, 20, message, 初始化完成开始处理数据...)); // 阶段2数据处理 int totalItems 100; for (int i 1; i totalItems; i) { Thread.sleep(50); // 模拟处理每条数据的耗时 int progress 20 (i * 60 / totalItems); // 进度从20%到80% SseController.sendMessage(clientId, status, Map.of(taskId, taskId, progress, progress, message, 正在处理第 i 条数据...)); } Thread.sleep(1000); // 模拟最终处理耗时 SseController.sendMessage(clientId, status, Map.of(taskId, taskId, progress, 95, message, 数据整合中...)); // 阶段3任务完成 Thread.sleep(500); SseController.sendMessage(clientId, complete, Map.of(taskId, taskId, progress, 100, message, 任务执行成功, downloadUrl, /api/download/ taskId)); log.info(耗时任务执行完毕taskId: {}, taskId); } catch (InterruptedException e) { Thread.currentThread().interrupt(); SseController.sendMessage(clientId, error, Map.of(taskId, taskId, message, 任务被中断)); log.error(任务被中断taskId: {}, taskId, e); } catch (Exception e) { SseController.sendMessage(clientId, error, Map.of(taskId, taskId, message, 任务执行失败: e.getMessage())); log.error(任务执行失败taskId: {}, taskId, e); } } }关键点解析Async异步执行这是核心耗时的任务绝对不能在SSE的连接线程即connect接口的线程中执行。否则你会阻塞这个连接线程导致它无法及时响应其他请求甚至无法发送后续的SSE消息。Async注解将方法提交到Spring的异步线程池中执行立即返回从而释放了HTTP连接线程。进度计算与推送在任务的各个关键节点通过调用SseController.sendMessage推送不同事件类型status,complete,error的消息。消息内容通常用JSON格式便于前端解析。进度百分比需要根据业务逻辑合理计算。异常处理务必在异步任务中捕获所有异常并通过SSE通道将错误信息推送给客户端。如果异常未被捕获任务会静默失败用户将收不到任何反馈。4.3 配置层启用异步与线程池调优要让Async生效并保证系统稳定我们必须进行配置。Configuration EnableAsync // 启用Spring的异步执行能力 public class AsyncConfig { Bean(taskExecutor) public TaskExecutor taskExecutor() { ThreadPoolTaskExecutor executor new ThreadPoolTaskExecutor(); // 核心线程数即使空闲也保留的线程数 executor.setCorePoolSize(5); // 最大线程数队列满后能创建的最大线程数 executor.setMaxPoolSize(20); // 队列容量核心线程满后新任务进入队列等待 executor.setQueueCapacity(100); // 线程名前缀 executor.setThreadNamePrefix(sse-task-); // 拒绝策略当线程池和队列都满时新任务的处理策略 // CallerRunsPolicy: 由调用者线程这里是Tomcat的HTTP线程自己执行这是一种简单的降级 executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy()); // 非核心线程空闲存活时间秒 executor.setKeepAliveSeconds(60); executor.initialize(); return executor; } }关键点解析EnableAsync这个注解必须加它是开启异步功能的开关。线程池参数这是性能与稳定的关键。不能使用默认的SimpleAsyncTaskExecutor它为每个任务新建线程否则高并发下线程数会爆炸。CorePoolSize和MaxPoolSize需要根据你的服务器资源和任务特性来设定。对于IO密集型如我们的模拟任务可以设大一些。QueueCapacity是缓冲队列。设置太小容易触发拒绝策略设置太大会消耗内存并增加延迟。拒绝策略这里用了CallerRunsPolicy当池和队列满时任务会在调用者线程即Tomcat工作线程中运行。这保证了任务不会被丢弃但会影响到HTTP线程处理新请求的能力。另一种常见策略是AbortPolicy直接抛出异常你需要根据业务重要性来选择。线程命名给线程设置清晰的前缀在排查问题如用jstack看线程堆栈时非常有用。4.4 前端页面使用EventSource接收事件后端准备好了我们还需要一个简单的前端页面来测试。创建一个index.html。!DOCTYPE html html langzh-CN head meta charsetUTF-8 titleSSE流式响应测试/title /head body h2SSE流式任务进度演示/h2 button onclickconnectSSE()连接SSE/button button onclickstartTask()开始模拟任务/button button onclickdisconnectSSE()断开连接/button br/br/ div连接状态: span idstatus未连接/span/div div任务进度: progress idprogress value0 max100/progress span idprogressText0%/span/div div最新消息: span idmessage-/span/div div ideventLog styleborder:1px solid #ccc; height:300px; overflow-y:scroll; padding:10px; margin-top:20px; strong事件日志/strongbr/ /div script let eventSource null; const clientId user_ Math.random().toString(36).substr(2, 9); // 生成一个随机客户端ID function connectSSE() { if (eventSource eventSource.readyState ! EventSource.CLOSED) { logEvent(SSE连接已存在); return; } // 连接后端SSE端点带上clientId参数 const url http://localhost:8080/api/sse/connect?clientId${clientId}; eventSource new EventSource(url); eventSource.onopen function(event) { document.getElementById(status).textContent 已连接; logEvent(SSE连接已建立。); }; // 监听通用消息未指定event字段的消息 eventSource.onmessage function(event) { logEvent(收到消息: ${event.data}); }; // 监听特定事件类型的消息 eventSource.addEventListener(connect, function(event) { logEvent([connect事件] ${event.data}); }); eventSource.addEventListener(status, function(event) { const data JSON.parse(event.data); document.getElementById(progress).value data.progress; document.getElementById(progressText).textContent data.progress %; document.getElementById(message).textContent data.message; logEvent([status事件] 进度: ${data.progress}%, 信息: ${data.message}); }); eventSource.addEventListener(complete, function(event) { const data JSON.parse(event.data); document.getElementById(message).textContent data.message; logEvent([complete事件] ${data.message} 下载地址: ${data.downloadUrl}); // 可以在这里触发文件下载等操作 }); eventSource.addEventListener(error, function(event) { const data JSON.parse(event.data); document.getElementById(message).textContent 错误: data.message; logEvent([error事件] ${data.message}); }); eventSource.onerror function(event) { document.getElementById(status).textContent 连接错误; logEvent(SSE连接发生错误或已关闭。); // EventSource会自动尝试重连 }; } function startTask() { if (!eventSource || eventSource.readyState ! EventSource.OPEN) { alert(请先连接SSE); return; } const taskId task_ Date.now(); // 发起一个普通的HTTP请求来启动后台任务 fetch(/api/task/start?clientId${clientId}taskId${taskId}, { method: POST }).then(response { if (response.ok) { logEvent(任务 ${taskId} 已开始执行。); } else { logEvent(启动任务失败: ${response.status}); } }).catch(err { logEvent(启动任务请求失败: ${err}); }); } function disconnectSSE() { if (eventSource) { eventSource.close(); document.getElementById(status).textContent 已断开; logEvent(SSE连接已手动关闭。); eventSource null; } } function logEvent(msg) { const logDiv document.getElementById(eventLog); logDiv.innerHTML [${new Date().toLocaleTimeString()}] ${msg}br/; logDiv.scrollTop logDiv.scrollHeight; // 自动滚动到底部 } /script /body /html同时需要在后端增加一个触发任务的接口RestController RequestMapping(/api/task) public class TaskController { Autowired private TaskService taskService; PostMapping(/start) public ResponseEntityString startTask(RequestParam String clientId, RequestParam String taskId) { taskService.executeLongRunningTask(clientId, taskId); return ResponseEntity.ok(Task started: taskId); } }前端关键点解析EventSource对象这是浏览器原生API用于创建SSE连接。传入的URL就是我们的/api/sse/connect端点。事件监听onmessage监听所有未指定event字段的消息。addEventListener监听特定event类型如status,complete的消息。这是我们业务推送的主要方式。onopen和onerror监听连接状态。连接管理前端需要维护eventSource对象并在页面卸载时或用户主动操作时调用close()方法以通知服务器清理资源。虽然服务器端有超时清理但主动关闭是更好的实践。启动任务注意启动任务是通过另一个普通的HTTP接口/api/task/start触发的而不是通过SSE连接。SSE连接只用于接收服务器推送的消息。这是一个清晰的职责分离。5. 部署与生产环境下的关键考量把代码跑起来只是第一步。要真正在生产环境使用SSE有几个绕不开的坎必须跨过去。5.1 连接数限制与服务器优化一个SSE连接就是一个长期的HTTP连接。像Tomcat这样的Servlet容器其对并发连接数是有限制的受限于最大线程数maxThreads和连接器配置。默认配置可能只能处理一两百个并发连接。优化建议调整Servlet容器配置以SpringBoot内嵌Tomcat为例在application.yml中server: tomcat: max-connections: 10000 # 最大连接数 max-threads: 200 # 最大工作线程数 min-spare-threads: 10 # 最小空闲线程数增加max-connections和max-threads可以支持更多并发SSE连接。但要注意线程是昂贵的资源线程数过多会导致大量的上下文切换反而降低性能。SSE连接在等待消息期间线程实际上是被挂起的在异步模式下所以对线程的占用不像同步请求那么严重但依然需要合理评估。考虑使用Netty或Undertow对于需要维持大量长连接的场景如消息推送平台可以考虑将SpringBoot的默认容器从Tomcat切换到Netty或Undertow。它们在处理高并发、非阻塞IO方面有更好的设计。SpringBoot WebFlux响应式编程模型默认使用Netty天生适合这种流式、异步的场景。使用反向代理在生产环境中应用前面通常会有Nginx这样的反向代理。你需要配置Nginx支持代理SSE流。location /api/sse/ { proxy_pass http://backend-server; proxy_set_header Connection ; proxy_http_version 1.1; # 必须使用HTTP/1.1 chunked_transfer_encoding off; # 对于某些代理可能需要关闭分块传输编码 proxy_buffering off; # 关键关闭代理缓冲让数据立即转发 proxy_cache off; # 关闭缓存 proxy_read_timeout 3600s; # 设置一个很长的读超时时间 }proxy_buffering off;这一行至关重要。如果Nginx开启了缓冲它会尝试接收完整个后端响应再转发给客户端这就破坏了SSE的“流式”特性客户端会等到所有数据缓冲完才一次性收到。5.2 心跳机制与连接保活网络环境复杂中间可能经过网关、代理、防火墙。这些中间设备为了节省资源可能会关闭长时间没有数据交互的空闲连接。为了解决这个问题我们需要实现心跳机制。心跳就是服务器定期比如每30秒向客户端发送一条没有业务含义的消息例如只包含一个冒号的注释行:\n\n目的只有一个告诉网络中间件和客户端“这个连接还活着”。在SpringBoot中我们可以用一个后台定时任务来实现Component Slf4j public class SseHeartbeatTask { Scheduled(fixedDelay 30000) // 每30秒执行一次 public void sendHeartbeat() { SseController.getEmitterMap().forEach((clientId, emitter) - { if (emitter ! null) { try { // 发送一个注释作为心跳客户端EventSource会忽略它 emitter.send(SseEmitter.event().comment(heartbeat)); } catch (IOException e) { log.debug(发送心跳到客户端 {} 失败连接可能已断开, clientId); // 发送失败可以从Map中移除或者由回调函数处理 } } }); } }同时需要在启动类或配置类上加上EnableScheduling来启用定时任务。这样即使后端长时间没有业务消息推送连接也能保持活跃。5.3 客户端重连与消息可靠性浏览器端的EventSource对象内置了断线重连机制。当连接意外关闭时它会自动尝试重新连接。但是这里有两个问题重连间隔默认的重连时间可能不理想。我们可以在服务器端发送消息时通过retry字段来建议客户端重连的等待时间毫秒。emitter.send(SseEmitter.event().data(Hello).reconnectTime(5000L)); // 建议5秒后重连消息丢失与重复如果连接在服务器发送消息后、客户端接收前中断这条消息就丢失了。更复杂的场景下需要实现消息ID和断点续传。原理是服务器发送每条消息时都带一个递增的id。客户端断线重连时会在请求头中带上最后一次收到的消息IDLast-Event-ID。服务器收到后可以从这个ID之后开始发送消息。// 服务器端发送带ID的消息 String messageId generateNextId(); emitter.send(SseEmitter.event().id(messageId).data(Important Data)); // 在connect接口中可以读取Last-Event-ID头 GetMapping(/connect) public SseEmitter connect(RequestParam String clientId, RequestHeader(value Last-Event-ID, required false) String lastEventId) { // 如果lastEventId不为空可以查询并发送遗漏的消息... }实现完整的消息可靠性保障如确保至少一次、恰好一次送达会引入很大的复杂度通常需要引入消息队列如RabbitMQ, Kafka来持久化消息这超出了基础SSE的范畴。对于进度通知这类允许少量丢失的场景简单的重连机制通常已足够。6. 常见问题排查与性能调优心得在实际开发和压测过程中我遇到了不少典型问题这里总结一下排查思路和优化点。问题一客户端收不到消息或者消息延迟很久才一次性收到。排查步骤检查响应头首先用浏览器开发者工具或curl -i查看SSE接口的响应头确认Content-Type是text/event-stream。检查代理缓冲这是最常见的原因。如果你用了Nginx务必确认配置了proxy_buffering off;。其他代理如Apache, HAProxy也有类似配置。检查服务器端刷新虽然SseEmitter.send()方法内部会处理输出但在某些极端情况下确保在发送关键消息后调用emitter.flush()可以强制刷新缓冲区。检查客户端代码确认前端EventSource的事件监听器绑定正确没有JS错误。问题二连接数上去后服务器负载很高甚至出现OOM内存溢出。排查与优化监控连接数在SseController中记录emitterMap的size()或者通过JMX、Actuator端点监控。严格管理生命周期确保onCompletion、onTimeout、onError回调中一定将emitter从Map中移除。这是防止内存泄漏的生命线。合理设置超时不要设置0无限超时。根据业务场景设置一个合理的超时时间如30分钟让不活跃的连接能被自动清理。优化消息体积SSE消息是文本格式对于复杂数据使用紧凑的JSON避免发送冗余信息。可以考虑对消息进行压缩虽然SSE本身不支持但可以在应用层对data字段的JSON字符串进行gzip后再Base64但会增加客户端复杂度需权衡。评估线程池回顾AsyncConfig中的线程池配置。如果任务都是IO等待型的可以适当调大maxPoolSize和queueCapacity。使用监控工具如VisualVM, Prometheus观察线程池的活动线程数、队列大小避免任务堆积。问题三在分布式部署多台应用服务器时消息无法推送到正确的客户端。问题根源我们的emitterMap是存储在单个应用实例的内存中的。如果用户A连接到了服务器1而触发任务的请求被负载均衡到了服务器2那么服务器2上的TaskService无法找到服务器1内存中的那个emitter推送就会失败。解决方案这就需要引入外部存储来共享连接状态。常见的方案有Redis Pub/Sub每个应用实例订阅一个以clientId命名的频道。当需要向某个客户端推送消息时向对应的Redis频道发布消息。持有该客户端连接的应用实例收到消息后再通过本地的emitter发送出去。这种方式实现相对简单。消息队列如RabbitMQ为每个客户端创建一个队列原理类似。WebSocket集群解决方案如果系统已经使用了Spring的WebSocket且配置了STOMP代理中继如RabbitMQ可以借鉴其思路。但对于纯SSE使用Redis是更轻量的选择。实现分布式SSE会显著增加系统的复杂度因此需要根据实际业务规模和架构需求来决定是否必要。对于中小型应用通过负载均衡器的“会话保持”Session Affinity功能将同一用户的请求尽量路由到同一台后端服务器可以在一定程度上缓解这个问题但这并非高可用架构。7. 进阶思考SSE与WebSocket、HTTP/2 Server Push的对比选型在项目后期我们可能会思考SSE是不是所有场景下的最优解这里简单对比一下其他流式/推送技术。SSE vs WebSocketSSE优势协议简单基于HTTP自动重连浏览器原生支持与现有HTTP基础设施认证、缓存、代理兼容性好。WebSocket优势真正的全双工延迟极低适合高频、双向交互场景如在线游戏、实时协作编辑。选型建议服务器向客户端的单向数据流如通知、日志、进度首选SSE需要客户端频繁向服务器发送数据的双向交互场景选WebSocket。SSE vs HTTP/2 Server PushHTTP/2 Server Push 允许服务器主动向客户端推送资源如CSS, JS文件但它是在单个连接上多路复用的主要目的是优化页面加载而不是用于应用程序数据的实时推送。它缺乏SSE那种“事件流”的语义和客户端API。两者解决的问题域不同SSE在应用数据推送方面更成熟、更专用。SSE vs 长轮询Long Polling长轮询是“伪实时”它需要客户端不断发起新请求。SSE建立一次连接即可持续接收在连接管理、服务器压力和实时性上都优于长轮询。最终技术选型没有银弹。SSE以其简洁、高效和良好的浏览器兼容性在服务器推送领域占据着独特而重要的位置。这次用SpringBoot实现SSE服务端的经历让我深刻体会到将异步处理、连接管理和资源清理这些细节处理好就能将一个看似简单的技术点变成提升用户体验的利器。
)
