Spring AI流式输出实战：Flux+SSE从原理到落地，告别聊天等待

项目上线前的一轮联调里测试同事发来一句评价问一句话要转三秒圈等得心慌。当时的模拟项目X还没做流式输出用户的提问走的是一个最朴素的同步链路后端先把请求发给大模型等模型把完整回答生成完一次性打包返回前端再渲染。三秒里既有模型真实生成答案的时间也有我们主动制造的空白等待。后来我把这套流程改成了 Spring AI 的流式输出后端基于 Flux 把模型逐 token 吐出的内容推给浏览器前端按 SSE 协议解析后以打字机形式逐字呈现。同样一份回答页面从干等三秒变成了边等边出字体验差异非常直观。这篇文章就把这条链路从原理到落地、再到上线后踩坑的过程完整记录下来适合正在做 AI 应用集成、想优化聊天体验的后端工程师和全栈开发者。1. 干等三秒到底是谁的锅先算清楚延迟账1.1 大模型吐字是一粒一粒的不是一口气写完的先说一个容易被忽视的事实现在主流的大模型文本生成方式是自回归的。所谓自回归大白话就是每预测一个词都要先看前面已经生成的所有词。token 可以理解成模型输出的最小单位中文场景下大约一到两个 token 对应一个汉字。比如模型生成你好很高兴认识你这句话它在内部并不是一次性给出整句话而是先生成你再看上下文生成好然后继续生成、很……每一步都依赖前一步的结果。也就是说模型本来就是一字一句往外吐的生成过程天然是流式的。问题就在这里模型明明是流式生成的我们的网页却要等它把所有词都吐完再把整段话一次性展示。等于把本来可以渐进呈现的信息人为地压成了一个不可分割的整体。用户看到的不是回答正在生成而是页面卡住了最后突然蹦出一大段文字。这背后的产品逻辑是懒惰的一一早期接入大模型时大家沿用了传统 RPC 接口的思路请求、等待、响应。同步接口写起来最省事Controller 里一行代码拿到 String 或者 ChatResponse 就返回前端也不用处理事件流自然也没人意识到等待体验有多糟糕。1.2 体验账其实很好算首字时间比总耗时重要从用户体验的角度AI 回答的延迟应该拆成两个指标看首字时间TTFTTime To First Token从用户发出请求到页面上出现第一个字的时间。总完成时间整段回答全部显示完毕的时间。同步调用下首字时间几乎等于完整生成时间。一个需要 3 秒生成的回答用户整整 3 秒看不到任何内容也不知道服务到底有没有在干活。流式输出则完全不同首字时间通常能压到 500 毫秒以内甚至在网络顺畅时 300 毫秒就能出第一个字剩下的 2.5 秒用户看到的是文字不断蹦出来视觉上会觉得系统一直在回应。实践中一个 200 字的中文回答大约需要生成 100 到 200 个 token市面上主流模型平台每秒能输出 20 到 60 个 token算下来完整回答大约要 3 到 8 秒。同步接法意味着这 3 到 8 秒全程白屏流式接法则让用户在前 0.5 秒就开始进入阅读状态。算完这笔账之后你去看现在市面上的 AI 对话产品清一色都是打字机效果本质都是在掩盖后端真实生成时间。没人愿意面对一个转圈的空白页大家都希望看到内容在生长。所以流式输出不是炫技而是 AI 应用的基础体验门槛。2. 流式传输的组合拳SSE 协议与 Flux 的职责划分2.1 SSE 轻量又实用WebSocket 在这里是杀鸡用牛刀SSEServer-Sent Events是 HTML5 标准里定义的服务端推送技术它建立在普通 HTTP 连接之上。客户端发起请求后服务端可以在同一个连接里持续向客户端写文本块连接保持打开直到服务端主动关闭或者客户端断开。很多人一听到服务端推送就想到 WebSocket但在 AI 对话场景里WebSocket 是杀鸡用牛刀。WebSocket 是全双工协议支持双向任意消息但需要协议升级握手、自己实现心跳和重连逻辑接入网关和负载均衡时还要额外配置转发策略。而 AI 对话的核心需求其实只有一个方向模型把 token 不断推给浏览器。这是典型的单向持续输出场景SSE 天然合适。SSE 的优势很直接走的就是 HTTP各种网关、代理、负载均衡器基本都认识浏览器原生支持断线自动重连这在高并发网络环境下能省下不少心智负担协议本身非常轻量服务端只要往响应里写 text/event-stream 格式的文本即可不需要维护一个常驻的长连接池。2.2 Flux 在其中的角色一个异步产出元素的管道Flux 是 Project Reactor 中的响应式流类型代表一个可以异步产生 0 到 N 个元素的序列。理解 Flux 最简单的方式是拿水管做类比Flux 是水龙头SSE 是水管模型服务是水源。水龙头一打开水滴token就顺着水管持续送到浏览器水杯里水不用先攒满一桶再倒给你。Spring AI 的流式接口返回的正是 Flux 。每个 ChatResponse 里封装了模型输出的一小段内容以及对应的元信息。Spring Web 框架拿到 Flux 之后会一边从流里取元素一边按 SSE 格式把元素序列化到 HTTP 响应体里。这里有个概念需要弄清楚Flux 是惰性的。调用 stream() 方法只会返回一个流的定义真正开始请求模型、产生数据是在对象被订阅之后才发生的。订阅动作通常由 Web 框架在响应提交阶段触发所以从接口看你只是在方法里 return 一个 Flux背后是被动拉动的数据流。2.3 一条真实 SSE 数据流的解剖理解了角色划分再看一条真实的 SSE 数据就非常清晰。Spring 把 Flux 序列化成 SSE 后响应体大概是这个样式data:{content:你} data:{content:好} data:{content:}关键规则有两条每条消息以data:开头后面跟着按某种格式序列化后的数据消息之间必须用一个空行隔开。前端拿到字节流之后就是靠空行来判断一条消息在哪里结束、下一条从哪里开始。如果有需要还可以把返回类型升级成 FluxServerSentEvent...这样每条消息能携带 event、id、retry 等字段前端可以根据事件名区分开始生成、正在生成、生成完成、发生错误等不同阶段。注意SSE 格式里空行是消息分隔符如果序列化配置有问题把空行吞掉了前端会把多条消息合并成一条解析结果就会错乱。出现这种情况建议先抓原始响应体确认每个data:后面是不是都有空行。3. 后端落地在 Spring AI 里把 Flux 接出来3.1 从 ChatClient 的 stream() 说起Spring AI 的流式输出比很多人想象的简单核心就一句话把同步调用的 call() 换成 stream()。以当前主流版本为基础的写法先注入 ChatClient这个东西相当于一个更高层的对话编排工具。配置好之后代码大概是这样的import reactor.core.publisher.Flux; import org.springframework.ai.chat.ChatClient; import org.springframework.ai.chat.model.ChatResponse; Service public class ChatStreamService { private final ChatClient chatClient; public ChatStreamService(ChatClient.Builder chatClientBuilder) { this.chatClient chatClientBuilder.build(); } public FluxChatResponse streamChat(String message) { return chatClient.prompt() .user(message) .stream() .chatResponse(); } }其中.stream().chatResponse()返回 Flux 表示订阅后每个 token 到达都会触发一次 onNext携带一小段增量内容。如果用更底层的 ChatModel也可以直接调chatModel.stream(message)但不太建议直接从 ChatModel 起步。真正做项目时很快会遇到多轮对话记忆、Prompt 模板、工具调用这些需求ChatClient 在体系上把这些编排功能都收拢了后面扩展时改动的代码量会小很多。3.2 Controller 层两种响应写法以及为什么我推荐统一事件结构服务层拿到 Flux 之后Controller 需要把它暴露成 HTTP 接口。最简单的写法是直接返回 Flux PostMapping(value /chat/stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxChatResponse chatStream(RequestBody ChatRequest request) { return chatService.streamChat(request.message()); }这种写法对前端足够友好浏览器收到的每个 data 就是一条序列化后的 ChatResponse JSON。但生产环境里我更推荐第二种写法先定义一个统一的事件结构再通过 Flux 返回。比如public record StreamEvent(String type, String content) { public static StreamEvent start() { return new StreamEvent(start, ); } public static StreamEvent delta(String content) { return new StreamEvent(delta, content); } public static StreamEvent done() { return new StreamEvent(done, ); } public static StreamEvent error(String message) { return new StreamEvent(error, message); } }PostMapping(value /chat/stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxServerSentEventStreamEvent chatStream(RequestBody ChatRequest request) { return Flux.concat( Flux.just(ServerSentEvent.builder(StreamEvent.start()).event(message).build()), chatService.streamChat(request.message()) .map(resp - ServerSentEvent .builder(StreamEvent.delta(resp.getResult().getOutputText())) .event(message).build()), Flux.just(ServerSentEvent.builder(StreamEvent.done()).event(message).build()) ).onErrorResume(e - Flux.just( ServerSentEvent.builder(StreamEvent.error(e.getMessage())).event(error).build() )); }这样做的核心原因是流式接口的错误处理逻辑和普通接口完全不同。HTTP 状态码只能表达连接建立阶段的错误一旦流已经开始传输中途模型服务超时、上下文超长、参数非法再拿状态码去表达就晚了。最佳实践是在流中间发一个类型为 error 的事件前端收到后再决定是弹提示还是自动重试。统一事件结构能让前端逻辑大幅简化只需要根据 type 分支处理即可。3.3 最容易忽略的那几个参数流式接口跑通只是第一步真正影响生产稳定性的是几个看不见的参数。第一模型采样参数。temperature 控制随机性maxTokens 控制生成长度上限。但要注意不同模型服务商对 maxTokens 的语义解释有细微差异有的把它理解为生成 token 数上限有的则是上下文总长上限。不看文档直接照搬很容易出现回答莫名被截断的情况。我建议上线前用一个长问题把小参数组合都压一遍搞清楚边界。第二HTTP 客户端超时。模型平台的门户超时配置通常默认 60 秒流式长回答很容易超时。合理的做法是把 connectTimeout 设短一点比如 5 秒主要用来快速暴露网络不可达把 readTimeout 设得很长比如 300 秒因为流式模型生成过程中本来就需要长时间等待数据块到达。Spring 的底层 HTTP 客户端配置可以通过对应的 Client 定制器来控制别用全局默认值。第三响应缓冲。这部分单独放到后面第五节详细说因为它是上线时最容易踩的隐形坑。4. 前端打字机把字节流转成人味4.1 选 EventSource 还是 fetch 流式读取后端把流接出来了前端有两条路可以拿到数据。维度EventSourcefetch 流式读取请求方法仅 GET任意方法自定义请求头不支持支持断线重连原生支持需要自己实现识别 SSE 格式自动解析手动解析适用场景简单订阅、无需鉴权AI 对话、带鉴权、POST 请求AI 对话接口通常需要携带较长的用户输入也普遍要有登录态所以实际生产里我基本都用 fetch 流式读取。EventSource 的封闭性太强不支持 POST、不支持自定义 Header这两点足以让它在 AI 场景里出局。它适合的是那种用 GET 就能订阅、没有鉴权头的轻量通知场景。用 fetch 读取 SSE 流的核心代码async function streamChat(message) { const resp await fetch(/chat/stream, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ message }) }); if (!resp.ok || !resp.body) { throw new Error(连接失败${resp.status}); } const reader resp.body.getReader(); const decoder new TextDecoder(utf-8); let buffer ; while (true) { const { done, value } await reader.read(); if (done) break; buffer decoder.decode(value, { stream: true }); buffer handleSSEBlock(buffer); } }4.2 增量渲染的正确姿势这里有个很多新手都会踩的坑fetch 的 ReadableStream 返回的是二进制块这些块和 SSE 消息并不对齐。一条 SSE 消息可能被拆分成多个网络包也可能多个消息挤在同一个网络包里。如果不做粘包处理直接按块解析前端就会经常拿到半截 JSON要么解析失败要么丢数据。我习惯用一个 buffer 变量做粘包先把没解析完的文本暂存下来找到空行\n\n作为一条完整消息的分界点用完后剩下的部分继续留在 buffer 里。function handleSSEBlock(buffer) { const parts buffer.split(\n\n); const remain parts.pop(); // 最后一段可能是不完整的留到下一轮 for (const part of parts) { const dataLine part.split(\n).find(line line.startsWith(data:)); if (!dataLine) continue; const payload dataLine.slice(5).trim(); if (!payload || payload [DONE]) continue; const event JSON.parse(payload); if (event.type delta) { scheduleRender(event.content); } } return remain; }拿到增量内容之后渲染频率也要控制。token 到达频率可以非常高如果每来一个 chunk 就立刻 setState 一次React 会疯狂触发重渲染。尤其对话页面上还挂着一整条历史消息列表diff 成本很高。我习惯用 requestAnimationFrame 合并渲染let pendingText ; let rafId null; function scheduleRender(chunk) { pendingText chunk; if (rafId) return; rafId requestAnimationFrame(() { setDisplayText(prev prev pendingText); pendingText ; rafId null; }); }打字机光标不需要额外的 JS一段 CSS 就能搞定零成本且性能好。在文本末尾放一个跟随光标元素用 border-right 加闪烁动画即可。4.3 中断与重联聊天场景特有的头疼事用户随时可能会点击停止生成。前端停止的关键是调用 reader.cancel()或者用 AbortController 把整个 fetch 中断掉。仅仅不继续渲染是不够的如果底层 TCP 连接还开着后端可能继续调用模型接口钱也会继续烧。重联策略则看产品定位。对于一次性的问答对话网络抖动中断后直接提示用户点击重试从空内容重新开始是最稳妥的。SSE 协议本身支持用 Last-Event-ID 续传但 AI 场景里服务端通常不会保存历史 token 片段实际多半全量重来。与其追求复杂的断点续传不如把重试按钮和重新生成做成同一个入口交互反而更统一。5. 上线前实测四个坑一个都别漏5.1 第一个坑网关和容器超时把流掐断本地联调一切正常一上测试环境流式输出跑了一分多钟就被切断了页面卡住不再出字。查到最后问题出在前置网关的默认读取超时。HTTP 请求开始响应后网关对读取超时的计时方式各有不同如果它是从请求开始算总时长那么长回答必然中招。解决方案要分两层网关层把流式接口的路径纳入超时豁免名单或者设置更长的读超时按 5 分钟起步比较稳妥。服务层确认容器支持异步流式响应。如果用 Spring MVC 而不是 WebFlux返回 Flux 时底层线程模型依然是 Tomcat 的 servlet 线程高峰期多个长连接会各自占住线程所以要评估线程池容量WebFlux 则全程事件驱动压测表现通常更稳。5.2 第二个坑输出缓冲把流憋住了这个坑非常隐蔽表现是页面前几秒一动不动然后突然一次性出现一大段文字打字机效果完全失效。原因是某些容器或中间件默认开了输出缓冲要攒够一定字节数才往客户端刷一次。流式数据量小、间隔短很容易被缓冲住。解法是显式设置响应头关闭缓冲。常见手段是设置X-Accel-Buffering: no并让响应头带上Cache-Control: no-cache防止浏览器或代理把流缓存成整块。定位问题的时候可以从网络面板看 TTFB首字节时间——如果请求发出后几百毫秒内迟迟没有字节下来大概率就是缓冲的锅。5.3 第三个坑客户端断开后服务端还在生成用户中途关掉页面前端连接已经断开但如果后端 Flux 的订阅没有感知到下游关闭模型服务商那头的调用会继续执行费用照扣。解决思路是在流式管线里叠加 doOnCancelpublic FluxChatResponse streamChat(String message) { return chatClient.prompt() .user(message) .stream() .chatResponse() .doOnCancel(() - log.warn(客户端断开停止生成)); }同时配合模型服务商提供的取消接口在 onCancel 回调里主动发起取消请求。需要注意的是doOnCancel 并不保证一定执行如果应用进程被强杀或者网络发生分区TCP 断开可能感知不到。所以更保险的手段是依赖模型平台的 idle 超时机制或者定期发心跳。生产环境里我还加了异常断开次数的统计日志方便后续评估模型调用成本。5.4 第四个坑前端高频刷新导致的低端机卡顿联调环境用的都是开发机性能足够好问题不明显。一旦跑到用户手上的低端安卓机打字机变成了滚筒洗衣机文字滚动一顿一顿的。根因不在流式接口而在前端渲染。优化手段有三板斧一是只更新最后一个消息节点避免整棵消息树 diff二是把流式文本单独拆成轻量组件不要和页面主干耦合三是控制渲染频率用 requestAnimationFrame 限流到每秒 30 帧左右即可满足人眼感知不要每来一个 token 都立刻重渲染一次页面。6. 进阶玩法让流式输出可观测、可控制6.1 在流里做统计首字时间和总耗时流式接口上线之后一定要能拿数据说话。首字时间等于订阅之后第一次 onNext 的时间总耗时等于收到完成信号的时间。Reactor 的钩子可以直接采集public FluxChatResponse streamChatWithMetrics(String message) { long[] startTime {0}; long[] firstTokenTime {0}; return chatClient.prompt() .user(message) .stream() .chatResponse() .doOnSubscribe(s - startTime[0] System.currentTimeMillis()) .doOnNext(resp - { if (firstTokenTime[0] 0) { firstTokenTime[0] System.currentTimeMillis(); log.info(首字时间: {}ms, firstTokenTime[0] - startTime[0]); } }) .doOnComplete(() - log.info( 流式回答完成总耗时: {}ms, System.currentTimeMillis() - startTime[0])); }把这些指标上报到监控平台之后你才会真正看到不同模型服务商在不同时段的延迟表现。实测中有些服务商在高峰期的首字时间会从 300ms 飙到 2 秒没有统计的话这些劣化根本发现不了。6.2 背压、限流与并发控制Flux 支持背压订阅方可以通过控制请求速率来调节拉取速度。但在模型服务场景下token 的生成速度由服务端决定前端再节流也改变不了生成节奏所以背压在 AI 对话里更多是防止下游处理不过来的手段。真正要防的是同一路请求把模型服务打满因此接入层要做并发限制。可以用 Semaphore 做一个简单的限流闸口private final Semaphore chatSlots new Semaphore(10); public FluxChatResponse guardedChat(String message) { if (!chatSlots.tryAcquire()) { return Flux.error(new RuntimeException(当前会话繁忙请稍后再试)); } return chatService.streamChat(message) .doFinally(signal - chatSlots.release()); }Semaphore 方案的好处是简单直接坏处是释放动作依赖 doFinally一旦出现不可控异常信号量可能泄漏。所以要在整体上再加一层兜底超时超过一定时间强制释放连接和信号量避免并发额度被慢慢耗尽。6.3 给每个流一个身份链路追踪排查流式问题的最大痛点是日志对不上。用户反馈回答到一半断了后端日志里面对应的可能是几百条零散的 token 输出根本不知道哪条属于哪次会话。建议从接口入口生成一个会话 ID用日志框架的 MDC 透传到整条链路的日志里。同时在 Flux 的每个 onNext 里把累计字符数和 token 数打出来日志大致长这样[conversation8f3a21, stream1] 已输出 128 字符累计 token 计数 96 [conversation8f3a21, stream1] 已输出 140 字符累计 token 计数 105一旦用户报障直接按会话 ID 过滤日志立刻能看到这个流是从哪个 token 开始变得异常的排查效率完全不在一个数量级。最后说回开头那次测试。改完流式输出之后同样的模型、同样的生成时间页面在 0.4 秒左右就开始出字了测试同事再也没提过干等三秒。这套链路走下来我最大的体会是流式输出不只是接口层把 call 换成 stream 那么简单它是一整套从传输协议、响应式管线、前端渲染到可观测性的协同改造。把这套链路打通之后你焦虑的不再是给用户一个完整答案而是让用户看到一个正在思考的过程。