企业官网建设流程全解析

从写第一行 Agent 后端代码到现在，我踩过最深的坑，就是“结果怎么回给前端”。轮询方案看起来简单——前端隔几秒拉一次任务状态，但 Agent 任务的执行时间随模型结构和工具链路波动很大，轮询间隔根本没法定：定短了接口被打爆，定长了思考过程永远展示不出来。后来我决定彻底重构，用 Spring Boot 3 内置的 SseEmitter 做 SSE 长连接，配合 Redis 做跨实例广播和消息轨迹持久化，把流式思考、工具调用全过程实时推到用户面前。这套架构在项目里跑了三个多月，线上很稳，我把整个设计和踩坑过程完整写出来。适合正被轮询和不稳定长连接折磨的后端同学，也适合想了解生产级流式推送怎么落地的读者。

1. 从轮询到 SSE：先搞明白为什么非换不可

1.1 轮询方案到底卡在哪

先回顾一下最朴素的实现。Agent 任务提交后，后端把任务状态写进数据库，前端每两秒请求一次查询接口。这个方案在请求频率低、任务耗时固定的时候勉强能用，但放到 Agent 场景里就非常别扭。

Agent 执行一次任务，中间要经历大模型分析上下文、生成思考过程、决定调用哪个工具、等待工具执行完再继续生成。每一步耗时都不可控。思考可能只要几百毫秒，某个外部工具可能要跑二十秒，模型后续生成可能又花十几秒。你让前端怎么定轮询间隔？设成 2 秒，工具在跑的时候，前端一直在空轮询；设成 10 秒，思考过程中的增量输出全被吞了，用户在页面只能干等一个旋转的加载图标，然后突然整段结果一次性蹦出来。

我接手之前的智能体平台时，线上就是这个状态。某个工具接口偶发变慢，一次任务最久跑到四十多秒，而前端超时设了三十秒，任务还没跑完，请求先断了。前端再发起重试，状态又因为接口幂等没做好而重复通知用户，搞得用户以为系统在执行同一件事执行了两次。

更致命的是数据库压力。一个 Agent 页面同时挂几十个用户，每个用户每两秒打一次查询接口，一个列表页还要带上工具执行日志、消息状态、运行耗时这些信息，查询 SQL 越写越复杂。高峰期数据库 CPU 直接飙到 80%。后来我统计了一下，一个用户一次 Agent 任务，轮询产生的查询请求平均有三十多次，而真正有价值的变更事件不超过十次。这就是典型的高频低效。

我当时的结论是：只要 Agent 是流式生成的，轮询在原理上就不成立。因为轮询是“客户端主动去要”，而流式输出的本质是“服务端主动推”。

1.2 为什么不直接上 WebSocket

换长连接，很多人第一反应是 WebSocket。确实，WebSocket 能双向通信，看起来什么都能干。但实际做 Agent 流式输出时，你会发现大部分场景只需要一个方向：服务端把事件推给前端。WebSocket 最大的问题，在于它把问题变复杂了，却没有带来对应的收益。

首先，WebSocket 是一个独立于 HTTP 的协议，握手之后走完全不同的帧格式。你需要在项目里多维护一套连接管理、心跳探测、消息编解码、前端 SDK 封装，后端和前端各加一堆配置。如果团队里前端资源紧张，这套东西的学习和维护成本是很实在的。

其次，WebSocket 没有原生的断线重连和事件恢复机制。连接一断，中间收到的消息就丢了，你需要自己实现消息序号、重连后增量拉取。而 SSE 在这方面占了便宜：底层还是 HTTP，浏览器原生的EventSource自带自动重连，断线重连时会自动带上Last-Event-ID头。你只需要在后端按这个 ID 做一次事件补推，断线续传就完成了。

SSE 上做不到的，是客户端通过同一连接主动往服务端发数据。但在 Agent 交互场景里，客户端至多需要发一个“取消任务”的指令。这个指令完全没必要走无状态的长连接，直接用一个普通的POST /api/agent/cancel接口就完了，内部通过 Redis 标记一下会话状态，正在执行的任务线程轮询取消标记后自行终止。

所以我后来把方案定为：正文指令走 REST 接口，流式结果走 SSE。两个通道各司其职，互不干扰。这比把所有能力塞进一个 WebSocket 通道里干净得多。

1.3 Redis 在流式体系里扛什么活

决定用 SSE 之后，面临的下一个问题是：生产环境不可能单实例部署。用户请求经过网关或负载均衡，可能会打到任意一台后端机器上。SSE 连接建立之后，这个长连接就绑定在某台机器的某个线程上。但 Agent 后台任务在哪个机器上跑，是不确定的。即使一开始是同一台机器，任务执行到一半服务器重启、负载均衡把重连请求转发到另一台机器，都可能发生。

所以，SSE 连接不能只存在于单机内存里，必须有一个“全局层”来支撑三件事：

第一，跨实例广播。A 机器上的 Agent 执行线程产生事件后，要能通知到所有机器——只有持有该会话连接的机器才需要真正推给前端。这里需要 Redis Pub/Sub 做消息广播，所有后端实例都订阅同一个 channel，收到消息后判断本地有没有对应会话，有就推。

第二，事件持久化。SSE 断线重连时要补推事件，必须有一份全局的、按顺序排列的事件账本。单机内存队列不行，跨实例没法访问；数据库表可以，但不够轻量且写入频繁。我用 Redis Stream，每个会话一个独立的 Stream，天然有序、支持按 ID 读取，非常适合做事件账本。

第三，会话状态与限流。哪些会话还活着、每个客户端当前最后收到的事件 ID 是多少、某用户在一段时间内能发起多少次 Agent 请求，这些元状态都用 Redis 来存。

一句话总结：SSE 负责和浏览器搞好关系，Redis 负责跨实例把话说清楚。这套分工是整个架构的核心。

2. 生产级流式中枢的架构拆解

2.1 系统模块划分

我把这套系统拆成了五个模块，每个模块职责单一，互相之间只通过事件接口协作。

HTTP 接口层负责接收 Agent 请求、建立 SSE 连接、接收取消指令。它不关心 Agent 内部怎么执行，只负责把前端请求翻译成内部事件。

会话管理层负责维护每个用户会话对应的SseEmitter实例。它管理连接的生命周期、心跳、超时、优雅关闭，是前端与后端之间的“连接仓库”。

事件中心负责定义事件类型、封装事件对象，并向 Redis Stream 和 Redis Pub/Sub 发布事件。它是所有事件流出的唯一出口。之所以做成唯一出口，是为了保证每个事件无论从哪条路径产生，都会走一遍持久化和广播流程，不会出现某个模块自己直接用 SseEmitter 发消息导致日志缺失的情况。

Agent 执行引擎负责真正运行大模型调用和工具编排。它订阅事件中心发布的任务指令，在执行过程中不断产生思考、增量、工具调用、工具结果事件，再回传给事件中心。

工具调用中枢是单独拆出来的模块。它维护所有可被 Agent 调用的工具注册表，负责参数校验、权限校验、超时控制、结果回收，确保外部系统能力以标准接口进入 Agent 执行链路。

2.2 事件模型设计

所有流式推送的事件统一抽象成下面这个结构：

public record AgentEvent( String eventId, // 全局唯一事件 ID，用于断点续传 String sessionId, // 会话 ID AgentEventType type, // 事件类型 String payload, // 具体内容，JSON 字符串 long timestamp ) {}

事件类型我定义好了六种，足够覆盖 Agent 流式交互的全部阶段：

有人会问，思考过程和最终答案的增量有什么区别？这不是多此一举，而是两个完全不同的事件流。Agent 的思考过程可能包含大量中间推理，但未必是给用户看的内容；而最终答案是要展示在页面正文里的。如果混在一起，前端很难区分哪段是思考、哪段是正式回答。分开定义，既方便 UI 差异化呈现，也让后端在控制展示策略时更灵活。

2.3 核心数据流走一遍

把整个流程串起来看一遍。前端新建EventSource，发起 SSE 连接请求。网关层根据负载策略把请求打到某台后端实例，实例上的会话管理器创建SseEmitter，把连接注册到本地会话表，同时在 Redis 中登记会话信息。

紧接着，前端通过普通POST接口提交 Agent 任务。任务消息通过 Redis Pub/Sub 广播到所有实例，某台实例的 Agent 执行引擎消费到任务消息，开始执行主流程。

执行引擎每产生一个事件，就调用事件中心的publish方法。这个方法会做两步操作：先把事件追加到 Redis Stream 做持久化，再通过 Redis Pub/Sub 广播事件。所有实例的订阅监听器收到广播后，查本地会话表，只有真正持有这台前端连接的实例会把事件通过SseEmitter推出去。

一旦 Agent 决定调用工具，执行引擎发出TOOL_CALL事件，然后等待工具返回。工具执行完成，执行引擎把结果拼进上下文，继续后续的模型调用，这一个循环直到任务结束后发出DONE事件。

如果中途连接断开，浏览器侧EventSource会自动重连，并在请求头带上Last-Event-ID。后端读到这个值后，去 Redis Stream 查询该 ID 之后的所有事件，一次性补推给前端，再继续接收新事件。整个过程不需要前端写额外代码。

3. 基于 SseEmitter 的流式通道落地

3.1 SseEmitter 基础使用姿势

Spring Boot 3 原生支持 SSE，核心类就是SseEmitter。Controller 里的写法非常简单：

@RestController @RequestMapping("/api/agent") public class AgentStreamController { private final SessionManager sessionManager; @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter stream(@RequestParam String sessionId) { SseEmitter emitter = new SseEmitter(300_000L); sessionManager.register(sessionId, emitter); return emitter; } }

这里有两个细节容易翻车。第一，produces必须写TEXT_EVENT_STREAM_VALUE，否则 Spring MVC 不按 SSE 的响应方式处理，浏览器拿到的响应可能不是一个流。第二，SseEmitter构造器可以传超时时间，单位毫秒，我这里传了三十分钟。如果不显式设置，Spring Boot 底层会按容器默认超时来算，通常很快就把连接回收了，前端根本等不到 Agent 跑完。

SseEmitter的发送方式也值得单独提一下。推荐使用带事件名的SseEventBuilder：

emitter.send(SseEmitter.event() .name("agent/thinking") .data(Map.of("content", "用户的问题是...")));

event().name()设置的名称，到了前端会自动对应到addEventListener里的事件名。如果不用name，事件会被归类到默认的message事件，前端需要统一在onmessage里按 payload 类型去分流，反而麻烦。

还要注意一点，同一个SseEmitter实例上的发送操作是线程安全的，但如果你同时从多个线程往里发事件，不保证业务层面的顺序。所以我的原则是：一个会话同时只有一个发送线程，事件先经过一个内存队列，由发送线程按顺序从队列里取出来发。

3.2 会话注册与生命周期管理

连接建立之后，SessionManager要维护一张“会话 ID → SseEmitter”的本地映射。这里选ConcurrentHashMap就够了，因为本地会话表不需要跨实例共享，每个实例只需要知道自己持有的连接。

完整的管理逻辑不能只 put 就完事，要同时挂上一堆回调：

public void register(String sessionId, SseEmitter emitter) { // 如果之前有同会话的旧连接，先处理掉 SseEmitter previous = localSessions.put(sessionId, emitter); if (previous != null) { previous.complete(); } emitter.onCompletion(() -> localSessions.remove(sessionId, emitter)); emitter.onTimeout(() -> { localSessions.remove(sessionId, emitter); emitter.complete(); }); emitter.onError(e -> { localSessions.remove(sessionId, emitter); emitter.completeWithError(e); }); }

这段代码里最关键的是previous.complete()。用户可能因为网络抖动或浏览器刷新，旧的 SSE 连接还没断、新的连接就已经建立。如果不先把旧连接掐掉，同一个会话就有两个活跃 emitter，消息会被重复推送，前端页面出现内容错乱。这个坑我遇到过两次，一次是重复渲染，一次是内存里积累了死连接。

另外，onCompletion、onTimeout、onError三个回调都要记得清理。有一段时间线上出现内存缓慢增长，排查到最后就是SseEmitter超时后没有任何回调去移除本地映射，导致表里堆满了已失效的 emitter。这个问题在连接数大的时候尤其明显。

3.3 心跳与超时参数调优

SSE 底层是 HTTP 长连接，中间经过的设备和代理如果长时间看不到数据，会按各自的空闲超时策略把连接关掉。尤其像某些 IDC 的硬件负载均衡，默认空闲超时可能只有五分钟，一旦连接上没有数据流过，就会被强制断开。更麻烦的是，Agent 的思考阶段可能真的会“沉默”一段时间——模型在思考下一步怎么走，工具在等待外部接口响应，都是几十秒不发任何事件。

这时候必须做心跳。每十五到二十秒往连接上发一个注释帧，不算业务事件，但能让链路知道连接还活着。

ScheduledExecutorService heartbeatExecutor = Executors.newSingleThreadScheduledExecutor(); private void startHeartbeat(SseEmitter emitter) { heartbeatExecutor.scheduleAtFixedRate(() -> { try { emitter.send(SseEmitter.event().comment("heartbeat")); } catch (IOException e) { emitter.completeWithError(e); } }, 0, 15, TimeUnit.SECONDS); }

comment("heartbeat")发送的是一条注释帧，浏览器不会把它当成业务事件，因此不会触发onmessage，也能起到探活作用。心跳线程用一个全局共享的调度线程池就行，不要每个连接都 new 一个线程，否则连接数上千时线程数会爆炸。

参数调优给一组我实测合理的目标值，可以直接抄作业：

心跳间隔不要设得太短，比如五秒一次。虽然能更有效地保活，但没必要地增加了无效流量，而且在部分浏览器上，注释帧如果频率过高，反而可能导致渲染线程频繁处理空事件，影响页面性能。

3.4 前端 EventSource 对接细节

前端对接的核心代码很短：

const es = new EventSource('/api/agent/stream?sessionId=' + sessionId); es.addEventListener('agent/thinking', (e) => { const data = JSON.parse(e.data); renderThinking(data.content); }); es.addEventListener('agent/token_delta', (e) => { const data = JSON.parse(e.data); appendResult(data.content); }); es.addEventListener('agent/tool_call', (e) => { const data = JSON.parse(e.data); renderToolCall(data.name, data.args); }); es.addEventListener('agent/done', (e) => { es.close(); }); es.onerror = () => { // 浏览器会自动重连，这里只需处理重连事件 console.log('连接异常，准备重连'); };

有一个经验是：EventSource的自动重连是浏览器内置的，默认延迟三秒，但如果你在服务端主动调用了emitter.complete()，前端不会再重连。这正好是正常结束的语义。如果你要终止一个异常连接，也建议先推送一个ERROR事件，再调用complete()，这样前端能弹出错误提示而不是静默重连后看到一片空白。

另外，前端不要自己去写setInterval轮询“检查连接状态”。EventSource内置了readyState状态位，从CONNECTING到OPEN再到CLOSED，你只需要在onerror回调里做 UI 状态提示就行。手动加轮询只会和浏览器自动重连机制打架，出现重复连接。

4. 工具调用中枢：Agent 的双手落地

4.1 工具注册机制

工具调用中枢要解决的核心问题是：模型怎么知道有哪些工具可以用、参数长什么样、调用的入口在哪。我采用的是“注解扫描 + 运行时注册表”的方式。

先定义一个工具注解：

@Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) public @interface AgentTool { String name(); // 工具名，模型调用时用的名字 String description(); // 工具描述，模型根据描述决定是否使用 }

然后在业务代码里写一个工具方法：

@Component public class OrderTools { @AgentTool(name = "query_order_status", description = "根据订单号查询物流状态") public String queryOrderStatus(@ToolParam(description = "订单号") String orderId) { return orderService.queryStatus(orderId); } }

核心机制就藏在ToolRegistry里。我实现了一个BeanPostProcessor，在 Spring 容器初始化每个 Bean 之后扫描其方法，发现带@AgentTool注解的方法就注册到工具表：

@Component public class ToolAnnotationProcessor implements BeanPostProcessor { private final ToolRegistry toolRegistry; @Override public Object postProcessAfterInitialization(Object bean, String beanName) { for (Method method : ReflectionUtils.getAllDeclaredMethods(bean.getClass())) { AgentTool annotation = AnnotatedElementUtils.findMergedAnnotation(method, AgentTool.class); if (annotation != null) { ToolDefinition definition = new ToolDefinition( annotation.name(), annotation.description(), generateJsonSchema(method), new MethodToolInvoker(bean, method) ); toolRegistry.register(definition); } } return bean; } }

用这种方式，新增一个工具只需要在任意 Spring Bean 里写一个带注解的方法，不用改注册表、不用改 Agent 执行引擎。模型侧的 prompt 内容会自动从注册表生成。工具多了以后，这种“零配置新增”的体验非常重要。某公司有一个内部的订单查询系统，接入这套机制后，新增一个查询接口只花了几分钟。

4.2 调用编排与流式反馈

模型决定调用工具时，事件中心会收到一条TOOL_CALL事件，里面包含工具名和 JSON 格式的参数。工具调用中枢拿到后，分四步处理。

第一步是解析和校验。我用 JSON Schema 校验参数，工具方法上的@ToolParam注解会生成 schema，例如orderId字段会被标记为string且required。如果参数缺了，直接返回错误事件，不会真的去调用工具方法。

第二步是异步执行。工具方法可能耗时较长，不能占着 Agent 执行线程等待。我用CompletableFuture把工具调用丢进独立的线程池：

CompletableFuture<ToolResult> future = CompletableFuture .supplyAsync(() -> invoker.invoke(jsonArgs), toolExecutor) .orTimeout(45, TimeUnit.SECONDS); future.whenComplete((result, err) -> { AgentEvent event = err != null ? AgentEvent.toolError(sessionId, toolName, err.getMessage()) : AgentEvent.toolResult(sessionId, toolName, result.getSummary()); eventCenter.publish(event); });

这里有两个细节值得说。第一，orTimeout必须加。工具方法外部接口如果挂死了，不能无限等下去。四十五秒是我根据线上外部接口 P99 耗时倒推的，你有更长或者更短的接口可以按实际调。

第二，工具执行期间，前端需要有一个“转场”反馈。TOOL_CALL事件发出后，模型已经生成了一部分内容，前端展示“正在调用工具”，如果工具没跑完，什么都不发，用户又会觉得卡住了。所以我增加了TOOL_CALL_PENDING状态事件，每五秒推送一次工具执行进度。这个事件不要求前端在页面展示复杂内容，只需要显示一个轻量的轮播状态，比如“正在等待订单接口返回”。体验上明显比干等好很多。

第三步是结果回流。工具返回结果后，事件中心发布TOOL_RESULT事件，前端替换进度提示。同时，工具结果会被回写到 Agent 执行上下文中，模型会基于结果继续生成后续内容。

第四步是错误兜底。工具抛异常、超时、参数错误，都要转换成结构化的错误数据传给模型，让模型有机会根据错误信息重新规划路径，而不是直接让整个任务失败。比如查询订单失败后，模型可能会说“订单接口暂时不可用，建议稍后再试”，这比直接抛异常体验好太多了。

4.3 参数校验与安全控制

能随便让 Agent 调用工具听起来很爽，但如果不加控制，就是灾难。我踩过一个特别典型的坑：某个测试同学在对话里问“把订单金额取整”，模型不知道怎么理解，居然尝试调用了一个内部订单修改工具，传了个非法参数，好在参数校验挡住了，没有产生实际修改。从这以后，我把安全控制做成了硬性要求，三个层面缺一不可。

第一层是参数白名单校验。每个工具的参数名必须在 schema 列表里，类型必须匹配。这一层主要是挡住模型幻觉产生的乱传参。

第二层是权限校验。工具注册的时候可以带一个权限标识，执行前通过上下文判断当前用户是否有权限调用。比如“退款”工具普通用户永远不能直接触发，只有客服角色通过专门的指令才能到达。权限校验放在工具中枢里，不要在工具方法内部自己写，否则每个工具都要重复一遍 if 判断。

第三层是审计。所有工具调用事件，包括请求参数、执行耗时、执行结果摘要、错误信息，必须写入日志和 Redis Stream。如果后续出了线上问题，要能回溯到具体哪一次对话调用了哪个工具、传了什么参数、返回了什么结果。Agent 的工具调用不是普通接口调用，错误的影响面更大，审计日志是唯一能还原现场的线索。

5. Redis 在跨实例广播与持久化中的关键角色

5.1 用 Pub/Sub 打通多实例实时通道

现在假设系统有两台后端实例。用户 A 的 SSE 连接打到了实例 1，而 Agent 任务执行线程可能跑在实例 2 上。如果实例 2 直接持有连接，那任务结果怎么回到实例 1？答案是：实例 2 把事件发布到 Redis，实例 1 订阅到之后再推送。

我用 Redis Pub/Sub 作为实时广播通道。会话级别的频道命名是agent:channel:{sessionId}。事件中心发布的代码如下：

public void publish(String sessionId, AgentEvent event) { // 1. 先写入 Stream，确保持久化 streamGateway.append(sessionId, event); // 2. 再通过 Pub/Sub 广播 String json = objectMapper.writeValueAsString(event); stringRedisTemplate.convertAndSend("agent:channel:" + sessionId, json); }

每一台后端实例都要配置一个 Redis 消息监听容器，订阅所有可能出现的会话频道。但问题来了，频道名是动态的，你不能在配置类里写死。对策是用频道模式匹配，比如agent:channel:*：

@Bean public RedisMessageListenerContainer container(RedisConnectionFactory factory) { RedisMessageListenerContainer container = new RedisMessageListenerContainer(); container.setConnectionFactory(factory); container.addMessageListener(listener(), new PatternTopic("agent:channel:*")); return container; }

监听器内部的逻辑也很简单：

public void onMessage(Message message, byte[] pattern) { String channel = new String(message.getChannel()); String sessionId = channel.substring(channel.lastIndexOf(':') + 1); AgentEvent event = objectMapper.readValue(message.getBody(), AgentEvent.class); SseEmitter emitter = sessionManager.getLocal(sessionId); if (emitter != null) { emitter.send(SseEmitter.event().name(event.type().getEventName()) .data(event.payload())); } }

重点在于：每个实例都收到广播，但只有本地有对应会话的实例才会真正往客户端推数据。这样做的好处是无状态化——每个实例只需要管理自己手上的连接，不需要一个全局的连接路由表。

5.2 用 Stream 做事件账本与断点续传

Pub/Sub 有一个致命弱点：消息不持久化。如果某台实例在订阅容器启动之前就开始执行任务，或者某个会话的所有实例都还没有订阅频道，那广播出去的消息就丢了。对于实时阶段，丢了可能只是连接还没建好，影响不大；但一旦连接中途断开需要补推，没有账本一切都无从谈起。

所以我在广播之前，先把事件写进 Redis Stream。每个会话一个 Stream，key 是agent:stream:{sessionId}。追加事件用的是StreamRecords：

public String append(String sessionId, AgentEvent event) { Map<String, String> fields = Map.of( "eventId", event.eventId(), "type", event.type().name(), "payload", event.payload(), "timestamp", String.valueOf(event.timestamp()) ); return stringRedisTemplate.opsForStream().add( StreamRecords.mapToString(fields) .withStreamKey("agent:stream:" + sessionId)) .getValue(); }

Redis Stream 为每条消息生成一个自增的 ID。由于所有实例都往同一个 Stream 追加消息，这个 ID 天然就是全局有序的，正好可以充当断点续传的游标。

断线重连时，后端拿到 HTTP 请求头里的Last-Event-ID，用XREAD读取该 ID 之后的所有消息并补推：

public List<AgentEvent> readFrom(String sessionId, String lastEventId) { var readResult = stringRedisTemplate.opsForStream().read(numericKey -> numericKey.fromStream(StreamOffset.create( "agent:stream:" + sessionId, ReadOffset.from(lastEventId))) ); // 解析返回的 records 并映射回 AgentEvent }

前端EventSource在重连时默认会带上Last-Event-ID，但你用SseEmitter.event()时，Spring 并不会自动把发送事件的 ID 映射成 SSE 规范的id:字段，至少我在 Spring Boot 3 里测试时是这样的。所以一个务实做法是：后端手动把eventId放进事件数据的 JSON 里，前端拿到事件后手动记录最后收到的 ID，重连时通过 URL 参数传回：

let lastEventId = localStorage.getItem('lastEventId'); const es = new EventSource('/api/agent/stream?sessionId=' + sessionId + '&lastEventId=' + lastEventId); // 每次收到事件时更新 es.addEventListener('agent/token_delta', (e) => { const data = JSON.parse(e.data); lastEventId = data.eventId; localStorage.setItem('lastEventId', lastEventId); });

后端在读lastEventId参数时，做一次readFrom补推，再继续走正常订阅流程。这个方案我实测在丢包率较高的网络环境下，能恢复绝大部分中间事件，体验上比单纯等浏览器重连然后从空状态继续强太多。

5.3 广播与持久化的配套细节

Stream 不做清理会越积越大。单个会话的 Stream 在任务结束后就没用了，必须设置过期策略。我在DONE事件发布后，会执行清理：把agent:stream:{sessionId}的 TTL 设为 24 小时，同时删除对应的 Pub/Sub 订阅频道里的元数据。不过要小心，不能立刻del，因为如果客户端在最后一刻断线重连，还需要补最后的DONE事件。留 24 小时是折中方案，既能覆盖大概率的重连场景，又不会长期占内存。

序列化方面，Redis Stream 的字段值我用 JSON 字符串存。有人可能会把整个事件序列化成二进制，这浪费了 Stream 天然支持多字段的优势。拆成type、payload、timestamp字段，后续如果要按类型过滤或者做时间范围查询，就不需要把整条 JSON 捞出来再解析，直接读字段就行。

顺序方面要注意，Redis Stream 在一个 key 内部的消息 ID 是单调递增的，所以同一会话的事件顺序是可靠的。但 Redis Pub/Sub 的通知顺序并不能严格保证与 Stream 写入顺序一致，因为每一次convertAndSend都是一次独立的网络往返，可能因为网络抖动导致后发先至。我在实践中发现这个概率极低，业务上没有造成实际影响。如果你有强顺序要求，可以让订阅方收到广播后不直接推送，而是先解析事件 ID 做一次排序确认，稳妥但增加了延迟。我的建议是：信任 Redis 内部的顺序，实测足够。

6. 生产级可靠性：把断流问题摁死

6.1 这么多场景都会断流

SSE 看起来简单，实际到了生产环境，断流的场景比我预想的多得多。我归一下类：

第一类是连接空闲超时。问题不在业务代码，而在于中间链路。从 Tomcat 连接池到 Nginx，再到云厂商的负载均衡，每一层都有空闲超时策略，默认值往往只有几十秒到几分钟。Agent 任务在思考阶段完全可能几秒钟不发任何事件，中间设备就把连接掐了。心跳能解决大部分空闲超时问题。

第二类是服务实例重启。发布新版本时，实例上的 SSE 连接全部被断开。前端浏览器会自动重连，但如果新请求被负载均衡转发到另一台没有该连接上下文的实例，就相当于连接从零开始，中间事件全部丢失。所以断线续传必须配合SessionManager的本地会话设计：重连请求打到任何一台实例都行，新实例从 Redis Stream 里补数据。

第三类是网络抖动。弱网环境下移动端断连特别频繁，EventSource自动重连倒是方便，但重连间隙产生的事件还是会丢。这也是我对 Stream 持久化如此看重的原因。

第四类是服务端超时配置不当。SseEmitter 默认超时如果设得太短，Agent 还没跑完连接就被服务端主动关闭，这不是断流，而是“主动断流”，用户端的表现是转圈转到一半突然结束。

6.2 断线重连与事件补推

断线重连我概括成“三步走”。第一步，浏览器发现连接断开，触发onerror，自动进入重连流程。第二步，重连请求带上lastEventId参数，后端从 Redis Stream 读取该 ID 之后的事件，按顺序补推。第三步，补推完成后，继续监听后续广播事件。

这里有一个容易忽略的点：如果补推期间，Agent 还在继续产生新事件，那事件中心往 Stream 追加新事件和 Redis 广播之间可能有时间差。补推线程刚读完某个 ID 位置，新事件已经追加进去了，但因为广播线程和补推线程并发，前端可能会先收到由广播推送的新事件，再收到由补推线程推送的旧事件，导致顺序错乱。

我的规避方案是：给每个会话加一个本地内存事件队列，补推和新广播的事件都先进队列，由会话发送线程统一从队列取出来按顺序推。这样无论补推线程和消费者线程哪个先到，经过队列排序后，前端永远能看到有序事件流。代价是引入了额外的内存占用，但每个会话的队列在DONE事件后就清理掉了，压力可控。

6.3 Nginx 与网关层的 SSE 配置

如果你的服务前面有 Nginx，默认配置下 SSE 一定跑不顺。原因在于 Nginx 默认开启了响应缓冲，攒满 4KB 或等后端关闭才把数据发给客户端，SSE 事件流就完全失去了“流”的意义。

需要在 Nginx 的 location 里加上一组关键配置：

location /api/agent/stream { proxy_pass http://backend; proxy_http_version 1.1; proxy_set_header Connection ''; proxy_buffering off; proxy_cache off; proxy_read_timeout 3600s; proxy_send_timeout 3600s; proxy_buffers 16 4k; proxy_buffer_size 2k; }

解释一下几个关键项。proxy_buffering off是重中之重，关掉缓冲，数据到了网关就往下游转发。proxy_http_version 1.1配合proxy_set_header Connection ''是标准的 HTTP 长连接配置，如果不这么干，Nginx 在上游取不到 SSE 响应之后可能立刻断开到后端的连接。proxy_read_timeout设成 3600 秒，是为了防止 Nginx 在连接空闲时主动断开。

云上负载均衡也有类似配置项，有些云厂商的 SLB 对 SSE 长连接默认空闲超时只有几十秒，必须同步调大。实际操作中我养成了一个习惯：每调一次网关层配置，就在前端页面放一个三十分钟的静默观察窗口，看是否出现空闲断流，确保心跳和网关超时配置没有冲突。

6.4 优雅关闭与连接配额

服务发布要进行优雅关闭，不能让正在使用的用户看到连接瞬间断开的错误。Spring Boot 的优雅停机配合 Spring MVC 的异步请求处理，可以在停机时先把已处理的请求放行，但 SSE 长连接不是普通请求，需要额外处理。

我的做法是在ApplicationListener<ContextClosedEvent>里，遍历本地会话表，逐个发送一个SYSTEM_SHUTDOWN事件，告诉前端“服务即将升级，请稍后重试”，然后调用emitter.complete()主动关闭连接。前端收到SYSTEM_SHUTDOWN后，礼貌地给用户展示“正在重新连接”，然后发起新一轮连接。这样就避免了停机时前端因为连接异常中断而弹错误提示的尴尬。

连接配额方面，SseEmitter 本质上是占用 Servlet 异步线程的资源。如果某个实例上的连接数过多，会影响同实例其他接口的可用线程池。我加了一层连接数的动态检查：当单实例的活跃连接超过阈值（比如 2000，根据实例规格调整），新的 SSE 连接请求直接返回429 TOO_MANY_REQUESTS响应，前端收到该状态码后可以提示“当前服务繁忙”。这个防护在流量高峰时很管用，比让连接无限增长直到 OOM 强得多。

7. 常见问题速查表与避坑实录

7.1 一页速查表

把我在实际排障过程中遇到的高频问题整理成了下面的表格，供大家直接参考：

7.2 踩坑实录：我差点被心跳线程坑死

有一个问题花了我一个下午加一个晚上才定位清楚。当时连接数冲到四千多，CPU 使用率也飙上去了。一开始以为是大模型接口调用太频繁，后来排查线程池才发现，每个SseEmitter连接我都new了一个ScheduledExecutorService来发心跳，四千多个连接就有四千多个线程在定时跑，Tomcat 已经疯了。

这是一个典型的过度设计。心跳是高频且统一的行为，用全局共享的一个ScheduledThreadPoolExecutor或者干脆用单向延迟队列实现循环发送，线程数只需要几条。修复之后 CPU 直接降了三分之二。这个案例时刻提醒我：凡是“每个连接一个线程”的设计，在生产环境一定会被连接数打脸。

另一个印象深的坑是在浏览器端。某一个版本的 Chrome 对EventSource的自动重连策略做了调整，重连间隔会根据网络状态动态拉长。如果线路上游控制的是一条不稳定的隧道，前端可能在断线后等一分钟才会重建连接，这段时间用户看到的页面就是一直转圈。我们的解法是在onerror回调里做 UI 文案提醒，给用户一个“手动重新连接”按钮，避免用户干等。

7.3 关于“给用户看什么”的思考

技术全部打通之后，产品层面会遇到一个新的问题：Agent 思考过程到底要不要原样展示？我的建议是不要。模型思考过程往往冗长、跳跃、包含大量与最终答案无关的中间推理，直接展示会让用户觉得 Agent 很“混乱”。更好的做法是，把思考过程截取关键节点，以简短的摘要事件推给前端，比如“正在分析用户的订单问题”。而真正的答案正文，才用TOKEN_DELTA逐字展示。

这个设计决定了前端的渲染策略和事件表里的字段都必须提前规划好。技术层面，思考摘要和正文增量走的是不同的事件类型，天然可以分开控制；产品层面，这给了你极大的灵活性——想展示细节就多推几条THINKING事件，想收敛就把THINKING事件降频。这套架构本身不限制你，限制你的只有业务判断力。

结尾：一些实际体会

整个方案落地之后，我最明显的感受是：SSE 本身不难，难的是把 Agent 流式交互当成一个完整的链路来设计。用户看到的只是文字流式输出，但背后是连接管理、事件模型、跨实例广播、持久化账本、断点续传、工具编排这些环节协同工作。每一个环节单拎出来都不复杂，但组合在一起时，任何一个环节掉链子都会破坏体验。

如果只让我给出一条建议，那一定是：不要等到线上出了断流问题再去补可靠性设计。从第一行SseEmitter代码开始，就把超时、清理、心跳、Stream 账本这四个基础件写好。后面再加实例、加工具、加会话量时，你只需要顺势扩展，而不是推倒重来。这套架构里最值钱的不是某个具体 API 的用法，而是把“连接当作有生命周期的对象”来管理的这种意识。只要抓住了这一点，哪怕后面换了框架、换了中间件，整个思路依然可以迁移复用。