1. 项目概述
Server-Sent Events（SSE）是一种基于HTTP的轻量级实时通信技术，它允许服务器主动向客户端推送数据。与WebSocket不同，SSE是单向通信（服务器→客户端），特别适合需要服务器推送但客户端不需要频繁发送数据的场景。
我在实际项目中多次使用SSE实现以下功能：
- 实时通知系统
- 股票行情推送
- 日志实时监控
- 进度条更新
SSE相比轮询和WebSocket有以下优势：
- 协议简单，基于普通HTTP
- 自动重连机制
- 浏览器原生支持
- 更低的资源消耗
2. 核心实现解析
2.1 依赖配置
虽然Spring Web MVC已经内置SSE支持，但我建议添加以下依赖以获得更好的开发体验：
<dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>cn.hutool</groupId> <artifactId>hutool-all</artifactId> <version>5.8.12</version> </dependency>注意：Lombok可以简化代码，Hutool提供了优秀的工具类，但都不是SSE必需的
2.2 SSE核心类设计
我优化后的SseClient实现如下：
@Slf4j @Component public class SseClient { // 使用ConcurrentHashMap保证线程安全 private static final Map<String, 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 控制器实现
优化后的Controller：
@RestController @RequestMapping("/sse") public class SseController { @Autowired private SseClient sseClient; @GetMapping("/connect") public SseEmitter connect(@RequestParam String clientId) { return sseClient.createConnection(clientId); } @PostMapping("/send") public ResponseEntity<String> 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> <title>SSE Demo</title> <style> #messages { height: 300px; overflow-y: scroll; border: 1px solid #ccc; padding: 10px; } </style> </head> <body> <h1>SSE实时消息</h1> <div id="messages"></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> </html>3.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 结合WebSocket
SSE适合服务器推送，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
| 指标 | SSE | WebSocket |
|---|---|---|
| 方向 | 单向 | 双向 |
| 协议 | HTTP | WS/WSS |
| 连接开销 | 低 | 中等 |
| 浏览器支持 | 良好 | 优秀 |
| 消息格式 | 文本 | 二进制/文本 |
9. 实际案例分享
9.1 股票行情推送系统
实现方案：
- 使用SSE推送实时行情
- 每5秒推送一次数据
- 客户端可以订阅特定股票
@RestController public class StockController { @GetMapping("/stocks/stream") public SseEmitter streamStocks(@RequestParam List<String> symbols) { SseEmitter emitter = new SseEmitter(); ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor(); executor.scheduleAtFixedRate(() -> { List<StockQuote> 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或专业的消息队列系统。
