1. 项目背景与核心痛点为什么SSE的“永久连接”是个伪命题在构建现代Web应用特别是需要实时数据推送的场景时Server-Sent EventsSSE因其协议简单、天然支持浏览器端EventSource API而备受青睐。很多开发者包括我自己在早期都曾被其“长连接”特性所吸引认为只要建立了连接就能一劳永逸地实现服务端到客户端的单向数据流。Spring Boot通过SseEmitter这个类将SSE的服务器端实现封装得非常友好几行代码就能跑通一个Demo。然而真实的线上环境很快就会给你上一课。你会发现那些在本地开发环境跑得好好的连接在线上时不时就“僵死”了——服务端还在孜孜不倦地准备数据但客户端早已因为网络波动、页面关闭、设备休眠等原因悄无声息地离线了。更棘手的是SseEmitter本身并没有提供一个可靠的、主动的机制来即时感知这种离线状态。它的onCompletion和onTimeout回调往往是在连接已经彻底终结如超时、显式关闭后才会触发对于网络瞬断、客户端异常崩溃这类“静默离线”几乎是无效的。这就引出了我们面临的核心痛点一个“永久存活”的SSE连接在客户端离线后会变成一个持续消耗服务器资源内存、线程的“僵尸连接”。服务端会持续为这个不存在的客户端维护SseEmitter实例尝试发送数据这些操作会占用连接池、线程池资源甚至可能引发内存泄漏。当这种僵尸连接积累到一定数量轻则导致服务响应变慢重则直接拖垮整个应用。所以我们真正要解决的不是一个技术上的“永久连接”而是一个工程上的“健壮连接”。目标是在享受SSE便捷性的同时确保连接池的清洁及时清理无效连接释放资源。这不仅仅是优化更是线上服务稳定性的基石。2. SseEmitter 工作机制与生命周期管理盲区要解决问题必须先透彻理解工具本身。Spring Boot的SseEmitter是ResponseBodyEmitter的子类它本质上是对异步请求处理的一种封装。当我们创建一个SseEmitter并返回给Spring MVC时框架会持有一个指向该对象的引用并保持HTTP连接打开允许我们通过send()方法异步地多次写入数据。它的生命周期大致由以下几个关键点构成创建与绑定在Controller中实例化SseEmitter通常我们会设置一个超时时间例如new SseEmitter(30_000L)。这个超时时间至关重要它决定了连接在没有任何活动的情况下能保持多久。数据发送通过emitter.send(data)方法发送事件。这里的data可以是String、MediaType定义的对象等。每次发送都会重置内部的“最后活动时间”。完成与超时连接正常结束时如客户端调用EventSource.close()会触发onCompletion回调。如果连接在超时时间内没有任何发送或接收活动则会触发onTimeout回调。错误处理发送过程中发生IO异常等错误会触发onError回调。问题的盲区就在于**“活动”的定义**。SseEmitter及其底层的机制判断连接是否活跃主要依据是否有数据成功写入到网络通道。但是当客户端网络断开时服务端的操作系统TCP栈可能不会立即感知取决于TCP Keep-Alive和系统配置或者感知到错误但Spring的抽象层没有以我们能捕获的方式立即抛出。此时服务端调用emitter.send(data)数据会进入缓冲区但最终会因为无法送达而由底层触发IO错误。从调用send()到错误回调触发这中间可能存在一个不可忽视的时间窗口。在这个窗口期内连接在逻辑上已经是“僵尸”状态但我们无法通过API立即得知。此外onTimeout回调的触发依赖于我们设置的超时时间。如果我们为了维持“永久”连接而设置一个极长的超时如Long.MAX_VALUE那么这个本应作为安全网的机制也失效了。因此单纯依赖SseEmitter内置的生命周期回调无法满足对客户端离线状态的实时、可靠判断需求。我们必须引入额外的“心跳”或“健康检查”机制。3. 实战方案心跳机制 连接管理器实现可靠探活经过多个项目的迭代我认为最有效且易于实施的方案是“应用层心跳”配合一个中央化的“连接管理器”。这个方案不依赖于特定的TCP层行为而是在我们自己的业务逻辑里定义连接的活性。3.1 核心架构设计整个方案围绕两个核心组件展开心跳事件客户端定期向服务端发送一个特殊的事件比如type: ping。服务端收到后更新该连接的最后活跃时间戳。连接管理器一个全局的单例组件负责存储所有活跃的SseEmitter及其元数据如客户端ID、最后心跳时间。它同时会启动一个后台定时任务扫描所有连接清理那些超过一定时间未发送心跳的“僵尸连接”。3.2 详细实现步骤3.2.1 定义连接与管理器首先我们创建一个连接持有类和一个管理器。import org.springframework.web.servlet.mvc.method.annotation.SseEmitter; import java.util.Map; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Executors; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; /** * 连接信息封装 */ public class SseConnection { private String clientId; private SseEmitter emitter; private long lastHeartbeatTime; // 构造函数、getter、setter 省略... public void updateHeartbeat() { this.lastHeartbeatTime System.currentTimeMillis(); } public boolean isTimeout(long timeoutMillis) { return System.currentTimeMillis() - lastHeartbeatTime timeoutMillis; } } /** * 连接管理器单例可通过Component注入 */ Component public class SseConnectionManager { private final MapString, SseConnection connectionMap new ConcurrentHashMap(); private final ScheduledExecutorService scheduler Executors.newSingleThreadScheduledExecutor(); // 心跳超时时间例如 45秒 private static final long HEARTBEAT_TIMEOUT_MS 45_000L; // 清理任务执行间隔例如 30秒 private static final long CLEANUP_INTERVAL_MS 30_000L; PostConstruct public void init() { // 项目启动后开始定时清理任务 scheduler.scheduleAtFixedRate(this::cleanupStaleConnections, CLEANUP_INTERVAL_MS, CLEANUP_INTERVAL_MS, TimeUnit.MILLISECONDS); } PreDestroy public void destroy() { scheduler.shutdown(); } /** * 注册新连接 */ public void addConnection(String clientId, SseEmitter emitter) { SseConnection conn new SseConnection(clientId, emitter); conn.updateHeartbeat(); // 创建时即更新心跳 connectionMap.put(clientId, conn); // 设置SseEmitter的完成和超时回调用于被动清理 emitter.onCompletion(() - removeConnection(clientId)); emitter.onTimeout(() - removeConnection(clientId)); emitter.onError((e) - removeConnection(clientId)); } /** * 更新客户端心跳 */ public void updateHeartbeat(String clientId) { SseConnection conn connectionMap.get(clientId); if (conn ! null) { conn.updateHeartbeat(); } } /** * 向特定客户端发送数据 */ public void sendToClient(String clientId, Object data) { SseConnection conn connectionMap.get(clientId); if (conn ! null conn.getEmitter() ! null) { try { conn.getEmitter().send(data); // 发送数据也算是一种活跃可以更新心跳时间可选 // conn.updateHeartbeat(); } catch (Exception e) { // 发送失败很可能连接已失效直接移除 removeConnection(clientId); } } } /** * 定时清理僵尸连接 */ private void cleanupStaleConnections() { long now System.currentTimeMillis(); connectionMap.entrySet().removeIf(entry - { SseConnection conn entry.getValue(); if (conn.isTimeout(HEARTBEAT_TIMEOUT_MS)) { // 主动完成并清理资源 try { conn.getEmitter().complete(); } catch (Exception ignored) {} log.warn(清理僵尸SSE连接ClientId: {}, conn.getClientId()); return true; // 移除该条目 } return false; }); } private void removeConnection(String clientId) { SseConnection removed connectionMap.remove(clientId); if (removed ! null) { log.info(SSE连接被移除ClientId: {}, clientId); } } }3.2.2 实现SSE控制器与心跳接口接下来在Controller中我们提供连接建立和心跳接收的端点。RestController RequestMapping(/sse) public class SseController { Autowired private SseConnectionManager connectionManager; /** * 客户端连接入口 * param clientId 客户端唯一标识可以从请求参数或Header中获取 */ GetMapping(path /connect, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter connect(RequestParam String clientId) { // 设置一个较长的超时时间因为我们将用应用层心跳管理活性 SseEmitter emitter new SseEmitter(360_000L); // 例如6分钟 // 注册连接到管理器 connectionManager.addConnection(clientId, emitter); // 立即发送一个欢迎事件或连接成功事件 try { emitter.send(SseEmitter.event().name(connect).data({\status\:\connected\})); } catch (IOException e) { // 初始发送失败可能客户端立即断开了交给管理器回调处理 } return emitter; } /** * 客户端心跳接口 */ PostMapping(/heartbeat) public ResponseEntity? heartbeat(RequestParam String clientId) { connectionManager.updateHeartbeat(clientId); return ResponseEntity.ok().build(); } /** * 业务数据推送示例接口 */ PostMapping(/push) public ResponseEntity? pushMessage(RequestParam String clientId, RequestBody String message) { connectionManager.sendToClient(clientId, SseEmitter.event().data(message)); return ResponseEntity.ok().build(); } }3.2.3 客户端实现要点客户端需要做两件事建立连接和定期发送心跳。// 使用 EventSource const clientId generateUniqueClientId(); // 生成或从服务器获取一个唯一ID const eventSource new EventSource(/sse/connect?clientId${clientId}); eventSource.onopen (event) { console.log(SSE连接已建立); // 启动心跳定时器 startHeartbeat(clientId); }; eventSource.onmessage (event) { console.log(收到数据:, event.data); // 处理业务数据... }; eventSource.onerror (event) { console.error(SSE连接错误, event); // 发生错误尝试重连 eventSource.close(); setTimeout(() { // 重新执行连接逻辑 }, 5000); }; // 心跳函数 function startHeartbeat(clientId) { setInterval(() { // 使用fetch或axios发送心跳请求 fetch(/sse/heartbeat?clientId${clientId}, { method: POST }) .catch(err console.warn(心跳发送失败可能网络不佳, err)); }, 30000); // 每30秒发送一次心跳应小于服务端的超时时间(45秒) }关键设计点客户端心跳间隔30秒必须小于服务端定义的连接超时时间45秒。这样即使某一次心跳因网络抖动失败在下一次心跳成功之前连接还不会被判定为超时提供了容错空间。这个“时间差”是系统鲁棒性的关键。4. 方案优化与生产环境进阶考量上面的方案已经能解决90%的问题但在高并发、高可用的生产环境中我们还需要考虑更多。4.1 分布式环境下的连接管理上述SseConnectionManager是基于内存的ConcurrentHashMap这在单机部署时没问题。但在微服务或集群部署时客户端可能连接到任意一个服务实例。A实例上的连接管理器并不知道B实例上维护的连接。解决方案是引入一个外部集中式存储例如Redis。存储结构使用Redis的Hash或Sorted Set。Key可以是sse:connectionsField是clientIdValue是序列化的连接信息如最后心跳时间、所在服务器实例标识。心跳更新客户端发送心跳时服务端实例将对应clientId的最后心跳时间戳写入Redis。全局清理需要一个独立的定时任务可以部署在其中一个实例上或使用分布式调度框架定期扫描Redis中所有连接信息清理超时的记录。同时每个服务实例也需要监听Redis中自己负责的连接是否被清理然后本地执行emitter.complete()。实例标识连接信息中需要包含serverInstanceId这样在清理时可以通知到正确的服务实例去关闭本地连接。这引入了复杂度但换来了水平扩展的能力。4.2 心跳机制的强化与容错双向心跳上述是客户端主动上报心跳。更健壮的做法是服务端也主动下探。服务端可以定期比如每隔20秒向所有活跃连接发送一个特殊的“服务器ping”事件。客户端收到后必须在一个很短的时间窗口内如5秒回复一个“pong”事件。这能更快地发现单向网络中断。自适应心跳间隔在连接稳定时可以适当拉长心跳间隔如60秒以减少请求数。当检测到网络不稳定心跳失败或延迟高时自动缩短间隔如15秒进行更密集的探活。心跳与业务数据合并如果业务数据推送非常频繁可以省略独立的心跳包将每次成功的业务数据发送视为一次有效的心跳。这需要在管理器的sendToClient方法成功发送后也调用updateHeartbeat。4.3 资源清理与异常处理细节SseEmitter.complete()与SseEmitter.completeWithError()在清理连接时务必调用complete()方法。这会给客户端发送一个SSE流结束的信号一个空的data:消息并触发本地的onCompletion回调确保Spring MVC框架能正确回收相关资源。如果是因为错误可以使用completeWithError()。并发修改问题在cleanupStaleConnections方法中我们遍历connectionMap并可能移除元素。使用ConcurrentHashMap和entrySet().removeIf()是线程安全的。但在高并发下仍需注意在sendToClient等方法中获取到的SseConnection可能刚被清理任务移除因此需要进行判空。连接数监控与告警将connectionMap.size()通过Micrometer等工具暴露为监控指标。设置告警阈值当活跃连接数异常增长可能内存泄漏或骤降可能网络或服务问题时能及时通知运维人员。4.4 与WebSocket的选型再思考在实现了如此复杂的心跳和管理逻辑后你可能会问这和WebSocket还有多大区别确实SSE在需要服务端主动向客户端推送的场景下其简洁性优势在应对“永久连接”管理时被削弱了。选型建议坚持使用SSE如果你的场景主要是服务端向客户端单向推送如新闻推送、告警通知、股票价格且客户端主要是现代浏览器SSE的协议简单、自动重连、与HTTP基础设施兼容性好如身份认证、代理依然是巨大优势。本文的方案就是为此保驾护航。考虑WebSocket如果你的应用是双向、高频、低延迟的交互如聊天室、协同编辑、实时游戏或者客户端环境复杂如某些不支持EventSource的旧浏览器或移动端框架那么WebSocket是更自然的选择。WebSocket协议本身提供了Ping/Pong帧用于保活虽然同样需要自己管理连接池但双向通信的模型更匹配这类需求。5. 常见踩坑点与调试技巧在实际落地过程中我遇到了不少坑这里分享出来帮你避雷。SseEmitter超时时间设置不当不要设置为Long.MAX_VALUE。即使有心跳也应设置一个合理的上限如10-30分钟作为最后的防线防止某些极端情况下心跳逻辑失效导致连接永远无法释放。客户端忘记发送心跳这是最常见的问题。务必在前端代码中确保心跳定时器被正确启动并且在页面unload或连接onerror时清除定时器避免内存泄漏。Nginx等代理服务器超时SSE连接会长期占用一个HTTP连接。你需要确保Nginx的proxy_read_timeout、proxy_send_timeout等配置值足够大且大于你的心跳超时时间否则代理层会主动断开连接。通常需要配置为proxy_read_timeout 3600s;这样的大值。浏览器连接数限制HTTP/1.1下浏览器对同一域名有并发连接数限制通常是6个。如果你的页面同时打开了多个SSE连接可能会阻塞其他资源的加载。可以考虑使用HTTP/2或者将SSE服务部署在独立的子域名下。调试工具在浏览器开发者工具的“网络”(Network)选项卡中筛选EventStream类型可以直观地看到SSE事件流、连接状态和传输的数据。这是调试前端连接问题最有效的工具。在服务端通过日志详细记录连接的创建、心跳更新、数据发送和清理事件对于排查问题至关重要。实现一个健壮的、能应对客户端各种离线情况的SSE服务远不止几行SseEmitter代码那么简单。它要求我们对HTTP长连接的本质、Spring的异步处理模型、网络的不确定性有更深的理解。通过引入应用层心跳和主动连接管理我们能够有效地解决“僵尸连接”问题构建出稳定可靠的实时数据推送服务。这套方案虽然增加了一些复杂度但换来的系统稳定性和资源可控性在生产环境中是完全值得的。
