Spring Boot 3 + SSE + Redis：构建生产级Agent流式交互中枢

如果你还在用轮询的方式去拉 Agent 的思考进度，我劝你快停手。最近我在打磨一个基于 Spring Boot 3 的 Agent 流式交互中枢，把原本的短轮询全部换成了 SSE + Redis 的方案，终于不用再看着前端每隔几百毫秒发一次请求了。这篇就把完整的设计思路、核心代码和一路踩过的坑都掰开聊聊，特别是流式思考过程、工具调用状态同步，以及断线续传这些问题，都是生产环境里真正绕不开的关卡。

先说清楚这篇适合谁。如果你正在做 Agent 应用，或者想把大模型输出从“等结果一起返回”改成“边思考边输出”的体验，又或者你已经搭过 SSE 但总觉得不稳固、一上线就断流，那这篇很值得读完。我会尽量把方案选型的逻辑、生产化的细节、以及一些常规文档里不会写的坑也一起讲透。

1. 为什么放弃轮询：Agent 流式交互的现状与痛点

1.1 轮询方案为什么撑不起 Agent 场景

很多团队刚做 Agent 应用时，第一反应就是轮询。客户端每隔一两秒调一次接口，问“任务好了吗”，服务端查任务状态，返回“还在跑”。这个方案在小流量、短耗时任务里还能凑合，但放到 Agent 场景里几乎是一场灾难。

Agent 的一次完整执行往往要经历多轮大模型推理、工具选择、工具参数生成、工具结果回传、再推理。慢的时候一个任务能跑几十秒甚至几分钟。如果客户端每 1 秒轮询一次，几十秒下来就是几十个请求。这些请求大部分都是无效的，因为状态没变化，但服务端还是得花资源去查 Redis、查数据库，一路把空结果返回回去。到了并发量稍微高一点，系统响应变慢，又反过来拉长任务执行时间，形成恶性循环。

更难受的是，轮询没法去体现“流式思考”的过程。Agent 中间会输出推理片段、会中途调用工具、会修正计划，这些过程如果只是在任务结束时一次性返回，用户根本不知道当前卡在哪一步。体验上就是一个黑盒，用户只能干等。哪怕你给轮询接口加上了 step、progress 这种字段，也只是把状态粒度强行拉高，本质上还是客户端主动去“猜测”状态变化的时机，不仅实现别扭，前端代码也会变得非常难维护。

1.2 SSE 与 WebSocket 的取舍，以及 Redis 的角色

既然要推流，常见的选择无非两个：SSE 和 WebSocket。我最终选的是 SSE，也就是 Server-Sent Events。核心原因很简单：Agent 场景下，大部分数据流向都是单向的，就是服务端主动把事件推给客户端。大模型在逐步吐 token，工具调用状态在逐步变化，客户端在这个过程中几乎不需要向服务端发消息，最多就是被动接收。

SSE 是建立在 HTTP 之上的轻量协议，浏览器端原生支持EventSource，会自动做重连，服务端实现也非常简单。相比之下，WebSocket 虽然支持双向通信，但握手、协议、反向代理、防火墙穿透都要多一重配置和调试成本。如果只是为了单向推送，WebSocket 有点杀鸡用牛刀。再加上 SSE 本身有Last-Event-ID这种断线续传机制，配合 Redis 缓存事件，可以让客户端重连后立即补拉缺失的消息，这正好命中 Agent 场景里“思考过程不能丢”的诉求。

那 Redis 在里面的角色又是什么？简单说，SSE 只解决了“连接怎么推”的问题，但生产环境里 Agent 服务通常不止一个实例。用户连到了 A 实例，任务却在 B 实例上跑，那 B 实例生成的事件怎么推到用户的长连接上？这就需要一个跨实例的消息通道。Redis Pub/Sub 在这里扮演了轻量级消息总线的角色，所有携带着任务进度、思考片段、工具调用事件的业务消息都发布到同一个频道上，各节点订阅后，识别出哪些事件属于自己节点上的会话，再通过对应 SSE 连接推给客户端。再加上 Redis 本身还能存任务状态、会话映射、消息幂等标识，一套组合下来，既不引入额外中间件，又能解决多实例推送的基本问题。

2. 生产级架构设计：从单机长连到多实例消息中枢

2.1 整体数据流：一次请求如何变成一段流式响应

在设计这套中枢之前，我先明确了一个原则：任何 Agent 执行过程都不能和某台机器的本地变量绑定。任务状态、事件消息、会话映射全部放在 Redis 里，SSE 连接只是“可移动的出口”。这样任何一个实例挂了，任务还可以被其他实例接管，客户端重连后也能拿到完整上下文。

整体数据流大致是这样的。客户端发起一次 Agent 调用，通过 REST 接口或网关进入 Web 层；Web 层立即返回一个sessionId，同时建立 SSE 连接，连接请求里带上这个sessionId。请求参数被封装成任务消息，写入 Redis 队列或直接发布到 Redis Channel 中。某个空闲的 Agent worker 实例消费到任务消息，开始执行大模型推理和工具调用，每产生一段思考文本、一个工具调用请求、一个工具执行结果，都会组装成统一的事件结构，发布到 Redis 的一个全局频道中。同时，所有 Spring Boot 实例都订阅了这个频道，收到事件后判断sessionId是否落在自己节点的连接表里，是则通过 SseEmitter 推给客户端。

这个模型的关键在于：Agent worker 不需要知道客户端长连接在哪，Web 层也不需要关心任务执行细节。两者之间靠 Redis 解耦，靠事件协议对齐。这样还带来一个额外好处：Web 层可以独立扩展，Agent worker 也可以独立扩展，没有单点瓶颈。实测下来，单是去掉轮询，整体请求量就下降了 80% 以上，系统压力小了很多。

2.2 Redis 在流式中枢里的三类核心职责

聊清楚 Redis 的作用，我习惯把它拆成三层：消息通道、状态存储、会话路由。

第一层是消息通道。我主要用的是 Redis Pub/Sub，全局频道agent:events，发布各类事件。为什么选 Pub/Sub 而不是 Streams 或者普通队列？因为它足够轻，语义上就是“广播给所有订阅者”，非常适合多实例同时监听并做本连接路由。但要注意，Redis Pub/Sub 的消息不持久化，如果订阅端短暂断连，消息会丢。这个问题后面第 5 章会再展开，如果对可靠性要求很高，可以把事件同时写入 Streams，SSE 只是加了一个快速分发通道。

第二层是状态存储。我用Hash结构存每个任务的状态，比如status: running / tool_calling / done，以及record: 已经发送过的事件偏移量或事件 ID 列表。客户端断线重连后，Web 层从 Redis 里查到这个任务的当前状态，再决定是从头补发还是只补最后几条。

第三层是会话路由。我用Hash结构存sessionId到nodeId的映射，节点启动时注册自己，节点下线时清理过期映射。当一个实例收到 Redis 广播的事件后，只需要查本地连接表，不需要每次都查 Redis。但如果事件来自另一个实例，本机查不到连接，就直接忽略，不产生额外开销。这个小设计在高并发下非常关键，避免了每个实例都拿着全局关系表到处找连接。

2.3 Spring Boot 3 异步模型选择：SseEmitter 还是响应式流

Spring Boot 3 中实现 SSE 主流有两种姿势：一种是 Spring MVC 的SseEmitter，另一种是 Spring WebFlux 的Flux<ServerSentEvent>。如果你是新项目，且愿意全部切到 WebFlux，响应式流在数据拉取和背压上有天然优势，但代价是整套技术栈都要转向 Reactor，数据库访问、Redis 客户端也要跟着换，改造成本不小。

我的方案是继续用 Spring MVC +SseEmitter，因为原来的业务代码大量依赖同步的接口和工具类，强行换 WebFlux 会让复杂度爆炸。SseEmitter本质上就是 Servlet 异步线程模型下提供的 SSE 封装，使用起来非常直接：接口收到请求后立即返回，线程从 Tomcat 线程池中释放，连接由异步线程和底层 Netty/容器配合维护。随后在另一个线程里往 SseEmitter 写事件即可。

不过这里有一个极其重要的变化：Spring Boot 3.2 之后对虚拟线程支持更好了，默认的 Tomcat 线程池可以被替换成虚拟线程。起初我以为把所有普通接口改成虚拟线程就行了，后来才发现 SseEmitter 的异步写和虚拟线程池之间如果配合不当，很容易把连接池任务队列撑爆。这个坑在第 5 章会详细说。

3. 核心模块实现：SSE 控制器、Redis 订阅与协议设计

3.1 用 SseEmitter 搭一个可落地的 SSE 端点

先看最基础的控制层实现。定义一个连接端点，客户端传入sessionId，服务端为这个会话创建一个SseEmitter，设置超时时间，然后存入本地连接表。之后所有针对该会话的事件都会通过这个 SseEmitter 推送到客户端。

@RestController @RequestMapping("/api/agent") public class AgentSseController { private final ConcurrentHashMap<String, SseEmitter> liveEmitters = new ConcurrentHashMap<>(); @GetMapping(value = "/stream/{sessionId}", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter stream(@PathVariable String sessionId) { SseEmitter emitter = new SseEmitter(0L); // 0L 表示不超时，需根据场景调整 liveEmitters.put(sessionId, emitter); emitter.onCompletion(() -> liveEmitters.remove(sessionId)); emitter.onTimeout(() -> liveEmitters.remove(sessionId)); emitter.onError(e -> liveEmitters.remove(sessionId)); return emitter; } }

这里注意几点。第一，SseEmitter本身是 Spring MVC 提供的异步返回对象，控制器返回它之后，Tomcat 不会立刻关闭连接，而是脱离当前请求线程，进入异步模式。第二，new SseEmitter(0L)意味着永不超时，但生产环境最好别直接用 0，而是设置一个合理的超时时间，比如半小时，避免客户端一直挂住不释放。第三，liveEmitters是本地缓存，只代表当前节点上的连接，节点重启后必然丢失，所以会话级恢复逻辑必须依赖 Redis，而不是这里。

3.2 把 Redis Pub/Sub 事件桥接到 SSE 推送通道

连接建立好了，接下来的关键是把 Redis 广播事件桥接给对应的 SseEmitter。我的实现是：定义 Redis 消息监听容器，监听agent:events频道，收到消息后解析事件体，根据事件里的sessionId判断当前节点有没有对应的 SseEmitter，有就发往客户端。

@Component public class RedisEventBridge implements MessageListener { private final ObjectMapper objectMapper; private final AgentSseController sseController; @Override public void onMessage(Message message, byte[] pattern) { String payload = new String(message.getBody(), StandardCharsets.UTF_8); AgentEvent event = objectMapper.readValue(payload, AgentEvent.class); sseController.push(event.sessionId(), event); } }

而在AgentSseController里需要补一个push方法，负责把事件序列化成ServerSentEvent格式写出去。这里有个小技巧：每次写入前先发送一条注释行作为心跳，保持连接活跃，同时避免代理节点因为“长时间没有数据而超时断开”。具体会在第 4 章讲。

public void push(String sessionId, AgentEvent event) { SseEmitter emitter = liveEmitters.get(sessionId); if (emitter == null) return; try { ServerSentEvent<AgentEvent> sseEvent = ServerSentEvent.builder(event) .id(String.valueOf(event.eventId())) .event(event.type()) .build(); emitter.send(sseEvent); } catch (Exception e) { liveEmitters.remove(sessionId); } }

在配置 Redis 监听容器时，要注意线程池配置。默认的 RedisMessageListenerContainer 在事件量大的时候，如果只用一个 listener 线程，很容易成为瓶颈。我建议在容器初始化时设置多个执行线程，同时给不同的 Agent 事件类型拆到不同频道，避免单一频道的消费压力。

3.3 流式思考与工具调用的事件协议设计

要想让前端流畅展示 Agent 的思考过程，事件协议必须提前设计好，不能事到临头再拼。我的协议里事件类型一共有 5 种：thinking、tool_call、tool_result、final_answer、error。每种事件结构上都包含sessionId、eventId、type、data、timestamp。eventId用来做幂等和断线续传，data是业务载荷。

{ "sessionId": "xxxx-xxxx", "eventId": 12, "type": "tool_call", "data": { "toolName": "web_search", "arguments": "{\"keyword\":\"spring boot sse\"}", "callId": "call_01" }, "timestamp": 1712345678901 }

事件粒度也很重要。thinking事件可以按 token 片段推，也可以按完整句子推。推得太细，Redis 和网络开销成倍增长；推得太粗，前端动画会卡。我的经验是，大模型每生成 30-50 个字推送一次完整片段，同时附带一个delta标记，由前端自己决定是替换还是追加。工具调用事件则必须一次推完整结构，包含工具名、参数和调用 ID，方便前端展示工具运行状态。

这里尤其要设计好final_answer的data。它既要包含最终答案文本，也要包含这次 Agent 执行过程中所有工具调用的摘要列表。这样客户端即使在重连后没有收到历史事件，也能通过final_answer感知整体执行结果。

4. 生产级优化：断线续传、心跳保活与多实例水平扩展

4.1 会话状态映射：客户端重连后如何续上

断线续传是生产级 SSE 方案里最容易被忽略的一块。客户端网络抖动、代理重连、页面刷新，任何一个场景都会导致原有 SSE 连接断开。如果重连后直接新建一条连接，就会丢失中间事件，用户发现“思考过程突然断了几步”，体验非常割裂。

解决思路是给每个会话维护一个“已推送事件 ID”的水位线。Redis 里存一份sessionId -> 已推送事件ID列表，客户端重连时在请求头或查询参数里带上Last-Event-ID，Web 层拿到这个 ID 后，从 Redis 里把应用户缺失的事件全部补推一次。

实际落地时，我会把事件同时写入 Redis 的一个 Stream 结构，以sessionId作为 key 维度，每条事件都有自己的 ID，用于增量读取。这样即使事件已经通过 Pub/Sub 广播过了，也不会在断线续传时丢失。补推的优先级要控制好：先补状态类事件，再补增量文本事件，最后补终态事件，避免前端因为乱序渲染出错。测试下来，这个续传机制对用户体感改善非常明显，至少不会再出现白屏或半个回答卡在页面上的情况。

4.2 心跳与代理缓冲：让长连接不再无端断开

SSE 长连接在生产环境第一个遇到的敌人，往往是网络链路中的各种代理层。Nginx、网关、防火墙都有空闲超时，如果一段时间内没有任何数据帧通过，底层就会把连接判定为僵尸连接并断开。要避免这个问题，只能在应用层定期发送心跳。

最简单的心跳方式是定时往 SseEmitter 发送一个注释行，格式是: ping\n\n。这种注释事件浏览器和 EventSource 会忽略，但能维持连接活跃。我用一个全局的ScheduledExecutorService每隔 15 秒遍历所有活跃连接发送心跳。但要注意，不能只发心跳不发业务事件，否则虽然连接没断，用户也可能因为长时间没有真实内容而以为页面卡死。所以心跳间隔最好比代理超时时间短一半以上。

另一个非常常见的坑是 Nginx 默认开启proxy_buffering，导致 SSE 数据被缓冲到一定大小才返回。表现就是前端事件不是实时展示，而是一段一段地蹦出来。解决办法是在响应头里加上X-Accel-Buffering: no，同时关闭代理缓冲。Spring 里可以直接写过滤器统一设置这个头，下面这个配置可以参考。

@Component public class SseHeaderFilter implements Filter { @Override public void doFilter(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException { if (request instanceof HttpServletRequest httpRequest && httpRequest.getRequestURI().startsWith("/api/agent/stream")) { HttpServletResponse httpResponse = (HttpServletResponse) response; httpResponse.setHeader("X-Accel-Buffering", "no"); httpResponse.setHeader("Cache-Control", "no-cache"); httpResponse.setHeader("Connection", "keep-alive"); } chain.doFilter(request, response); } }

4.3 断流排查速查：常见根因和解决顺序

做了这套中枢之后，我总结了一张断流排查表。每次线上报“SSE 断流”，我会按这个顺序查，基本能快速定位到根因。

这张表里每一条都是我真实踩过的。尤其是第一条，如果 Nginx 和 Spring Boot 两端的超时配置不一致，SSE 很容易在刚建立连接没多久就被掐断。所以生产部署前，先统一网关层与应用层的心跳时机，能少踩一大半坑。

5. 常见问题与避坑实录

5.1 虚拟线程和 SseEmitter 一起用，我差点踩坏连接池

Spring Boot 3.2 开始支持虚拟线程，很多朋友直接spring.threads.virtual.enabled=true把 Tomcat 线程换掉，感觉很爽。但虚拟线程不是万能的，特别是它和 SseEmitter 的写操作配合不当，容易导致瞬间创建大量虚拟线程，把底层线程工厂都打满。

我的教训是：不要把所有接口都无脑放在虚拟线程上。SseEmitter 一旦建立，请求线程已经释放，后续的推送发生在业务处理线程中，这里的线程模型和 Tomcat 的请求线程无关。对于真正的 Agent 执行任务，我用的是一个普通的ThreadPoolTaskExecutor，核心线程数和最大线程数都设了上限，避免突发大量任务时资源失控。虚拟线程在我的场景里主要用来跑那些本来会阻塞 IO 的普通同步接口，比如查询历史记录、获取工具列表。

如果你非要在 SseEmitter 的 push 路径上使用虚拟线程，一定要限制并发的推送任务数量，否则一个上游慢接口就能瞬间创建几千个虚拟线程，内存和调度开销都会爆。推荐做法是维护一个信号量，控制在途推送任务数。

5.2 Redis Pub/Sub 的丢消息问题，替代方案是什么

Redis Pub/Sub 的语义是 fire-and-forget，发布时不关心有没有订阅者在听，订阅者掉线期间的消息直接丢失。这在大多数 Agent 场景下其实是可以接受的，因为丢的只是实时增量，重连后可以通过持久化事件补发。但如果你的业务要求“每个工具调用结果都不能丢”，那 Pub/Sub 就不够用了。

我最终的方案是双写：业务事件先写入 Redis Stream，然后同时在 Pub/Sub 频道发一条广播，让在线连接尽快收到。如果某个节点掉线，它重新启动后可以通过 Redis Stream 从上次的消费偏移量补拉事件，再推给已经重连上的客户端。这样做的好处是实时性和可靠性都有了保障，缺点是实现复杂度增加不少。对于刚开始做 SSE 的方案，我建议先只用 Pub/Sub，把完整流程跑通，再按需升级为 Streams。

这里再提一句，不要把 Redis Streams 和 Kafka 混为一谈。Redis Streams 更适合中小规模的消息持久化需求，操作简单，不引入额外中间件，但海量吞吐和复杂分区还是得交给真正的消息队列。

5.3 客户端重连时的消息幂等控制

SSE 自带自动重连，浏览器 EventSource 在连接断开后会自动重新发起请求，并带上Last-Event-ID头。但很多自研前端工具不会自动带这个头，导致补推逻辑失效。所以服务端需要兼容两种重连方式：一种是通过Last-Event-ID头，另一种是通过 URL 参数里的lastEventId。

防重复也一样重要。客户端收到事件后，需要根据eventId做去重。前端的事件处理函数里，可以维护一个已处理 ID 的 Set，超过一定数量后只保留最近 N 条。服务端推送时也要注意：如果一个事件既被实时推送过，又因为重连被补推了一次，前端拿到两个相同eventId的事件时不能二次渲染工具调用卡片。我在协议里给每个事件加了类型和顺序标识，前端按eventId排序后只处理递增事件，能有效避免重复和乱序。

6. 从这套方案里学到的：坐标定位与扩展方向

6.1 稳定运行后的个人体会

这套 SSE + Redis 的中枢结构，我先是用了几个月，把验证服务和核心闭环都跑通，然后逐步完善成生产级方案。最深的体会是：SSE 选型本身没问题，但真正决定体验的是细节，比如心跳、续传、幂等、代理配置。任何一个环节没做扎实，用户都会用“又断流了”“又没动静了”来反馈。

在实际部署中，我建议你先把单个链路跑稳，再考虑水平扩展。不要让多实例路由、Redis Streams 这些高级特性一开始就叠加进来。先用单实例 + Pub/Sub 把整个流程打通，观察运行 2 周后有没有断流和消息丢失问题，之后再逐步加节点、补全持久化。很多团队一上来就追求完美架构，结果排查问题时被多实例和消息补偿搞得焦头烂额。

6.2 下一步可以怎么演进

如果后续想继续往生产级方向走，我比较推荐两个扩展点。第一，把 Redis Pub/Sub 升级为 Redis Streams + 消费组，这样多个 Agent worker 可以共同消费同一批任务消息，而不是简单广播，任务调度层面的可靠性会大幅提升。第二，SSE 毕竟只解决了服务端到客户端单向推送，如果未来产品需要客户端取消、人工介入、动态调整工具调用参数，就要引入一条反向通道，这时再考虑 WebSocket 或者建立一个单向的反向 HTTP 接口也不迟。

对我来说，这套方案带来的最大价值，是彻底把 Agent 的黑盒执行过程变成了可观察、可追溯的透明流程。只要把事件协议持续维护好，后面无论是接监控、做审计，还是给用户回放历史执行过程，都有现成的数据基础。个人感觉，这比单纯去追新框架新特性要实用得多。