摘要在生成式 AI 与大语言模型LLM爆发的当下传统“请求-等待-响应”的 HTTP 短轮询与同步阻塞模式彻底失效。一个动辄需要推理数秒甚至数十秒的请求若采用同步等待不仅会给用户带来长达数秒的白屏焦虑更会迅速耗尽服务器的连接池与内存资源。SSEServer-Sent Events服务端推送事件凭借其基于标准 HTTP 协议、轻量级、单向流式传输及原生断线重连等特性成为了大模型实时交互Stream Generation的绝对首选方案。本文将从大模型场景下的响应痛点切入深度拆解 SSE 的底层协议流转与网络机制对比 WebSocket 与 Chunked Encoding系统性讲解前端对 SSE 流式数据的精准解析与缓冲区Buffer状态机处理以及后端基于Python (FastAPI / Asyncio)与Java (Spring WebFlux / Reactive)的高并发异步执行引擎实现最后给出生产级别的背压控制、连接池治理与断线重连避坑指南。前言AI 时代下的 HTTP 响应范式转移在传统的 Web 应用开发中数据传输通常遵循标准的请求-响应Request-Response模式。客户端发起 HTTP 请求服务端处理业务逻辑并返回完整的 JSON 或 HTML随后连接关闭。然而大语言模型LLM的推理机制是自回归生成Autoregressive Generation——模型并不是瞬间产生整段文本而是一个 Token 一个 Token 地顺序预测。在生产环境中如果等待大模型将 1000 个 Token 全部生成完毕再统一返回给客户端通常需要3 秒到 15 秒。这种传统的同步阻塞交互会带来两个致命瓶颈极其恶劣的用户体验用户面对数秒乃至数十秒的白屏或 Loading 转圈极大增加了放弃使用的概率。服务端连接与内存雪崩高并发场景下海量 HTTP 请求被同步挂起Blocked等待模型推理完成。这会迅速打爆 Web 服务器如 Tomcat、Gunicorn的线程池导致系统崩溃。为了解决这一痛点流式传输Streaming应运而生。通过在模型每生成一个 Token 时立即将其推送到客户端用户的首字延迟TTFT, Time To First Token可以从数秒骤降至100~200 毫秒。在众多的实时流式传输技术中SSEServer-Sent Events凭借其专为“服务端向客户端单向推送”设计的特性成为了 OpenAI、Anthropic、DeepSeek 等大厂 API 的事实标准。一、 为什么是大模型交互的绝对首选SSE 底层原理与技术选型1.1 SSEServer-Sent Events核心原理SSEServer-Sent Events是 HTML5 标准规范的一部分。它的核心思想极其简单利用 HTTP 协议的长连接Keep-Alive由服务端持续向客户端发送基于纯文本的数据流Text Stream。从网络层来看SSE 依然建立在标准的 HTTP/1.1 或 HTTP/2 协议之上使用普通的 TCP 连接不需要经过额外的 WebSocket 握手升级。SSE 的 HTTP 头部特征当服务端准备建立 SSE 流式响应时必须返回带有特定 Header 的 HTTP 响应HTTP/1.1 200 OK Content-Type: text/event-stream; charsetutf-8 Cache-Control: no-cache Connection: keep-alive X-Accel-Buffering: noContent-Type: text/event-stream关键头部声明该响应为一个持续的事件流告知浏览器或客户端不要等待连接关闭而是边接收边解析。Cache-Control: no-cache禁止浏览器或中间网关如 Nginx、CDN缓存事件数据确保实效性。Connection: keep-alive保持 HTTP 保持长连接不关闭。X-Accel-Buffering: no专为 Nginx 设计关闭 Nginx 的反向代理响应缓冲区Buffering使得服务端发送的每一个 Chunk 能实时穿透 Nginx 送达前端。SSE 消息文本格式SSE 的数据传输格式是严格规定的纯文本行UTF-8 编码每一条消息由若干个 key-value 行组成以换行符\n分隔消息与消息之间必须以连续两个换行符\n\n结尾。常见的字段格式如下字段名作用描述示例data事件的具体数据负载。若有多行data:客户端会自动将其拼接并用\n分隔。data: {content: 你好}event自定义事件类型。客户端可据此进行不同的事件监听如message、error、done。event: updateid事件的唯一标识符。用于断线重连时发送给服务端实现从断点恢复。id: 10024retry告知客户端在连接中断后重新尝试建立连接的等待间隔时间毫秒。retry: 3000:以冒号开头的行会被视为注释客户端会自动忽略。常用于**发送心跳包Ping**防止连接超时。: heartbeat一段标准的 SSE 数据流示例如下event: message id: 1 data: {token: 你好, finish: false} event: message id: 2 data: {token: 我是, finish: false} event: message id: 3 data: {token: 人工智能, finish: false} event: done data: [DONE]1.2 SSE vs WebSocket vs HTTP Chunked 深度对比在实时通信领域开发者常在SSE、WebSocket与HTTP Chunked Encoding之间进行选型。下表梳理了三者的技术差异评估维度SSE (Server-Sent Events)WebSocketHTTP Chunked Encoding数据流向单向服务端 ➔ 客户端全双工双向收发单向服务端 ➔ 客户端底层协议标准 HTTP / HTTPSWebSocket 协议依赖Upgrade握手标准 HTTP / HTTPS数据格式UTF-8 文本流格式化结构二进制ArrayBuffer/Blob或文本纯原始字节流无统一格式规范断线重连原生支持带Last-Event-ID需手动在应用层实现心跳与重连无原生重连机制需重新发起 HTTP网关防火墙兼容性极佳80/443 端口无缝穿透 Nginx/CDN一般部分严格的企业防火墙会拦截 WebSocket极佳浏览器 API 接入原生EventSource/fetch原生WebSocketfetchReadableStream资源消耗极低复用 HTTP/2 多路复用中等需维护全双工长连接状态极低为什么 LLM 生成式场景优先选择 SSE 而非 WebSocket业务契合度高大模型对话场景绝大多数情况下是“用户输入一句话一个 HTTP 请求 ➔ 模型持续推出来一段话持续单向响应”。这是典型的单向下行流完全不需要 WebSocket 的双向全双工开销。极佳的网关与防火墙兼容性WebSocket 使用ws://或wss://协议在很多企业级防火墙、代理网关或 CDN 中可能会被阻断或需要特殊配置而 SSE 只是普通的 HTTP 响应能够无缝兼容现有的 Nginx、HAProxy、 Cloudflare 及微服务网关。HTTP/2 HTTP/3 赋能在 HTTP/2 环境下多个 SSE 请求可以完全复用同一条 TCP 连接多路复用 Multiplexing极大节省了客户端与服务端的连接资源。二、 前端视角流式数据的精准解析与缓冲区状态机尽管浏览器提供了原生的EventSourceAPI 用于接入 SSE但在真实的生产级 LLM 应用开发中原生的EventSource存在两大致命缺陷只支持 GET 请求不支持 POST 请求。而大模型 Prompt 通常很长放在 GET 请求的 URL 参数中极易超出浏览器 URL 长度限制。无法自定义 HTTP Header无法在请求头中携带Authorization: Bearer Token进行身份鉴权。因此现代前端React / Vue / Next.js普遍采用fetchReadableStream配合流式解析缓冲区Stream Buffer来构建高可用的 SSE 客户端。[服务端 SSE 字节流] │ ▼ [ ReadableStreamReader (Uint8Array) ] │ ▼ [ TextDecoder (拼接未完结的 UTF-8 字节) ] │ ▼ [ 流式缓冲区 (Stream Buffer 字符串拼接) ] │ 寻找 \n\n 消息切分点 ▼ [ 消息提取与状态机解析 (data: {...}) ] │ ▼ [ UI 增量渲染 (React/Vue State) ]2.1 解决网络粘包与半包基于状态机的缓冲区解析在 TCP 传输层数据是以字节块Chunk形式分批送达的。这意味着客户端每次通过reader.read()读取到的文本绝不刚好是一条完整的 SSE 消息粘包Nack Packet一次read()读取到了包含 3 条完整data:消息的文本。半包Half Packet一次read()读取到的文本恰好在data: {token: 你好处断开了剩下的半截数据在下一个 Chunk 才会送达。如果不经过缓冲区处理直接JSON.parse()必定会导致前端疯狂抛出SyntaxError解析异常生产级前端 SSE 解析器代码实现TypeScript下面展示一套零依赖、能够优雅处理粘包与半包的流式解析器export interface SSEMessage { event?: string; data: string; id?: string; retry?: number; } export class SSEStreamParser { private buffer: string ; private onMessage: (msg: SSEMessage) void; constructor(onMessage: (msg: SSEMessage) void) { self.onMessage onMessage; } /** * 逐步传入从 ReadableStream 读取到的 Chunk 文本 */ public feed(chunk: string): void { this.buffer chunk; // SSE 规定消息与消息之间必须以 \n\n 分隔 let delimiterIndex: int; while ((delimiterIndex this.buffer.indexOf(\n\n)) ! -1) { // 提取一条完整的消息文本块 const rawMessage this.buffer.slice(0, delimiterIndex); // 将剩余部分留在缓冲区中处理半包 this.buffer this.buffer.slice(delimiterIndex 2); // 解析单条消息文本块 this.parseRawMessage(rawMessage); } } private parseRawMessage(rawMessage: string): void { const lines rawMessage.split(\n); const message: SSEMessage { data: }; let hasData false; for (const line of lines) { if (line.startsWith(:)) { // 心跳包或注释忽略 continue; } const colonIndex line.indexOf(:); if (colonIndex -1) continue; const field line.slice(0, colonIndex).trim(); let value line.slice(colonIndex 1); if (value.startsWith( )) { value value.slice(1); // 移除冒号后的首个空格 } switch (field) { case event: message.event value; break; case data: // 如果单条消息中有多个 data: 行按规范用换行符连接 message.data (hasData ? \n : ) value; hasData true; break; case id: message.id value; break; case retry: message.retry parseInt(value, 10); break; } } if (hasData) { this.onMessage(message); } } }2.2 前端 Fetch 流式接入完整实战结合上面的解析器使用标准fetch接口对接后端 POST 接口async function fetchLLMStream(prompt: str, onToken: (text: str) void) { const controller new AbortController(); // 用于取消请求 try { const response await fetch(/api/v1/chat/completions, { method: POST, headers: { Content-Type: json, Authorization: Bearer sk-xxx }, body: JSON.stringify({ prompt, stream: true }), signal: controller.signal }); if (!response.ok || !response.body) { throw new Error(HTTP 错误: ${response.status}); } const reader response.body.getReader(); const decoder new TextDecoder(utf-8); // 创建流式解析器 const parser new SSEStreamParser((msg: SSEMessage) { if (msg.data [DONE]) { console.log(流式传输结束); return; } try { const payload JSON.parse(msg.data); if (payload.content) { onToken(payload.content); // 触发 UI 更新 } } catch (err) { console.error(JSON 解析失败:, err, msg.data); } }); // 循环读取流数据 while (true) { const { done, value } await reader.read(); if (done) break; // 将 Uint8Array 转换为字符串并输入解析器 const chunkText decoder.decode(value, { stream: true }); parser.feed(chunkText); } } catch (error) { if (error.name AbortError) { console.log(用户取消了流式请求); } else { console.error(流式请求异常:, error); } } }三、 后端架构基于 Python (FastAPI / Asyncio) 的高并发异步执行引擎在后端服务中实现 SSE 流式推送的重中之重是异步非阻塞Async / Non-blocking。如果在同步阻塞的 Web 框架如传统的 Flask、Django 或 Gunicorn 同步 Worker中编写 SSE一个 SSE 连接就会独占一个操作系统线程。一旦并发数十个请求服务器的线程资源就会被全部耗尽直接拒绝后续服务Python 的FastAPI配合Asyncio事件循环使得单线程能够利用协程Coroutine并发管理数千个 SSE 长连接。[ FastAPI 协程事件循环 (Event Loop) ] │ ┌─────────────────────────┼─────────────────────────┐ ▼ ▼ ▼ [ 客户端 1 (SSE 请求) ] [ 客户端 2 (SSE 请求) ] [ 客户端 3 (SSE 请求) ] │ │ │ Async Iterable 生成器 Async Iterable 生成器 Async Iterable 生成器 │ │ │ ▼ ▼ ▼ [ 异步调用大模型 API ] [ 异步调用大模型 API ] [ 异步调用大模型 API ] (await client.stream()) (await client.stream()) (await client.stream())3.1 生产级 FastAPI 异步 SSE 引擎实现下面的示例展示了如何利用EventSourceResponse或 FastAPI 原生StreamingResponse结合 Python 异步生成器AsyncGenerator实现高并发的流式接口import asyncio import json import logging from typing import AsyncGenerator from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from openai import AsyncOpenAI logging.basicConfig(levellogging.INFO) logger logging.getLogger(SSE-Engine) app FastAPI(titleProduction SSE LLM Engine) # 初始化 OpenAI 异步客户端 async_client AsyncOpenAI(api_keysk-your-openai-key) class ChatRequest(BaseModel): prompt: str model: str gpt-4o-mini async def llm_event_generator(prompt: str, model: str) - AsyncGenerator[str, None]: 异步生成器从 LLM 流式读取 Token并包装为标准 SSE 格式输出 try: # 发起异步流式请求不阻塞事件循环 response_stream await async_client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], streamTrue ) counter 0 async for chunk in response_stream: if chunk.choices and len(chunk.choices) 0: delta chunk.choices[0].delta.content if delta: counter 1 payload { id: counter, content: delta } # 遵循标准 SSE 格式输出data: JSON\n\n yield fdata: {json.dumps(payload, ensure_asciiFalse)}\n\n # 出让控制权给事件循环防止 CPU 密集型任务卡死事件循环 await asyncio.sleep(0) # 发送结束标记 yield data: [DONE]\n\n except asyncio.CancelledError: logger.warning(客户端断开连接取消下游 LLM 推理任务) # 此处可编写清理资源或打断下游 API 调用的逻辑 raise except Exception as e: logger.error(f流式生成异常: {str(e)}) error_payload {error: Internal Processing Error, details: str(e)} yield fevent: error\ndata: {json.dumps(error_payload)}\n\n app.post(/api/v1/chat/stream) async def stream_chat(request: ChatRequest): SSE 流式响应接口 if not request.prompt.strip(): raise HTTPException(status_code400, detailPrompt 不能为空) return StreamingResponse( llm_event_generator(request.prompt, request.model), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 告知 Nginx 禁用响应缓冲 } ) if __name__ __main__: import uvicorn # 使用 uvicorn 运行 ASGI 高性能服务 uvicorn.run(app, host0.0.0.0, port8000, workers4)四、 后端架构基于 Java (Spring WebFlux / Reactive) 的响应式非阻塞 SSE对于大型企业级 Java 体系而言传统的 Spring MVC (Servlet 架构) 基于“每个请求一个线程Thread-per-request”模型。如果在 Spring MVC 中使用SseEmitter在极高并发下依然会导致 Tomcat 线程池枯竭。Spring WebFlux结合Project Reactor提供了响应式Reactive非阻塞编程模型。基于 Netty 事件驱动网络库仅用极少量的线程即可支撑数万级别的并发 SSE 流式连接。[ HTTP 请求 ] ── [ Netty EventLoop Thread ] │ ▼ [ FluxServerSentEventT 响应式流 ] │ ▼ (非阻塞 Reactor 管道) [ 实时推送流至客户端 ]4.1 Spring WebFlux SSE 生产级代码实现1. Maven 依赖dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-webflux/artifactId /dependency dependency groupIdio.projectreactor/groupId artifactIdreactor-core/artifactId /dependency /dependencies2. Service 与 Controller 实现package com.example.sse.service; import org.springframework.http.codec.ServerSentEvent; import org.springframework.stereotype.Service; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.time.Duration; import java.util.Map; import java.util.UUID; Service public class LlmStreamService { /** * 模拟响应式获取 LLM 流式输出真实场景中替换为 Async WebClient 调用 */ public FluxServerSentEventMapString, Object streamLlmResponse(String prompt) { String[] tokens (你好 我是 基于 Spring WebFlux 响应式 架构 的 大模型 流式 引擎 。).split( ); // 模拟以 100ms 间隔产生 Token 的响应式流 return Flux.interval(Duration.ofMillis(100)) .take(tokens.length) .map(index - { String token tokens[index.intValue()]; MapString, Object data Map.of( id, index, content, token ); return ServerSentEvent.MapString, Objectbuilder() .id(UUID.randomUUID().toString()) .event(message) .data(data) .build(); }) .concatWith(Mono.just( ServerSentEvent.MapString, Objectbuilder() .event(done) .data(Map.of(content, [DONE])) .build() )); } }package com.example.sse.controller; import com.example.sse.service.LlmStreamService; import org.springframework.http.MediaType; import org.springframework.http.codec.ServerSentEvent; import org.springframework.web.bind.annotation.*; import reactor.core.publisher.Flux; import java.util.Map; RestController RequestMapping(/api/v1) public class LlmStreamController { private final LlmStreamService llmStreamService; public LlmStreamController(LlmStreamService llmStreamService) { this.llmStreamService llmStreamService; } PostMapping(value /chat/stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxServerSentEventMapString, Object streamChat(RequestBody MapString, String request) { String prompt request.get(prompt); if (prompt null || prompt.isBlank()) { return Flux.error(new IllegalArgumentException(Prompt 不能为空)); } return llmStreamService.streamLlmResponse(prompt); } }五、 生产级落地避坑指南与高可用治理在真实的大规模生产环境中仅仅写通 SSE 代码是远远不够的。中间网关阻塞、连接无故断开、内存泄露与超时控制等问题是每一个工程架构师必须解决的硬骨头。[ 客户端 ] ──(Keep-Alive / Heartbeat)──► [ Nginx 网关 (proxy_buffering off) ] ──► [ SSE 微服务 ]5.1 网关层优化Nginx / CDN 响应缓冲与超时拦截在许多生产事故中开发者发现本地测试时 SSE 是一字一字吐出来的一上生产环境就变成了“白屏几秒后突然砸出一大段文本”。这 100% 是由于 Nginx 或 CDN 开启了响应缓冲Response BufferingNginx 核心配置解析在负责反向代理的 Nginx 配置中必须对 SSE 路径进行如下特殊配置location /api/v1/chat/stream { proxy_pass http://llm_backend_cluster; # 1. 关键关闭响应缓冲确保流式 Token 实时穿透 proxy_buffering off; proxy_cache off; # 2. 关键关闭 HTTP Chunk 缓存 chunked_transfer_encoding on; # 3. 关键防止 Nginx 因默认 60s 无数据交互而强行断开长连接 proxy_read_timeout 3600s; proxy_send_timeout 3600s; # 4. 支持 HTTP/1.1 长连接 proxy_http_version 1.1; proxy_set_header Connection ; proxy_set_header Host $host; # 5. 禁用 gzip 压缩防止 Gzip 收集满一定字节数才刷盘 gzip off; }5.2 心跳保活机制Heartbeat / Keep-Alive大模型在处理复杂 Reasoning 逻辑如 DeepSeek-R1 或 OpenAI o1时可能需要“思考”长达 10~30 秒在此期间不会向客户端输出任何 Token。在长达数十秒的静默期内客户端、中间防火墙、Nginx 或 Load Balancer如 AWS ALB极易判定连接已超时并主动强行切断 TCP 连接。解决方案定时发送 SSE 注释心跳包服务端必须在一个独立的定时任务或协程中每隔5~10 秒向客户端推送一个 SSE 规范允许的注释行Commentasync def safe_event_generator(): while True: # 产生业务数据的同时如果超过 5s 无数据生成心跳包 # SSE 注释行格式冒号开头客户端会静默忽略但能维持 TCP 活性 yield : heartbeat\n\n await asyncio.sleep(5)5.3 背压控制Backpressure Management与连接泄露背压问题Backpressure大模型生成 Token 的速度远快于前端客户端或网络较差的移动端接收和渲染的速度。如果服务端不加限制地推流会导致后端服务器的内存缓冲区Buffer无限膨胀最终导致 OOM内存溢出。解法在 Java WebFlux 中充分利用 Reactor 的.onBackpressureDrop()或.onBackpressureBuffer(1024)进行策略管控在 Python 中使用带容量限制的asyncio.Queue(maxsize100)来平衡生产与消费速度。客户端主动断开连接处理Client Abort Handling当用户在界面点击“停止生成”或直接关闭网页时客户端会发送 TCP RST 包断开连接。解法后端生成器必须监听连接中断事件如 Python 捕捉asyncio.CancelledError并立即向下游的大模型推理集群如 vLLM, Ollama 或 API发送 Abort 取消请求避免模型在后台继续无效地消耗昂贵的 GPU 算力总结在 AI 原生应用的新时代SSEServer-Sent Events凭借其基于标准 HTTP、极低开销、原生格式规范与良好的网关穿透性成为了连接大模型自回归推理能力与前端用户实时体验的最稳固桥梁。要构建一个生产级的高并发 SSE 架构关键在于前端抛弃简陋的EventSource采用fetchReadableStream配合基于缓冲区的状态机精准防御粘包与半包。后端坚决弃用传统的阻塞式线程池模式全面转向Python Asyncio或Java Spring WebFlux响应式非阻塞架构确保数万并发长连接下的系统稳健。运维与网关调优 Nginx 反向代理配置关闭 Buffer设置心跳保活与客户端中断响应机制切实保障 GPU 算力与服务稳定性。通过本文提供的整套实战方案开发者可以搭建出低延迟、高并发、生产级可靠的大模型流式交互平台。 SSE流式解析与异步执行
摘要在生成式 AI 与大语言模型LLM爆发的当下传统“请求-等待-响应”的 HTTP 短轮询与同步阻塞模式彻底失效。一个动辄需要推理数秒甚至数十秒的请求若采用同步等待不仅会给用户带来长达数秒的白屏焦虑更会迅速耗尽服务器的连接池与内存资源。SSEServer-Sent Events服务端推送事件凭借其基于标准 HTTP 协议、轻量级、单向流式传输及原生断线重连等特性成为了大模型实时交互Stream Generation的绝对首选方案。本文将从大模型场景下的响应痛点切入深度拆解 SSE 的底层协议流转与网络机制对比 WebSocket 与 Chunked Encoding系统性讲解前端对 SSE 流式数据的精准解析与缓冲区Buffer状态机处理以及后端基于Python (FastAPI / Asyncio)与Java (Spring WebFlux / Reactive)的高并发异步执行引擎实现最后给出生产级别的背压控制、连接池治理与断线重连避坑指南。前言AI 时代下的 HTTP 响应范式转移在传统的 Web 应用开发中数据传输通常遵循标准的请求-响应Request-Response模式。客户端发起 HTTP 请求服务端处理业务逻辑并返回完整的 JSON 或 HTML随后连接关闭。然而大语言模型LLM的推理机制是自回归生成Autoregressive Generation——模型并不是瞬间产生整段文本而是一个 Token 一个 Token 地顺序预测。在生产环境中如果等待大模型将 1000 个 Token 全部生成完毕再统一返回给客户端通常需要3 秒到 15 秒。这种传统的同步阻塞交互会带来两个致命瓶颈极其恶劣的用户体验用户面对数秒乃至数十秒的白屏或 Loading 转圈极大增加了放弃使用的概率。服务端连接与内存雪崩高并发场景下海量 HTTP 请求被同步挂起Blocked等待模型推理完成。这会迅速打爆 Web 服务器如 Tomcat、Gunicorn的线程池导致系统崩溃。为了解决这一痛点流式传输Streaming应运而生。通过在模型每生成一个 Token 时立即将其推送到客户端用户的首字延迟TTFT, Time To First Token可以从数秒骤降至100~200 毫秒。在众多的实时流式传输技术中SSEServer-Sent Events凭借其专为“服务端向客户端单向推送”设计的特性成为了 OpenAI、Anthropic、DeepSeek 等大厂 API 的事实标准。一、 为什么是大模型交互的绝对首选SSE 底层原理与技术选型1.1 SSEServer-Sent Events核心原理SSEServer-Sent Events是 HTML5 标准规范的一部分。它的核心思想极其简单利用 HTTP 协议的长连接Keep-Alive由服务端持续向客户端发送基于纯文本的数据流Text Stream。从网络层来看SSE 依然建立在标准的 HTTP/1.1 或 HTTP/2 协议之上使用普通的 TCP 连接不需要经过额外的 WebSocket 握手升级。SSE 的 HTTP 头部特征当服务端准备建立 SSE 流式响应时必须返回带有特定 Header 的 HTTP 响应HTTP/1.1 200 OK Content-Type: text/event-stream; charsetutf-8 Cache-Control: no-cache Connection: keep-alive X-Accel-Buffering: noContent-Type: text/event-stream关键头部声明该响应为一个持续的事件流告知浏览器或客户端不要等待连接关闭而是边接收边解析。Cache-Control: no-cache禁止浏览器或中间网关如 Nginx、CDN缓存事件数据确保实效性。Connection: keep-alive保持 HTTP 保持长连接不关闭。X-Accel-Buffering: no专为 Nginx 设计关闭 Nginx 的反向代理响应缓冲区Buffering使得服务端发送的每一个 Chunk 能实时穿透 Nginx 送达前端。SSE 消息文本格式SSE 的数据传输格式是严格规定的纯文本行UTF-8 编码每一条消息由若干个 key-value 行组成以换行符\n分隔消息与消息之间必须以连续两个换行符\n\n结尾。常见的字段格式如下字段名作用描述示例data事件的具体数据负载。若有多行data:客户端会自动将其拼接并用\n分隔。data: {content: 你好}event自定义事件类型。客户端可据此进行不同的事件监听如message、error、done。event: updateid事件的唯一标识符。用于断线重连时发送给服务端实现从断点恢复。id: 10024retry告知客户端在连接中断后重新尝试建立连接的等待间隔时间毫秒。retry: 3000:以冒号开头的行会被视为注释客户端会自动忽略。常用于**发送心跳包Ping**防止连接超时。: heartbeat一段标准的 SSE 数据流示例如下event: message id: 1 data: {token: 你好, finish: false} event: message id: 2 data: {token: 我是, finish: false} event: message id: 3 data: {token: 人工智能, finish: false} event: done data: [DONE]1.2 SSE vs WebSocket vs HTTP Chunked 深度对比在实时通信领域开发者常在SSE、WebSocket与HTTP Chunked Encoding之间进行选型。下表梳理了三者的技术差异评估维度SSE (Server-Sent Events)WebSocketHTTP Chunked Encoding数据流向单向服务端 ➔ 客户端全双工双向收发单向服务端 ➔ 客户端底层协议标准 HTTP / HTTPSWebSocket 协议依赖Upgrade握手标准 HTTP / HTTPS数据格式UTF-8 文本流格式化结构二进制ArrayBuffer/Blob或文本纯原始字节流无统一格式规范断线重连原生支持带Last-Event-ID需手动在应用层实现心跳与重连无原生重连机制需重新发起 HTTP网关防火墙兼容性极佳80/443 端口无缝穿透 Nginx/CDN一般部分严格的企业防火墙会拦截 WebSocket极佳浏览器 API 接入原生EventSource/fetch原生WebSocketfetchReadableStream资源消耗极低复用 HTTP/2 多路复用中等需维护全双工长连接状态极低为什么 LLM 生成式场景优先选择 SSE 而非 WebSocket业务契合度高大模型对话场景绝大多数情况下是“用户输入一句话一个 HTTP 请求 ➔ 模型持续推出来一段话持续单向响应”。这是典型的单向下行流完全不需要 WebSocket 的双向全双工开销。极佳的网关与防火墙兼容性WebSocket 使用ws://或wss://协议在很多企业级防火墙、代理网关或 CDN 中可能会被阻断或需要特殊配置而 SSE 只是普通的 HTTP 响应能够无缝兼容现有的 Nginx、HAProxy、 Cloudflare 及微服务网关。HTTP/2 HTTP/3 赋能在 HTTP/2 环境下多个 SSE 请求可以完全复用同一条 TCP 连接多路复用 Multiplexing极大节省了客户端与服务端的连接资源。二、 前端视角流式数据的精准解析与缓冲区状态机尽管浏览器提供了原生的EventSourceAPI 用于接入 SSE但在真实的生产级 LLM 应用开发中原生的EventSource存在两大致命缺陷只支持 GET 请求不支持 POST 请求。而大模型 Prompt 通常很长放在 GET 请求的 URL 参数中极易超出浏览器 URL 长度限制。无法自定义 HTTP Header无法在请求头中携带Authorization: Bearer Token进行身份鉴权。因此现代前端React / Vue / Next.js普遍采用fetchReadableStream配合流式解析缓冲区Stream Buffer来构建高可用的 SSE 客户端。[服务端 SSE 字节流] │ ▼ [ ReadableStreamReader (Uint8Array) ] │ ▼ [ TextDecoder (拼接未完结的 UTF-8 字节) ] │ ▼ [ 流式缓冲区 (Stream Buffer 字符串拼接) ] │ 寻找 \n\n 消息切分点 ▼ [ 消息提取与状态机解析 (data: {...}) ] │ ▼ [ UI 增量渲染 (React/Vue State) ]2.1 解决网络粘包与半包基于状态机的缓冲区解析在 TCP 传输层数据是以字节块Chunk形式分批送达的。这意味着客户端每次通过reader.read()读取到的文本绝不刚好是一条完整的 SSE 消息粘包Nack Packet一次read()读取到了包含 3 条完整data:消息的文本。半包Half Packet一次read()读取到的文本恰好在data: {token: 你好处断开了剩下的半截数据在下一个 Chunk 才会送达。如果不经过缓冲区处理直接JSON.parse()必定会导致前端疯狂抛出SyntaxError解析异常生产级前端 SSE 解析器代码实现TypeScript下面展示一套零依赖、能够优雅处理粘包与半包的流式解析器export interface SSEMessage { event?: string; data: string; id?: string; retry?: number; } export class SSEStreamParser { private buffer: string ; private onMessage: (msg: SSEMessage) void; constructor(onMessage: (msg: SSEMessage) void) { self.onMessage onMessage; } /** * 逐步传入从 ReadableStream 读取到的 Chunk 文本 */ public feed(chunk: string): void { this.buffer chunk; // SSE 规定消息与消息之间必须以 \n\n 分隔 let delimiterIndex: int; while ((delimiterIndex this.buffer.indexOf(\n\n)) ! -1) { // 提取一条完整的消息文本块 const rawMessage this.buffer.slice(0, delimiterIndex); // 将剩余部分留在缓冲区中处理半包 this.buffer this.buffer.slice(delimiterIndex 2); // 解析单条消息文本块 this.parseRawMessage(rawMessage); } } private parseRawMessage(rawMessage: string): void { const lines rawMessage.split(\n); const message: SSEMessage { data: }; let hasData false; for (const line of lines) { if (line.startsWith(:)) { // 心跳包或注释忽略 continue; } const colonIndex line.indexOf(:); if (colonIndex -1) continue; const field line.slice(0, colonIndex).trim(); let value line.slice(colonIndex 1); if (value.startsWith( )) { value value.slice(1); // 移除冒号后的首个空格 } switch (field) { case event: message.event value; break; case data: // 如果单条消息中有多个 data: 行按规范用换行符连接 message.data (hasData ? \n : ) value; hasData true; break; case id: message.id value; break; case retry: message.retry parseInt(value, 10); break; } } if (hasData) { this.onMessage(message); } } }2.2 前端 Fetch 流式接入完整实战结合上面的解析器使用标准fetch接口对接后端 POST 接口async function fetchLLMStream(prompt: str, onToken: (text: str) void) { const controller new AbortController(); // 用于取消请求 try { const response await fetch(/api/v1/chat/completions, { method: POST, headers: { Content-Type: json, Authorization: Bearer sk-xxx }, body: JSON.stringify({ prompt, stream: true }), signal: controller.signal }); if (!response.ok || !response.body) { throw new Error(HTTP 错误: ${response.status}); } const reader response.body.getReader(); const decoder new TextDecoder(utf-8); // 创建流式解析器 const parser new SSEStreamParser((msg: SSEMessage) { if (msg.data [DONE]) { console.log(流式传输结束); return; } try { const payload JSON.parse(msg.data); if (payload.content) { onToken(payload.content); // 触发 UI 更新 } } catch (err) { console.error(JSON 解析失败:, err, msg.data); } }); // 循环读取流数据 while (true) { const { done, value } await reader.read(); if (done) break; // 将 Uint8Array 转换为字符串并输入解析器 const chunkText decoder.decode(value, { stream: true }); parser.feed(chunkText); } } catch (error) { if (error.name AbortError) { console.log(用户取消了流式请求); } else { console.error(流式请求异常:, error); } } }三、 后端架构基于 Python (FastAPI / Asyncio) 的高并发异步执行引擎在后端服务中实现 SSE 流式推送的重中之重是异步非阻塞Async / Non-blocking。如果在同步阻塞的 Web 框架如传统的 Flask、Django 或 Gunicorn 同步 Worker中编写 SSE一个 SSE 连接就会独占一个操作系统线程。一旦并发数十个请求服务器的线程资源就会被全部耗尽直接拒绝后续服务Python 的FastAPI配合Asyncio事件循环使得单线程能够利用协程Coroutine并发管理数千个 SSE 长连接。[ FastAPI 协程事件循环 (Event Loop) ] │ ┌─────────────────────────┼─────────────────────────┐ ▼ ▼ ▼ [ 客户端 1 (SSE 请求) ] [ 客户端 2 (SSE 请求) ] [ 客户端 3 (SSE 请求) ] │ │ │ Async Iterable 生成器 Async Iterable 生成器 Async Iterable 生成器 │ │ │ ▼ ▼ ▼ [ 异步调用大模型 API ] [ 异步调用大模型 API ] [ 异步调用大模型 API ] (await client.stream()) (await client.stream()) (await client.stream())3.1 生产级 FastAPI 异步 SSE 引擎实现下面的示例展示了如何利用EventSourceResponse或 FastAPI 原生StreamingResponse结合 Python 异步生成器AsyncGenerator实现高并发的流式接口import asyncio import json import logging from typing import AsyncGenerator from fastapi import FastAPI, HTTPException from fastapi.responses import StreamingResponse from pydantic import BaseModel from openai import AsyncOpenAI logging.basicConfig(levellogging.INFO) logger logging.getLogger(SSE-Engine) app FastAPI(titleProduction SSE LLM Engine) # 初始化 OpenAI 异步客户端 async_client AsyncOpenAI(api_keysk-your-openai-key) class ChatRequest(BaseModel): prompt: str model: str gpt-4o-mini async def llm_event_generator(prompt: str, model: str) - AsyncGenerator[str, None]: 异步生成器从 LLM 流式读取 Token并包装为标准 SSE 格式输出 try: # 发起异步流式请求不阻塞事件循环 response_stream await async_client.chat.completions.create( modelmodel, messages[{role: user, content: prompt}], streamTrue ) counter 0 async for chunk in response_stream: if chunk.choices and len(chunk.choices) 0: delta chunk.choices[0].delta.content if delta: counter 1 payload { id: counter, content: delta } # 遵循标准 SSE 格式输出data: JSON\n\n yield fdata: {json.dumps(payload, ensure_asciiFalse)}\n\n # 出让控制权给事件循环防止 CPU 密集型任务卡死事件循环 await asyncio.sleep(0) # 发送结束标记 yield data: [DONE]\n\n except asyncio.CancelledError: logger.warning(客户端断开连接取消下游 LLM 推理任务) # 此处可编写清理资源或打断下游 API 调用的逻辑 raise except Exception as e: logger.error(f流式生成异常: {str(e)}) error_payload {error: Internal Processing Error, details: str(e)} yield fevent: error\ndata: {json.dumps(error_payload)}\n\n app.post(/api/v1/chat/stream) async def stream_chat(request: ChatRequest): SSE 流式响应接口 if not request.prompt.strip(): raise HTTPException(status_code400, detailPrompt 不能为空) return StreamingResponse( llm_event_generator(request.prompt, request.model), media_typetext/event-stream, headers{ Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no, # 告知 Nginx 禁用响应缓冲 } ) if __name__ __main__: import uvicorn # 使用 uvicorn 运行 ASGI 高性能服务 uvicorn.run(app, host0.0.0.0, port8000, workers4)四、 后端架构基于 Java (Spring WebFlux / Reactive) 的响应式非阻塞 SSE对于大型企业级 Java 体系而言传统的 Spring MVC (Servlet 架构) 基于“每个请求一个线程Thread-per-request”模型。如果在 Spring MVC 中使用SseEmitter在极高并发下依然会导致 Tomcat 线程池枯竭。Spring WebFlux结合Project Reactor提供了响应式Reactive非阻塞编程模型。基于 Netty 事件驱动网络库仅用极少量的线程即可支撑数万级别的并发 SSE 流式连接。[ HTTP 请求 ] ── [ Netty EventLoop Thread ] │ ▼ [ FluxServerSentEventT 响应式流 ] │ ▼ (非阻塞 Reactor 管道) [ 实时推送流至客户端 ]4.1 Spring WebFlux SSE 生产级代码实现1. Maven 依赖dependencies dependency groupIdorg.springframework.boot/groupId artifactIdspring-boot-starter-webflux/artifactId /dependency dependency groupIdio.projectreactor/groupId artifactIdreactor-core/artifactId /dependency /dependencies2. Service 与 Controller 实现package com.example.sse.service; import org.springframework.http.codec.ServerSentEvent; import org.springframework.stereotype.Service; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; import java.time.Duration; import java.util.Map; import java.util.UUID; Service public class LlmStreamService { /** * 模拟响应式获取 LLM 流式输出真实场景中替换为 Async WebClient 调用 */ public FluxServerSentEventMapString, Object streamLlmResponse(String prompt) { String[] tokens (你好 我是 基于 Spring WebFlux 响应式 架构 的 大模型 流式 引擎 。).split( ); // 模拟以 100ms 间隔产生 Token 的响应式流 return Flux.interval(Duration.ofMillis(100)) .take(tokens.length) .map(index - { String token tokens[index.intValue()]; MapString, Object data Map.of( id, index, content, token ); return ServerSentEvent.MapString, Objectbuilder() .id(UUID.randomUUID().toString()) .event(message) .data(data) .build(); }) .concatWith(Mono.just( ServerSentEvent.MapString, Objectbuilder() .event(done) .data(Map.of(content, [DONE])) .build() )); } }package com.example.sse.controller; import com.example.sse.service.LlmStreamService; import org.springframework.http.MediaType; import org.springframework.http.codec.ServerSentEvent; import org.springframework.web.bind.annotation.*; import reactor.core.publisher.Flux; import java.util.Map; RestController RequestMapping(/api/v1) public class LlmStreamController { private final LlmStreamService llmStreamService; public LlmStreamController(LlmStreamService llmStreamService) { this.llmStreamService llmStreamService; } PostMapping(value /chat/stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public FluxServerSentEventMapString, Object streamChat(RequestBody MapString, String request) { String prompt request.get(prompt); if (prompt null || prompt.isBlank()) { return Flux.error(new IllegalArgumentException(Prompt 不能为空)); } return llmStreamService.streamLlmResponse(prompt); } }五、 生产级落地避坑指南与高可用治理在真实的大规模生产环境中仅仅写通 SSE 代码是远远不够的。中间网关阻塞、连接无故断开、内存泄露与超时控制等问题是每一个工程架构师必须解决的硬骨头。[ 客户端 ] ──(Keep-Alive / Heartbeat)──► [ Nginx 网关 (proxy_buffering off) ] ──► [ SSE 微服务 ]5.1 网关层优化Nginx / CDN 响应缓冲与超时拦截在许多生产事故中开发者发现本地测试时 SSE 是一字一字吐出来的一上生产环境就变成了“白屏几秒后突然砸出一大段文本”。这 100% 是由于 Nginx 或 CDN 开启了响应缓冲Response BufferingNginx 核心配置解析在负责反向代理的 Nginx 配置中必须对 SSE 路径进行如下特殊配置location /api/v1/chat/stream { proxy_pass http://llm_backend_cluster; # 1. 关键关闭响应缓冲确保流式 Token 实时穿透 proxy_buffering off; proxy_cache off; # 2. 关键关闭 HTTP Chunk 缓存 chunked_transfer_encoding on; # 3. 关键防止 Nginx 因默认 60s 无数据交互而强行断开长连接 proxy_read_timeout 3600s; proxy_send_timeout 3600s; # 4. 支持 HTTP/1.1 长连接 proxy_http_version 1.1; proxy_set_header Connection ; proxy_set_header Host $host; # 5. 禁用 gzip 压缩防止 Gzip 收集满一定字节数才刷盘 gzip off; }5.2 心跳保活机制Heartbeat / Keep-Alive大模型在处理复杂 Reasoning 逻辑如 DeepSeek-R1 或 OpenAI o1时可能需要“思考”长达 10~30 秒在此期间不会向客户端输出任何 Token。在长达数十秒的静默期内客户端、中间防火墙、Nginx 或 Load Balancer如 AWS ALB极易判定连接已超时并主动强行切断 TCP 连接。解决方案定时发送 SSE 注释心跳包服务端必须在一个独立的定时任务或协程中每隔5~10 秒向客户端推送一个 SSE 规范允许的注释行Commentasync def safe_event_generator(): while True: # 产生业务数据的同时如果超过 5s 无数据生成心跳包 # SSE 注释行格式冒号开头客户端会静默忽略但能维持 TCP 活性 yield : heartbeat\n\n await asyncio.sleep(5)5.3 背压控制Backpressure Management与连接泄露背压问题Backpressure大模型生成 Token 的速度远快于前端客户端或网络较差的移动端接收和渲染的速度。如果服务端不加限制地推流会导致后端服务器的内存缓冲区Buffer无限膨胀最终导致 OOM内存溢出。解法在 Java WebFlux 中充分利用 Reactor 的.onBackpressureDrop()或.onBackpressureBuffer(1024)进行策略管控在 Python 中使用带容量限制的asyncio.Queue(maxsize100)来平衡生产与消费速度。客户端主动断开连接处理Client Abort Handling当用户在界面点击“停止生成”或直接关闭网页时客户端会发送 TCP RST 包断开连接。解法后端生成器必须监听连接中断事件如 Python 捕捉asyncio.CancelledError并立即向下游的大模型推理集群如 vLLM, Ollama 或 API发送 Abort 取消请求避免模型在后台继续无效地消耗昂贵的 GPU 算力总结在 AI 原生应用的新时代SSEServer-Sent Events凭借其基于标准 HTTP、极低开销、原生格式规范与良好的网关穿透性成为了连接大模型自回归推理能力与前端用户实时体验的最稳固桥梁。要构建一个生产级的高并发 SSE 架构关键在于前端抛弃简陋的EventSource采用fetchReadableStream配合基于缓冲区的状态机精准防御粘包与半包。后端坚决弃用传统的阻塞式线程池模式全面转向Python Asyncio或Java Spring WebFlux响应式非阻塞架构确保数万并发长连接下的系统稳健。运维与网关调优 Nginx 反向代理配置关闭 Buffer设置心跳保活与客户端中断响应机制切实保障 GPU 算力与服务稳定性。通过本文提供的整套实战方案开发者可以搭建出低延迟、高并发、生产级可靠的大模型流式交互平台。