1. 项目概述Server-Sent EventsSSE是一种基于HTTP的轻量级实时通信技术它允许服务器主动向客户端推送数据。与WebSocket不同SSE是单向通信服务器→客户端特别适合需要服务器推送但客户端不需要频繁发送数据的场景。我在实际项目中多次使用SSE实现以下功能实时通知系统股票行情推送日志实时监控进度条更新SSE相比轮询和WebSocket有以下优势协议简单基于普通HTTP自动重连机制浏览器原生支持更低的资源消耗2. 核心实现解析2.1 依赖配置虽然Spring Web MVC已经内置SSE支持但我建议添加以下依赖以获得更好的开发体验dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId optionaltrue/optional /dependency dependency groupIdcn.hutool/groupId artifactIdhutool-all/artifactId version5.8.12/version /dependency注意Lombok可以简化代码Hutool提供了优秀的工具类但都不是SSE必需的2.2 SSE核心类设计我优化后的SseClient实现如下Slf4j Component public class SseClient { // 使用ConcurrentHashMap保证线程安全 private static final MapString, SseEmitter CONNECTIONS new ConcurrentHashMap(); // 默认心跳间隔(毫秒) private static final long HEARTBEAT_INTERVAL 30_000; /** * 创建SSE连接 * param clientId 客户端唯一标识 * return SseEmitter实例 */ public SseEmitter createConnection(String clientId) { // 设置1小时超时(根据业务调整) SseEmitter emitter new SseEmitter(3_600_000L); // 添加初始心跳 sendHeartbeat(emitter, clientId); // 设置回调 emitter.onCompletion(() - { log.info([{}]连接正常关闭, clientId); CONNECTIONS.remove(clientId); }); emitter.onTimeout(() - { log.warn([{}]连接超时, clientId); CONNECTIONS.remove(clientId); }); emitter.onError(e - { log.error([{}]连接异常: {}, clientId, e.getMessage()); CONNECTIONS.remove(clientId); }); CONNECTIONS.put(clientId, emitter); log.info([{}]创建SSE连接成功当前连接数: {}, clientId, CONNECTIONS.size()); return emitter; } /** * 发送消息 */ public boolean sendMessage(String clientId, String eventName, Object data) { SseEmitter emitter CONNECTIONS.get(clientId); if (emitter null) { log.warn([{}]连接不存在, clientId); return false; } try { SseEmitter.SseEventBuilder event SseEmitter.event() .id(UUID.randomUUID().toString()) .name(eventName) .data(data) .reconnectTime(5_000L); emitter.send(event); return true; } catch (IOException e) { log.error([{}]消息发送失败: {}, clientId, e.getMessage()); CONNECTIONS.remove(clientId); emitter.completeWithError(e); return false; } } /** * 关闭连接 */ public void closeConnection(String clientId) { SseEmitter emitter CONNECTIONS.remove(clientId); if (emitter ! null) { emitter.complete(); log.info([{}]连接已关闭, clientId); } } /** * 发送心跳保持连接 */ private void sendHeartbeat(SseEmitter emitter, String clientId) { ScheduledExecutorService scheduler Executors.newSingleThreadScheduledExecutor(); scheduler.scheduleAtFixedRate(() - { try { emitter.send(SseEmitter.event() .comment(heartbeat) .reconnectTime(HEARTBEAT_INTERVAL)); } catch (Exception e) { scheduler.shutdown(); } }, 0, HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS); } }关键改进点增加了心跳机制防止长时间无数据导致连接断开使用UUID作为事件ID避免重复更完善的错误处理和日志记录线程安全的连接管理2.3 控制器实现优化后的ControllerRestController RequestMapping(/sse) public class SseController { Autowired private SseClient sseClient; GetMapping(/connect) public SseEmitter connect(RequestParam String clientId) { return sseClient.createConnection(clientId); } PostMapping(/send) public ResponseEntityString sendMessage( RequestParam String clientId, RequestParam(required false) String event, RequestBody Object data) { boolean success sseClient.sendMessage(clientId, event, data); return success ? ResponseEntity.ok(消息发送成功) : ResponseEntity.status(404).body(客户端不存在); } GetMapping(/close) public void closeConnection(RequestParam String clientId) { sseClient.closeConnection(clientId); } }3. 前端实现细节3.1 基础实现!DOCTYPE html html head titleSSE Demo/title style #messages { height: 300px; overflow-y: scroll; border: 1px solid #ccc; padding: 10px; } /style /head body h1SSE实时消息/h1 div idmessages/div script const clientId generateClientId(); let eventSource; // 生成客户端ID function generateClientId() { return client- Math.random().toString(36).substr(2, 9); } // 初始化SSE连接 function initSSE() { if (typeof EventSource ! undefined) { eventSource new EventSource(/sse/connect?clientId${clientId}); eventSource.addEventListener(open, (e) { addMessage(系统, 连接已建立); }); eventSource.addEventListener(error, (e) { if (e.eventPhase EventSource.CLOSED) { addMessage(系统, 连接已关闭); } else { addMessage(系统, 连接错误); } }); // 默认消息处理器 eventSource.onmessage (e) { addMessage(服务器, e.data); }; // 自定义事件处理器 eventSource.addEventListener(custom-event, (e) { const data JSON.parse(e.data); addMessage(data.sender, data.message); }); } else { alert(您的浏览器不支持SSE); } } // 添加消息到界面 function addMessage(sender, message) { const messagesDiv document.getElementById(messages); const messageElement document.createElement(div); messageElement.innerHTML strong${sender}:/strong ${message}; messagesDiv.appendChild(messageElement); messagesDiv.scrollTop messagesDiv.scrollHeight; } // 页加载时初始化 window.onload initSSE; // 页面关闭时断开连接 window.onbeforeunload () { if (eventSource) { eventSource.close(); fetch(/sse/close?clientId${clientId}); } }; /script /body /html3.2 高级功能实现3.2.1 断线重连let reconnectAttempts 0; const MAX_RECONNECT_ATTEMPTS 5; const RECONNECT_DELAY 3000; function initSSE() { eventSource new EventSource(/sse/connect?clientId${clientId}); eventSource.addEventListener(error, (e) { if (e.eventPhase EventSource.CLOSED) { if (reconnectAttempts MAX_RECONNECT_ATTEMPTS) { reconnectAttempts; setTimeout(initSSE, RECONNECT_DELAY); addMessage(系统, 连接断开尝试重连(${reconnectAttempts}/${MAX_RECONNECT_ATTEMPTS})); } else { addMessage(系统, 重连次数已达上限); } } }); }3.2.2 消息类型处理eventSource.addEventListener(notification, (e) { const notification JSON.parse(e.data); showNotification(notification.title, notification.content); }); eventSource.addEventListener(progress, (e) { const progress JSON.parse(e.data); updateProgressBar(progress.value); });4. 生产环境注意事项4.1 性能优化连接管理实现连接池管理设置合理的超时时间定期清理无效连接消息广播public void broadcast(String eventName, Object data) { CONNECTIONS.forEach((clientId, emitter) - { try { emitter.send(SseEmitter.event() .name(eventName) .data(data)); } catch (IOException e) { log.warn([{}]广播消息失败, clientId); CONNECTIONS.remove(clientId); } }); }4.2 安全考虑认证授权GetMapping(/connect) public SseEmitter connect( RequestParam String clientId, RequestHeader(Authorization) String token) { if (!authService.validateToken(token)) { throw new UnauthorizedException(); } return sseClient.createConnection(clientId); }HTTPS必须使用HTTPS保证通信安全CSRF防护虽然SSE是GET请求但仍需考虑CSRF防护4.3 监控与日志建议记录以下指标当前活跃连接数消息发送成功率平均连接时长错误类型统计5. 常见问题解决方案5.1 连接限制问题浏览器对单个域的SSE连接数有限制通常6个解决方案使用不同子域分散连接合并多个数据流到一个连接5.2 代理服务器问题某些代理服务器会缓冲SSE响应导致实时性降低。解决方案设置响应头response.setHeader(X-Accel-Buffering, no);配置代理服务器不缓冲SSE响应5.3 移动端问题移动设备在网络切换时可能导致连接中断。解决方案实现更激进的重连策略在应用恢复时检查连接状态6. 测试策略6.1 单元测试SpringBootTest public class SseClientTest { Autowired private SseClient sseClient; Test void testCreateConnection() { SseEmitter emitter sseClient.createConnection(test-client); assertNotNull(emitter); assertEquals(1, SseClient.getConnectionCount()); } Test void testSendMessage() throws IOException { String clientId test-client; SseEmitter emitter sseClient.createConnection(clientId); MockAsyncContext asyncContext new MockAsyncContext(); emitter.initialize(asyncContext); assertTrue(sseClient.sendMessage(clientId, test, Hello)); } }6.2 压力测试使用JMeter模拟100-1000个并发连接不同频率的消息推送长时间连接稳定性7. 扩展应用场景7.1 结合WebSocketSSE适合服务器推送WebSocket适合双向通信可以组合使用RestController public class CombinedController { Autowired private SseClient sseClient; Autowired private WebSocketHandler webSocketHandler; GetMapping(/notifications) public SseEmitter getNotifications(RequestParam String userId) { return sseClient.createConnection(notifications- userId); } MessageMapping(/chat) public void handleChatMessage(ChatMessage message) { webSocketHandler.broadcast(message); } }7.2 微服务架构在微服务中使用SSE实现事件通知RestController public class OrderController { PostMapping(/orders) public Order createOrder(RequestBody OrderRequest request) { Order order orderService.createOrder(request); eventPublisher.publishEvent(new OrderCreatedEvent(order)); return order; } GetMapping(/orders/events) public SseEmitter orderEvents(RequestParam String userId) { SseEmitter emitter new SseEmitter(); eventPublisher.subscribe(event - { if (event instanceof OrderCreatedEvent ((OrderCreatedEvent) event).getOrder().getUserId().equals(userId)) { emitter.send(event); } }); return emitter; } }8. 性能对比8.1 SSE vs 轮询指标SSE轮询延迟低(~100ms)高(取决于轮询间隔)服务器负载低高带宽消耗低高实现复杂度中等简单8.2 SSE vs WebSocket指标SSEWebSocket方向单向双向协议HTTPWS/WSS连接开销低中等浏览器支持良好优秀消息格式文本二进制/文本9. 实际案例分享9.1 股票行情推送系统实现方案使用SSE推送实时行情每5秒推送一次数据客户端可以订阅特定股票RestController public class StockController { GetMapping(/stocks/stream) public SseEmitter streamStocks(RequestParam ListString symbols) { SseEmitter emitter new SseEmitter(); ScheduledExecutorService executor Executors.newSingleThreadScheduledExecutor(); executor.scheduleAtFixedRate(() - { ListStockQuote quotes stockService.getQuotes(symbols); emitter.send(quotes); }, 0, 5, TimeUnit.SECONDS); emitter.onCompletion(() - executor.shutdown()); return emitter; } }9.2 实时日志监控实现方案后端收集日志事件通过SSE推送到前端支持过滤和搜索RestController public class LogController { GetMapping(/logs/tail) public SseEmitter tailLogs( RequestParam(required false) String level, RequestParam(required false) String keyword) { SseEmitter emitter new SseEmitter(); logAppender.addListener(event - { if (matchesFilter(event, level, keyword)) { emitter.send(event); } }); return emitter; } }10. 最佳实践总结连接管理为每个连接设置合理的超时时间实现心跳机制保持连接及时清理无效连接错误处理捕获所有IO异常实现自动重连机制记录详细的错误日志性能优化避免频繁创建销毁连接合并多个小消息为一个大消息使用压缩减少带宽安全实践必须使用HTTPS实现认证授权限制连接频率监控指标活跃连接数消息吞吐量错误率在实际项目中SSE特别适合以下场景实时通知系统监控仪表盘实时数据可视化进度反馈对于更复杂的实时交互需求建议考虑WebSocket或专业的消息队列系统。 
长河君 · 2026/11/4 9:26:28 · 阅读 1.2k SSE技术实现实时通信：原理、优化与实践
1. 项目概述Server-Sent EventsSSE是一种基于HTTP的轻量级实时通信技术它允许服务器主动向客户端推送数据。与WebSocket不同SSE是单向通信服务器→客户端特别适合需要服务器推送但客户端不需要频繁发送数据的场景。我在实际项目中多次使用SSE实现以下功能实时通知系统股票行情推送日志实时监控进度条更新SSE相比轮询和WebSocket有以下优势协议简单基于普通HTTP自动重连机制浏览器原生支持更低的资源消耗2. 核心实现解析2.1 依赖配置虽然Spring Web MVC已经内置SSE支持但我建议添加以下依赖以获得更好的开发体验dependency groupIdorg.projectlombok/groupId artifactIdlombok/artifactId optionaltrue/optional /dependency dependency groupIdcn.hutool/groupId artifactIdhutool-all/artifactId version5.8.12/version /dependency注意Lombok可以简化代码Hutool提供了优秀的工具类但都不是SSE必需的2.2 SSE核心类设计我优化后的SseClient实现如下Slf4j Component public class SseClient { // 使用ConcurrentHashMap保证线程安全 private static final MapString, SseEmitter CONNECTIONS new ConcurrentHashMap(); // 默认心跳间隔(毫秒) private static final long HEARTBEAT_INTERVAL 30_000; /** * 创建SSE连接 * param clientId 客户端唯一标识 * return SseEmitter实例 */ public SseEmitter createConnection(String clientId) { // 设置1小时超时(根据业务调整) SseEmitter emitter new SseEmitter(3_600_000L); // 添加初始心跳 sendHeartbeat(emitter, clientId); // 设置回调 emitter.onCompletion(() - { log.info([{}]连接正常关闭, clientId); CONNECTIONS.remove(clientId); }); emitter.onTimeout(() - { log.warn([{}]连接超时, clientId); CONNECTIONS.remove(clientId); }); emitter.onError(e - { log.error([{}]连接异常: {}, clientId, e.getMessage()); CONNECTIONS.remove(clientId); }); CONNECTIONS.put(clientId, emitter); log.info([{}]创建SSE连接成功当前连接数: {}, clientId, CONNECTIONS.size()); return emitter; } /** * 发送消息 */ public boolean sendMessage(String clientId, String eventName, Object data) { SseEmitter emitter CONNECTIONS.get(clientId); if (emitter null) { log.warn([{}]连接不存在, clientId); return false; } try { SseEmitter.SseEventBuilder event SseEmitter.event() .id(UUID.randomUUID().toString()) .name(eventName) .data(data) .reconnectTime(5_000L); emitter.send(event); return true; } catch (IOException e) { log.error([{}]消息发送失败: {}, clientId, e.getMessage()); CONNECTIONS.remove(clientId); emitter.completeWithError(e); return false; } } /** * 关闭连接 */ public void closeConnection(String clientId) { SseEmitter emitter CONNECTIONS.remove(clientId); if (emitter ! null) { emitter.complete(); log.info([{}]连接已关闭, clientId); } } /** * 发送心跳保持连接 */ private void sendHeartbeat(SseEmitter emitter, String clientId) { ScheduledExecutorService scheduler Executors.newSingleThreadScheduledExecutor(); scheduler.scheduleAtFixedRate(() - { try { emitter.send(SseEmitter.event() .comment(heartbeat) .reconnectTime(HEARTBEAT_INTERVAL)); } catch (Exception e) { scheduler.shutdown(); } }, 0, HEARTBEAT_INTERVAL, TimeUnit.MILLISECONDS); } }关键改进点增加了心跳机制防止长时间无数据导致连接断开使用UUID作为事件ID避免重复更完善的错误处理和日志记录线程安全的连接管理2.3 控制器实现优化后的ControllerRestController RequestMapping(/sse) public class SseController { Autowired private SseClient sseClient; GetMapping(/connect) public SseEmitter connect(RequestParam String clientId) { return sseClient.createConnection(clientId); } PostMapping(/send) public ResponseEntityString sendMessage( RequestParam String clientId, RequestParam(required false) String event, RequestBody Object data) { boolean success sseClient.sendMessage(clientId, event, data); return success ? ResponseEntity.ok(消息发送成功) : ResponseEntity.status(404).body(客户端不存在); } GetMapping(/close) public void closeConnection(RequestParam String clientId) { sseClient.closeConnection(clientId); } }3. 前端实现细节3.1 基础实现!DOCTYPE html html head titleSSE Demo/title style #messages { height: 300px; overflow-y: scroll; border: 1px solid #ccc; padding: 10px; } /style /head body h1SSE实时消息/h1 div idmessages/div script const clientId generateClientId(); let eventSource; // 生成客户端ID function generateClientId() { return client- Math.random().toString(36).substr(2, 9); } // 初始化SSE连接 function initSSE() { if (typeof EventSource ! undefined) { eventSource new EventSource(/sse/connect?clientId${clientId}); eventSource.addEventListener(open, (e) { addMessage(系统, 连接已建立); }); eventSource.addEventListener(error, (e) { if (e.eventPhase EventSource.CLOSED) { addMessage(系统, 连接已关闭); } else { addMessage(系统, 连接错误); } }); // 默认消息处理器 eventSource.onmessage (e) { addMessage(服务器, e.data); }; // 自定义事件处理器 eventSource.addEventListener(custom-event, (e) { const data JSON.parse(e.data); addMessage(data.sender, data.message); }); } else { alert(您的浏览器不支持SSE); } } // 添加消息到界面 function addMessage(sender, message) { const messagesDiv document.getElementById(messages); const messageElement document.createElement(div); messageElement.innerHTML strong${sender}:/strong ${message}; messagesDiv.appendChild(messageElement); messagesDiv.scrollTop messagesDiv.scrollHeight; } // 页加载时初始化 window.onload initSSE; // 页面关闭时断开连接 window.onbeforeunload () { if (eventSource) { eventSource.close(); fetch(/sse/close?clientId${clientId}); } }; /script /body /html3.2 高级功能实现3.2.1 断线重连let reconnectAttempts 0; const MAX_RECONNECT_ATTEMPTS 5; const RECONNECT_DELAY 3000; function initSSE() { eventSource new EventSource(/sse/connect?clientId${clientId}); eventSource.addEventListener(error, (e) { if (e.eventPhase EventSource.CLOSED) { if (reconnectAttempts MAX_RECONNECT_ATTEMPTS) { reconnectAttempts; setTimeout(initSSE, RECONNECT_DELAY); addMessage(系统, 连接断开尝试重连(${reconnectAttempts}/${MAX_RECONNECT_ATTEMPTS})); } else { addMessage(系统, 重连次数已达上限); } } }); }3.2.2 消息类型处理eventSource.addEventListener(notification, (e) { const notification JSON.parse(e.data); showNotification(notification.title, notification.content); }); eventSource.addEventListener(progress, (e) { const progress JSON.parse(e.data); updateProgressBar(progress.value); });4. 生产环境注意事项4.1 性能优化连接管理实现连接池管理设置合理的超时时间定期清理无效连接消息广播public void broadcast(String eventName, Object data) { CONNECTIONS.forEach((clientId, emitter) - { try { emitter.send(SseEmitter.event() .name(eventName) .data(data)); } catch (IOException e) { log.warn([{}]广播消息失败, clientId); CONNECTIONS.remove(clientId); } }); }4.2 安全考虑认证授权GetMapping(/connect) public SseEmitter connect( RequestParam String clientId, RequestHeader(Authorization) String token) { if (!authService.validateToken(token)) { throw new UnauthorizedException(); } return sseClient.createConnection(clientId); }HTTPS必须使用HTTPS保证通信安全CSRF防护虽然SSE是GET请求但仍需考虑CSRF防护4.3 监控与日志建议记录以下指标当前活跃连接数消息发送成功率平均连接时长错误类型统计5. 常见问题解决方案5.1 连接限制问题浏览器对单个域的SSE连接数有限制通常6个解决方案使用不同子域分散连接合并多个数据流到一个连接5.2 代理服务器问题某些代理服务器会缓冲SSE响应导致实时性降低。解决方案设置响应头response.setHeader(X-Accel-Buffering, no);配置代理服务器不缓冲SSE响应5.3 移动端问题移动设备在网络切换时可能导致连接中断。解决方案实现更激进的重连策略在应用恢复时检查连接状态6. 测试策略6.1 单元测试SpringBootTest public class SseClientTest { Autowired private SseClient sseClient; Test void testCreateConnection() { SseEmitter emitter sseClient.createConnection(test-client); assertNotNull(emitter); assertEquals(1, SseClient.getConnectionCount()); } Test void testSendMessage() throws IOException { String clientId test-client; SseEmitter emitter sseClient.createConnection(clientId); MockAsyncContext asyncContext new MockAsyncContext(); emitter.initialize(asyncContext); assertTrue(sseClient.sendMessage(clientId, test, Hello)); } }6.2 压力测试使用JMeter模拟100-1000个并发连接不同频率的消息推送长时间连接稳定性7. 扩展应用场景7.1 结合WebSocketSSE适合服务器推送WebSocket适合双向通信可以组合使用RestController public class CombinedController { Autowired private SseClient sseClient; Autowired private WebSocketHandler webSocketHandler; GetMapping(/notifications) public SseEmitter getNotifications(RequestParam String userId) { return sseClient.createConnection(notifications- userId); } MessageMapping(/chat) public void handleChatMessage(ChatMessage message) { webSocketHandler.broadcast(message); } }7.2 微服务架构在微服务中使用SSE实现事件通知RestController public class OrderController { PostMapping(/orders) public Order createOrder(RequestBody OrderRequest request) { Order order orderService.createOrder(request); eventPublisher.publishEvent(new OrderCreatedEvent(order)); return order; } GetMapping(/orders/events) public SseEmitter orderEvents(RequestParam String userId) { SseEmitter emitter new SseEmitter(); eventPublisher.subscribe(event - { if (event instanceof OrderCreatedEvent ((OrderCreatedEvent) event).getOrder().getUserId().equals(userId)) { emitter.send(event); } }); return emitter; } }8. 性能对比8.1 SSE vs 轮询指标SSE轮询延迟低(~100ms)高(取决于轮询间隔)服务器负载低高带宽消耗低高实现复杂度中等简单8.2 SSE vs WebSocket指标SSEWebSocket方向单向双向协议HTTPWS/WSS连接开销低中等浏览器支持良好优秀消息格式文本二进制/文本9. 实际案例分享9.1 股票行情推送系统实现方案使用SSE推送实时行情每5秒推送一次数据客户端可以订阅特定股票RestController public class StockController { GetMapping(/stocks/stream) public SseEmitter streamStocks(RequestParam ListString symbols) { SseEmitter emitter new SseEmitter(); ScheduledExecutorService executor Executors.newSingleThreadScheduledExecutor(); executor.scheduleAtFixedRate(() - { ListStockQuote quotes stockService.getQuotes(symbols); emitter.send(quotes); }, 0, 5, TimeUnit.SECONDS); emitter.onCompletion(() - executor.shutdown()); return emitter; } }9.2 实时日志监控实现方案后端收集日志事件通过SSE推送到前端支持过滤和搜索RestController public class LogController { GetMapping(/logs/tail) public SseEmitter tailLogs( RequestParam(required false) String level, RequestParam(required false) String keyword) { SseEmitter emitter new SseEmitter(); logAppender.addListener(event - { if (matchesFilter(event, level, keyword)) { emitter.send(event); } }); return emitter; } }10. 最佳实践总结连接管理为每个连接设置合理的超时时间实现心跳机制保持连接及时清理无效连接错误处理捕获所有IO异常实现自动重连机制记录详细的错误日志性能优化避免频繁创建销毁连接合并多个小消息为一个大消息使用压缩减少带宽安全实践必须使用HTTPS实现认证授权限制连接频率监控指标活跃连接数消息吞吐量错误率在实际项目中SSE特别适合以下场景实时通知系统监控仪表盘实时数据可视化进度反馈对于更复杂的实时交互需求建议考虑WebSocket或专业的消息队列系统。