上个月我把一个 Agent 应用从“用户问一句、等三秒、整段答案一次性吐出来”改成了“思考过程和工具调用全程流式直播”。改造完当天产品同事一句话让我愣了半分钟“刷新页面之后刚才已经显示过的思考过程还能恢复吗”不能。因为最初的实现里流式数据只存在每个 SseEmitter 背后——一个注定随着连接断开就消失的内存临时对象。这就是这篇博客的由来。在 Spring Boot 3 里用 SSEServer-Sent Events承载 Agent 的流式思考与工具调用输出再引入 Redis 作为跨实例、跨连接的状态兜底与断线补偿层让流式体验真正达到生产级。文章不会只贴能跑的 Demo而是把心跳、超时、补发、多实例同步、工具调用事件编排这些藏在“生产级”三个字后面的细节全部摊开讲。1. 流式输出为什么非 SSE 不可轮询、长轮询、WebSocket 的真实账本1.1 先算一笔轮询的糊涂账很多团队第一次做 Agent 对话流式化时第一反应是“那就让前端隔一秒轮询一次接口把生成过程中的 token 增量拉回来”。这个方案写起来最顺手但它有一个根本矛盾延迟和压力不可兼得。假设大模型生成速度是每秒 30 个 token前端每 1 秒轮询一次用户看到的文字就是 1 秒一跳的幻灯片中间没有任何平滑感。如果想把它压到 200 毫秒一跳就要把轮询频率放到 5 次每秒等于每个会话每秒打 5 个 HTTP 请求。Agent 场景还有一个特别致命的问题工具调用阶段的耗时不可预测。搜索接口可能 200 毫秒返回也可能 3 秒超时外部 HTTP 调用可能卡 5 秒。轮询接口如果按固定频率打要么在等待期间疯狂空转要么设置很长的轮询间隔让用户看着“正在思考”发呆。长轮询Long Polling能稍微缓解延迟问题但代价是服务端每个请求都被挂住一个线程而且请求超时后客户端要立刻重新发起连接实现一个可靠的断线续传逻辑相当繁琐。核心问题始终没解决这是“客户端主动拉”的模型而大模型输出是“服务端被动推”的自然形态。1.2 SSE 的取舍单向推送才是 Agent 思考流的形状SSE 全称 Server-Sent Events是一个基于 HTTP 长连接的单向推送协议。它的数据格式极其简单Content-Type 是text/event-stream每行data:前缀加内容空行分隔一条事件。正是这种简单让它成为 Agent 流式输出的天然载体。你可以用 curl 直接调试一个 SSE 端点看到的就是纯文本流的原始事件这对于排查问题来说简直是救命级的便利curl -N https://api.example.com/agent/session-123/stream event: thinking_delta id: 10 data: {d:我需要先查一下订单状态} event: tool_call_start id: 20 data: {tool:queryOrderStatus,args:{orderId:A10001}}对比 WebSocket它是全双工协议能做双向通信但为此付出了高昂的复杂度——帧协议、掩码、二进制分片、连接握手、心跳机制全都要自己处理。而 Agent 流式场景里浏览器到服务器的主要消息只有一个“取消生成”或者“发送下一轮输入”这些完全可以通过普通 POST 接口解决。非要让一条 SSE 连接承担双向通信属于杀鸡用牛刀。我见过的很多生产事故也是这么来的有人图省事把流式输出全部换成 WebSocket结果连接状态管理、断线重连、协议升级失败这些问题一个不少最后还不如老老实实回到 HTTP 语义。决策标准很简单只要主体交互是从服务端单向流向浏览器SSE 就是成本最低、心智负担最小的方案。1.3 但 SSE 有两个天生的生产级短板SSE 看起来很美好可真把它放到生产环境立刻会暴露两个问题。第一个短板是连接生命周期不可控。浏览器和服务器之间隔着一堆反向代理、负载均衡、网关这些中间层普遍有连接空闲超时。一个 Agent 会话里模型思考可能持续 30 秒甚至更久这段时间内没有任何字节从服务端发出代理层很可能直接把连接掐掉。浏览器 EventSource 会自动重连但重连之后会话上下文已经丢了——这就是“断流”现象最典型的来源。第二个短板是连接本身不携带业务状态。SseEmitter 对象只存在于创建它的那个 JVM 实例里进程重启、扩容、负载均衡切换实例都会导致连接背后的状态全部清零。而 Agent 的思考过程、工具调用记录这些真正有价值的数据恰恰是跨连接、跨实例需要保留的。所以结论很清楚SSE 负责“管道”但管道本身不靠谱必须有一层东西负责“记忆”。这就是我把 Redis 拉进来的根本原因。2. 中枢的整体形态Agent 执行引擎、SSE 网关与 Redis 的三角色划分2.1 链路总览与角色划分整个中枢系统有四个参与方浏览器前端、SSE 网关Spring Boot 应用、Agent 执行引擎、Redis。完整链路是这样的前端通过EventSource建立到/agent/{sessionId}/stream的 SSE 连接这条连接由 Spring Boot 里的 SseEmitter 承载。Agent 执行引擎在后台跑主循环可能是单独一个线程池里的任务也可能是虚拟线程。它负责调用大模型 SDK、解析流式 token、调用各类工具。引擎每产生一个事件思考增量、工具调用开始、工具调用结束、最终回答增量不是直接塞给 SseEmitter而是先写成一条带序号的事件写入 Redis。Redis 中的事件再通过订阅/消费机制被推送回本实例对应的那个 SseEmitter最终由 Emitter 送给浏览器。从职责上拆Redis 在这个架构里干了三件完全不同的活临时事件日志存储、跨实例事件中转、断线补发的数据源。这三件事绑定在一起本质上都是因为“Agent 执行引擎”和“浏览器连接的 SSE 网关”在物理上不一定是同一个进程。2.2 为什么必须引入 Redis 而不是进程内内存缓存有朋友可能会问我单机部署Agent 引擎和 SSE 网关就在同一个 Spring Boot 进程里直接用ConcurrentHashMapString, SseEmitter存一下不就行了吗为什么还要费劲引入 Redis单机小规模确实可以但你要接受三个现实连接粘滞问题。一旦部署两台实例做负载均衡浏览器先连上实例 A下一次请求被分发到实例 BB 上根本没有这个 session 的 SseEmitter。你只能被迫开启负载均衡的会话粘滞sticky session而它会让流量分配严重不均。热部署与重启丢失。平滑发布时旧实例的连接被强制断开SseEmitter 对象直接从堆内存里蒸发前端自动重连后拿不到任何历史状态用户看到的是一块空白。无法支撑 Agent 引擎独立演进。如果后面想把 Agent 执行引擎拆成独立服务、按压力单独扩容进程内缓存方案就彻底走不通了。Redis 在中间增加了一次写操作换来的是状态和连接解耦。执行引擎不用关心浏览器连在哪台机器上只要把事件写进 RedisSSE 网关也不用关心 Agent 引擎跑在哪里只要从 Redis 拿到事件往本地连接上推。2.3 事件协议设计用事件类型区分思考、工具调用与终稿生产级系统里SSE 不只是“往浏览器推几行字”而是要把 Agent 的整个执行过程以结构化事件的方式直播出去。我强烈建议从一开始就把事件协议定清楚而不是把文本直接塞进data:里裸奔。一个够用的事件协议至少包含这些字段字段类型说明eventIdlong全局递增序号用于断线补发typestring事件类型如 thinking_delta / tool_call_start / tool_call_delta / tool_call_end / answer_delta / done / errorsessionIdstring会话标识dataobject具体载荷按类型不同结构不同tslong毫秒时间戳SSE 协议本身支持事件类型所以最终发到浏览器的数据块长这样event: tool_call_start id: 20 data: {tool:queryOrderStatus,args:{orderId:A10001},sessionId:s-123}前端只需要按event类型注册不同的事件监听器各自更新对应的 UI 区域思考过程面板、工具调用卡片、正文答案区。这套协议一旦定下来后端引擎和前端渲染就可以并行开发互不阻塞。3. SseEmitter 实战心跳保活、超时策略与发射失败的最后一公里3.1 用 SseEmitter 还是响应式 Flux这是个权衡Spring Boot 3 里做 SSE 有两条路Spring MVC 自带的 SseEmitter以及 WebFlux 里的FluxServerSentEvent。我给的答案是Servlet 栈项目老老实实用 SseEmitter。SseEmitter 底层是 Servlet 3.1 的异步响应机制。Controller 方法返回 SseEmitter 之后Tomcat 线程池就释放回线程池里真正的推送操作发生在业务线程或者虚拟线程中不会长期占用 Servlet 线程。这对 Spring Boot 3 的默认 MVC 架构最友好不用为了一个流式接口把整个项目切到响应式编程模型。Flux 方案确实代码风格更函数式适合全链路响应式架构。但如果你的项目里其他接口都是传统阻塞式 JDBC、RestTemplate那为了 SSE 引入 WebFlux会让同一个服务里同时存在两套编程模型排查问题的时候会非常痛苦。3.2 心跳保活与超时策略的完整配置SseEmitter 有一点非常容易踩坑它有一个构造函数参数叫 timeout单位毫秒默认值是 30 秒。意思是说如果 30 秒内没有通过这个 Emitter 发送任何数据连接就会自动超时关闭。在大模型思考期间30 秒没输出是很常见的事所以裸写new SseEmitter()基本等于天然断流。这里有两条路可以走。第一条是在创建时设置一个较大的总超时时间比如 180 秒同时配合心跳防止代理层空闲回收RestController public class AgentStreamController { private final SseSessionRegistry registry; GetMapping(/agent/{sessionId}/stream) public SseEmitter stream(PathVariable String sessionId, RequestHeader(value Last-Event-ID, required false) String lastEventId) { SseEmitter emitter new SseEmitter(180_000L); registry.register(sessionId, emitter); // 连接关闭时清理注册表 emitter.onCompletion(() - registry.remove(sessionId, emitter)); emitter.onTimeout(() - registry.remove(sessionId, emitter)); emitter.onError(e - registry.remove(sessionId, emitter)); return emitter; } }第二条是全局心跳任务。每 15 秒向所有存活的 Emitter 发送一个 SSE 注释行注释行不会触发浏览器的任何事件但能维持 TCP 连接上的字节流动让代理层认为连接是活跃的Component public class SseHeartbeatTask { private final SseSessionRegistry registry; public SseHeartbeatTask(SseSessionRegistry registry) { ScheduledExecutorService scheduler Executors.newSingleThreadScheduledExecutor(); scheduler.scheduleAtFixedRate(this::beat, 15, 15, TimeUnit.SECONDS); } private void beat() { registry.allEmitters().forEach(emitter - { try { emitter.send(SseEmitter.event().comment(heartbeat)); } catch (IOException ex) { // 发送失败说明连接已经断了等待 onError 回调清理即可 registry.markStale(emitter); } }); } }心跳间隔不建议低于 15 秒。太频繁反而会在慢客户端上制造不必要的 CPU 和网络开销而大多数反向代理的空闲超时默认是 60 秒以上15 秒一次完全够用。3.3 发送失败的处理sse 断流的最后一公里生产环境里你会遇到大量 send 抛IOException的场景用户关了浏览器、手机切换网络、代理层主动回收连接。SseEmitter 的 send 方法一旦抛出异常再继续使用这个连接就已经不可恢复。处理原则只有一个及时清理别让僵尸 Emitter 在注册表里越积越多。上面代码里的markStale就是做这事。我见过最典型的故障就是团队忘了在 onError 回调里移出注册表结果连接池被没断开的僵尸连接占满新会话全部连不上。还有第二个细节同一个 SseEmitter 不能被多个线程并发 send。表面上它内部有同步但两个线程同时发送很容易出现事件顺序交错或者线程阻塞。我实践下来的做法是为每个 session 维护一个单独的单线程 Executor所有推送任务按序提交确保事件顺序严格一致ExecutorService executor Executors.newSingleThreadExecutor(); executor.submit(() - { try { emitter.send(event); } catch (IOException e) { registry.remove(sessionId, emitter); } });4. Redis 兜底层事件日志、断线补发与多实例同步的实现细节4.1 按 sessionId 组织的环形事件日志Redis 这块我试过两种存储结构一个是 List INCR 自行管理递增序号另一个是 Redis Stream 直接复用自带的 ID。如果你对 Redis 还不熟悉用 List INCR 更直观写事件时执行RPUSH agent:stream:{sessionId}:events 20|tool_call_start|{json}。维护序号INCR agent:stream:{sessionId}:seq得到的就是全局递增事件 ID。需要补发时LRANGE agent:stream:{sessionId}:events 0 -1全量取回或者用LINDEX定位到指定序号对应的偏移量。如果团队对 Redis Stream 够熟我推荐直接用 Stream。它自动生成的时间戳 ID 本身就是近似单调递增的天然适合当 SSE 的 id 用XADD agent:stream:s-123 * type thinking_delta data {d:我需要先查一下订单状态}每次写入事件时同时EXPIRE agent:stream:s-123 900刷新 TTL整体保留 15 分钟覆盖绝大多数模型思考加工具调用时间。事件日志不用做永久存储生产经验告诉我保留最近 200 到 500 条足够撑过断线重连的场景。4.2 断线重连与 Last-Event-ID 补发机制SSE 协议自带一个很优秀的能力浏览器断线重连时会自动带上Last-Event-ID请求头里面是它最后收到的那条事件 id。服务端拿到这个 id从 Redis 里把之后的所有事件补发回去就实现了无缝续传。关键实现点在于补发逻辑。不能用 String 简单比较因为顺序由序号定义public ListSseEvent replaySince(String sessionId, long lastEventId) { ListString events redisTemplate.opsForList() .range(agent:stream: sessionId :events, 0, -1); return events.stream() .map(this::deserializeEvent) .filter(e - e.getEventId() lastEventId) .collect(Collectors.toList()); }在补发之后再把后续实时事件继续推给同一个 Emitter 就行。这里最容易踩的坑是前端页面刷新不等于 EventSource 自动重连。EventSource 的自动重连浏览器内存里还有 Last-Event-ID而用户 F5 刷新页面后整个 JS 上下文没了Last-Event-ID 也跟着丢了。所以页面重新加载时前端应该从sessionStorage里手动把上一个会话的 lastEventId 拿出来通过查询参数传给服务端const lastId sessionStorage.getItem(last-event-id-${sessionId}) || ; const es new EventSource(/agent/${sessionId}/stream?lastEventId${lastId});服务端把查询参数的 lastEventId 和请求头的 Last-Event-ID 统一处理一视同仁。4.3 多实例部署下的跨节点转发多实例时核心矛盾的根源是这个Agent 引擎在某台实例上跑但浏览器的 SSE 连接可能挂在另一台实例上。要跨实例传递事件Redis Stream 就要当消息总线用了。我的做法是每个实例启动一个消费者线程执行XREAD BLOCK 5000 STREAMS agent:stream:s-123 $只关心本实例注册了 SseEmitter 的那些 sessionId。拿到事件后查本地注册表如果有对应连接就直接推送没有就丢弃因为一定有其他实例在消费同一批事件。如果担心多个实例重复消费可以把 SessionId 和实例 ID 的映射也存进 Redis Hash写入事件前先查一下这个 sessionId 落在哪个实例只有归属的那个实例才需要消费。这个映射表在连接建立时写入连接关闭时删除。4.4 TTL 与内存控制生产环境最容易犯的错误是事件日志只写不删。一个会话如果持续了 10 分钟产生了 300 条事件假设每条 200 字节才 60 KB。看起来不多但如果压测环境下同时有 1000 个活动会话Redis 内存就会多出几十 MB。真正的麻烦来自异常会话——前端异常退出但服务端没收到关闭信号注册表和日志就会残留。我的经验是给所有 session 维度相关的 Key 统一加 TTL写入时刷新。用 Redis 的EXPIRE命令EXPIRE agent:stream:s-123 900 EXPIRE agent:stream:s-123:seq 900同时定期扫描已过期但还在注册表的 SseEmitter做一次“延迟清理”。TTL 一定要覆盖一个完整会话的最长时长模型思考、工具调用、网络抖动之后的重连补发全都要算进去。5. 工具调用的流式编排思考、调用、回传全程直播的事件协议5.1 Agent 主循环拆解两轮推理之间发生了什么先看一个典型的 Agent 会话是什么样的。用户问“我的订单到哪里了”Agent 内部其实走了多轮循环第一轮推理大模型流式输出“我需要先查一下订单状态”并发出调用queryOrderStatus工具的决定。参数收集模型给出完整 JSON 参数{orderId:A10001}。工具执行引擎调用订单查询服务。结果回填把工具返回的数据注入上下文发起第二轮推理。第二轮推理流式输出“您的订单已经发货了”结束。传统实现是在步骤 5 结束之后一次性把最终答案返回给用户。流式改造要做的是把步骤 1、2、3、4、5 全部以事件形式实时推送出去。5.2 工具调用阶段的 SSE 事件编排完整的工具调用事件流应该长这样event: thinking_delta id: 11 data: {d:我需要先查一下订单状态} event: tool_call_start id: 20 data: {tool:queryOrderStatus,args:{orderId:A10001},argsState:complete} event: tool_call_end id: 30 data: {tool:queryOrderStatus,summary:订单状态已发货,resultSize:256} event: thinking_delta id: 40 data: {d:订单已经发出去了接下来帮用户确认物流信息} event: answer_delta id: 50 data: {d:您的订单} event: answer_delta id: 51 data: {d:已经发货了} event: done id: 60 data: {finishReason:stop}这里有三个细节需要重点说明。第一工具入参可能会很长。一次性把一个几千字符的 JSON 塞进data:字段SSE 连接一次就要承载一个大包浏览器解析出来再渲染中间会明显卡顿。我的做法是tool_call_start只推摘要字段完整参数要做脱敏和截断只保留必要信息。第二工具执行本身可能是流式的。比如搜索工具会一批一批返回结果数据库查询可能边查边能出量。遇到这种情况可以额外定义tool_call_delta事件把工具中间结果以增量方式推送。但要注意推送结果的频次不要超过每秒几次否则前端 DOM 更新会成为新的瓶颈。第三为什么要在工具执行期间不断推 thinking_delta。用户等待工具返回的这段时间最容易流失。工具跑 3 秒用户的耐心可能已经到头了。如果 3 秒内收到一两条“正在查询中”的思考增量感知会好很多。5.3 前端渲染层的按事件累积状态机前端 EventSource 的 API 本身就支持自定义事件监听这一步做起来很直接const es new EventSource(/agent/${sessionId}/stream); es.addEventListener(thinking_delta, (e) { const { d } JSON.parse(e.data); thinkingPanel.appendText(d); }); es.addEventListener(tool_call_start, (e) { const { tool, args } JSON.parse(e.data); toolPanel.renderToolCall({ tool, args }); }); es.addEventListener(tool_call_delta, (e) { const { delta } JSON.parse(e.data); toolPanel.appendResult(delta); }); es.addEventListener(tool_call_end, (e) { const { summary } JSON.parse(e.data); toolPanel.finishToolCall({ summary }); }); es.addEventListener(answer_delta, (e) { const { d } JSON.parse(e.data); answerPanel.appendText(d); }); es.addEventListener(done, () es.close());重点补充一个渲染层的注意事项善用事件 id不要自己维护计数器。EventSource 断线重连后浏览器负责携带 Last-Event-ID前端只需要在每次事件回调里顺手把它写进 sessionStorage 即可。如果自己另外维护一个“已渲染到的位置”重连后很容易和浏览器自身的重连机制打架。6. 生产化加固清单反代配置、背压治理与可观测性6.1 反代层配置与心跳联动就算应用层把 SseEmitter 和心跳都做得明明白白最坑的一层往往在反向代理。以常见的 Nginx 为例默认的proxy_read_timeout60 秒会把超过 60 秒没有任何响应数据的连接掐断所以必须显式调大并关闭缓冲location /agent/ { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; proxy_send_timeout 300s; chunked_transfer_encoding off; }proxy_buffering off很关键。如果开着缓冲SSE 数据会攒在一个缓冲区里攒到一定大小才发给客户端导致流式效果消失用户看到的是 5 秒一蹦。如果顾虑性能想开缓冲就必须把缓冲上限调大但建议一开始就关掉语义最干净。6.2 背压控制与连接数治理流式系统还有个隐蔽的坑叫背压大模型吐 token 的速率可能非常快一瞬间几百个条事件涌进来而浏览器渲染速度有限的慢客户端处理不过来。如果服务端不管不顾地 sendTCP 发送缓冲区塞满连接就被掐断了。我常用的对策是给每个 session 的事件通道加一个有界队列。Agent 引擎产生的所有事件先进队列SSE 写线程从队列里取出来发。队列满了直接拒绝写入并把这个 session 标记为“需要客户端重连续传”这样至少不会把连接打死。全局连接数同样要治理。单实例能支撑的并发 SSE 连接是有限的Tomcat 默认最大线程数、文件描述符都受限制所以要做两层限制指标建议阈值处理方式单实例最大 SSE 活跃连接数2000超过后拒绝新连接返回 503单 session 最大推送队列长度500超过后丢弃事件标记重连连续心跳失败次数3主动移除连接6.3 可观测性与故障演练线上没有可观测性的流式系统等于盲开飞机。至少要埋这四类指标活跃连接数当前所有 SseEmitter 的数量能直接反映负载。事件推送总数和速率判断是否有人恶意刷连接。Redis 事件日志堆积量如果某个 session 的日志暴涨而消费慢说明消费链路有瓶颈。断线重连率这个指标最能反映用户体验重连率超过 5% 就要回头看心跳和代理配置。日志方面我强烈建议把 sessionId 作为贯穿日志的核心字段让 Agent 执行引擎、SSE 推送、Redis 写入三条链路的日志都能用同一个 sessionId 串起来查。最后分享一个我们踩坑后总结的土办法在压测环境里专门写一个脚本模拟“kill -9 实例后浏览器自动重连”的场景。第一次跑的时候你会发现即使心跳、补发逻辑都写了还是会出现事件乱序或者重复推送。真正处理完这四个来源——网络抖动、进程重启、负载均衡切换、前端刷新——这个系统才敢说真的达到了生产级。流式架构的复杂度不会消失只会转移到边界上而边界恰恰在 Redis 和 SSE 的交接处。