★ FEATURED ARTICLE 搞大模型接入的这几年最怕产品经理跑过来跟我说“李工为什么我们问一个问题页面要转圈死等二十秒才把整段话吐出来隔壁竞品都是一个字一个字蹦出来的。”很多刚从传统 CRUD 转过来的兄弟第一反应是在 Controller 里调完ChatClient.call()拿到一整个字符串再通过 JSON 序列化丢给前端。大模型推理不是查单表主键TTFT首字响应时间如果长达几秒甚至十几秒用户早就把页面关了。要把大模型的 Token 生成节奏平滑地透传给终端从最底层的模型提供商接口、Spring AI 框架层、WebFlux 服务层一直到前端浏览器必须建立一条基于响应式背压的非阻塞全双工或单向流管道。今天我们就来拆透基于 Spring AI、Spring WebFlux 与 Server-Sent EventsSSE的全链路流式透传实操把其中的缓冲陷阱、连接管理和异常兜底一次性讲清楚。全链路透传的技术拓扑与核心痛点先看全链路架构[前端浏览器/App] │ EventSource / fetch(ReadableStream) ▼ [Nginx / API Gateway] (必须关闭 response buffering) │ HTTP/1.1 or HTTP/2 ▼ [Spring WebFlux Service] (Spring AI ChatClient.stream()) │ Reactor FluxChatResponse ▼ [LLM Provider API] (OpenAI / 本地 Ollama / 智谱等)很多团队在对接时容易遇到三个典型“假流式”现象服务内全缓冲在 WebFlux Controller 里虽然返回了Flux但是中间用了一个阻塞式方法比如blockFirst()或者误用了collectList()流直接退化成同步大对象。中间件截流代码写得非常规范本地单机测试也是逐字打印但一部署到测试环境走 Nginx 反向代理前端依然是等全部生成完毕后“哗”地一下整屏弹出。原因就是 Nginx 默认开启了 proxy buffering。连接悬挂与取消信号丢失用户问到一半把浏览器标签页关掉了但后端的 Spring AI 还在吭哧吭哧向大模型请求 Token计费不仅一直在扣连接池和线程资源也被空耗。响应式 Controller标准 SSE 封装在 Spring AI 中ChatClient提供了.stream()方法其返回值为FluxChatResponse。在 WebFlux 中暴露 SSE 接口最稳妥的方式是使用 Spring 框架提供的ServerSentEventT包装对象明确声明媒体类型为text/event-stream。看完整的控制器实现package com.yali.ai.controller; import org.springframework.ai.chat.client.ChatClient; import org.springframework.ai.chat.model.ChatResponse; import org.springframework.http.MediaType; import org.springframework.http.codec.ServerSentEvent; import org.springframework.web.bind.annotation.GetMapping; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import reactor.core.publisher.Flux; import reactor.core.publisher.SignalType; import java.time.Duration; import java.util.concurrent.atomic.AtomicInteger; RestController RequestMapping(/api/v1/chat) public class StreamingChatController { private final ChatClient chatClient; public StreamingChatController(ChatClient.Builder chatClientBuilder) { this.chatClient chatClientBuilder.build(); } GetMapping(value /stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxServerSentEventString streamChat(RequestParam(prompt) String prompt) { AtomicInteger eventId new AtomicInteger(0); return chatClient.prompt() .user(prompt) .stream() .chatResponse() .map(response - { String content response.getResult().getOutput().getContent(); // 过滤空内容防止前端解析无意义空包 return content ! null ? content : ; }) .filter(text - !text.isEmpty()) .map(chunk - ServerSentEvent.Stringbuilder() .id(String.valueOf(eventId.incrementAndGet())) .event(message) .data(chunk) .build()) // 增加心跳与超时兜底防止客户端在长思考阶段判定连接死掉 .timeout(Duration.ofSeconds(60)) .doFinally(signalType - { if (signalType SignalType.CANCEL) { System.out.println(客户端主动断开连接及时终止上游 LLM 流订阅); } }) .onErrorResume(e - { System.err.println(流式生成异常: e.getMessage()); return Flux.just(ServerSentEvent.Stringbuilder() .event(error) .data(服务暂时繁忙生成已被中断) .build()); }); } }这里有几个细节值得深究Event 格式契约使用ServerSentEvent.builder()可以规范输出标准的id、event与data。前端如果使用标准的浏览器EventSource可以通过addEventListener(message)和addEventListener(error)区分正常文本与业务异常。取消信号感知doFinally当客户端网络中断或主动调用EventSource.close()时WebFlux 会触发下游的 Cancel 信号。由于 Reactor 是惰性推拉模型Cancel 信号会顺着管道向上游传播通知 Spring AI 底层的 HTTP Client如 WebClient立刻中断与 LLM 服务商的 Socket停止计费。网关与反向代理层的“去缓冲”配置很多同学写完 Java 代码后本地联调一切丝滑一上预发网关就歇菜。根源在于反向代理层。如果你的系统前置了 Nginx必须在反代路由中显式关闭缓冲并调整协议版本location /api/v1/chat/ { proxy_pass http://backend_webflux_cluster; # 核心配置关闭响应缓冲实现即收即发 proxy_buffering off; proxy_cache off; # 保持长连接 proxy_http_version 1.1; proxy_set_header Connection ; proxy_set_header Host $host; # SSE 连接容易长时间空闲调大读写超时 proxy_read_timeout 300s; proxy_send_timeout 300s; # 禁用分块传输编码缓冲 chunked_transfer_encoding on; }如果是使用 Spring Cloud Gateway默认的NettyRoutingFilter在处理FluxDataBuffer时是流式转发的但如果你的全局 Filter 中有对 Request/Response Body 进行解包做签名验签、全量日志打印的切面那恭喜你你已经把流给截断了千万不要在网关拦截器里去DataBufferUtils.join()任何全量读取 Response 的逻辑都会直接把 SSE 变成批处理。前端消耗流式数据的两种姿势前端在对接 SSE 时传统做法是原生EventSource但原生的EventSource默认只支持GET请求且无法在 Header 里携带自定义的 Authorization Token。在实际大模型应用中用户 Prompt 往往很长并且可能附带文件上下文我们推荐使用现代浏览器的fetchAPI 结合ReadableStream来实现流式读取async function fetchStreamAnswer(prompt, token) { const response await fetch(/api/v1/chat/stream?prompt encodeURIComponent(prompt), { headers: { Authorization: Bearer token, Accept: text/event-stream } }); const reader response.body.getReader(); const decoder new TextDecoder(utf-8); let done false; while (!done) { const { value, done: readerDone } await reader.read(); done readerDone; if (value) { const chunk decoder.decode(value, { stream: true }); // 解析 SSE 协议规范中的 data: 内容 const lines chunk.split(\n); for (const line of lines) { if (line.startsWith(data:)) { const content line.replace(/^data:/, ); process.stdout.write(content); // 实时渲染到界面 } } } } }架构师的上线自查清单最后分享一下我带队上线大模型流式接口时的上线检查表连接数上限评估SSE 是长连接Tomcat 这种一个请求霸占一个线程的 Servlet 容器在千人并发下分分钟打爆线程池。必须使用 Netty WebFlux 响应式非阻塞容器单机支撑上万长连接才可能稳住。首字延迟监控在 Metrics 中埋点从收到用户请求到下游 LLM 吐出第一个 chunk 的时间作为系统的核心 SLA 监控项。背压与限流当客户端是老旧手机或弱网环境时消费速度如果远低于大模型吐字速度Netty 的发送缓冲区会被撑爆容易引发 DirectMemory 堆外溢出。背压策略必须明确后续我们会专门开一讲细聊慢客户端的堆外治理。流式架构看似只是“打字机”效果的视觉糖果背后实则是从网络协议、网关中间件到响应式执行栈的全面重构。把每一环的流控和缓冲调教顺畅系统才能在激增的 AI 访问量前稳如磐石。
阅读完成 · 觉得有帮助？
