AI 每日学习 — Spring Boot 中基于 SSE 的大模型流式响应：从协议到生产落地

2026-07-15 每日技术学习

今日主题

Spring Boot 中基于 SSE 的大模型流式响应：从协议到生产落地

所属领域：Java AI / 网络

为什么后端架构师需要掌握它：
流式输出（Streaming）是所有 Chat 类、AI 助手、代码助手产品的标配体验。从协议层面（HTTP/1.1 chunked → SSE → WebSocket），到 Java 侧（Servlet 3.1 async → WebFlux → WebClient），再到生产问题（首 Token 延迟、背压、用户断网、超时控制、Token 计量、可观测性），任何一个细节没处理好都会让 AI 产品的体验从"惊艳"变成"难用"。这是把"调用大模型 API"变成"生产级 AI 应用"的第一道工程门槛。

1. 核心问题

学完后需要回答以下 5 个问题：

1. 为什么 LLM 推理必须用流式响应？一次性返回和流式返回在体验、成本、架构上有什么本质差异？
2. SSE（Server-Sent Events）和 WebSocket 在大模型场景下分别适合什么场景？为什么大多数 Chat 产品默认选 SSE？
3. Spring MVC（Servlet 3.1 async）和 Spring WebFlux 两种栈实现 SSE 有什么区别？在团队技术栈统一时该如何选择？
4. 流式响应中的"半截响应"“用户断网”“服务端超时”"Token 超限"这些异常如何在 Java 侧被优雅处理？
5. 流式场景下，Token 计量、日志追踪、限流、熔断该怎么设计？和普通 HTTP 接口的差异在哪里？

2. 前置知识

理解本主题只需掌握以下基础，本文不再展开：

• HTTP/1.1 基础：请求/响应模型、Content-Type、Transfer-Encoding: chunked。
• Servlet 3.1 异步处理：AsyncContext、startAsync()的基本概念（不要求精通）。
• Spring Boot Web：知道@RestController、@GetMapping即可。
• 大模型 API 基本概念：知道 Chat Completion 有stream=true参数即可，不需要了解训练原理。
• 响应式编程基础：理解 Mono/Flux 是"数据流"而不是"集合"即可。

3. 原理讲解

3.1 为什么必须流式？

大模型推理是逐 Token 自回归生成：每生成一个 Token，都依赖之前所有 Token。70B 级别模型生成 1000 字回答，可能需要 5~15 秒。

如果走"一次性返回"：

客户端 ──请求──> 服务端 ──[5~15 秒沉默]──> 返回完整 JSON

• 用户在前 5 秒看不到任何东西，感知延迟 = 总耗时。
• 一旦网络中断，前面 12 秒的等待全部作废。
• 服务端必须缓存完整结果，内存压力 = 并发数 × 单次响应大小。
• 无法做"打字机效果"，体验差。

如果走"流式返回"：

客户端 ──请求──> 服务端 ──data: xxxx\n\n──> data: yyyy\n\n──> ... ──> data: [DONE]\n\n ↑ 首 Token 延迟通常 200~800ms

• 用户几乎立刻看到第一个字，感知延迟 ≈ 首 Token 延迟。
• 中途断网只损失未送达部分，可重连续传（SSE 协议支持Last-Event-ID，但 LLM 场景一般不续传，而是允许重发）。
• 服务端边生成边推送，内存占用低。
• 支持"打字机效果"，符合人类对话习惯。

3.2 SSE vs WebSocket

架构师结论：Chat 类、AI 助手、代码补全 99% 用 SSE 就够了。只有"AI 协作白板""语音实时对话"这种双向高频才需要 WebSocket。

3.3 SSE 协议格式

SSE 非常简洁，本质是带特殊前缀的 chunked 响应：

HTTP/1.1 200 OK Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive X-Accel-Buffering: no data: {"choices":[{"delta":{"content":"你"}}]} data: {"choices":[{"delta":{"content":"好"}}]} data: {"choices":[{"delta":{"content":"，"}}]} data: [DONE]

要点：

• 每个事件以data:开头，两个换行（\n\n）表示事件结束。
• data: [DONE]是 OpenAI 风格的结束标记（不同厂商可能不同）。
• X-Accel-Buffering: no是告诉 Nginx不要缓冲，否则客户端要等缓冲区满才看到内容。
• Cache-Control: no-cache防止 CDN 缓存。
• Content-Type 必须是text/event-stream。

3.4 Java 侧实现路线

在 Spring Boot 中实现 SSE 有两条路线：

路线 A：Spring MVC + Servlet 3.1 异步（推荐用于已有 Spring MVC 项目）

• 使用SseEmitter，Spring 内置，零额外依赖。
• 兼容团队现有技术栈、线程模型、过滤器链路。
• 适合从 0 到 1 接入 LLM。

路线 B：Spring WebFlux（推荐用于新建项目或高并发 AI 网关）

• 使用Flux<ServerSentEvent<String>>返回值。
• 天然背压、线程开销低、适合高并发。
• 学习曲线较陡，团队需要熟悉响应式编程。

架构选择标准：

• 团队没有响应式经验 →选 A。
• 系统已经是 WebFlux 栈 → 选 B。
• 并发要求 > 5000 持续流式连接 → 认真评估 B，并压测验证。
• 混合场景（同一项目里既有传统接口又有流式接口）→ 选 A，统一 MVC 即可。

3.5 容易踩的坑

1. Nginx 缓冲：默认proxy_buffering on，会把 SSE 攒满 4KB 才发给客户端 → 关闭：proxy_buffering off;。
2. 反向代理超时：默认proxy_read_timeout 60s，流式响应经常超时 → 改为300s或更大。
3. Spring 线程池耗尽：Tomcat 默认最大 200 线程，每个 SSE 连接占一个线程直到结束 → 必须配置spring.mvc.async.request-timeout和合理的线程池。
4. 客户端断网不感知：浏览器关闭页面时，SSE 连接不会立即断开，需要心跳保活。
5. 首 Token 延迟被埋：总耗时看着正常，但首 Token 1 秒后才到，原因是 prompt 太大或服务端在拼上下文。

4. Java 工程实践

4.1 完整代码示例（Spring MVC + SseEmitter + WebClient）

Maven 依赖（Spring Boot 3.2+）：

<dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-webflux</artifactId><!-- 用 WebClient --></dependency>

Controller：把大模型流式响应原样转发给前端

@RestController@RequestMapping("/ai")@RequiredArgsConstructorpublicclassChatController{privatefinalChatStreamServicechatStreamService;/** * 客户端（浏览器/APP）调用此接口 * 接收：POST application/json，返回：text/event-stream */@PostMapping(value="/chat/stream",produces=MediaType.TEXT_EVENT_STREAM_VALUE)publicSseEmitterstreamChat(@RequestBodyChatRequestrequest,@RequestHeader(value="X-User-Id")StringuserId){// 1. 设置超时（5 分钟，覆盖最长推理时间）SseEmitteremitter=newSseEmitter(5*60*1000L);// 2. 客户端断开时清理资源（关键！）emitter.onCompletion(()->log.info("SSE completed, userId={}",userId));emitter.onTimeout(()->{log.warn("SSE timeout, userId={}",userId);emitter.complete();});emitter.onError(ex->log.error("SSE error, userId={}",userId,ex));// 3. 异步调用大模型并推送chatStreamService.streamToEmitter(request,userId,emitter);returnemitter;}}

Service：调用 OpenAI 兼容协议并流式推送

@Service@RequiredArgsConstructorpublicclassChatStreamService{privatefinalWebClientopenAIClient;// 注入配置好的 WebClientprivatefinalTokenMetertokenMeter;// 你的 Token 计量组件privatefinalChatHistoryRepositoryhistoryRepository;// 持久化对话历史publicvoidstreamToEmitter(ChatRequestrequest,StringuserId,SseEmitteremitter){// 1. 构造大模型请求Map<String,Object>body=Map.of("model",request.getModel(),"stream",true,"messages",request.getMessages());// 2. 累积完整回答（用于持久化 & Token 统计）StringBuilderfullAnswer=newStringBuilder();// 3. 订阅大模型的 Flux<String>（每个元素是一个 SSE data 行）openAIClient.post().uri("/v1/chat/completions").bodyValue(body).retrieve().bodyToFlux(String.class).filter(line->line.startsWith("data:")).map(line->line.substring(5).trim()).filter(data->!"[DONE]".equals(data)).doOnNext(data->{try{// 3.1 解析 delta.contentJsonNodedelta=objectMapper.readTree(data).path("choices").path(0).path("delta").path("content");if(!delta.isMissingNode()&&!delta.isNull()){Stringtoken=delta.asText();fullAnswer.append(token);// 3.2 推送到前端emitter.send(SseEmitter.event().name("message").data(token));}}catch(Exceptione){log.error("Failed to parse stream chunk",e);emitter.completeWithError(e);}}).doOnComplete(()->{// 4. 完整响应结束：持久化 + Token 计量try{emitter.send(SseEmitter.event().name("done").data(""));emitter.complete();historyRepository.save(userId,request,fullAnswer.toString());tokenMeter.record(userId,request.getModel(),fullAnswer.length()/2);// 粗估}catch(Exceptione){log.error("Failed to finalize stream",e);}}).doOnError(ex->{log.error("LLM stream error, userId={}",userId,ex);try{emitter.send(SseEmitter.event().name("error").data("AI 服务暂时不可用"));emitter.completeWithError(ex);}catch(IOExceptionignored){}}).subscribe();// 必须订阅才会触发}}

WebClient 配置（重要！）：

@ConfigurationpublicclassWebClientConfig{@BeanpublicWebClientopenAIClient(@Value("${llm.base-url}")StringbaseUrl,@Value("${llm.api-key}")StringapiKey){returnWebClient.builder().baseUrl(baseUrl).defaultHeader(HttpHeaders.AUTHORIZATION,"Bearer "+apiKey).clientConnector(newReactorClientHttpConnector(HttpClient.create().responseTimeout(Duration.ofMinutes(5))// 推理可能很久.keepAlive(true))).codecs(c->c.defaultCodecs().maxInMemorySize(1024*1024)).build();}}

4.2 关键配置项

4.3 推荐的异常处理与日志

• traceId 贯穿：用 MDC 把traceId注入到大模型请求和大模型响应日志，方便排查。
• 分级日志：首 Token 延迟、每 100 Token 一次进度日志、结束时的总耗时和 Token 数。
• 结构化日志：JSON 格式，包含userId、model、promptTokens、completionTokens、durationMs、firstTokenLatencyMs。
• 客户端断网：通过 SseEmitter 的onCompletion/onError回调清理大模型侧的连接（调用其 cancel）。
• 超时分层：连接超时（10s）、首 Token 超时（30s）、整体超时（5min），三者必须独立设置。

5. 生产环境案例

案例：AI 客服上线首日，“打字机效果"变成"整段刷新”

业务背景：某电商接入 GPT-4 做智能客服，前端是 Web 页面。

问题表现：

• 用户输入问题后，页面空白 8~10 秒，然后突然一次性显示整段回答。
• 部分用户反馈"页面卡死"。
• 服务端 CPU 不高，但 Tomcat 线程数被打满。

可能原因：

1. Nginx 开了proxy_buffering，把 SSE 攒满 4KB 才下发。
2. Tomcat 线程被打满，导致后续请求排队。
3. 大模型本身不是流式调用（忘记传stream=true）。

排查步骤：

1. 直接curl -N调用 AI 接口 → 发现数据是分批返回的 → 排除大模型侧。
2. 浏览器 F12 看 Network → 发现响应头是Transfer-Encoding: chunked但 content 不是text/event-stream→SseEmitter 没生效？
3. 仔细看代码，发现 Controller 写的是produces = MediaType.APPLICATION_JSON_VALUE→SSE 退化成普通 chunked JSON 数组。
4. 同时检查 Nginx 配置 →proxy_buffering on是默认开启的。
5. 检查 Tomcat 线程数 → 200 个线程被 200 个未释放的 SSE 连接占满。

解决方案：

1. 修正produces = MediaType.TEXT_EVENT_STREAM_VALUE。
2. Nginx 关闭缓冲：proxy_buffering off; proxy_read_timeout 300s;。
3. Tomcat 线程数从 200 提到 500。
4. 加入心跳：每 15 秒发一个: ping\n\n（SSE 注释行，浏览器会忽略），防止中间网络设备超时断开。

验证方式：

• curl -N看到一行一行返回。
• 浏览器 F12 看 Network → Response 是event-stream类型，且 0.5 秒内出现第一条数据。
• 压测 500 并发持续流式，Tomcat 线程数稳定在 500 附近且无 OOM。

如何避免再次发生：

• 接入 AI 流式接口的 checklist 必须包含 SSE 头部、Nginx 配置、超时配置。
• 性能测试用例必须包含"持续 30 分钟 100 并发流式"场景。
• 监控指标必须包含ai_first_token_latency_ms、ai_stream_active_connections。

6. 技术选型与权衡

6.1 优点（流式 + SSE）

• 用户体验优秀：首 Token 延迟低，感知流畅。
• 资源利用率高：边生成边推送，内存压力小。
• 协议简单：HTTP 即可，调试方便（curl -N 就能看）。
• 浏览器支持好：原生EventSourceAPI，无需第三方库。

6.2 缺点

• 服务端线程模型挑战：每个连接长期占用一个线程。
• 中间件兼容：需要 Nginx、网关、CDN 都正确配置。
• 双向通信不行：客户端中途想"取消"或"修改问题"需要走另外的 HTTP 接口。
• 重连后无法续传：LLM 是无状态推理，重连通常意味着从头开始（不像视频流可以续传字节）。

6.3 适用场景

• Chat 类应用（网页聊天、客服、AI 助手）。
• 代码补全、AI 搜索、文档问答。
• 任何"长文本生成 + 用户等待"的场景。

6.4 不适用场景

• 极短响应（< 200ms）：流式开销不划算，普通 JSON 即可。
• 双向实时交互（语音对讲、协作编辑）：用 WebSocket。
• 移动端弱网且需要离线缓存：流式不利于缓存。

6.5 替代方案

• WebSocket：需要双向时才考虑。
• HTTP/2 + gRPC streaming：微服务内部、跨服务流式调用比 SSE 更高效，但前端集成成本高。
• 长轮询（Long Polling）：老式兼容方案，移动端老系统才用。

6.6 AI 主题额外权衡

• 云端模型 vs 私有化：私有化部署 Ollama/vLLM 时，Nginx 反向代理配置同样关键。
• 大模型 vs 小模型：小模型首 Token 更快（200ms 以内），流式的相对收益小一些，但用户感知差异小。
• 是否保留人工确认：客服、医生等高风险场景，必须在 SSE 推送完毕后再由人工审核。

7. 常见错误

1. 错误认知：“SSE 就是 WebSocket 的简化版”——SSE 是 HTTP 之上的单向协议，WebSocket 是独立协议，选哪个取决于是否需要双向。

2. Nginx 默认配置上线：SSE 必须关闭proxy_buffering、调大proxy_read_timeout，否则等于没优化。默认配置就是错的。

3. 不设置超时：SseEmitter不设超时默认 30 秒，推理稍长就断开。每个值都要根据业务显式设置。

4. 忘记清理资源：客户端断开后大模型侧的连接没取消 → Token 仍然在扣费 → 资损。必须在onCompletion/onError中 cancel 上游。

5. 没有首 Token 延迟监控：只监控总耗时，首 Token 5 秒才到用户看不到，体验崩塌，但 P99 看着正常。必须独立打点。

6. 线程池配置不足：SSE 连接长期占线程，Tomcat 默认 200 线程根本不够。按"峰值并发 × 平均连接时长"估算。

7. 缺乏背压机制：客户端网络慢时，WebFlux 背压会丢数据。需要监控onBackpressureBuffer警告。

8. 架构师视角总结

什么时候用：

• 任何 LLM 长文本输出场景。
• 团队能接受 SSE 的中间件配置和线程模型。

什么时候不用：

• 响应时间 < 200ms 的短请求。
• 双向实时交互（用 WebSocket）。
• 客户端是不支持 SSE 的老 IE/老 App（极少）。

评审方案时重点问：

• 流式响应的超时设置是多少？分几层？
• 中间件（Nginx/网关/CDN）的缓冲和超时配置改了吗？
• 线程池容量是否按峰值并发 × 平均连接时长估算？
• 首 Token 延迟如何监控？告警阈值是什么？
• 客户端断网时，大模型侧的资源如何释放？谁来 cancel？

故障时先检查：

1. 大模型侧是否真的流式（stream=true）？→curl -N验证。
2. Nginx/网关是否关闭缓冲？
3. Tomcat 线程是否被打满？→/actuator/metrics/executor.active。
4. 首 Token 延迟是不是过高？→ 看埋点。
5. 大模型供应商是否限流或故障？→ 多供应商切换。

必须监控的指标：

• ai_first_token_latency_ms（P50/P95/P99）
• ai_stream_total_duration_ms
• ai_stream_active_connections
• ai_stream_error_rate（按错误类型分类：超时/解析失败/网络中断/限流）
• ai_stream_token_usage（按 user/model 维度）

必须提前设计兜底：

• 大模型供应商故障 → 多供应商 + 智能路由。
• 客户端断网 → 大模型侧连接清理 + 不重试。
• 长推理超时 → 渐进式超时（首 Token 30s，整体 5min）。
• 用户输入超长 Prompt → 提前截断 + 提示用户。

AI 主题额外：

• 该功能是否真的需要大模型？短回答规则引擎是否更便宜稳定？
• 模型失败后系统如何降级？→ 静态兜底文案 / 传统检索 / 人工客服。
• 如何控制模型不确定性？→ 输出后做格式校验、敏感词过滤。
• 如何控制成本？→ 按用户/租户限流、缓存相似问题、上下文压缩。
• 如何控制数据权限？→ Prompt 注入过滤、用户身份携带、知识库租户隔离。

9. 今日练习

题目（先不要看答案）

1. 概念理解题：
为什么 ChatGPT 网页版使用 SSE 而不是 WebSocket？反过来，什么场景下你应该选 WebSocket？请列出至少 2 个判断维度。

2. 生产故障分析题：
某天 AI 助手的 P99 首 Token 延迟从 800ms 飙升到 6 秒，但 P99 总耗时只增加了 1 秒（从 8 秒到 9 秒）。最可能的 3 个原因是什么？分别如何排查？

3. 系统设计题：
请设计一个统一 AI 流式网关，要求：

• 支持接入多家大模型供应商（OpenAI、DeepSeek、Anthropic、自研模型）
• 支持 SSE 透传到前端
• 支持按用户/租户限流（如每用户 60 req/min）
• 支持失败自动切换供应商
• 记录首 Token 延迟、Token 用量等关键指标

请画出架构图（组件 + 数据流），并说明关键设计取舍。

4. Java 实践题：
在 Spring Boot 3 项目中实现一个/ai/chat/stream接口：

• 输入：{"model":"gpt-4","messages":[{"role":"user","content":"你好"}]}
• 输出：SSE 格式，逐 Token 推送大模型回答
• 要求：处理客户端断连、处理大模型超时、记录首 Token 延迟到 Micrometer

完成后用curl -N -X POST http://localhost:8080/ai/chat/stream -H "Content-Type: application/json" -d '{...}'验证。

参考答案（先独立思考再来看）

答案 1（概念理解）

判断维度：

• 通信方向：单向（服务端→客户端）选 SSE；双向选 WebSocket。LLM 场景是"问一次、答一次"，SSE 足够。
• 协议兼容：SSE 是 HTTP，无需特殊代理支持；WebSocket 需要 Upgrade 头，部分企业代理会拦截。SSE 更适合 ToB 场景。
• 自动重连：浏览器原生EventSource自动重连，WebSocket 需要手写重连逻辑。
• 服务器实现复杂度：SSE 用普通 HTTP 框架即可，WebSocket 需要异步 I/O。
• 选 WebSocket 的场景：多人协作编辑、实时语音/视频、AI 协作白板（用户和 AI 互相打断）。

答案 2（故障分析）

最可能 3 个原因：

1. 大模型供应商侧拥塞：上游推理服务排队。→ 排查：看供应商 dashboard、看其他调用方是否也慢、对比不同供应商的延迟。
2. Prompt 上下文过大：每次请求带了过多历史消息，导致 prefill 阶段慢。→ 排查：打点prompt_tokensP95，看是否暴增；检查是否有上下文压缩逻辑失效。
3. 本地线程池/连接池耗尽：WebClient 连接池或 Tomcat 线程池不够，新请求在排队。→ 排查：看http.client.requestsactive 数、看 Tomcatthreads.busy。

答案 3（系统设计）

架构图：

┌────────────┐ │ Browser │ └─────┬──────┘ │ SSE ▼ ┌──────────────────┐ │ API Gateway │ (限流、鉴权、灰度) └─────────┬────────┘ │ ▼ ┌──────────────────┐ │ AI Gateway │ ← 统一 SSE 输出 │ ┌──────────────┐ │ │ │ 限流(Redis) │ │ │ │ 路由(规则) │ │ → 供应商 A (OpenAI) │ │ 熔断(Resilience4j) │ │ → 供应商 B (DeepSeek) │ │ 监控(Micrometer) │ │ → 供应商 C (自研) │ └──────────────┘ │ └──────────────────┘

关键取舍：

• 路由策略：默认按模型名路由（gpt-4 → OpenAI）；故障时按错误率切换；按用户等级分配（VIP 走更好的模型）。
• 限流位置：网关层 Redis 滑动窗口，按 userId+tenantId 限流。
• 熔断：Resilience4j，5 秒内错误率 > 50% 触发熔断 30 秒。
• 成本控制：Token 计量写到 Kafka，下游对账。
• 统一 SSE：无论后端是什么协议（HTTP chunked、SSE、WebSocket），网关对外统一 SSE。

答案 4（Java 实践）

参考上面的 Controller + Service 代码，关键点：

• produces = TEXT_EVENT_STREAM_VALUE不能少。
• SseEmitter必须设置超时和onCompletion/onError回调。
• WebClient 的responseTimeout要 ≥ SseEmitter 超时。
• 首 Token 延迟通过System.nanoTime()在第一个doOnNext中打点。
• 客户端断连时emitter.onCompletion触发，需要调用 WebClient 的 subscription dispose。

10. 延伸学习

后续推荐学习（按依赖顺序）：

1. Spring AI Streaming API（官方文档）：Spring 官方对 LLM 流式响应的封装，比手写 WebClient 更优雅。掌握StreamingChatClient、ChatResponse流式版本。
2. WebFlux 响应式 SSE（Spring 官方文档）：理解Flux<ServerSentEvent<T>>的背压机制，对比 MVC 的SseEmitter。
3. Nginx 反向代理与 SSE 调优（Nginx 官方文档）：深入理解proxy_buffering、proxy_cache、gzip对 SSE 的影响（gzip 会破坏 SSE 格式！）。
4. Resilience4j 熔断在流式场景的应用（官方文档）：流式响应的"中途失败"如何被熔断器识别和处理。

关联说明：

• Spring AI 是今日手写 WebClient 的"工业级封装"，学完能少写很多样板代码。
• WebFlux 是更高并发的替代方案，10K 并发以上必须评估。
• Nginx 调优是上线的"最后一公里"，本地测试没问题不代表生产没问题。
• 熔断是 AI 网关的必备能力，明天会进一步展开。

11. 今日知识卡片（≤200 字）

核心原理：LLM 自回归生成是"逐 Token 长任务"，SSE 让服务端边生成边推送，把感知延迟从"总耗时"降到"首 Token 延迟"（通常 200~800ms）。

生产实践建议：SSE 三件套必做——(1)produces = text/event-stream不能少；(2) Nginx 关闭proxy_buffering并调大proxy_read_timeout到 300s+；(3) 客户端断连必须在onCompletion中 cancel 大模型侧连接，否则 Token 仍扣费。

常见风险：Tomcat 线程数（默认 200）被持续 SSE 连接耗尽，导致全站 502。必须按"峰值并发 × 平均连接时长"估算线程池。

架构设计判断标准：单向长文本推送选 SSE；只有"双向实时"才选 WebSocket。SSE 不是"弱化版 WebSocket"，而是"为单向流式而生的 HTTP 协议"。