最近这半年凡是接过大模型服务的开发者基本都遇到过同一个场面调用接口返回的不是一坨完整的 JSON而是一行一行往外蹦的文本。浏览器的 Network 面板里挂着一条特别长的 pending 请求状态码 200但响应迟迟不结束看起来像服务器卡住了。实际上这就是 SSE 在干活。SSE全称 Server-Sent Events服务端推送事件。大模型接口的“打字机效果”依赖它AI 编程助手的上下文增量输出依赖它很多 AI Agent 让用户实时看到工具调用的过程也是用它把中间状态一层层推给前端。可以说在 AI 应用这一波浪潮里SSE 几乎是默认的流式传输方案。这篇文章我会从协议原理、前后端实现、流式解析、断连排查这几个维度把 SSE 从“听说过”讲到“能直接用到项目里”。不管你是后端、前端还是全栈只要在做和 AI 相关的项目这篇都值得花十分钟看完。1. 大模型都在用它SSE 凭什么成了 AI 应用的默认选择1.1 从“一问一答”到“边说边写”AI 应用需要怎样的实时性先想一个问题为什么传统接口很少用 SSE到了 AI 时代它反而成了主流传统 API 是“一次性交易”你发一个请求服务器算完把完整结果一次性返回连接关闭。拿一个普通查询接口举例数据库查 200ms接口 250ms 返回用户没感知因为时间足够短。但大模型生成一段回答短则 1-2 秒长则几十秒。如果让用户对着空白页面等 30 秒然后“啪”地弹出一整段回复这个体验基本是不可接受的。更关键的是大模型领域的“首字延迟”TTFT指标。用户发出 prompt 之后模型可能要在 GPU 上跑几百毫秒甚至几秒才能吐第一个 token。第一个 token 出来得越早用户就越觉得系统“快”。后面的 token 是一个一个生成的整个过程天然就是一个流式序列。SSE 恰好能把这种“边生成边推送”的能力以极低的成本实现出来。一句话总结AI 应用不仅要结果正确还要让用户感知到“它在思考、它在写”这种过程化的体验只有流式协议给得了。而 SSE 就是目前实现流式输出门槛最低、兼容性最稳的方案。1.2 SSE 不是新东西但 AI 让它焕发第二春SSE 协议本身并不新。它随着 HTML5 规范一起出现浏览器端对应的是EventSource接口。十多年前它的典型场景是股票行情推送、实时新闻提醒、服务端通知这类“服务端主动往客户端推数据”的业务。当时 WebSocket 横空出世把大家的目光都吸引走了。WebSocket 能双向通信能传二进制看起来什么都能干SSE 一度被当成“简化版的弱鸡方案”存在感很低。结果 AI 时代一来SSE 反而成了真香选择。原因很直接大模型场景只需要“服务端往客户端推”用户输入是一次性的不需要一条真正意义上的双向长连接。SSE 建立在普通 HTTP 之上不需要像 WebSocket 那样升级协议、做握手握手状态维护任何一个普通的 HTTP 服务端都能输出 SSE。SSE 的文本格式足够简单天然适合 JSON 数据逐段传输。模型输出的 token 按行塞进data:字段前端拿到就渲染整个链路没有额外的序列化和解析成本。我最近看到不少同学的热门搜索里同时出现“java 实现 sse”“vue python sse”“通知 sse”说明大家是真的在项目里遇到了这个需求才回头补这块知识。这不奇怪因为大模型接口都这么干你绕不开。2. SSE 协议拆解一个文本流能玩出什么花2.1 协议层的三个关键设计SSE 本质上还是 HTTP 响应只是它把“一次性返回完整的响应体”改成了“持续不断地往响应体里写内容”。一个标准的 SSE 响应长这样HTTP/1.1 200 OK Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive data: {content: 你} data: {content: 好} data: {content: }几个关键点Content-Type必须是text/event-stream这是浏览器识别 SSE 的唯一标志。连接保持打开服务端不断向客户端写入文本块。每条消息以两个换行符\n\n结尾消息之间靠空行分隔。这和普通 HTTP 响应最大的区别在于普通响应写完就关闭SSE 响应会一直挂着服务端每次有东西要推就往响应流里追加一段文本。很多人会拿 SSE 和 WebSocket 对比。最简单的理解方式WebSocket 是“双向电话”服务端和客户端谁都能发起对话SSE 是“单向广播”只有服务端能主动说话客户端只能竖起耳朵听。在 AI 场景里客户端根本不需要主动推东西给服务端所以 SSE 够用而且省掉了很多 WebSocket 需要处理的复杂度。2.2 事件格式的细节与坑SSE 的每条消息由若干字段组成最常见的四个字段是字段作用说明data消息主体可以有多行多行会被拼接成一个字符串event事件类型不写默认为message浏览器端可通过addEventListener监听不同事件id事件编号用于断线重连时告诉服务端“我收到哪了”retry重连间隔告诉浏览器断线后等待多少毫秒再重连实际项目里最容易踩坑的是data里直接包含换行。比如 AI 回答里出现了一段带换行的 Markdown 代码块如果不对内容做转义直接把换行塞进data字段解析就会被破坏。按照 SSE 规范data字段本身可以出现多行但这一行的语义是“同一消息的延续”需要用换行符分隔的多个data:行来表达data: 第一行内容 data: 第二行内容这条消息最终会被解析成第一行内容\n第二行内容。如果内容是动态生成的尤其是 AI 模型输出的文本处理时要格外小心要么把内容里的\n转义成字面量要么服务端统一做多行data处理否则前端拿到的数据会对不齐。另外还有一个很容易忽略的用法注释行。SSE 允许以冒号:开头的一行作为注释例如: ping注释行不会被当作事件处理浏览器会直接忽略。但它在真实项目中特别有用——服务端可以把它当心跳包发出去既能维持连接不被代理切断又不会在业务层产生多余事件。这个技巧我在后面的保活章节会再展开。2.3 自动重连比 WebSocket 省心的地方SSE 有一个原生机制是被很多人忽略的好东西自动重连。只要连接非正常关闭浏览器端的EventSource会自动重新发起请求不需要写任何重连逻辑。如果服务端在断线前发过id字段重连时请求头里会自动带上Last-Event-ID服务端根据这个编号就能判断客户端漏了哪些事件然后把缺口补上。这一点在真实场景里价值非常大。移动端网络不稳定用户偶尔切个 Wi-Fi如果是 WebSocket 方案断了就断了通常要推倒重来或者自己写整套重连状态机。SSE 只要在服务端稍微配合一下事件编号断线续传的成本低一个数量级。当然用fetch自己做流式读取的时候自动重连和Last-Event-ID这些能力都要自己实现。这也是很多封装库存在的原因——后面我会单开一节讲。3. 真实项目里的 SSE 实现方案Java、Python、Vue 全链路3.1 后端 JavaSseEmitter 的正确用法Spring 框架从 4.2 开始提供了SseEmitter专门用来输出 SSE 响应。我在实际项目里的最小实现大概是这样的RestController RequestMapping(/api/ai) public class ChatController { private final ExecutorService executor Executors.newFixedThreadPool(16); GetMapping(value /chat/stream, produces text/event-stream) public SseEmitter chatStream(RequestParam String prompt) { // 超时时间设为 0表示永不超时 SseEmitter emitter new SseEmitter(0L); executor.execute(() - { try { ListString tokens mockAiGenerate(prompt); for (String token : tokens) { // 这里可以是 JSON 字符串 emitter.send(SseEmitter.event() .id(UUID.randomUUID().toString()) .data({\content\:\ token \})); Thread.sleep(50); } emitter.complete(); } catch (Exception e) { emitter.completeWithError(e); } }); return emitter; } }这段代码有几个细节值得展开produces text/event-stream是为了明确接口的返回类型很多代理层依赖这个 Content-Type 做识别。new SseEmitter(0L)里的 0 表示不超时。默认构造器会带上 30 秒超时AI 场景下模型思考时间可能超过 30 秒不设超时是必须的。SseEmitter.event()是 Spring 提供的构造器能生成带id、data、event的 SSE 事件块。相比直接send()一个字符串业务代码会清晰很多。Thread.sleep(50)是模拟模型逐 token 返回。真实项目里这里应该是调用大模型 SDK 的流式接口把回调拿到的 token 逐个 send。注意SseEmitter的异步是建立在服务端自己管理线程池上的。你调用了emitter.send()之后响应数据才会真正写到底层连接。如果线程池被耗尽或者发送失败客户端看到的就是连接被掐断、报idle timeout之类的错误。生产环境建议单独维护一个线程池不要让业务线程池和 SSE 发送线程混在一起。3.2 PythonFlask 和 FastAPI 各有各的写法Python 生态里实现 SSE 也很简单。Flask 可以通过Response生成器来写from flask import Flask, Response import json import time app Flask(__name__) def mock_ai_generate(prompt): # 模拟模型流式返回 for i in range(10): yield {content: ftoken-{i}, prompt: prompt} app.route(/api/ai/chat/stream) def chat_stream(): def generate(): for token in mock_ai_generate(request.args.get(prompt, )): data json.dumps(token, ensure_asciiFalse) yield fdata: {data}\n\n time.sleep(0.05) yield data: [DONE]\n\n return Response(generate(), mimetypetext/event-stream)Flask 的核心是generate()生成器函数。函数里的yield每次产生一段文本Response把它包装成流式 HTTP 响应。注意这里模拟的[DONE]是很多大模型厂商约定的结束标记前端解析到它就知道流结束了。如果用 FastAPI更推荐直接返回StreamingResponse或者用第三方库sse-starlette。我自己在 FastAPI 项目里习惯直接用StreamingResponse因为依赖少、可控制性强from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app FastAPI() app.post(/api/ai/chat/stream) async def chat_stream(request: Request): async def event_generator(): # 这里从 request body 里解析 prompt再调用模型流式接口 async for token in fetch_tokens_from_model(): yield fdata: {json.dumps({content: token}, ensure_asciiFalse)}\n\n if await request.is_disconnected(): break yield data: [DONE]\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)FastAPI 的StreamingResponse和 Flask 思路一致但因为是异步框架request.is_disconnected()可以帮我们检测客户端是否已经断开。这一句在真实项目中很关键客户端关掉页面后服务端还在傻傻地跑模型生成纯属浪费 GPU 资源。加上这个判断一旦感知到断连立刻停掉生成器省下来的算力可不少。3.3 前端 Vue为什么我用 fetch 而不是 EventSource前端接受 SSE 的方式有两种一种是用浏览器原生EventSource一种是用fetch自己解析流式响应。EventSource简单几行代码就能跑起来const eventSource new EventSource(/api/ai/chat/stream); eventSource.onmessage (event) { console.log(event.data); };但它在真实 AI 场景里几乎不可用因为EventSource只支持 GET 请求不能在请求头里自定义Authorization和Content-Type。大模型接口通常要带 API Key 或者鉴权 Token你至少得自定义一个请求头所以大多数 AI 项目都用fetch方案。在 Vue 里我会这样封装一个最小可用的流式接收函数async function fetchSSE(url, options) { const response await fetch(url, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${token} }, body: JSON.stringify(options.body) }); if (!response.ok) { throw new Error(HTTP ${response.status}); } const reader response.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 }); // SSE 消息以空行分隔 const events buffer.split(\n\n); buffer events.pop(); // 最后一段可能不完整留到下一轮 for (const rawEvent of events) { const parsed parseSSEEvent(rawEvent); if (parsed.data [DONE]) { return; } onMessage?.(parsed); } } } function parseSSEEvent(rawEvent) { const lines rawEvent.split(\n); const dataLines []; let event message; for (const line of lines) { if (line.startsWith(event:)) { event line.slice(6).trim(); } else if (line.startsWith(data:)) { dataLines.push(line.slice(5).trimStart()); } } return { event, data: dataLines.join(\n) }; }这里的核心是buffer拼接逻辑。reader.read()每次拿到的 chunk 不一定刚好是一条完整消息可能半条、可能好几条都挤在一起所以不能直接解析。正确做法是先把 chunk 解码后丢进buffer按\n\n切分最后残留的不完整数据留到下一轮继续拼。这个模式几乎适用于所有基于流的文本协议解析不止 SSE。4. 流式解析与封装把 SSE 接入业务逻辑的工程细节4.1 流式消息解析的完整链路很多同学第一次处理 SSE 都会问data:后面那堆乱七八糟的字符串到底该怎么解析成规范的对象这个问题的答案取决于你对协议格式的理解有多深。一条完整的 SSE 事件流从前端视角看是这样的data: {type: thinking, content: 正在分析问题} data: {type: delta, content: 你} data: {type: delta, content: 好} data: [DONE]解析步骤可以拆解成五层解码网络层拿到的是字节流先用TextDecoder解码成字符串注意要开stream: true模式否则多字节字符比如中文在 chunk 边界处会被拆坏。缓冲解码后的字符串进buffer等待完整消息。切分用\n\n或\r\n\r\n把 buffer 切成一个个事件块。最后一个不完整的事件块留在 buffer 里继续拼接。字段解析对每个事件块按行拆分处理data:、event:、id:、retry:等字段。多个data:行拼接成一个完整数据体。业务处理把解析出的数据交给业务层驱动 UI 更新。第五层最容易写乱。因为大模型接口的 SSE 事件不同厂商格式不一样有的会在开头发一个 role 说明有的中间穿插思考过程有的以[DONE]结尾有的干脆不发结束标记靠服务端主动断连接。所以封装的时候onMessage回调不应该直接绑定 UI 渲染而应该先做一层“归一化”把不同厂商的格式统一成本项目内部的消息结构再交给页面层处理。否则换个模型厂商前端代码就要重写一遍。4.2 那个让人头大的报错stream disconnected before completion: idle timeout waiting for sse热门搜索里有一条很扎眼的报错stream disconnected before completion: idle timeout waiting for sse。我不止一次在项目里遇到过这种问题而且它出现的时机总是“看起来一切都正常但等了几十秒突然断了”。先说结论这个报错的意思是在 SSE 连接上等待服务端消息的时间超过了空闲超时阈值连接被某层组件强制关闭。至于是哪个组件断的需要按链路排查。我的一次真实排查经历是这样的项目里接了一个大模型接口前端用 fetch 读取流式响应。一开始输出正常但用户提问一个特别复杂的问题时模型在推理阶段可能要沉默十几秒甚至更久然后浏览器就报了idle timeout waiting for sse。我用了三步定位第一步确认是不是服务端问题。查看服务端日志发现模型调用还在正常执行说明服务端进程没挂也没主动断开连接。第二步确认是不是网关/代理问题。项目前端经过 Nginx 转发proxy_read_timeout默认 60 秒但模型推理时间不稳定一旦沉默超过 60 秒Nginx 就会主动断开。这就解释了为什么短问题没事、长问题必断。第三步改了 Nginx 配置还不够因为问题还可能出现在云厂商的负载均衡层或者 API 网关层。我开始在服务端加心跳每 15 秒往 SSE 流里写一行注释: ping\n\n。这样代理层的“最近一次数据活动”计时器不断被刷新空闲超时就不会触发了。这个解决方案几乎是 SSE 项目的标配服务端定期发心跳把所有中间层都“喂饱”。心跳内容用注释行最好因为不用修改业务数据格式浏览器解析时自动忽略非常干净。如果心跳加了还是断那就要看是不是客户端自己的超时设置有问题。比如fetch如果用了AbortController的 timeout或者某些前端框架有默认请求超时都会掐掉长期不响应数据的请求。这时候在客户端也把超时策略解除只依赖业务层的空转检测问题基本都能解决。4.3 封装一个通用 SSE 客户端经历过几次踩坑之后我养成了一个习惯SSE 的客户端一定不能散落在页面组件里必须抽成一个独立的模块。我通常会封装一个SSEClient基类暴露几个核心能力connect(url, options)建立连接支持 GET 和 POST。on(type, handler)监听不同类型的 SSE 事件。disconnect()主动断开。自动重连机制断线后按指数退避策略重试重试次数可配置。回调统一处理[DONE]结束标记、错误事件、网络异常都收敛到内部处理。TypeScript 类的大致结构长这样export class SSEClient { private reader?: ReadableStreamDefaultReaderUint8Array; private controller new AbortController(); private buffer ; private retryCount 0; private reconnectTimer?: ReturnTypetypeof setTimeout; constructor(private url: string, private options: RequestInit { onEvent: (event: string, data: string) void }) {} async connect() { const response await fetch(this.url, { ...this.options, signal: this.controller.signal, }); if (!response.ok) { throw new Error(SSE connection failed: ${response.status}); } this.reader response.body!.getReader(); const decoder new TextDecoder(utf-8); while (true) { const { done, value } await this.reader.read(); if (done) { this.handleDisconnect(); return; } this.buffer decoder.decode(value, { stream: true }); const events this.buffer.split(\n\n); this.buffer events.pop() ?? ; for (const rawEvent of events) { const { event, data } this.parseEvent(rawEvent); this.options.onEvent(event, data); } } } private parseEvent(rawEvent: string) { // 解析 data: / event: 等字段 } private handleDisconnect() { if (this.controller.signal.aborted) return; const delay Math.min(1000 * 2 ** this.retryCount, 30000); this.reconnectTimer setTimeout(() { this.retryCount; this.connect(); }, delay); } disconnect() { this.controller.abort(); clearTimeout(this.reconnectTimer); } }这里的指数退避逻辑是参考 TCP 重传的经典思路第一次等 1 秒第二次 2 秒第三次 4 秒最高封顶 30 秒。既不会在服务端瞬时异常时疯狂打请求又能保证恢复后尽快自动接上。5. 踩坑总结代理缓冲、心跳保活、选型边界5.1 Nginx 和网关层缓冲导致流式“失效”这个问题几乎每个接 SSE 的人都遇到过后端明明写了流式输出前端看到的效果却是一段话攒了半天突然一次性出现完全没有打字机的流畅感。排查方向先看 Nginx。Nginx 默认开启了proxy_buffering它会把上游响应整个缓冲到本地攒够了一定大小或者上游关闭连接之后才一次性转发给客户端。这在普通接口场景下问题不大但对 SSE 是致命的——流式效果等于没了而且用户的等待体验会变得很差。我的解决方案是在 Nginx 配置里为流式接口关掉缓冲同时禁用代理缓存并调大读取超时location /api/ai/ { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_read_timeout 60s; proxy_send_timeout 60s; proxy_set_header Connection ; proxy_http_version 1.1; }proxy_buffering off让上游数据及时透传proxy_http_version 1.1是为了避免 HTTP 1.0 的 keep-alive 限制。如果你的后端应用整体走的是云负载均衡注意看服务商有没有类似的缓冲开关比如阿里云 SLB 的“响应压缩”和“缓冲”设置。另外后端应用自身也可能开启响应缓冲。比如 Spring Boot 里如果加了一些拦截器或者 server 配置把server.servle t.response.characterEncoding之类的设置改得太激进也可能导致流式失效。总的来说凡是链路里存在“攒一批再发”的默认行为都要检查一遍。5.2 心跳间隔怎么定才合理心跳是 SSE 项目里绕不开的话题。心跳间隔太短服务端每秒发一个 ping虽然客户端会忽略但网络开销和日志噪音不小心跳间隔太长又可能越过代理的空闲超时阈值连接被掐断。我的经验是心跳间隔设为代理空闲超时时间的一半。比如 Nginxproxy_read_timeout是 60 秒心跳就设 20-30 秒网关层如果是 30 秒心跳就设 10-15 秒。这个比例给网络抖动留下了足够余量又不会造成无谓的请求量。有些大模型 SDK 或 OctoAI 类 SDK 在自己包里提供了 keepalive 配置底层就是定时发注释行。如果你是自己封装后端流式接口建议把这部分做成可配置项不要写死。因为在开发环境、测试环境、生产环境的代理策略通常不一样硬编码一个间隔很可能换个环境就出问题。5.3 什么时候别用 SSE虽然 SSE 在 AI 场景里很顺手但它不是万能的。如果在项目里遇到下面几类需求大概率不合适需要服务端和客户端双向实时通信。比如在线协作编辑用户 A 的每次按键都要同步给用户 B同时 B 也可能反过来给 A 推数据这种场景用 WebSocket 更合理。需要传输二进制数据。SSE 本质是基于文本的虽然可以通过 base64 编码塞二进制内容但效率和原生二进制支持差距很大。WebSocket 直接支持二进制帧这才是它的主场。连接规模极高且长时间挂载。SSE 每个连接占一个 HTTP 长连接服务端连接数容易成为瓶颈。虽然 WebSocket 也占连接但 SSE 通常还会配合 HTTP 连接池、负载均衡等因素架构上的成本需要提前评估。需要向后兼容老浏览器。现代浏览器对 SSE 的支持没问题但如果你要兼容旧版 IE 或某些嵌入式 WebView就得引入 polyfill 或者 fallback 方案多出来的工作量可能比想象中多。到了这里SSE 的原理、实现、踩坑和选型基本都过了一遍。最后分享一个个人经验我后来在项目里把 SSE 解析逻辑抽成了一个通用的“行式流解析器”不跟具体的 SSE 字段绑定只负责 buffer 拼接和按分割符切块。这样即使以后接口从 SSE 换成了别的基于行的流式协议解析层几乎不用动只需要换一层字段解析。这个抽象思路比死磕某一个接口的格式要长远得多。
SSE协议：大模型流式输出与AI应用实时推送的工程实践
最近这半年凡是接过大模型服务的开发者基本都遇到过同一个场面调用接口返回的不是一坨完整的 JSON而是一行一行往外蹦的文本。浏览器的 Network 面板里挂着一条特别长的 pending 请求状态码 200但响应迟迟不结束看起来像服务器卡住了。实际上这就是 SSE 在干活。SSE全称 Server-Sent Events服务端推送事件。大模型接口的“打字机效果”依赖它AI 编程助手的上下文增量输出依赖它很多 AI Agent 让用户实时看到工具调用的过程也是用它把中间状态一层层推给前端。可以说在 AI 应用这一波浪潮里SSE 几乎是默认的流式传输方案。这篇文章我会从协议原理、前后端实现、流式解析、断连排查这几个维度把 SSE 从“听说过”讲到“能直接用到项目里”。不管你是后端、前端还是全栈只要在做和 AI 相关的项目这篇都值得花十分钟看完。1. 大模型都在用它SSE 凭什么成了 AI 应用的默认选择1.1 从“一问一答”到“边说边写”AI 应用需要怎样的实时性先想一个问题为什么传统接口很少用 SSE到了 AI 时代它反而成了主流传统 API 是“一次性交易”你发一个请求服务器算完把完整结果一次性返回连接关闭。拿一个普通查询接口举例数据库查 200ms接口 250ms 返回用户没感知因为时间足够短。但大模型生成一段回答短则 1-2 秒长则几十秒。如果让用户对着空白页面等 30 秒然后“啪”地弹出一整段回复这个体验基本是不可接受的。更关键的是大模型领域的“首字延迟”TTFT指标。用户发出 prompt 之后模型可能要在 GPU 上跑几百毫秒甚至几秒才能吐第一个 token。第一个 token 出来得越早用户就越觉得系统“快”。后面的 token 是一个一个生成的整个过程天然就是一个流式序列。SSE 恰好能把这种“边生成边推送”的能力以极低的成本实现出来。一句话总结AI 应用不仅要结果正确还要让用户感知到“它在思考、它在写”这种过程化的体验只有流式协议给得了。而 SSE 就是目前实现流式输出门槛最低、兼容性最稳的方案。1.2 SSE 不是新东西但 AI 让它焕发第二春SSE 协议本身并不新。它随着 HTML5 规范一起出现浏览器端对应的是EventSource接口。十多年前它的典型场景是股票行情推送、实时新闻提醒、服务端通知这类“服务端主动往客户端推数据”的业务。当时 WebSocket 横空出世把大家的目光都吸引走了。WebSocket 能双向通信能传二进制看起来什么都能干SSE 一度被当成“简化版的弱鸡方案”存在感很低。结果 AI 时代一来SSE 反而成了真香选择。原因很直接大模型场景只需要“服务端往客户端推”用户输入是一次性的不需要一条真正意义上的双向长连接。SSE 建立在普通 HTTP 之上不需要像 WebSocket 那样升级协议、做握手握手状态维护任何一个普通的 HTTP 服务端都能输出 SSE。SSE 的文本格式足够简单天然适合 JSON 数据逐段传输。模型输出的 token 按行塞进data:字段前端拿到就渲染整个链路没有额外的序列化和解析成本。我最近看到不少同学的热门搜索里同时出现“java 实现 sse”“vue python sse”“通知 sse”说明大家是真的在项目里遇到了这个需求才回头补这块知识。这不奇怪因为大模型接口都这么干你绕不开。2. SSE 协议拆解一个文本流能玩出什么花2.1 协议层的三个关键设计SSE 本质上还是 HTTP 响应只是它把“一次性返回完整的响应体”改成了“持续不断地往响应体里写内容”。一个标准的 SSE 响应长这样HTTP/1.1 200 OK Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive data: {content: 你} data: {content: 好} data: {content: }几个关键点Content-Type必须是text/event-stream这是浏览器识别 SSE 的唯一标志。连接保持打开服务端不断向客户端写入文本块。每条消息以两个换行符\n\n结尾消息之间靠空行分隔。这和普通 HTTP 响应最大的区别在于普通响应写完就关闭SSE 响应会一直挂着服务端每次有东西要推就往响应流里追加一段文本。很多人会拿 SSE 和 WebSocket 对比。最简单的理解方式WebSocket 是“双向电话”服务端和客户端谁都能发起对话SSE 是“单向广播”只有服务端能主动说话客户端只能竖起耳朵听。在 AI 场景里客户端根本不需要主动推东西给服务端所以 SSE 够用而且省掉了很多 WebSocket 需要处理的复杂度。2.2 事件格式的细节与坑SSE 的每条消息由若干字段组成最常见的四个字段是字段作用说明data消息主体可以有多行多行会被拼接成一个字符串event事件类型不写默认为message浏览器端可通过addEventListener监听不同事件id事件编号用于断线重连时告诉服务端“我收到哪了”retry重连间隔告诉浏览器断线后等待多少毫秒再重连实际项目里最容易踩坑的是data里直接包含换行。比如 AI 回答里出现了一段带换行的 Markdown 代码块如果不对内容做转义直接把换行塞进data字段解析就会被破坏。按照 SSE 规范data字段本身可以出现多行但这一行的语义是“同一消息的延续”需要用换行符分隔的多个data:行来表达data: 第一行内容 data: 第二行内容这条消息最终会被解析成第一行内容\n第二行内容。如果内容是动态生成的尤其是 AI 模型输出的文本处理时要格外小心要么把内容里的\n转义成字面量要么服务端统一做多行data处理否则前端拿到的数据会对不齐。另外还有一个很容易忽略的用法注释行。SSE 允许以冒号:开头的一行作为注释例如: ping注释行不会被当作事件处理浏览器会直接忽略。但它在真实项目中特别有用——服务端可以把它当心跳包发出去既能维持连接不被代理切断又不会在业务层产生多余事件。这个技巧我在后面的保活章节会再展开。2.3 自动重连比 WebSocket 省心的地方SSE 有一个原生机制是被很多人忽略的好东西自动重连。只要连接非正常关闭浏览器端的EventSource会自动重新发起请求不需要写任何重连逻辑。如果服务端在断线前发过id字段重连时请求头里会自动带上Last-Event-ID服务端根据这个编号就能判断客户端漏了哪些事件然后把缺口补上。这一点在真实场景里价值非常大。移动端网络不稳定用户偶尔切个 Wi-Fi如果是 WebSocket 方案断了就断了通常要推倒重来或者自己写整套重连状态机。SSE 只要在服务端稍微配合一下事件编号断线续传的成本低一个数量级。当然用fetch自己做流式读取的时候自动重连和Last-Event-ID这些能力都要自己实现。这也是很多封装库存在的原因——后面我会单开一节讲。3. 真实项目里的 SSE 实现方案Java、Python、Vue 全链路3.1 后端 JavaSseEmitter 的正确用法Spring 框架从 4.2 开始提供了SseEmitter专门用来输出 SSE 响应。我在实际项目里的最小实现大概是这样的RestController RequestMapping(/api/ai) public class ChatController { private final ExecutorService executor Executors.newFixedThreadPool(16); GetMapping(value /chat/stream, produces text/event-stream) public SseEmitter chatStream(RequestParam String prompt) { // 超时时间设为 0表示永不超时 SseEmitter emitter new SseEmitter(0L); executor.execute(() - { try { ListString tokens mockAiGenerate(prompt); for (String token : tokens) { // 这里可以是 JSON 字符串 emitter.send(SseEmitter.event() .id(UUID.randomUUID().toString()) .data({\content\:\ token \})); Thread.sleep(50); } emitter.complete(); } catch (Exception e) { emitter.completeWithError(e); } }); return emitter; } }这段代码有几个细节值得展开produces text/event-stream是为了明确接口的返回类型很多代理层依赖这个 Content-Type 做识别。new SseEmitter(0L)里的 0 表示不超时。默认构造器会带上 30 秒超时AI 场景下模型思考时间可能超过 30 秒不设超时是必须的。SseEmitter.event()是 Spring 提供的构造器能生成带id、data、event的 SSE 事件块。相比直接send()一个字符串业务代码会清晰很多。Thread.sleep(50)是模拟模型逐 token 返回。真实项目里这里应该是调用大模型 SDK 的流式接口把回调拿到的 token 逐个 send。注意SseEmitter的异步是建立在服务端自己管理线程池上的。你调用了emitter.send()之后响应数据才会真正写到底层连接。如果线程池被耗尽或者发送失败客户端看到的就是连接被掐断、报idle timeout之类的错误。生产环境建议单独维护一个线程池不要让业务线程池和 SSE 发送线程混在一起。3.2 PythonFlask 和 FastAPI 各有各的写法Python 生态里实现 SSE 也很简单。Flask 可以通过Response生成器来写from flask import Flask, Response import json import time app Flask(__name__) def mock_ai_generate(prompt): # 模拟模型流式返回 for i in range(10): yield {content: ftoken-{i}, prompt: prompt} app.route(/api/ai/chat/stream) def chat_stream(): def generate(): for token in mock_ai_generate(request.args.get(prompt, )): data json.dumps(token, ensure_asciiFalse) yield fdata: {data}\n\n time.sleep(0.05) yield data: [DONE]\n\n return Response(generate(), mimetypetext/event-stream)Flask 的核心是generate()生成器函数。函数里的yield每次产生一段文本Response把它包装成流式 HTTP 响应。注意这里模拟的[DONE]是很多大模型厂商约定的结束标记前端解析到它就知道流结束了。如果用 FastAPI更推荐直接返回StreamingResponse或者用第三方库sse-starlette。我自己在 FastAPI 项目里习惯直接用StreamingResponse因为依赖少、可控制性强from fastapi import FastAPI, Request from fastapi.responses import StreamingResponse app FastAPI() app.post(/api/ai/chat/stream) async def chat_stream(request: Request): async def event_generator(): # 这里从 request body 里解析 prompt再调用模型流式接口 async for token in fetch_tokens_from_model(): yield fdata: {json.dumps({content: token}, ensure_asciiFalse)}\n\n if await request.is_disconnected(): break yield data: [DONE]\n\n return StreamingResponse(event_generator(), media_typetext/event-stream)FastAPI 的StreamingResponse和 Flask 思路一致但因为是异步框架request.is_disconnected()可以帮我们检测客户端是否已经断开。这一句在真实项目中很关键客户端关掉页面后服务端还在傻傻地跑模型生成纯属浪费 GPU 资源。加上这个判断一旦感知到断连立刻停掉生成器省下来的算力可不少。3.3 前端 Vue为什么我用 fetch 而不是 EventSource前端接受 SSE 的方式有两种一种是用浏览器原生EventSource一种是用fetch自己解析流式响应。EventSource简单几行代码就能跑起来const eventSource new EventSource(/api/ai/chat/stream); eventSource.onmessage (event) { console.log(event.data); };但它在真实 AI 场景里几乎不可用因为EventSource只支持 GET 请求不能在请求头里自定义Authorization和Content-Type。大模型接口通常要带 API Key 或者鉴权 Token你至少得自定义一个请求头所以大多数 AI 项目都用fetch方案。在 Vue 里我会这样封装一个最小可用的流式接收函数async function fetchSSE(url, options) { const response await fetch(url, { method: POST, headers: { Content-Type: application/json, Authorization: Bearer ${token} }, body: JSON.stringify(options.body) }); if (!response.ok) { throw new Error(HTTP ${response.status}); } const reader response.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 }); // SSE 消息以空行分隔 const events buffer.split(\n\n); buffer events.pop(); // 最后一段可能不完整留到下一轮 for (const rawEvent of events) { const parsed parseSSEEvent(rawEvent); if (parsed.data [DONE]) { return; } onMessage?.(parsed); } } } function parseSSEEvent(rawEvent) { const lines rawEvent.split(\n); const dataLines []; let event message; for (const line of lines) { if (line.startsWith(event:)) { event line.slice(6).trim(); } else if (line.startsWith(data:)) { dataLines.push(line.slice(5).trimStart()); } } return { event, data: dataLines.join(\n) }; }这里的核心是buffer拼接逻辑。reader.read()每次拿到的 chunk 不一定刚好是一条完整消息可能半条、可能好几条都挤在一起所以不能直接解析。正确做法是先把 chunk 解码后丢进buffer按\n\n切分最后残留的不完整数据留到下一轮继续拼。这个模式几乎适用于所有基于流的文本协议解析不止 SSE。4. 流式解析与封装把 SSE 接入业务逻辑的工程细节4.1 流式消息解析的完整链路很多同学第一次处理 SSE 都会问data:后面那堆乱七八糟的字符串到底该怎么解析成规范的对象这个问题的答案取决于你对协议格式的理解有多深。一条完整的 SSE 事件流从前端视角看是这样的data: {type: thinking, content: 正在分析问题} data: {type: delta, content: 你} data: {type: delta, content: 好} data: [DONE]解析步骤可以拆解成五层解码网络层拿到的是字节流先用TextDecoder解码成字符串注意要开stream: true模式否则多字节字符比如中文在 chunk 边界处会被拆坏。缓冲解码后的字符串进buffer等待完整消息。切分用\n\n或\r\n\r\n把 buffer 切成一个个事件块。最后一个不完整的事件块留在 buffer 里继续拼接。字段解析对每个事件块按行拆分处理data:、event:、id:、retry:等字段。多个data:行拼接成一个完整数据体。业务处理把解析出的数据交给业务层驱动 UI 更新。第五层最容易写乱。因为大模型接口的 SSE 事件不同厂商格式不一样有的会在开头发一个 role 说明有的中间穿插思考过程有的以[DONE]结尾有的干脆不发结束标记靠服务端主动断连接。所以封装的时候onMessage回调不应该直接绑定 UI 渲染而应该先做一层“归一化”把不同厂商的格式统一成本项目内部的消息结构再交给页面层处理。否则换个模型厂商前端代码就要重写一遍。4.2 那个让人头大的报错stream disconnected before completion: idle timeout waiting for sse热门搜索里有一条很扎眼的报错stream disconnected before completion: idle timeout waiting for sse。我不止一次在项目里遇到过这种问题而且它出现的时机总是“看起来一切都正常但等了几十秒突然断了”。先说结论这个报错的意思是在 SSE 连接上等待服务端消息的时间超过了空闲超时阈值连接被某层组件强制关闭。至于是哪个组件断的需要按链路排查。我的一次真实排查经历是这样的项目里接了一个大模型接口前端用 fetch 读取流式响应。一开始输出正常但用户提问一个特别复杂的问题时模型在推理阶段可能要沉默十几秒甚至更久然后浏览器就报了idle timeout waiting for sse。我用了三步定位第一步确认是不是服务端问题。查看服务端日志发现模型调用还在正常执行说明服务端进程没挂也没主动断开连接。第二步确认是不是网关/代理问题。项目前端经过 Nginx 转发proxy_read_timeout默认 60 秒但模型推理时间不稳定一旦沉默超过 60 秒Nginx 就会主动断开。这就解释了为什么短问题没事、长问题必断。第三步改了 Nginx 配置还不够因为问题还可能出现在云厂商的负载均衡层或者 API 网关层。我开始在服务端加心跳每 15 秒往 SSE 流里写一行注释: ping\n\n。这样代理层的“最近一次数据活动”计时器不断被刷新空闲超时就不会触发了。这个解决方案几乎是 SSE 项目的标配服务端定期发心跳把所有中间层都“喂饱”。心跳内容用注释行最好因为不用修改业务数据格式浏览器解析时自动忽略非常干净。如果心跳加了还是断那就要看是不是客户端自己的超时设置有问题。比如fetch如果用了AbortController的 timeout或者某些前端框架有默认请求超时都会掐掉长期不响应数据的请求。这时候在客户端也把超时策略解除只依赖业务层的空转检测问题基本都能解决。4.3 封装一个通用 SSE 客户端经历过几次踩坑之后我养成了一个习惯SSE 的客户端一定不能散落在页面组件里必须抽成一个独立的模块。我通常会封装一个SSEClient基类暴露几个核心能力connect(url, options)建立连接支持 GET 和 POST。on(type, handler)监听不同类型的 SSE 事件。disconnect()主动断开。自动重连机制断线后按指数退避策略重试重试次数可配置。回调统一处理[DONE]结束标记、错误事件、网络异常都收敛到内部处理。TypeScript 类的大致结构长这样export class SSEClient { private reader?: ReadableStreamDefaultReaderUint8Array; private controller new AbortController(); private buffer ; private retryCount 0; private reconnectTimer?: ReturnTypetypeof setTimeout; constructor(private url: string, private options: RequestInit { onEvent: (event: string, data: string) void }) {} async connect() { const response await fetch(this.url, { ...this.options, signal: this.controller.signal, }); if (!response.ok) { throw new Error(SSE connection failed: ${response.status}); } this.reader response.body!.getReader(); const decoder new TextDecoder(utf-8); while (true) { const { done, value } await this.reader.read(); if (done) { this.handleDisconnect(); return; } this.buffer decoder.decode(value, { stream: true }); const events this.buffer.split(\n\n); this.buffer events.pop() ?? ; for (const rawEvent of events) { const { event, data } this.parseEvent(rawEvent); this.options.onEvent(event, data); } } } private parseEvent(rawEvent: string) { // 解析 data: / event: 等字段 } private handleDisconnect() { if (this.controller.signal.aborted) return; const delay Math.min(1000 * 2 ** this.retryCount, 30000); this.reconnectTimer setTimeout(() { this.retryCount; this.connect(); }, delay); } disconnect() { this.controller.abort(); clearTimeout(this.reconnectTimer); } }这里的指数退避逻辑是参考 TCP 重传的经典思路第一次等 1 秒第二次 2 秒第三次 4 秒最高封顶 30 秒。既不会在服务端瞬时异常时疯狂打请求又能保证恢复后尽快自动接上。5. 踩坑总结代理缓冲、心跳保活、选型边界5.1 Nginx 和网关层缓冲导致流式“失效”这个问题几乎每个接 SSE 的人都遇到过后端明明写了流式输出前端看到的效果却是一段话攒了半天突然一次性出现完全没有打字机的流畅感。排查方向先看 Nginx。Nginx 默认开启了proxy_buffering它会把上游响应整个缓冲到本地攒够了一定大小或者上游关闭连接之后才一次性转发给客户端。这在普通接口场景下问题不大但对 SSE 是致命的——流式效果等于没了而且用户的等待体验会变得很差。我的解决方案是在 Nginx 配置里为流式接口关掉缓冲同时禁用代理缓存并调大读取超时location /api/ai/ { proxy_pass http://backend; proxy_buffering off; proxy_cache off; proxy_read_timeout 60s; proxy_send_timeout 60s; proxy_set_header Connection ; proxy_http_version 1.1; }proxy_buffering off让上游数据及时透传proxy_http_version 1.1是为了避免 HTTP 1.0 的 keep-alive 限制。如果你的后端应用整体走的是云负载均衡注意看服务商有没有类似的缓冲开关比如阿里云 SLB 的“响应压缩”和“缓冲”设置。另外后端应用自身也可能开启响应缓冲。比如 Spring Boot 里如果加了一些拦截器或者 server 配置把server.servle t.response.characterEncoding之类的设置改得太激进也可能导致流式失效。总的来说凡是链路里存在“攒一批再发”的默认行为都要检查一遍。5.2 心跳间隔怎么定才合理心跳是 SSE 项目里绕不开的话题。心跳间隔太短服务端每秒发一个 ping虽然客户端会忽略但网络开销和日志噪音不小心跳间隔太长又可能越过代理的空闲超时阈值连接被掐断。我的经验是心跳间隔设为代理空闲超时时间的一半。比如 Nginxproxy_read_timeout是 60 秒心跳就设 20-30 秒网关层如果是 30 秒心跳就设 10-15 秒。这个比例给网络抖动留下了足够余量又不会造成无谓的请求量。有些大模型 SDK 或 OctoAI 类 SDK 在自己包里提供了 keepalive 配置底层就是定时发注释行。如果你是自己封装后端流式接口建议把这部分做成可配置项不要写死。因为在开发环境、测试环境、生产环境的代理策略通常不一样硬编码一个间隔很可能换个环境就出问题。5.3 什么时候别用 SSE虽然 SSE 在 AI 场景里很顺手但它不是万能的。如果在项目里遇到下面几类需求大概率不合适需要服务端和客户端双向实时通信。比如在线协作编辑用户 A 的每次按键都要同步给用户 B同时 B 也可能反过来给 A 推数据这种场景用 WebSocket 更合理。需要传输二进制数据。SSE 本质是基于文本的虽然可以通过 base64 编码塞二进制内容但效率和原生二进制支持差距很大。WebSocket 直接支持二进制帧这才是它的主场。连接规模极高且长时间挂载。SSE 每个连接占一个 HTTP 长连接服务端连接数容易成为瓶颈。虽然 WebSocket 也占连接但 SSE 通常还会配合 HTTP 连接池、负载均衡等因素架构上的成本需要提前评估。需要向后兼容老浏览器。现代浏览器对 SSE 的支持没问题但如果你要兼容旧版 IE 或某些嵌入式 WebView就得引入 polyfill 或者 fallback 方案多出来的工作量可能比想象中多。到了这里SSE 的原理、实现、踩坑和选型基本都过了一遍。最后分享一个个人经验我后来在项目里把 SSE 解析逻辑抽成了一个通用的“行式流解析器”不跟具体的 SSE 字段绑定只负责 buffer 拼接和按分割符切块。这样即使以后接口从 SSE 换成了别的基于行的流式协议解析层几乎不用动只需要换一层字段解析。这个抽象思路比死磕某一个接口的格式要长远得多。
