1. 为什么选择SSE实现实时数据推送？

Server-Sent Events（SSE）是一种基于HTTP的轻量级实时通信协议，特别适合需要服务器向客户端持续推送数据的场景。与WebSocket相比，SSE具有几个显著优势：

首先，SSE采用标准的HTTP协议，不需要额外的协议升级握手过程。这意味着：

• 可以直接复用现有HTTP基础设施
• 更容易通过防火墙和代理服务器
• 天然支持跨域（通过CORS）
• 服务端实现更简单，不需要处理复杂的连接状态

其次，SSE内置了断线重连机制。客户端断开连接后会自动尝试重新连接，服务端可以通过事件ID实现消息续传。这在移动网络不稳定的场景下特别有用。

// 典型SSE响应头 response.setContentType("text/event-stream"); response.setCharacterEncoding("UTF-8"); response.setHeader("Cache-Control", "no-cache"); response.setHeader("Connection", "keep-alive");

注意：SSE是单向通信（服务端→客户端），如果需要双向通信应该选择WebSocket。但在实时监控、消息通知等场景下，SSE往往是更优解。

2. Spring Boot中实现SSE的核心组件

2.1 SseEmitter的工作原理

Spring 4.2+提供了SseEmitter类作为SSE的实现核心。其底层原理是：

1. 保持HTTP连接长时间开放（默认30秒超时）
2. 使用分块传输编码（chunked transfer encoding）
3. 按照SSE规范格式发送事件流

@GetMapping("/stream") public SseEmitter streamData() { SseEmitter emitter = new SseEmitter(180_000L); // 3分钟超时 // 业务逻辑... return emitter; }

关键参数说明：

• 超时时间：根据业务需求设置，过长会占用连接资源
• 缓冲区大小：默认256KB，大数据量需要调整
• 心跳机制：防止代理服务器断开空闲连接

2.2 事件消息的格式规范

SSE协议规定的事件格式包含：

• data：有效载荷（可多行）
• id：事件ID（用于断线续传）
• event：事件类型（前端可监听特定类型）
• retry：重试间隔（毫秒）

emitter.send(SseEmitter.event() .id("event-123") .name("priceUpdate") .data("{\"symbol\":\"AAPL\",\"price\":182.73}") .reconnectTime(5000));

实战经验：data字段建议使用JSON格式，前端更易解析。对于二进制数据需要先做Base64编码。

3. 完整实现方案与性能优化

3.1 服务端实现步骤

1. 创建Emitter仓库（管理活跃连接）

@Component public class SseEmitterRepository { private final Map<String, SseEmitter> emitters = new ConcurrentHashMap<>(); public void addEmitter(String clientId, SseEmitter emitter) { emitters.put(clientId, emitter); emitter.onCompletion(() -> removeEmitter(clientId)); emitter.onTimeout(() -> removeEmitter(clientId)); } // ...其他方法 }

2. 控制器处理订阅请求

@GetMapping("/subscribe/{clientId}") public SseEmitter subscribe(@PathVariable String clientId) { SseEmitter emitter = new SseEmitter(180_000L); emitterRepository.addEmitter(clientId, emitter); return emitter; }

3. 业务事件推送服务

@Service @RequiredArgsConstructor public class EventPushService { private final SseEmitterRepository emitterRepository; public void pushEvent(String clientId, Object data) { emitterRepository.getEmitter(clientId).ifPresent(emitter -> { try { emitter.send(SseEmitter.event() .data(data, MediaType.APPLICATION_JSON)); } catch (IOException e) { emitter.completeWithError(e); } }); } }

3.2 客户端实现方案

前端使用EventSource API：

const eventSource = new EventSource('/subscribe/user123'); eventSource.addEventListener('priceUpdate', (e) => { const data = JSON.parse(e.data); console.log('Received update:', data); }); eventSource.onerror = (err) => { console.error('SSE error:', err); // 自动重连逻辑已内置 };

移动端注意事项：

• Android需要使用OkHttp的SSE支持
• iOS NSURLSession原生支持SSE
• 需要处理应用退到后台时的连接保持

3.3 性能优化要点

1. 连接管理：

• 合理设置超时时间（通常1-5分钟）
• 实现心跳机制（每30秒发送注释行）

// 心跳任务 scheduler.scheduleAtFixedRate(() -> { emitters.forEach((id, emitter) -> { try { emitter.send(SseEmitter.event().comment("heartbeat")); } catch (IOException ignored) {} }); }, 30, 30, TimeUnit.SECONDS);

2. 资源释放：

• 实现onCompletion/onTimeout回调
• 定期清理僵尸连接
• 使用WeakReference存储Emitter

3. 负载均衡策略：

• 会话亲和性（sticky session）
• 使用Redis Pub/Sub跨实例广播

4. 生产环境问题排查指南

4.1 常见问题与解决方案

4.2 监控指标建议

1. 活跃连接数

# HELP sse_active_connections Current active SSE connections # TYPE sse_active_connections gauge sse_active_connections 42

2. 消息吞吐量

# HELP sse_messages_sent_total Total SSE messages sent # TYPE sse_messages_sent_total counter sse_messages_sent_total 1024

3. 错误率监控

try { emitter.send(event); } catch (IOException e) { metrics.counter("sse.errors").increment(); throw e; }

4.3 压力测试方案

使用JMeter模拟测试：

1. 配置HTTP请求头：

Accept: text/event-stream Cache-Control: no-cache

2. 添加响应断言：

• 检查Content-Type是否为text/event-stream
• 验证数据格式合规性

3. 关键测试指标：

• 单机最大并发连接数
• 消息延迟百分位
• 长时间连接的稳定性

5. 高级应用场景扩展

5.1 结合Spring Security实现鉴权

@GetMapping("/secure-stream") public SseEmitter secureStream(Principal principal) { if (!hasAccess(principal)) { throw new AccessDeniedException(); } SseEmitter emitter = new SseEmitter(); // ...其他逻辑 return emitter; }

安全注意事项：

• 避免在URL中暴露敏感参数
• 每个连接应该关联用户会话
• 考虑使用短时效token

5.2 与Spring Cloud Stream集成

@StreamListener("inputChannel") public void handleMessage(Message<String> message) { String clientId = message.getHeaders().get("clientId"); sseService.pushEvent(clientId, message.getPayload()); }

这种架构适合：

• 需要与其他消息系统（Kafka/RabbitMQ）集成
• 高吞吐量场景
• 需要消息持久化的业务

5.3 前端优化技巧

1. 消息队列缓冲：

let messageQueue = []; let isProcessing = false; eventSource.onmessage = (e) => { messageQueue.push(e.data); if (!isProcessing) { processQueue(); } }; function processQueue() { if (messageQueue.length === 0) return; isProcessing = true; const batch = messageQueue.splice(0, 10); // 批量处理逻辑... requestAnimationFrame(processQueue); }

2. 优雅降级策略：

• 检测SSE支持情况
• 自动回退到长轮询
• 提供手动刷新按钮

在实际项目中，SSE特别适合以下场景：

• 实时股票报价
• 新闻资讯推送
• 监控报警系统
• 聊天应用消息通知
• 实时协同编辑

我最近在金融项目中实践发现，对于每分钟数千次的价格更新，SSE相比WebSocket节省了约40%的服务器资源。关键在于合理设置批处理间隔和心跳机制，既保证实时性又避免资源浪费。