你有没有见过这样的场景：页面上一个“正在思考”的加载动画，用户等得都快睡着了，后端其实早就把结果算出来了，却怎么也推不过去。我接手某智能问答平台的改造时，面对的就是这种场面。旧系统用的是最传统的轮询方案：前端每两秒就向服务端问一次“Agent 思考完了没”，每次请求都要过网关、走鉴权、查会话状态，用户量一上来，数据库扛不住，页面也开始卡。更要命的是，Agent 内部一调用外部工具，响应慢一点，连接就断，用户看到的就是一个永远转圈的加载状态。

痛定思痛之后，我们用 Spring Boot 3 的 SseEmitter 配合 Redis Stream，把整套 Agent 的流式思考与工具调用过程，从“轮询黑洞”改造成一条可以被前端实时订阅的事件流。这套方案上线后，延迟从秒级降到毫秒级，服务端压力也小了一个量级。如果你正在搭 AI Agent 的实时反馈链路、做工具调用进度的可视化，或者单纯想搞清楚 SSE 到底能不能扛住生产级场景，这篇文章可以给你一个完整参考。

1. 告别轮询：通信架构选型与决策思考

1.1 轮询方案的三个致命伤

先聊聊旧系统的轮询到底痛在哪里。最直接的问题是延迟和无效请求。当时为了知道 Agent 是否思考完成，前端每两秒打一次接口，接口内部去查一条状态记录。如果有 100 个并发用户在跑，每秒钟就有 50 个左右的查询请求，其中绝大多数都查不到新状态，相当于把数据库和网关当成了“状态轮播器”在空转。

第二个问题是压力传导。每次轮询都会经过负载均衡、应用网关、鉴权过滤器，最终落到数据库查询上。Agent 单次推理如果耗时 3 分钟，一个用户就会产生 90 次轮询请求，整个链路都被无效流量填满。更尴尬的是，高峰期数据库查询变慢，轮询请求堆积，反而拖慢了真正有价值的业务请求。

第三个问题是断流体验差。网页端的轮询一旦遇到网络抖动或者代理超时，前端没有收到预期的结果，就只能报错让用户重试。没有推送能力，服务端只能被动等客户端来问，这在 Agent 这种“过程很长、结果分批产生”的场景里，体验几乎是无解的。后来我们试过长轮询，把请求挂住等数据，但本质上还是在“猜结果有没有好”，Tomcat 线程被大量占用，问题只是换了一种形式出现。

1.2 SSE 与 WebSocket 的对比：为什么选了“单行道”

一开始团队里也有人提议上 WebSocket，毕竟 WebSocket 听起来更“实时”。但深入分析场景之后，我发现对于 Agent 流式思考与工具调用这种业务，WebSocket 是杀鸡用牛刀，甚至还会引入更多麻烦。

先做个直观对比：

Agent 的核心场景里，绝大多数时间数据是从服务端流向客户端：思考片段、工具调用状态、答案 token，都是服务端在主动产出。客户端真正需要主动上报的东西很少，无非就是“取消当前生成”之类的控制指令，这种指令用普通 POST 请求就能实现，完全不需要为它维护一条全双工长连接。

生活里可以这样类比：WebSocket 是一条双向四车道的高速公路，但业务里 99% 的车流都是从服务端仓库往客户端送货；SSE 是一条直达的单车道专线，虽然只能单向跑，但跑得又快又稳，而且天然支持断线自动重新接上。既然双向能力几乎用不上，为什么要承担 WebSocket 的握手、心跳、重连、代理兼容等一系列额外成本呢？

1.3 SSE 在生产环境里的“隐藏优势”

除了协议本身的匹配度，SSE 在部署层面还占了便宜。它本质上是 HTTP 响应，Content-Type 是text/event-stream，这意味着现有的负载均衡、鉴权、日志链路基本可以复用。Nginx 这类反向代理对 SSE 的支持也相对简单，窗口期长一点的配置就能搞定。

另外，SSE 直接支持断线续传。客户端断开后，浏览器原生的 EventSource 会自动重连，并且可以带上Last-Event-ID请求头，服务端根据这个 ID 把缺失的事件重新推送一遍。这在 Agent 场景里太重要了：用户中途刷新页面，原本已经展示的思考过程和工具调用进度不应该丢。我们基于这个能力设计了完整的事件恢复方案，后面会详细讲。

Spring 生态里也有现成的SseEmitter，不需要引入额外的推送框架，集成成本非常低。这一点在技术选型时是决定性的——能用标准组件解决的事，就不应该自己造轮子。

2. 整体架构拆解：让 Agent 的思考过程变成数据流

2.1 三个核心模块的职责划分

决定用 SSE 之后，我面临的下一个问题是：事件流从哪里来，怎么保证多实例下不丢不重。于是我把系统划分成三个核心模块：

第一层是接入层，基于 Spring Boot 3 实现。它的职责是创建SseEmitter，维护连接注册表，接收客户端订阅请求，并且把上游产生的事件推送给对应会话。这个层不处理 Agent 的推理逻辑，只负责“通道”的稳定性。

第二层是 Agent 运行时。它负责决策循环：调用大模型生成思考，判断是否触发工具调用，执行工具，把结果回填上下文，继续下一轮推理。每一次状态变化都会生成一个事件对象，交到事件通道里。

第三层是 Redis 层。它承担三个职责：事件流持久化、会话状态与游标保存、分布式锁与幂等控制。为什么不让 Agent 运行时直接把事件推给 SseEmitter？因为一旦服务多实例部署，A 实例创建的连接，事件可能由 B 实例产生，必须有一个中间层来做事件的路由和暂存。Redis Stream 就是我们选的那个中间层。

客户端 <——> Spring Boot 3 接入层 <——> Redis Stream <——> Agent 运行时 | +——> 会话状态 Hash / 分布式锁 / 幂等标记

2.2 事件模型设计：定义一套 Agent 可观测协议

把 Agent 的思考过程变成数据流之后，最核心的问题是事件类型要清晰。我们参考了一些公开的 Agent 事件协议思路，结合业务场景最终设计了下面这套事件类型：

每个事件都统一携带这些字段：会话 ID、事件序列号、事件类型、事件产生时间、业务 JSON 数据。其中事件序列号是整个体系的关键，它既是断线续传的游标，也是消息乱序排查的线索。

这里特别说一下工具调用这种事件。Agent 在调用工具时，前端需要展示的不只是一个“工具执行中”的静态文字，而是尽量具体的动态信息。比如“正在调用代码解释器”“正在查询数据库”“正在请求天气接口”这样的实时反馈，会让用户明显感觉系统是“活”的，而不是卡住了。所以我们把工具调用拆成了三个事件：TOOL_CALL表示 Agent 决定调用什么，TOOL_EXECUTING表示执行开始，TOOL_RESULT表示执行完成。三段式拆法，配合审计日志，后续排查问题也非常方便。

2.3 断线续传的语义设计

SSE 自带的Last-Event-ID机制好归好，但如果业务里有多个事件流混在一起，单靠这个字段是不够的。我们在实现时做了一层包装：事件序列号是全会话递增的，客户端断线重连时，把之前收到的最大事件序列号通过请求参数传给服务端。服务端拿到这个序列号，从 Redis Stream 里拉取该会话后续的事件，重放给客户端。

这种设计能解决两个实际问题。第一，用户刷新页面后能恢复之前的进度，而不是从头开始重新生成。第二，跨实例重连时，哪怕客户端从负载均衡器落到了另一台机器，只要 Redis 里存着事件流，新实例就能无缝接管推送。这也是我强调“用 Redis Stream 而不用内存队列”的原因——进程重启了、实例挂了，事件流还在 Redis 里，不会跟着进程一起消失。

3. 核心实现：基于 SseEmitter 与 Redis Stream 的落地细节

3.1 Spring Boot 3 中 SseEmitter 的基本用法

先来看接入层最基础的代码形态。新建一个控制器，暴露两个接口，一个负责订阅事件流，一个负责触发 Agent 执行：

@RestController @RequestMapping("/agent/stream") public class AgentStreamController { private final Map<String, SseEmitter> emitterRegistry = new ConcurrentHashMap<>(); private final ExecutorService agentExecutor = Executors.newVirtualThreadPerTaskExecutor(); private final AgentRuntime agentRuntime; public AgentStreamController(AgentRuntime agentRuntime) { this.agentRuntime = agentRuntime; } @GetMapping(value = "/{sessionId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter subscribe(@PathVariable String sessionId) { SseEmitter emitter = new SseEmitter(0L); emitterRegistry.put(sessionId, emitter); emitter.onCompletion(() -> emitterRegistry.remove(sessionId)); emitter.onTimeout(() -> emitterRegistry.remove(sessionId)); emitter.onError(e -> emitterRegistry.remove(sessionId)); // 先发送一个连接就绪事件 sendEvent(emitter, "CONNECTED", Map.of("sessionId", sessionId)); return emitter; } @PostMapping("/{sessionId}/start") public void start(@PathVariable String sessionId, @RequestBody UserRequest request) { String traceId = UUID.randomUUID().toString(); agentExecutor.submit(() -> agentRuntime.execute(sessionId, request, traceId)); } private void sendEvent(SseEmitter emitter, String eventType, Object data) { try { emitter.send(SseEmitter.event() .name(eventType) .data(data, MediaType.APPLICATION_JSON)); } catch (IOException e) { emitter.complete(); throw new EventPushException("SSE 连接已断开", e); } } }

这里有几个细节值得展开。第一，new SseEmitter(0L)表示容器层面不设置超时时间，因为 Agent 任务可能长达几分钟甚至更久。不设置超时之后，就必须在应用层维护心跳，否则中间的网络设备会认为连接空闲而回收链路。第二，sendEvent方法里一旦抛出IOException，说明客户端已经断开，必须立刻调用complete()并清理注册表，否则内存里会堆积大量无用连接。这个清理动作很多人会忽略，后面我会专门讲它引发的生产事故。

第三点，我用了虚拟线程执行 Agent 任务。Spring Boot 3.2 以后，Java 21 的虚拟线程已经可以很方便地接入。Agent 这种场景是典型的 IO 密集型任务：调用大模型接口要等待网络返回，执行工具要等待外部服务响应，大量时间都阻塞在 IO 上。虚拟线程让 Agent 任务以非常低的成本并发执行，不用担心线程池耗尽的问题。如果是传统的平台线程，几十个 Agent 任务就可能把线程资源全部吃光。

3.2 Agent 主循环与事件推送的结合

要把“流式思考与工具调用中枢”跑起来，关键是 Agent 运行时在执行主循环时，每一步都要发射对应的事件。这里给一段简化但不失核心逻辑的伪代码：

public void execute(String sessionId, UserRequest request, String traceId) { AgentContext ctx = new AgentContext(sessionId, request, traceId); SseEmitter emitter = emitterRegistry.get(sessionId); if (emitter == null) { throw new IllegalStateException("会话尚未建立 SSE 连接"); } sendEvent(emitter, "AGENT_START", Map.of("question", request.getQuestion())); while (!ctx.isFinished()) { // 1. 大模型生成思考片段，这里以流式方式返回 LLMResponse llmResponse = llm.chat(ctx.buildMessages()); for (String fragment : llmResponse.getThinkingFragments()) { sendEvent(emitter, "THINKING_FRAGMENT", Map.of("fragment", fragment)); } // 2. 判断下一步动作 AgentAction action = llm.planNextAction(ctx.toSnapshot()); if (action == null) { ctx.markFinished(); break; } // 3. 如果是工具调用，发射事件并执行 if (action.isToolCall()) { ToolCall call = action.getToolCall(); sendEvent(emitter, "TOOL_CALL", Map.of( "toolName", call.getToolName(), "arguments", call.getArguments() )); sendEvent(emitter, "TOOL_EXECUTING", Map.of("toolName", call.getToolName())); ToolResult result = toolExecutor.execute(call); ctx.appendMessage(result.toMessage()); sendEvent(emitter, "TOOL_RESULT", Map.of( "toolName", call.getToolName(), "summary", result.summary() )); } } // 4. 最终答案流式输出 for (String token : ctx.getAnswerTokens()) { sendEvent(emitter, "ANSWER_TOKEN", Map.of("token", token)); } sendEvent(emitter, "COMPLETE", Map.of("answer", ctx.getFinalAnswer())); }

这段伪代码的核心逻辑就是循环：思考、决策、调用工具、拿结果、再思考。每一轮循环都产出一批事件，这些事件既推给了前端，也同时写进了 Redis Stream。这样即使前端错过了某些片段，后续也还能补回来。

这里我要提醒一个关键点：sendEvent里其实不应该直接操作原始SseEmitter，而应该走一层事件总线。我在项目里是先把事件写入 Redis Stream，再由推送模块异步从 Stream 里消费并推送给连接。直接推的问题在于，如果推送瞬间网络抖动，前端没收到，后面就没有任何补救手段了；但走 Redis Stream 之后，事件天然持久化了一份，可以事后重放。

3.3 写事件进 Redis Stream 的代码姿势

事件总线的第一步是把事件写入 Redis Stream。这一步看似简单，实际有很多细节。这是我实际项目里的一段精简代码：

String eventJson = objectMapper.writeValueAsString(event); StreamOperations<String, Object, Object> streamOps = redisTemplate.opsForStream(); MapRecord<String, Object, Object> record = StreamRecords.newRecord() .ofObject(eventJson) .withStreamKey(AgentEventStream.STREAM_KEY) .withId(RecordId.autoGenerate()); streamOps.add(record);

所有事件都追加到同一个 Stream Key 里，而不是每个会话一个 Key。这是因为 Redis Stream 本身就是按 ID 有序存储的，会话 ID 作为事件字段，后续消费时按会话过滤即可。如果每个会话建一个 Key，会话数量一多，Key 的规模会非常可怕，运维也不好管理。

消费端使用消费组更合理。每个应用实例都加入同一个消费组，但这里要特别注意的是：不能让多个实例同时消费同一会话的事件并重复推送。我采用的方案是，消费组里每个消费者只处理自己本地注册表中存在的SseEmitter对应的事件。换句话说，事件被某个实例消费后，先检查emitterRegistry里有没有目标 sessionId，有就推给连接，没有就忽略。这样天然解决了多实例下的连接归属问题。

3.4 推送失败的处理与连接清理

事件推给前端时，如果连接已经断开，send方法会抛异常。这时候除了把 emitter 清理掉，还有一个重要动作：不要 ack 消费组的消息，让它留在 Pending 列表里。等客户端重新连上来，这部分消息还能重新投递出来，实现断线续传。先 ack 再从 Stream 里删除的话，事件就彻底丢了。

实际消费组消费的核心代码类似这样：

List<MapRecord<String, Object, Object>> records = streamOps.read( Consumer.from("agent-stream-group", instanceId), StreamReadOptions.empty().count(50).block(Duration.ofSeconds(2)), StreamOffset.create(STREAM_KEY, ReadOffset.lastConsumed()) ); for (MapRecord<String, Object, Object> record : records) { AgentEvent event = objectMapper.readValue((String) record.getValue().get("event"), AgentEvent.class); SseEmitter emitter = emitterRegistry.get(event.getSessionId()); if (emitter != null) { try { sendEvent(emitter, event.getType(), event.getData()); streamOps.ack(STREAM_KEY, "agent-stream-group", record.getId()); } catch (Exception e) { // 推送失败，不 ack，等待重试或连接恢复 log.warn("事件推送失败，recordId={}, sessionId={}", record.getId(), event.getSessionId()); } } else { // 没有本地连接，说明连接已经换到别的实例，直接 ack 避免重复消费 streamOps.ack(STREAM_KEY, "agent-stream-group", record.getId()); } }

注意这个处理逻辑的顺序很重要：先找到连接再推送，推送成功才 ack。如果握手期间连接还没注册到当前实例，宁可让事件留在 Pending 里，也不能让多实例重复推送，否则前端会看到消息乱序。

4. Redis 的高级玩法：会话恢复、幂等与横向扩展

4.1 会话状态与游标恢复

前面提到断线续传依赖事件序列号，那这个序列号存在哪里？我放在 Redis 的 Hash 里。每个会话一条 Hash，字段包括lastSeq、agentContext、traceId、expireAt。客户端重连时带上上次收到的lastSeq，服务端从 Hash 里取出会话上下文，再结合 Redis Stream 拉取后续事件做重放。

这个方案我在生产里实测过，非常稳。有一次某个实例半夜被强制重启，当时有几十个 Agent 会话正在运行。实例重启后，客户端 EventSource 自动重连到另一台机器，新实例从 Redis 里恢复会话上下文，继续推送之前没发完的事件。用户感知是页面闪了一下，进度条自动恢复，整个过程没有一条消息丢失。

具体恢复流程可以这样实现：

1. 客户端重连时携带lastSeq。
2. 服务端读取会话 Hash，拿到agentContext。
3. 从 Redis Stream 中查询该会话 ID 下大于lastSeq的所有事件。
4. 按序列号升序重放给客户端。
5. 重放完成后，继续消费新产生的事件。

这套逻辑听着不复杂，但能跑通的前提是事件序列号本身可靠。序列号的生成我直接在应用内用原子自增实现，一个会话只会被一个实例的 Agent 任务处理，所以不会出现并发冲突。

4.2 工具调用幂等：防止重复执行的惨案

Agent 调用工具这个环节，是整个系统里最容易出生产事故的地方。为什么？因为大模型的决策不是百分之百可靠的。模型在调用外部工具时，如果第一次网络超时，它可能会生成第二次调用；如果工具返回结果异常，模型可能会重新发起同样的调用。这时候如果工具是“查询天气”还好，重复执行无非浪费点流量；但如果工具是发短信、扣费、下单，重复执行的后果就严重了。

所以我的做法是给每一次工具调用生成一个全局唯一的requestId，在TOOL_CALL事件里带给前端的同时，也写入 Redis。工具执行器在真正干活之前，先拿requestId去抢分布式锁：

Boolean locked = redisTemplate.opsForValue() .setIfAbsent("lock:tool:" + requestId, "1", Duration.ofMinutes(5)); if (!Boolean.TRUE.equals(locked)) { // 说明这个 requestId 已经被执行过，直接返回缓存结果 return cachedToolResult(requestId); }

锁的过期时间要设置得足够长，因为一个工具调用可能本身就要执行好几秒，甚至要回调外部系统。我之前把过期时间设成 20 秒，结果一个外部接口在高峰期要 30 秒才返回，锁提前释放，同一请求又被重复执行了一次。这个教训之后，我把过期时间改成了 5 分钟，并且额外加了一个“执行完成后主动释放锁”的逻辑，双保险。

还要配套一个结果缓存。工具执行完成后，把结果以requestId为 Key 存到 Redis，TTL 设 24 小时。这样哪怕是事件重放、消费重试、模型重复决策，只要requestId相同，都能从缓存里直接拿结果，不会再执行一次工具。

4.3 为什么是 Redis Stream，而不是 Pub/Sub

很多人一听到“事件总线”就会想到 Redis 的发布订阅。这里我明确说一下我的取舍：Pub/Sub 有一个致命的缺点——消息不持久化。消费者不在线，消息就直接丢了。在 Agent 需要断线续传的场景里，这是不可接受的。

Redis Stream 则不同。它本质上是一个按 ID 有序的持久化日志结构，支持消费组、消息确认、Pending 列表、重新消费。这些能力组合起来，正好满足我们对可靠性的要求。

当然，Stream 也有自己的问题：数据会一直堆积。所以生产环境里我开启了 Stream 的 maxlen 裁剪，只保留每个会话最近 1000 条事件。这个数量对 Agent 会话来说完全够用，思考片段加工具调用事件一般也就几十到几百条，1000 这条线给了充足的缓冲空间，又不会让 Redis 内存无限膨胀。

XADD stream:agent:events MAXLEN ~ 10000 * field1 value1 field2 value2

MAXLEN ~后面的波浪号表示近似裁剪，牺牲一点精确性换取更高的写入性能。在大量事件写入的场景下，这个优化非常实用。

5. 生产环境踩坑实录：断流、乱序与连接泄漏

5.1 Nginx 缓冲导致的“伪流式”

第一个要吐槽的坑，就是反向代理的缓冲。SSE 刚上线时，我接到前端同事投诉，说“你们这个流式跟轮询没区别，还是一条条蹦出来，跟挤牙膏一样”。排查半天，发现罪魁祸首是 Nginx 默认开启了proxy_buffering，把后端返回的事件全部缓冲起来，攒够了才交给客户端。

修正之后的 Nginx 配置长这样：

location /agent/stream { proxy_pass http://backend_upstream; proxy_http_version 1.1; proxy_set_header Connection ""; proxy_buffering off; proxy_cache off; proxy_read_timeout 3600s; proxy_send_timeout 3600s; }

三个关键点：proxy_buffering off关闭缓冲，让事件实时透传；proxy_read_timeout设置 3600 秒，防止长时间没有数据时被 Nginx 掐断；proxy_set_header Connection ""清掉默认的 keep-alive 头，避免上游连接被错误处理。

还有一个容易被忽略的坑：代理层开启 gzip 压缩也会影响 SSE。虽然压缩能省流量，但会把流式数据攒在压缩缓冲区里，导致前端等很久才收到一段内容。我直接对 SSE 路径关闭了 gzip，因为事件流里大多数是短小的 JSON 片段，压缩收益不高，反而是实时性更重要。

5.2 浏览器连接数限制

另一个陷阱在客户端。浏览器对同一个域名下的 HTTP/1.1 连接数限制是 6 个，SSE 是长连接，一旦建立就会一直占据一个连接额度。如果页面同时打开多个 Agent 会话，连接数很快就会耗尽，后续的普通请求都会被阻塞。

我们的解法有两个方向。第一个方向是升级 HTTP/2，多路复用技术让所有请求共享同一条连接，SSE 不再占单独的连接名额。第二个方向是代码层面做约束：页面无论打开多少会话，始终只保留一个 SSE 连接，服务端在事件流里用sessionId区分不同会话，前端再做分发。第一种适合基础设施已经支持 HTTP/2 的场景，第二种兼容性更好。我们最终两个都做了，以第二种为兜底方案。

5.3 多实例下的乱序与重复消费

多实例部署之后，最棘手的问题是消息重复消费。早期版本里，每个实例都独立消费 Redis Stream 里所有会话的事件，导致同一个会话的事件被好几台机器的推送模块同时处理，前端收到的事件顺序完全错乱。

我最后的解法是“本地连接归属判定”，在前面代码里已经体现过：消费组虽然由所有实例共同组成，但每条事件处理时，只有持有目标会话SseEmitter的实例才会真正推送，其他实例直接 ack 丢弃。这个做法的前提是，每个会话的 SSE 连接只存在于一台实例上，所以不会出现两台实例同时推送同一个会话的情况。

如果将来连接数进一步扩大，还可以考虑“按会话 ID 路由到固定实例”的强一致方案，用一致性哈希把同一个会话的所有事件都分发给同一台实例。目前我们的规模还没到非上不可的地步，但设计上已经预留了这个演进方向。

5.4 SseEmitter 连接泄漏与内存增长

最后说一个最隐蔽的坑。SseEmitter 是一个与客户端连接强关联的对象，如果客户端关闭了连接，而后端没有感知到，注册表里的 emitter 对象就会一直留在内存里。初期我们内存监控图上能看到emitterRegistry.size()稳定增长，就是这里漏了。

排查思路是这样的：先确认回调有没有写全。onCompletion、onTimeout、onError三个回调都必须把对应 sessionId 从注册表移除。然后检查推送逻辑，只要send方法抛出IOException，就立即标记连接已失效。最后我加了一个兜底的定时任务，每 60 秒扫描一次注册表，找出超过 5 分钟没有活跃心跳的 emitter，强制清理。

心跳本身也值得一提。SseEmitter 不主动发数据的时候，中间的网络设备可能会认为连接空闲并回收。我的做法是每 30 秒发一条comment类型的心跳消息，既不会干扰前端的事件解析，又能保持链路活跃。

emitter.send(SseEmitter.event().comment("heartbeat"));

这行代码非常简单，但它直接解决了大量“莫名其妙断流”的隐患。我曾经排查过一个线上问题：用户反馈 Agent 跑到一半连接断开，查遍所有日志都正常，最后发现就是没有心跳，连接被云厂商的负载均衡器在 60 秒静默期后回收了。

5.5 事件丢失的排查清单

最后整理一个实战排查清单，遇到“前端少了一段事件”的问题时，按这个顺序查，效率最高：

这套清单在我们团队内部已经沉淀成标准排障手册，每次遇到流式相关问题，先走一遍再深入排查，能省很多时间。

最后再说说我的实际体会

这套系统从设计到上线，再到扛住大流量活动，中间经历了不少折腾，但回头看，当时最重要的一个决策就是把通信方式从轮询改成 SSE。很多人觉得 WebSocket 才是实时推送的标配，但 Agent 这个场景里，服务端才是绝对的主角，单向的 SSE 反而更契合，实现难度和维护成本都低了一截。

如果你也在做类似的事，我的建议是先别急着上重型消息中间件，SSE 加 Redis Stream 这套组合已经能解决绝大多数需求。把事件模型设计清楚，把断线续传和幂等控制做扎实，再把连接生命周期管理好，这套“中枢”就能稳定跑很久。希望文章里的代码和踩坑经验，能让你少走一点我们走过的弯路。