★ FEATURED ARTICLE 1. 从轮询到流式Agent 交互体验的分水岭做过 AI Agent 应用的人大概都有过这种体验前端发一个问题过去页面转圈转十几秒然后啪一下把整段回答甩出来。用户盯着空白屏幕的那十几秒心里其实已经在怀疑这个产品是不是卡死了。更难受的是工具调用场景——Agent 需要先思考、再决定调哪个工具、等工具返回、再继续思考这一整套链路如果只能等全部跑完才返回那用户体验基本就是灾难。传统的做法是前端定时轮询后端接口问结果好了没好了没还没好吗。这种方案能跑通但问题很明显轮询间隔设短了请求量爆炸后端压力大设长了用户感知延迟高。而且每次轮询都是一次完整的 HTTP 往返哪怕后端什么都没产出这个开销也省不掉。更关键的是轮询天然无法表达流的语义——你没法告诉用户我现在正在思考第二步我正在调用搜索工具只能给一个笼统的处理中。SSEServer-Sent Events就是来解决这个问题的。它基于 HTTP 长连接服务端可以主动、单向地往客户端推送事件流浏览器原生支持EventSource断线还能自动重连。对于 Agent 这种服务端持续产出、客户端只负责展示的场景SSE 的契合度远高于轮询也比 WebSocket 轻量得多——毕竟 Agent 交互大多是单向推送不需要双向实时通信那么重的协议。但光有 SSE 还不够。生产环境的 Agent 有几个绕不开的硬骨头多实例部署时用户这次请求打到 A 实例下次重连可能打到 B 实例上下文怎么续Agent 思考到一半服务重启了已经推送的内容怎么恢复多个用户并发跑 Agent怎么保证每个流的事件不串台这些问题的答案都指向同一个组件——Redis。这篇内容就是围绕Spring Boot 3 SSE Redis这套组合把 Agent 流式思考与工具调用中枢从零搭起来。我会讲清楚每个技术选型背后的理由、SSE 在 Spring Boot 3 里的正确打开方式、Redis 在流式场景里到底承担哪些职责、工具调用的流式编排怎么做以及我在实际压测和线上运行中踩过的那些坑。适合已经写过基础 Spring Boot 接口、想往 Agent 应用方向深入的开发者也适合正在被轮询方案折磨、想换流式架构的团队参考。2. 为什么是 SSE 而不是 WebSocket 或轮询2.1 三种方案在 Agent 场景下的真实对比选型这件事最怕的就是别人用啥我用啥。我把轮询、WebSocket、SSE 三种方案放到 Agent 流式这个具体场景里从几个维度做了对比结论其实挺清晰的。维度轮询WebSocketSSE通信方向客户端拉双向服务端单向推协议开销每次完整 HTTP握手后帧协议单次 HTTP 长连接断线重连天然支持需自己实现浏览器自动重连服务端实现复杂度低高中多实例扩展无状态简单需粘性会话或网关需外部存储续传适合 Agent 流式勉强过度设计高度契合Agent 的核心交互模式是用户发一次请求服务端持续产出思考片段、工具调用事件、最终答案。这是典型的单向流。WebSocket 的双向能力在这里用不上反而要付出握手、心跳、连接管理的额外成本。轮询则是用反复问模拟持续推本质上是逆着场景来的。SSE 的定位刚好卡在中间比轮询高效比 WebSocket 轻。它就是一个Content-Type: text/event-stream的 HTTP 响应服务端不关闭连接持续往里写data:格式的文本块。浏览器端new EventSource(url)一挂onmessage就能收到推送。2.2 SSE 的协议细节别被简单骗了SSE 看着简单但协议里有几个细节不搞清楚线上一定出问题。事件流的标准格式是这样的event: thinking data: {step:1,content:正在分析用户意图} event: tool_call data: {tool:search,args:{q:Spring Boot 3 SSE}} event: answer data: {content:Spring Boot 3 中 SSE 的推荐实现方式是...}几个关键点每条消息以两个换行\n\n结尾这是消息分隔符。少一个换行客户端就认为消息还没结束会一直等。event:字段自定义事件类型客户端可以用addEventListener(thinking, ...)分别监听。不写event:则默认是message事件。data:可以多行多行会被拼接。但实际用的时候建议一行一个 JSON省得解析麻烦。id:字段用于断线续传客户端重连时会带上Last-Event-ID请求头服务端可以据此从上次的位置继续推。这个字段是 SSE 断线恢复的关键后面讲 Redis 续传时会重点用到。retry:字段告诉浏览器重连间隔默认 3 秒。注意SSE 默认有连接数限制HTTP/1.1 下同域名最多 6 个并发连接浏览器会把这个限制算进去。如果你的页面同时开多个 SSE很容易被卡住。生产环境建议走 HTTP/2多路复用就没这个问题了。2.3 Spring Boot 3 里 SSE 的两种实现路径Spring Boot 3 基于 Spring Framework 6对响应式支持更完善了。实现 SSE 有两条路路径一Spring MVC 的SseEmitter。这是最直接的方式返回一个SseEmitter对象通过它的send()方法推送事件。适合传统阻塞式代码和现有 Spring MVC 项目无缝集成。路径二Spring WebFlux 的FluxServerSentEvent。响应式流适合已经用 WebFlux 的项目背压处理更优雅。考虑到大多数 Agent 项目还是基于 Spring MVC 的毕竟要调各种阻塞式的工具、数据库、HTTP 客户端我选SseEmitter作为主线。它虽然不如 WebFlux 优雅但胜在直观、好调试、和现有生态兼容。一个最小的SseEmitter用法长这样GetMapping(value /stream/{taskId}, produces MediaType.TEXT_EVENT_STREAM_VALUE) public SseEmitter stream(PathVariable String taskId) { SseEmitter emitter new SseEmitter(0L); // 0 表示不超时 executor.execute(() - { try { emitter.send(SseEmitter.event().name(thinking).data(开始思考)); // ... 业务逻辑 emitter.complete(); } catch (IOException e) { emitter.completeWithError(e); } }); return emitter; }new SseEmitter(0L)里的0L表示永不超时。这个参数很关键——Agent 跑一个复杂任务可能几十秒甚至几分钟默认超时时间Tomcat 默认 30 秒左右根本不够必须显式设成 0 或者一个足够大的值。3. Redis 在流式中枢里到底扛了哪些活很多人以为 SSE 就是服务端推、客户端收加 Redis 是多余的。单机单用户确实如此但一旦上生产Redis 就从可选变成必需。它在 Agent 流式中枢里至少承担四个职责每一个都对应一个真实的线上问题。3.1 事件缓冲解决推得比消费快和重连丢事件SSE 推送是即时的但客户端消费不一定跟得上。网络抖动、客户端处理慢、页面切到后台都可能导致事件丢失。更麻烦的是断线重连——用户网络闪断一下重连上来中间那几秒的思考事件就没了用户看到的就是跳了一段。解法是双写每次emitter.send()的同时把事件追加到 Redis 的一个 List 或 Stream 里key 用agent:stream:{taskId}。客户端重连时带上Last-Event-ID服务端从 Redis 里把该 ID 之后的事件全部补推一遍。这里用Redis Stream比 List 更合适因为 Stream 天然支持按 ID 范围读取XRANGE而且有消费者组机制。List 虽然也能用LRANGE但 ID 语义要自己维护麻烦。// 推送事件时双写 public void emit(String taskId, String eventName, Object payload) { String eventId String.valueOf(System.currentTimeMillis()); SseEvent event SseEvent.builder() .id(eventId).name(eventName).data(toJson(payload)).build(); // 写 Redis Stream redisTemplate.opsForStream().add(agent:stream: taskId, Map.of(id, eventId, name, eventName, data, toJson(payload))); // 推给客户端 emitter.send(SseEmitter.event().id(eventId).name(eventName).data(toJson(payload))); }重连补推的逻辑就是读 Streampublic void replay(String taskId, String lastEventId, SseEmitter emitter) { ListMapRecordString, Object, Object records redisTemplate.opsForStream() .range(agent:stream: taskId, Range.rightOpen(0-0, lastEventId -0)); for (var record : records) { // 重新推给客户端 } }提示Stream 的 ID 格式是时间戳-序号Last-Event-ID如果直接用时间戳补推时要拼上-0才能正确匹配范围。这个细节不注意补推会漏掉同一毫秒内的事件。3.2 分布式锁防止同一个任务被重复执行Agent 任务通常是有状态的一个taskId对应一次完整的思考链路。如果用户手抖点了两次或者前端重试机制触发同一个任务可能被两个线程同时跑结果就是事件重复推送、工具被调用两次、状态错乱。Redis 分布式锁在这里就是守门员。任务启动前先抢锁抢到才执行抢不到直接返回任务进行中。public boolean tryLock(String taskId, long ttlSeconds) { Boolean ok redisTemplate.opsForValue() .setIfAbsent(agent:lock: taskId, 1, Duration.ofSeconds(ttlSeconds)); return Boolean.TRUE.equals(ok); }锁的 TTL 要设得比任务最长执行时间略大防止任务还没跑完锁就过期了。但也不能太大否则任务异常退出后锁迟迟不释放。我的经验是设成预估最长耗时 × 1.5同时配合一个看门狗线程定期续期。释放锁的时候千万别直接DEL要校验是不是自己持有的否则可能误删别人的锁。用 Lua 脚本保证原子性private static final String UNLOCK_LUA if redis.call(get, KEYS[1]) ARGV[1] then return redis.call(del, KEYS[1]) else return 0 end;3.3 状态快照让 Agent 的记忆跨实例存活Agent 的思考是有上下文的它记得前面聊了什么、调过哪些工具、拿到了什么结果。这些状态如果只放在 JVM 内存里多实例部署就废了——用户重连打到另一个实例Agent 直接失忆。把 Agent 的会话状态序列化后存 Rediskey 用agent:context:{sessionId}每次思考完一个步骤就更新一次。这样任何实例都能加载到完整上下文继续往下跑。这里有个序列化选型的坑。默认的 JDK 序列化又慢又占空间强烈建议换成 JSONJackson或者更紧凑的二进制格式。如果上下文里有复杂对象Jackson 需要配好ObjectMapper的多态类型处理否则反序列化会丢类型信息。Bean public RedisTemplateString, Object redisTemplate(RedisConnectionFactory factory) { RedisTemplateString, Object template new RedisTemplate(); template.setConnectionFactory(factory); Jackson2JsonRedisSerializerObject serializer new Jackson2JsonRedisSerializer(Object.class); ObjectMapper mapper new ObjectMapper(); mapper.activateDefaultTyping( LaissezFaireSubTypeValidator.instance, ObjectMapper.DefaultTyping.NON_FINAL); serializer.setObjectMapper(mapper); template.setValueSerializer(serializer); template.setHashValueSerializer(serializer); template.setKeySerializer(new StringRedisSerializer()); return template; }3.4 限流与并发控制别让一个用户拖垮整个服务Agent 任务通常很重——要调大模型、要跑工具、要占连接。如果不限流一个用户开十个任务或者恶意刷接口整个服务就挂了。Redis 的计数器 过期时间是实现限流最轻量的方式。比如限制每个用户同时最多 3 个活跃任务public boolean acquireSlot(String userId, int maxConcurrent) { String key agent:slots: userId; Long current redisTemplate.opsForValue().increment(key); if (current 1) { redisTemplate.expire(key, Duration.ofMinutes(10)); } if (current maxConcurrent) { redisTemplate.opsForValue().decrement(key); return false; } return true; }任务结束时decrement释放槽位。这个方案简单有效缺点是如果任务异常退出没释放槽位会一直占着所以必须配合 TTL 兜底。4. 流式思考与工具调用的编排实战前面讲的是基础设施这一节进入 Agent 的核心——怎么把思考和工具调用编排成一条流畅的事件流。4.1 事件模型设计让前端知道现在在干嘛Agent 跑一个任务中间会经历很多阶段。如果只推一种message事件前端没法区分这是思考过程还是这是最终答案展示效果就很粗糙。我的做法是定义一套事件类型每种类型对应前端的一种展示形态。事件类型含义前端展示thinking模型正在推理打字机效果的思考气泡tool_call决定调用某工具工具卡片显示工具名和参数tool_result工具返回结果工具卡片内展开结果answer最终答案片段主回答区域流式追加error出错红色错误提示done任务结束关闭 loading 状态这套事件模型的好处是前端可以针对每种事件做精细化的 UI 处理。比如tool_call事件可以让工具卡片先转圈等tool_result到了再展示内容用户能清楚看到 Agent 在干什么。事件的数据结构建议统一成{ id: 1718000000000, type: tool_call, timestamp: 1718000000000, payload: { toolName: web_search, args: {query: Spring Boot 3 SSE}, callId: call_abc123 } }callId很关键它把tool_call和对应的tool_result关联起来。Agent 可能并行调多个工具没有callId前端就分不清哪个结果对应哪个调用。4.2 思考链路的流式产出逐 token 还是逐步骤大模型的输出是逐 token 的但 Agent 的思考通常分两个层次模型内部的推理 token 流和Agent 编排层面的步骤流。逐 token 推送的好处是打字机效果拉满用户感觉响应很快。坏处是事件量巨大一个几百字的回答可能产生上千个事件Redis 写入和网络传输压力都不小。我的折中方案是分层推送思考阶段按步骤推一个思考步骤一个thinking事件不逐 token。最终答案阶段逐 token 或按小批次推保证打字机效果。这样既控制了事件量又保住了最终输出的流畅感。如果模型支持流式返回最终答案阶段可以直接把模型的 token 流转发出去中间不做缓冲。// 最终答案阶段转发模型 token 流 modelClient.stream(prompt, token - { emit(taskId, answer, Map.of(content, token)); });4.3 工具调用的流式编排串行、并行与嵌套工具调用是 Agent 最复杂的部分。简单场景是串行——调一个工具等结果再调下一个。复杂场景是并行——同时调多个工具谁先回来先处理谁。更复杂的是嵌套——工具本身又是个 Agent内部还有自己的工具调用。串行编排最直观用循环就能实现for (ToolCall call : plan.getToolCalls()) { emit(taskId, tool_call, call); ToolResult result toolExecutor.execute(call); emit(taskId, tool_result, result); context.add(result); }并行编排要用CompletableFuture或者响应式流。关键是每个工具的结果回来时要能正确关联到对应的callId并且所有工具都完成后才能进入下一步。ListCompletableFutureVoid futures plan.getToolCalls().stream() .map(call - CompletableFuture.runAsync(() - { emit(taskId, tool_call, call); ToolResult result toolExecutor.execute(call); emit(taskId, tool_result, result); }, executor)) .toList(); CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();注意并行工具调用时多个线程同时往同一个SseEmitter写数据SseEmitter本身不是线程安全的。必须加锁或者用一个单线程的事件队列串行化写入。我踩过这个坑——并发写导致事件内容交错前端解析直接崩了。嵌套编排的处理原则是事件透传 前缀隔离。子 Agent 的事件推上来时加一个parentCallId字段前端据此做层级展示。同时子 Agent 的事件 ID 要加前缀避免和父 Agent 的 ID 冲突。4.4 心跳与超时让长连接活着SSE 长连接最怕的就是中间有代理或网关空闲一段时间后连接被悄悄掐断而两端都不知道。表现就是用户看着看着流突然不动了。解法是定期发心跳。每隔 15-30 秒推一个注释行以:开头或者一个ping事件保持连接活跃。ScheduledExecutorService heartbeat Executors.newSingleThreadScheduledExecutor(); heartbeat.scheduleAtFixedRate(() - { try { emitter.send(SseEmitter.event().comment(keepalive)); } catch (IOException e) { // 连接已断清理资源 } }, 0, 20, TimeUnit.SECONDS);心跳间隔要小于网关的空闲超时时间。常见的 Nginx 默认proxy_read_timeout是 60 秒所以心跳设 20-30 秒比较稳妥。同时 Nginx 那边也要配proxy_buffering off否则 SSE 的数据会被 Nginx 缓冲推不出去。5. 生产环境踩过的坑与调优经验这一节是我最想写的部分。前面讲的都是应该怎么做这里讲的是实际做的时候会怎么翻车。5.1 SseEmitter 的线程安全与资源泄漏SseEmitter不是线程安全的这个前面提过。但更隐蔽的问题是资源泄漏。每个SseEmitter底层对应一个 HTTP 连接如果任务异常结束没有调用complete()或completeWithError()这个连接就一直挂着时间长了连接池耗尽新请求全部阻塞。我的做法是给每个SseEmitter注册回调确保任何路径下都能正确关闭emitter.onCompletion(() - cleanup(taskId)); emitter.onTimeout(() - { emit(taskId, error, Map.of(msg, 连接超时)); cleanup(taskId); }); emitter.onError(e - cleanup(taskId));cleanup里要做的事包括释放分布式锁、删除 Redis 里的临时状态、从活跃任务表里移除、关闭心跳线程。任何一项漏了都会在长时间运行后暴露问题。5.2 Redis 大 key 与内存膨胀Agent 的事件流如果全量存 Redis一个跑几分钟的任务可能产生几千条事件每条几百字节加起来就是几 MB。如果并发任务多Redis 内存涨得飞快。几个治理手段设置 Stream 的最大长度用XADD ... MAXLEN ~ 1000限制老事件自动淘汰。反正客户端重连补推只需要最近的事件。任务完成后延迟删除给客户端留一个补推窗口比如 5 分钟后再删。用EXPIRE设置 TTL 就行。大 payload 拆分存储工具返回的大结果比如网页全文不要直接塞进事件里存到单独的 key事件里只放引用 ID。// 限制 Stream 长度 redisTemplate.opsForStream().add( StreamRecords.mapBacked(record).withStreamKey(key), XAddOptions.maxlen(1000).approximateTrimming(true));5.3 断线重连的边界情况断线重连看着简单实际边界情况一堆重连时任务已经结束客户端带着Last-Event-ID重连但任务早就complete了。这时候服务端要能识别出来直接把剩余事件补推完然后发一个done事件关闭连接而不是傻等。重连时任务还没开始用户刷新页面任务 ID 还在但服务端还没开始跑。这时候要返回一个任务不存在或未开始的事件让前端重新发起。Last-Event-ID 格式不对客户端可能传个乱七八糟的值过来服务端解析时要容错解析失败就当从头补推。补推事件太多如果断线时间很长积压的事件可能上千条一次性推完会阻塞。要分批推每批之间Thread.sleep一下给客户端喘息时间。5.4 压测数据与容量规划最后分享一组我实测的数据供容量规划参考。单实例4 核 8GTomcat 默认线程池下指标数值稳定并发 SSE 连接数约 800-1000单连接事件推送频率上限约 50 事件/秒单任务平均事件数80-150 条Redis 单任务内存占用约 200KB-1MB端到端延迟首事件200-500ms超过 1000 并发连接后Tomcat 的线程模型开始吃力建议上响应式WebFlux或者加实例。Redis 这边单实例扛几万 QPS 没问题瓶颈通常在网络带宽和序列化开销上。还有一个容易被忽略的点SSE 连接会占用 Tomcat 的工作线程。如果用SseEmitter配合阻塞式业务代码每个连接会占一个线程直到任务结束。1000 个并发连接就是 1000 个线程内存和上下文切换开销都很大。解法是把业务逻辑丢到独立的线程池执行SseEmitter的请求线程尽早释放。或者干脆上 WebFlux用事件循环模型几千连接也就几个线程。我在实际项目里最后选的是Spring MVC 独立业务线程池 SseEmitter的组合兼顾了开发效率和资源利用率。WebFlux 虽然更省资源但和现有阻塞式工具链的整合成本太高不划算。这套方案上线跑了几个月日均几十万次 Agent 调用稳定性没问题。唯一需要持续关注的是 Redis 内存和 SSE 连接的清理这两个地方一旦有泄漏都是慢性的、累积性的问题等发现的时候往往已经晚了。
阅读完成 · 觉得有帮助？
