企业官网建设流程全解析

1. 为什么流式 Agent 的“最后一公里”总是翻车

做过 AI Agent 项目的人大概率都遇到过这种场景：前端页面已经渲染出了模型吐出的前几段文字，用户正看得入神，突然光标卡住不动了，浏览器控制台里静静躺着一行stream disconnected before completion: idle timeout waiting for sse。用户刷新页面，重新发起一次对话，结果前面那几百个 token 的思考过程全白费了。更糟的是，如果这个 Agent 还挂着工具调用（比如查数据库、调外部 API、写文件），你根本不知道那次调用到底执行成功了没有，重试会不会造成重复扣款或者重复写入。

这个问题的本质，不是模型不行，也不是前端写得烂，而是流式传输链路上缺少一个能扛住生产流量的中枢。轮询方案（前端每隔一秒问一次“有新内容吗”）在 demo 阶段能跑，一旦并发上来，QPS 直接爆炸，服务端和数据库都吃不消；而裸用 SSE 又太脆弱，连接一断，上下文就丢了，工具调用的中间状态也无从恢复。

我这段时间用Spring Boot 3 + SSE + Redis搭了一套 Agent 的流式思考与工具调用中枢，核心目标就三个：推得动、断得掉、接得回。推得动是指模型 token 能实时推到浏览器；断得掉是指连接异常时服务端能感知并清理资源；接得回是指客户端重连后能从断点继续，而不是从头再来。这套东西适合正在做 AI Agent 产品、需要处理长连接流式输出、又不想被轮询拖垮后端的中高级 Java 开发者参考。下面我把整套设计思路、关键代码、踩过的坑一次性讲清楚。

2. 整体架构设计与技术选型拆解

2.1 为什么是 SSE 而不是 WebSocket

很多人第一反应是流式输出就该用 WebSocket，双向通信嘛。但 Agent 的典型交互是单向推送为主：模型生成 token、工具执行进度、最终结果，这些都是服务端往客户端推。客户端要发的只是一个初始请求和偶尔的“停止生成”指令，用普通 HTTP POST 就够了。

SSE 的优势在这里非常明显：它基于标准 HTTP，天然支持断线重连（浏览器 EventSource 自带Last-Event-ID机制），不需要额外的协议升级，网关和负载均衡的兼容性也比 WebSocket 好得多。WebSocket 在 Nginx 后面要额外配Upgrade头，SSE 只要关掉缓冲就行。实测下来，SSE 在移动端弱网环境下的表现也更稳，因为它是纯文本流，代理层不容易把它当成异常连接掐掉。

当然 SSE 也有硬伤：它是单向的，而且默认不支持二进制。但 Agent 场景里我们传的就是文本 token 和 JSON 事件，完全够用。

2.2 Redis 在中枢里到底扮演什么角色

这是整套架构的关键。很多人以为 Redis 只是拿来存 session 的，其实在这套中枢里它承担了四个职责：

• 事件缓冲队列：模型生成的 token 不是直接写 SSE 的，而是先进入 Redis 的 Stream 结构。这样即使某个客户端连接断了，事件也不会丢，重连后可以按 offset 续读。
• 分布式锁：防止同一个会话被多个请求同时处理，导致工具重复调用。
• 会话状态存储：记录当前 Agent 执行到哪一步、调用了哪些工具、结果是什么。
• 心跳与在线状态：通过带 TTL 的 key 判断客户端是否还活着。

用 Redis Stream 而不是 List 或 Pub/Sub，是因为 Stream 支持消费者组和消息 ID 回溯，天然适合“断点续传”这个需求。Pub/Sub 消息发出去就没了，List 又不好做多消费者。

2.3 三层结构：接入层、编排层、执行层

我把整个中枢拆成三层，职责清晰，方便横向扩展：

接入层用SseEmitter而不是 WebFlux 的Flux，是因为团队里大部分人对阻塞式编程更熟，而且SseEmitter配合线程池已经能扛住几千并发。如果追求极致性能可以上 WebFlux，但复杂度会陡增，这里不展开。

3. 核心细节解析与实操要点

3.1 SseEmitter 的超时与心跳配置

SseEmitter默认超时时间很短，而且一旦超时就会触发onTimeout回调，连接直接断掉。生产环境必须显式设置一个足够长的超时，同时配合心跳保活。

// 超时设置为 0 表示永不超时，由我们自己控制生命周期 SseEmitter emitter = new SseEmitter(0L); // 注册回调，连接断开时清理 Redis 资源 emitter.onCompletion(() -> cleanup(sessionId)); emitter.onTimeout(() -> cleanup(sessionId)); emitter.onError(e -> cleanup(sessionId));

这里有个坑：0L虽然表示不超时，但很多网关（比如 Nginx）有自己的proxy_read_timeout，默认 60 秒。如果 60 秒内没有任何数据流动，网关会主动断开连接。所以心跳是必须的，我一般每 15 秒发一个注释帧：

// 心跳线程，每 15 秒发一次 scheduler.scheduleAtFixedRate(() -> { try { emitter.send(SseEmitter.event().comment("heartbeat")); } catch (IOException e) { // 发送失败说明连接已断，触发清理 emitter.completeWithError(e); } }, 0, 15, TimeUnit.SECONDS);

注意：心跳帧用comment而不是data，因为注释帧不会触发前端onmessage，不会污染业务数据。

3.2 Redis Stream 做事件缓冲的正确姿势

模型每生成一个 token 就往 Redis Stream 里写一条，这个操作必须足够轻。我用的是XADD命令，配合MAXLEN做容量控制，防止内存无限增长：

// 写入事件，限制 Stream 最大长度 10000 条 redisTemplate.opsForStream().add( StreamRecords.newRecord() .in("agent:stream:" + sessionId) .ofMap(Map.of( "type", "token", "content", token, "seq", String.valueOf(seq++) )) );

读取端用XREAD配合BLOCK做阻塞读取，这样 SSE 推送线程不用空转轮询：

// 从指定 offset 开始阻塞读取，超时 5 秒 List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().read( StreamReadOptions.empty().block(Duration.ofSeconds(5)).count(10), StreamOffset.create(streamKey, ReadOffset.from(lastId)) );

lastId就是断点续传的关键。客户端重连时带上Last-Event-ID，服务端解析出这个 ID，从它之后继续读，前面已经推过的 token 不会重复推。

3.3 分布式锁防止工具重复调用

Agent 调用工具是最容易出问题的地方。假设用户网络抖动，前端自动重连，结果触发了两次“查询订单”工具，如果这个工具有副作用（比如发短信），那就出大事了。

我用 Redis 的SET NX PX实现了一个简单的分布式锁，锁的粒度是“会话 + 工具名”：

String lockKey = "agent:lock:" + sessionId + ":" + toolName; Boolean locked = redisTemplate.opsForValue() .setIfAbsent(lockKey, requestId, Duration.ofSeconds(30)); if (Boolean.TRUE.equals(locked)) { try { return executeTool(toolName, params); } finally { // 只有持有锁的请求才能释放 if (requestId.equals(redisTemplate.opsForValue().get(lockKey))) { redisTemplate.delete(lockKey); } } } else { // 已有请求在执行，返回“处理中”状态 throw new ToolBusyException(toolName); }

这里释放锁时校验requestId很重要。如果不校验，A 请求超时释放了锁，B 请求拿到锁，结果 A 的 finally 又把 B 的锁删了，就会出现并发漏洞。

3.4 序列化方式的选择

Redis 默认用 JDK 序列化，存进去的东西在redis-cli里根本看不懂，排查问题极其痛苦。我统一用StringRedisSerializer做 key 序列化，value 用GenericJackson2JsonRedisSerializer：

@Bean public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory factory) { RedisTemplate<String, Object> template = new RedisTemplate<>(); template.setConnectionFactory(factory); template.setKeySerializer(new StringRedisSerializer()); template.setHashKeySerializer(new StringRedisSerializer()); template.setValueSerializer(new GenericJackson2JsonRedisSerializer()); template.setHashValueSerializer(new GenericJackson2JsonRedisSerializer()); template.afterPropertiesSet(); return template; }

这样在redis-cli里GET出来的就是可读的 JSON，调试效率提升一大截。

4. 实操过程与核心环节实现

4.1 环境准备与依赖引入

先确认版本：Spring Boot 3.2.x，JDK 17，Redis 7.x。Spring Boot 3 对 JDK 17 是硬性要求，别用 JDK 8 硬凑。

<dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-data-redis</artifactId> </dependency> <dependency> <groupId>org.apache.commons</groupId> <artifactId>commons-pool2</artifactId> </dependency>

commons-pool2是 Lettuce 连接池需要的，不加的话高并发下会频繁创建连接，性能很差。

Redis 连接配置：

spring: data: redis: host: 127.0.0.1 port: 6379 lettuce: pool: max-active: 50 max-idle: 20 min-idle: 5 max-wait: 3000ms timeout: 5000ms

踩坑记录：timeout一定要设，默认是无限等待。我遇到过 Redis 主从切换时连接卡死，整个 SSE 推送线程全部阻塞，最后服务假死。设了 5 秒超时后，至少能快速失败并触发重连。

4.2 SSE 接入端点的完整实现

@RestController @RequestMapping("/api/agent") public class AgentStreamController { @GetMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter stream(@RequestParam String sessionId, @RequestHeader(value = "Last-Event-ID", required = false) String lastEventId) { SseEmitter emitter = new SseEmitter(0L); String streamKey = "agent:stream:" + sessionId; // 断点续传：从 lastEventId 之后开始读 String startId = (lastEventId != null) ? lastEventId : "0"; // 提交到线程池异步推送 streamExecutor.submit(() -> pushLoop(emitter, streamKey, startId)); emitter.onCompletion(() -> log.info("SSE completed: {}", sessionId)); emitter.onTimeout(() -> log.warn("SSE timeout: {}", sessionId)); emitter.onError(e -> log.error("SSE error: {}", sessionId, e)); return emitter; } }

produces = MediaType.TEXT_EVENT_STREAM_VALUE这个必须加，否则浏览器不会按 SSE 解析。

4.3 推送循环与断点续传逻辑

private void pushLoop(SseEmitter emitter, String streamKey, String startId) { String lastId = startId; try { while (!Thread.currentThread().isInterrupted()) { List<MapRecord<String, Object, Object>> records = redisTemplate.opsForStream().read( StreamReadOptions.empty().block(Duration.ofSeconds(5)).count(20), StreamOffset.create(streamKey, ReadOffset.from(lastId)) ); if (records == null || records.isEmpty()) { // 没有新数据，发心跳 emitter.send(SseEmitter.event().comment("hb")); continue; } for (MapRecord<String, Object, Object> record : records) { Map<Object, Object> value = record.getValue(); String type = (String) value.get("type"); // 用 Redis 的消息 ID 作为 SSE 的 event id，实现断点续传 emitter.send(SseEmitter.event() .id(record.getId().getValue()) .name(type) .data(value.get("content"))); lastId = record.getId().getValue(); // 遇到结束标记就关闭连接 if ("done".equals(type)) { emitter.complete(); return; } } } } catch (Exception e) { emitter.completeWithError(e); } }

这里的关键是把 Redis Stream 的消息 ID 直接作为 SSE 的 event id。浏览器重连时会自动带上Last-Event-ID请求头，服务端解析后从该 ID 之后继续读，完美实现断点续传。

4.4 Agent 执行与事件写入

Agent 的执行逻辑放在一个独立的服务里，它不直接操作 SSE，只负责往 Redis Stream 写事件：

@Service public class AgentExecutor { public void execute(String sessionId, String userInput) { String streamKey = "agent:stream:" + sessionId; String lockKey = "agent:lock:" + sessionId; // 会话级锁，防止同一会话并发执行 Boolean locked = redisTemplate.opsForValue() .setIfAbsent(lockKey, "1", Duration.ofMinutes(5)); if (!Boolean.TRUE.equals(locked)) { writeEvent(streamKey, "error", "会话正在处理中"); return; } try { // 1. 写入思考开始事件 writeEvent(streamKey, "thinking", "开始分析问题..."); // 2. 流式调用模型 modelClient.streamChat(userInput, token -> { writeEvent(streamKey, "token", token); }); // 3. 判断是否需要工具调用 if (needToolCall()) { writeEvent(streamKey, "tool_start", toolName); Object result = toolRegistry.invoke(toolName, params); writeEvent(streamKey, "tool_result", JSON.toJSONString(result)); } // 4. 写入结束事件 writeEvent(streamKey, "done", "完成"); } finally { redisTemplate.delete(lockKey); } } private void writeEvent(String streamKey, String type, String content) { redisTemplate.opsForStream().add( StreamRecords.newRecord().in(streamKey) .ofMap(Map.of("type", type, "content", content)) ); // 设置 Stream 过期时间，防止内存泄漏 redisTemplate.expire(streamKey, Duration.ofHours(2)); } }

4.5 前端 EventSource 的正确用法

前端用原生EventSource就够了，不需要额外库：

const es = new EventSource(`/api/agent/stream?sessionId=${sessionId}`); es.addEventListener('token', (e) => { appendToPanel(e.data); }); es.addEventListener('tool_start', (e) => { showToolIndicator(e.data); }); es.addEventListener('done', (e) => { es.close(); }); es.onerror = () => { // EventSource 会自动重连，并带上 Last-Event-ID console.warn('连接中断，正在重连...'); };

注意：EventSource默认会自动重连，重连间隔由服务端retry字段控制。如果不希望自动重连，需要在onerror里手动es.close()。

5. 常见问题与排查技巧实录

5.1 idle timeout 到底是谁掐断了连接

stream disconnected before completion: idle timeout waiting for sse这个报错，90% 的情况是中间层网关干的，不是你的代码。排查顺序：

Nginx 配置示例：

location /api/agent/stream { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_read_timeout 300s; proxy_set_header Connection ''; chunked_transfer_encoding off; }

proxy_buffering off是必须的，否则 Nginx 会缓冲响应，SSE 就变成“攒一批发一批”，失去流式效果。

5.2 Redis command timed out 的连锁反应

redis command timed out; nested exception is io.lettuce.core.RedisCommandTimeoutException这个报错通常出现在 Redis 负载高或者网络抖动时。Lettuce 默认命令超时是 60 秒，太长了，会导致 SSE 推送线程长时间阻塞。

我的处理方案是三层防护：

1. 缩短命令超时：spring.data.redis.timeout=3000ms
2. 开启连接池：避免每次操作都新建连接
3. 降级策略：Redis 不可用时，直接返回错误事件，不要让线程挂死

try { redisTemplate.opsForStream().add(record); } catch (RedisCommandTimeoutException e) { log.error("Redis 写入超时，降级处理", e); // 直接推送到 emitter，不走 Redis fallbackPush(emitter, content); }

5.3 工具调用重复执行的排查

如果发现工具被调了两次，先查三件事：

• 前端是不是触发了两次请求（看 Network 面板）
• 分布式锁的 key 是否包含 sessionId 和 toolName
• 锁的过期时间是否太短，导致第一个请求还没执行完锁就失效了

我一般把工具执行的锁过期时间设为工具预估最大耗时的 2 倍。比如查询类工具预估 5 秒，锁就设 10 秒。

5.4 内存泄漏：Stream 不清理的后果

Redis Stream 如果不设过期，会一直堆积。我见过一个项目跑了三天，Redis 内存从 200MB 涨到 8GB，全是历史会话的 token 事件。

解决方案是双保险：

• 每次写入后EXPIRE设置 2 小时过期
• 定时任务扫描超过 24 小时未活跃的 Stream 并删除

@Scheduled(fixedRate = 3600000) public void cleanupStaleStreams() { Set<String> keys = redisTemplate.keys("agent:stream:*"); for (String key : keys) { Long ttl = redisTemplate.getExpire(key); if (ttl != null && ttl < 0) { redisTemplate.delete(key); } } }

注意：生产环境慎用keys命令，会阻塞 Redis。建议用SCAN替代，或者维护一个活跃会话的 Set 来遍历。

5.5 常见问题速查表

6. 生产级优化与扩展思路

6.1 线程池参数怎么定

SSE 推送是 IO 密集型任务，线程大部分时间在阻塞等待 Redis 数据。核心线程数可以设为CPU 核数 * 2，最大线程数设为CPU 核数 * 10，队列用有界队列防止 OOM：

@Bean("streamExecutor") public ThreadPoolExecutor streamExecutor() { return new ThreadPoolExecutor( 8, 40, 60L, TimeUnit.SECONDS, new LinkedBlockingQueue<>(500), new ThreadFactoryBuilder().setNameFormat("sse-push-%d").build(), new ThreadPoolExecutor.CallerRunsPolicy() ); }

CallerRunsPolicy是兜底策略，队列满了就让调用线程自己执行，至少不会丢任务。

6.2 多实例部署下的会话粘性问题

如果服务部署了多个实例，客户端重连时可能连到另一台机器。因为事件都存在 Redis 里，所以数据不会丢，但要注意：

• 分布式锁必须用 Redis 实现，不能用本地锁
• 心跳检测要基于 Redis 的 key，不能基于本地内存
• 会话状态全部外置到 Redis，服务实例保持无状态

这样任意实例都能处理任意会话，配合负载均衡的轮询策略也没问题。

6.3 工具调用的安全边界

Agent 调用工具是风险最高的环节。我的做法是：

• 白名单机制：只有注册过的工具才能被调用
• 参数校验：每个工具定义自己的参数 schema，调用前校验
• 超时控制：工具执行超过 30 秒强制中断
• 审计日志：每次工具调用记录 sessionId、工具名、参数、结果、耗时

public Object invoke(String toolName, Map<String, Object> params) { Tool tool = toolRegistry.get(toolName); if (tool == null) { throw new IllegalArgumentException("未注册的工具: " + toolName); } tool.validate(params); long start = System.currentTimeMillis(); try { return tool.execute(params); } finally { auditLog.record(sessionId, toolName, params, System.currentTimeMillis() - start); } }

6.4 后续可以怎么扩展

这套中枢目前跑得挺稳，但还有几个方向可以继续打磨。一是把 Redis Stream 换成 Kafka，适合事件量特别大的场景，不过运维成本会上去；二是引入事件回放机制，把整个 Agent 执行过程录下来，方便调试和复盘；三是给 SSE 加上压缩，token 多的时候能省不少带宽。这些等有实际需求了再上，不要为了技术而技术。

我个人在实际操作中的体会是，流式 Agent 最难的不是模型调用，而是状态管理。只要把 Redis 这一层用扎实了，断线重连、工具幂等、多实例部署这些问题都会迎刃而解。最后再分享一个小技巧：调试 SSE 的时候，用curl -N直接看原始流，比在浏览器里看 Network 面板清楚得多，curl -N http://localhost:8080/api/agent/stream?sessionId=test一跑，事件格式对不对一目了然。