在实际 Web 开发中当需要从服务端向客户端推送实时数据时很多人会首先想到 WebSocket。但 WebSocket 是一种全双工协议如果业务场景只是服务端单向推送比如新闻推送、股票价格、任务进度通知等使用 WebSocket 就显得有些重了。这时Server-Sent EventsSSE是一个更轻量、更简单的选择。SSE 基于 HTTP 协议服务端通过一个长连接以流的形式向客户端持续发送数据。客户端使用 EventSource API 接收数据实现起来非常简洁。不过SSE 在生产环境中也会遇到一些典型问题浏览器对同一域名的并发连接数限制可能导致 SSE 连接被阻塞网络不稳定时如何自动重连以及重连后如何避免重复接收消息。下面我们就围绕这几个核心问题从 SSE 的基本用法开始逐步讨论 HTTP/2 如何缓解连接数限制并设计一套可靠的断线重连与消息去重机制。1. 理解 SSE 的基本工作机制与适用场景SSE 是 HTML5 标准的一部分它允许服务端通过 HTTP 连接向客户端推送数据。与 WebSocket 不同SSE 是单向的只能由服务端向客户端发送数据。SSE 基于纯文本协议数据格式简单默认支持断线重连。1.1 SSE 协议格式与浏览器 API服务端响应的 SSE 数据流需要遵循特定的格式每行以\n或\r\n结尾。以data:开头的行表示消息内容。以event:开头的行定义事件类型可选。以id:开头的行设置消息 ID用于断线重连后的消息去重。以retry:开头的行指定重连时间毫秒。一个空行表示消息结束。服务端示例Python FastAPIfrom fastapi import FastAPI from fastapi.responses import StreamingResponse import asyncio app FastAPI() app.get(/stream) async def stream_data(): async def event_generator(): message_id 0 while True: message_id 1 yield fid: {message_id}\n yield event: message\n yield fdata: 这是第 {message_id} 条消息\n\n await asyncio.sleep(1) return StreamingResponse(event_generator(), media_typetext/plain)客户端使用 EventSource 接收消息const eventSource new EventSource(/stream); eventSource.onmessage function(event) { console.log(收到消息:, event.data); }; eventSource.addEventListener(message, function(event) { console.log(自定义事件消息:, event.data); }); eventSource.onerror function(event) { console.error(连接错误:, event); };1.2 SSE 的优缺点与适用场景SSE 的主要优势协议简单基于 HTTP无需额外协议升级。自动重连浏览器内置重连机制。文本友好适合推送文本数据如日志、通知。兼容性好大部分现代浏览器支持。SSE 的局限性单向通信只能服务端推客户端。文本协议二进制数据需要编码。连接数限制受浏览器同源连接数限制。代理问题某些代理服务器可能缓冲 SSE 流。适用场景实时通知、新闻推送、监控数据、进度更新等只需要服务端推送的场景。2. 浏览器连接数限制与 HTTP/2 的解决方案2.1 浏览器并发连接数限制问题浏览器为了性能考虑会对同一域名的并发 HTTP 连接数进行限制。不同浏览器的限制策略不同浏览器HTTP/1.1 连接数限制HTTP/2 连接数限制Chrome6多路复用单个连接Firefox6多路复用单个连接Safari6多路复用单个连接在 HTTP/1.1 环境下如果页面已经打开了 6 个连接如图片、API 请求等再建立 SSE 连接时就会被阻塞直到有其他连接释放。这会导致实时数据延迟影响用户体验。2.2 HTTP/2 的多路复用机制HTTP/2 通过多路复用技术在单个 TCP 连接上并行传输多个请求和响应。这意味着所有请求共享同一个连接不受连接数限制。减少了 TCP 握手和 TLS 握手的开销。头部压缩进一步减少了带宽占用。在 HTTP/2 环境下SSE 连接可以与其他请求共享同一个连接不会受到连接数限制的影响。2.3 服务端配置 HTTP/2 支持以 Nginx 为例配置 HTTP/2server { listen 443 ssl http2; server_name example.com; ssl_certificate /path/to/cert.pem; ssl_certificate_key /path/to/private.key; location /stream { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ; proxy_buffering off; proxy_cache off; } }Spring Boot 中启用 HTTP/2# application.properties server.http2.enabledtrue server.ssl.key-storeclasspath:keystore.p12 server.ssl.key-store-passwordpassword server.ssl.key-store-typePKCS12注意HTTP/2 需要 HTTPS在生产环境中必须配置 SSL 证书。2.4 检测和降级处理虽然 HTTP/2 能解决连接数问题但需要考虑兼容性。客户端可以检测 HTTP/2 支持情况function supportsHTTP2() { return window.performance window.performance.getEntriesByType(navigation)[0].nextHopProtocol h2; } if (!supportsHTTP2()) { console.warn(HTTP/2 not supported, SSE may be affected by connection limits); // 可以考虑减少其他并发请求或使用轮询降级 }3. 设计可靠的断线重连机制虽然浏览器内置了重连机制但在生产环境中需要更精细的控制包括重连策略、错误处理和资源清理。3.1 浏览器默认重连机制SSE 规范定义了自动重连机制连接断开后浏览器会自动尝试重连。重连间隔由服务端的retry:字段控制默认约 3 秒。重连时会携带最后收到的消息 IDLast-Event-ID头。但这种机制比较基础无法应对复杂的网络环境。3.2 增强型重连策略设计我们需要实现一个更智能的重连策略class RobustEventSource { constructor(url, options {}) { this.url url; this.options options; this.reconnectAttempts 0; this.maxReconnectAttempts options.maxReconnectAttempts || 5; this.reconnectDelay options.initialReconnectDelay || 1000; this.maxReconnectDelay options.maxReconnectDelay || 30000; this.connection null; this.lastEventId null; this.isManualClose false; this.connect(); } connect() { if (this.isManualClose) return; const url this.lastEventId ? ${this.url}?lastEventId${this.lastEventId} : this.url; this.connection new EventSource(url); this.connection.onopen () { console.log(SSE 连接已建立); this.reconnectAttempts 0; this.reconnectDelay this.options.initialReconnectDelay || 1000; }; this.connection.onmessage (event) { this.lastEventId event.lastEventId || this.lastEventId; if (this.options.onMessage) { this.options.onMessage(event); } }; this.connection.onerror (event) { console.error(SSE 连接错误, event); if (this.connection.readyState EventSource.CLOSED) { this.scheduleReconnect(); } }; } scheduleReconnect() { if (this.isManualClose || this.reconnectAttempts this.maxReconnectAttempts) { console.error(达到最大重连次数停止重连); if (this.options.onMaxRetries) { this.options.onMaxRetries(); } return; } this.reconnectAttempts; const delay Math.min(this.reconnectDelay * Math.pow(1.5, this.reconnectAttempts - 1), this.maxReconnectDelay); console.log(${delay}ms 后尝试第 ${this.reconnectAttempts} 次重连); setTimeout(() { this.connect(); }, delay); } close() { this.isManualClose true; if (this.connection) { this.connection.close(); } } } // 使用示例 const sse new RobustEventSource(/stream, { initialReconnectDelay: 1000, maxReconnectDelay: 30000, maxReconnectAttempts: 10, onMessage: (event) { console.log(收到消息:, event.data); // 处理业务逻辑 }, onMaxRetries: () { console.error(重连失败显示错误界面); // 显示错误提示提供手动重连按钮 } });3.3 服务端配合支持断线重连服务端需要正确处理Last-Event-ID头避免重复发送消息from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse import asyncio app FastAPI() class MessageStore: def __init__(self): self.messages [] self.max_size 1000 def add_message(self, message): self.messages.append({ id: len(self.messages) 1, data: message, timestamp: asyncio.get_event_loop().time() }) # 保持固定大小避免内存泄漏 if len(self.messages) self.max_size: self.messages self.messages[-self.max_size:] def get_messages_since(self, last_id): last_id int(last_id) if last_id else 0 return [msg for msg in self.messages if msg[id] last_id] message_store MessageStore() app.get(/stream) async def stream_data(request: Request): last_event_id request.headers.get(last-event-id) async def event_generator(): # 先发送错过的消息 if last_event_id: missed_messages message_store.get_messages_since(last_event_id) for msg in missed_messages: yield fid: {msg[id]}\n yield fdata: 补发消息: {msg[data]}\n\n await asyncio.sleep(0.1) # 避免发送过快 # 然后发送新消息 message_id message_store.messages[-1][id] if message_store.messages else 0 while True: message_id 1 new_message f实时消息 {message_id} message_store.add_message(new_message) yield fid: {message_id}\n yield fdata: {new_message}\n\n await asyncio.sleep(2) return StreamingResponse(event_generator(), media_typetext/event-stream)4. 实现消息去重与状态同步机制断线重连后最大的挑战是如何避免重复处理消息以及如何同步客户端状态。4.1 基于消息 ID 的去重方案最简单的去重方式是基于消息 IDclass DeduplicatedEventHandler { constructor() { this.processedIds new Set(); this.maxProcessedSize 1000; // 避免内存无限增长 } handleMessage(event, callback) { const messageId event.lastEventId; // 检查是否已处理过该消息 if (messageId this.processedIds.has(messageId)) { console.log(消息 ${messageId} 已处理跳过); return; } // 执行业务处理 callback(event); // 记录已处理的消息 ID if (messageId) { this.processedIds.add(messageId); this.cleanupOldIds(); } } cleanupOldIds() { if (this.processedIds.size this.maxProcessedSize) { const ids Array.from(this.processedIds).sort(); const removeCount this.processedIds.size - this.maxProcessedSize / 2; for (let i 0; i removeCount; i) { this.processedIds.delete(ids[i]); } } } } // 使用示例 const deduplicator new DeduplicatedEventHandler(); const sse new RobustEventSource(/stream, { onMessage: (event) { deduplicator.handleMessage(event, (msg) { console.log(处理新消息:, msg.data); // 业务逻辑 }); } });4.2 基于时间戳的增量同步对于有时间顺序的数据可以使用时间戳进行增量同步class TimestampBasedSync { constructor() { this.lastSyncTime Date.now(); } async syncMessages() { try { const response await fetch(/api/messages?since${this.lastSyncTime}); const messages await response.json(); if (messages.length 0) { // 处理新消息 messages.forEach(msg { this.processMessage(msg); }); // 更新最后同步时间 this.lastSyncTime Math.max(...messages.map(msg msg.timestamp)); } } catch (error) { console.error(同步失败:, error); } } processMessage(message) { // 业务处理逻辑 console.log(处理消息:, message); } // 重连成功后调用 onReconnect() { this.syncMessages(); } }4.3 服务端消息存储与查询优化服务端需要高效存储和查询历史消息from typing import List, Dict import time import bisect class TimeWindowMessageStore: def __init__(self, window_size: int 3600): # 默认保存1小时 self.window_size window_size self.messages: List[Dict] [] self.message_dict: Dict[int, Dict] {} # 按ID快速查找 def add_message(self, data: str) - int: message_id int(time.time() * 1000) # 使用时间戳作为ID message { id: message_id, data: data, timestamp: time.time() } self.messages.append(message) self.message_dict[message_id] message # 清理过期消息 self.cleanup_old_messages() return message_id def get_messages_since(self, last_id: int, limit: int 100) - List[Dict]: # 使用二分查找提高查询效率 if not self.messages: return [] # 找到第一个大于last_id的消息索引 start_index 0 for i, msg in enumerate(self.messages): if msg[id] last_id: start_index i break return self.messages[start_index:start_index limit] def cleanup_old_messages(self): current_time time.time() cutoff_time current_time - self.window_size # 删除过期消息 while self.messages and self.messages[0][timestamp] cutoff_time: old_msg self.messages.pop(0) self.message_dict.pop(old_msg[id], None)5. 生产环境的最佳实践与故障排查5.1 连接监控与健康检查在生产环境中需要监控 SSE 连接的健康状态class ConnectionMonitor { constructor(sseInstance, checkInterval 30000) { // 30秒检查一次 this.sse sseInstance; this.checkInterval checkInterval; this.lastMessageTime Date.now(); this.monitorTimer null; this.maxSilentTime 60000; // 最大静默时间1分钟 this.startMonitoring(); } startMonitoring() { // 监听消息接收 this.sse.connection.onmessage (event) { this.lastMessageTime Date.now(); // 原有的消息处理逻辑 if (this.sse.options.onMessage) { this.sse.options.onMessage(event); } }; // 定期检查连接状态 this.monitorTimer setInterval(() { this.checkConnectionHealth(); }, this.checkInterval); } checkConnectionHealth() { const silentTime Date.now() - this.lastMessageTime; if (silentTime this.maxSilentTime) { console.warn(连接已静默 ${silentTime}ms可能已断开); this.sse.connection.close(); // 触发重连逻辑 } } stopMonitoring() { if (this.monitorTimer) { clearInterval(this.monitorTimer); } } }5.2 错误处理与降级方案当 SSE 不可用时需要有降级方案class RealTimeDataManager { constructor(url, options {}) { this.url url; this.options options; this.currentStrategy sse; this.fallbackStrategies [sse, long-polling, short-polling]; this.strategyIndex 0; this.start(); } start() { const strategy this.fallbackStrategies[this.strategyIndex]; try { switch (strategy) { case sse: this.startSSE(); break; case long-polling: this.startLongPolling(); break; case short-polling: this.startShortPolling(); break; } this.currentStrategy strategy; } catch (error) { this.fallbackToNextStrategy(); } } startSSE() { this.sse new RobustEventSource(this.url, { onMessage: this.options.onMessage, onMaxRetries: () { this.fallbackToNextStrategy(); } }); } startLongPolling() { console.log(降级到长轮询); this.pollingInterval setInterval(async () { try { const response await fetch(this.url); const data await response.json(); if (this.options.onMessage) { this.options.onMessage({ data: JSON.stringify(data) }); } } catch (error) { console.error(长轮询失败:, error); } }, 5000); // 5秒轮询 } startShortPolling() { console.log(降级到短轮询); this.pollingInterval setInterval(async () { try { const response await fetch(this.url); const data await response.json(); if (this.options.onMessage) { this.options.onMessage({ data: JSON.stringify(data) }); } } catch (error) { console.error(短轮询失败:, error); } }, 1000); // 1秒轮询 } fallbackToNextStrategy() { this.strategyIndex; if (this.strategyIndex this.fallbackStrategies.length) { console.error(所有策略都失败了); if (this.options.onCompleteFailure) { this.options.onCompleteFailure(); } return; } // 清理当前策略 this.cleanupCurrentStrategy(); // 尝试下一个策略 this.start(); } cleanupCurrentStrategy() { if (this.sse) { this.sse.close(); this.sse null; } if (this.pollingInterval) { clearInterval(this.pollingInterval); this.pollingInterval null; } } }5.3 常见问题排查指南问题现象可能原因检查方式解决方案连接立即断开服务端响应格式错误检查网络面板查看响应内容确保响应头为text/event-stream数据格式正确收不到消息代理服务器缓冲检查代理配置设置X-Accel-Buffering: no头禁用代理缓冲重连频繁网络不稳定或服务端问题查看错误日志调整重连策略检查服务端资源消息重复去重逻辑失效检查消息ID生成和去重逻辑确保消息ID唯一完善去重机制内存泄漏消息存储未清理监控内存使用设置合理的消息存储大小限制5.4 性能优化建议连接复用使用 HTTP/2 减少连接数压力。消息压缩对大量文本数据启用 gzip 压缩。批量发送对高频小消息进行批量处理。心跳机制定期发送心跳包保持连接活跃。资源清理页面卸载时主动关闭连接。// 页面卸载时清理资源 window.addEventListener(beforeunload, () { if (sse) { sse.close(); } }); // 心跳机制 setInterval(() { if (sse sse.connection.readyState EventSource.OPEN) { // 可以发送一个特殊的心跳消息 // 或者依赖监控机制检测连接状态 } }, 30000);SSE 为实时数据推送提供了一种简单有效的解决方案。通过 HTTP/2 解决连接数限制设计合理的重连和去重机制再配合完善的监控和降级方案可以在生产环境中构建出稳定可靠的实时通信系统。实际项目中还需要根据具体业务需求调整参数和策略特别是在消息频率、数据量和网络环境方面做好充分的测试和优化。