1. 流式解析到底在解决什么问题先把场景说清楚。你调用一个大模型接口问它“帮我写一段快速排序”如果走传统的请求-响应模式客户端发一个 POST服务端把整段回答在内存里拼完再一次性返回。用户看到的就是转圈、转圈、转圈然后“啪”一下整段文字全出来。短回答还好一旦回答上千字等待时间可能十几秒体验非常糟糕。流式解析要解决的就是这个“等待焦虑”。它让服务端每生成一小段内容就立刻推给客户端客户端边收边渲染用户看到文字像打字机一样一个个蹦出来。这背后涉及三个层面的工程问题传输协议怎么选、数据流怎么切分、前端怎么消费和渲染。这三个问题串起来就是“流式解析工程化”这个标题真正要覆盖的范围。热搜词里出现了 SSE、Web Streams API、TransformStream、OpenAI这几个词基本勾勒出了当前主流方案的技术轮廓。SSE 负责传输Web Streams API 负责在浏览器端处理数据流TransformStream 负责在管道中做转换OpenAI 的接口则是这套方案最典型的应用场景。我下面会把这几个环节拆开讲每个环节都给出可复现的代码和踩坑记录。这篇文章适合谁看如果你正在做 AI 对话类产品的前端或全栈开发或者你已经在用 SSE 但总觉得哪里不对劲——比如流断了不知道怎么恢复、中文乱码、多个流并发时状态混乱——那这篇内容应该能帮到你。我会从协议选型讲到代码落地再讲到线上排查尽量把每个决策背后的“为什么”说清楚。2. 传输协议选型为什么 SSE 成了默认答案2.1 SSE、WebSocket、轮询三者的真实取舍在流式场景里可选的传输方式主要有三种短轮询、WebSocket、SSE。很多人一上来就觉得 WebSocket 更“高级”但实际上在 AI 对话这个场景里SSE 才是更合适的选择。我把三者的关键差异列出来维度短轮询WebSocketSSE通信方向客户端拉全双工服务端推协议HTTP独立协议HTTP自动重连需自己实现需自己实现浏览器内置实现复杂度低高低代理兼容性好一般好适合场景低频更新双向实时单向流式推送AI 对话的本质是“客户端发一次请求服务端持续推回答”这是典型的单向推送。WebSocket 的全双工能力在这里是浪费的反而带来了额外的连接管理成本。短轮询则会产生大量无效请求延迟也不可控。SSE 基于 HTTP天然穿透大多数代理和网关浏览器还内置了重连机制工程上最省心。注意SSE 是单向的客户端不能通过同一个连接发消息。如果你需要中途打断生成得用另一个 HTTP 请求去通知服务端或者用 AbortController 直接断开连接。2.2 SSE 协议格式的细节SSE 的报文格式看起来简单但有几个细节如果没注意会导致解析失败。一个标准的 SSE 事件长这样data: {choices:[{delta:{content:你}}]} data: {choices:[{delta:{content:好}}]} data: [DONE]每条消息以data:开头以两个换行符\n\n结束。注意这个双换行是必须的它是事件的分隔符。如果服务端只发了一个换行浏览器会认为事件还没结束继续等待后续数据。还有一个容易忽略的点SSE 支持event:、id:、retry:等字段。id字段用于断线重连时告诉服务端从哪里继续retry用于指定重连间隔。但在 AI 对话场景里我们通常不需要这些因为每次对话是独立的断了就重新发起。OpenAI 的流式接口返回的就是标准 SSE 格式每个 chunk 是一个 JSON最后以data: [DONE]结束。这个[DONE]不是 JSON是一个特殊标记解析的时候要单独处理否则JSON.parse会直接抛异常。2.3 为什么不用 fetch 的 responseType: stream有人会想既然 SSE 就是 HTTP 流那我直接用 fetch 拿 response.body 不就行了确实可以而且现代浏览器里 fetch 的 response.body 就是一个 ReadableStream。但这里有个关键区别原生 EventSource 会自动帮你做事件切分、重连、状态管理而 fetch 方案需要你自己处理这些。那为什么很多项目还是选了 fetch 而不是 EventSource因为 EventSource 有两个硬伤第一它只支持 GET 请求没法带复杂的 POST body第二它不能自定义请求头没法传 Authorization。而调用大模型接口通常需要 POST 加自定义 header所以 fetch ReadableStream 成了更实际的选择。这就引出了下一个话题拿到 ReadableStream 之后怎么把它变成一个个可渲染的文本片段。3. Web Streams API把字节流变成可读文本3.1 ReadableStream 的基本消费方式fetch 返回的 response.body 是一个 ReadableStream里面是 Uint8Array 类型的字节块。最原始的消费方式是用 getReader()const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ prompt: 你好 }) }); const reader response.body.getReader(); const decoder new TextDecoder(utf-8); while (true) { const { done, value } await reader.read(); if (done) break; const text decoder.decode(value, { stream: true }); console.log(text); }这里有个关键参数decoder.decode(value, { stream: true })。如果不传stream: true当一个多字节字符比如中文被切分到两个 chunk 里时解码会出错出现乱码。传了stream: true之后TextDecoder 会缓存不完整的字节序列等下一个 chunk 来了再一起解码。这个细节我在实际项目里踩过坑中文回答偶尔出现“”就是因为这个。3.2 用 TransformStream 做管道化处理上面的写法能用但把所有逻辑堆在一个 while 循环里代码会越来越乱。更好的方式是用 TransformStream 把处理逻辑拆成独立的管道阶段。Web Streams API 提供了pipeThrough方法可以把多个 TransformStream 串起来const decoder new TextDecoder(utf-8); const splitSSE new TransformStream({ transform(chunk, controller) { const text decoder.decode(chunk, { stream: true }); // 按双换行切分事件 const events text.split(\n\n); for (const event of events) { if (event.trim()) controller.enqueue(event); } } }); const parseJSON new TransformStream({ transform(event, controller) { const line event.trim(); if (line.startsWith(data: )) { const data line.slice(6); if (data [DONE]) { controller.enqueue({ done: true }); return; } try { const parsed JSON.parse(data); const content parsed.choices?.[0]?.delta?.content; if (content) controller.enqueue({ content }); } catch (e) { // 忽略解析失败的 chunk } } } }); const stream response.body .pipeThrough(splitSSE) .pipeThrough(parseJSON); const reader stream.getReader(); while (true) { const { done, value } await reader.read(); if (done) break; if (value.content) { appendToDOM(value.content); } }这种管道化的写法有几个好处每个 TransformStream 只负责一件事方便单独测试可以灵活增删中间环节比如加一个统计 token 数的环节代码可读性明显提升。3.3 处理跨 chunk 边界的问题上面的splitSSE有一个隐藏 bug如果一个 SSE 事件被切分到两个 chunk 里split(\n\n)会把不完整的事件也切出来。比如第一个 chunk 结尾是data: {cho第二个 chunk 开头是ices:...}直接切分会得到两个残缺的片段。正确的做法是维护一个缓冲区只处理完整的事件let buffer ; const splitSSE new TransformStream({ transform(chunk, controller) { buffer decoder.decode(chunk, { stream: true }); const parts buffer.split(\n\n); // 最后一段可能不完整留在缓冲区 buffer parts.pop(); for (const part of parts) { if (part.trim()) controller.enqueue(part); } }, flush(controller) { // 流结束时处理剩余数据 if (buffer.trim()) controller.enqueue(buffer); } });这个buffer的维护是流式解析里最容易出错的地方。我见过不少项目因为没处理跨 chunk 边界导致长回答偶尔丢字或者 JSON 解析失败。判断标准很简单如果服务端返回的 chunk 大小不固定就必须做缓冲。4. 前端渲染与状态管理4.1 增量渲染的性能考量拿到文本片段之后最直接的做法是每来一个片段就innerHTML content或者setState(prev prev content)。这在短回答上没问题但回答长了之后频繁的 DOM 操作和 React 重渲染会明显卡顿。我的做法是用一个缓冲区加 requestAnimationFrame 做批量更新let pending ; let rafId null; function appendToDOM(text) { pending text; if (rafId) return; rafId requestAnimationFrame(() { document.getElementById(output).textContent pending; pending ; rafId null; }); }这样每帧最多更新一次 DOM即使服务端每秒推几十个 chunk渲染压力也可控。在 React 里可以用类似思路把流式内容存在 ref 里用 useSyncExternalStore 或者定时 flush 到 state。4.2 中断生成与 AbortController用户点了“停止生成”按钮你得真的把请求断掉不然服务端还在跑白白消耗 token。AbortController 是标准做法const controller new AbortController(); fetch(/api/chat, { signal: controller.signal, // ... }); // 用户点击停止 stopButton.onclick () controller.abort();abort 之后reader.read() 会抛出一个 AbortError需要在 catch 里单独处理不要当成真正的错误上报。另外要注意abort 只是断开了客户端连接服务端是否停止生成取决于服务端的实现。如果服务端用的是 OpenAI 的流式接口客户端断开后服务端的写入会失败通常也会跟着停止但这不算强保证。4.3 多轮对话的状态隔离一个页面上可能有多个对话同时进行或者用户快速切换对话。这时候要确保每个流的 reader 和 buffer 是独立的不能共用全局变量。我习惯把每个流封装成一个类或者闭包function createStreamSession(onChunk, onDone, onError) { let buffer ; let controller new AbortController(); async function start(payload) { const response await fetch(/api/chat, { method: POST, signal: controller.signal, headers: { Content-Type: application/json }, body: JSON.stringify(payload) }); // ... 消费流 } function stop() { controller.abort(); } return { start, stop }; }每个会话持有自己的 controller 和 buffer切换对话时调用对应的 stop状态就不会串。5. 服务端实现要点5.1 设置正确的响应头服务端要返回 SSE必须设置这几个响应头Content-Type: text/event-stream Cache-Control: no-cache Connection: keep-alive X-Accel-Buffering: noX-Accel-Buffering: no是给 Nginx 看的告诉它不要缓冲这个响应。如果不设Nginx 默认会缓冲导致客户端收不到实时数据等整个响应结束才一次性收到。这个坑我在线上环境遇到过本地测试正常一上 Nginx 就变成“假流式”。5.2 转发上游流式响应如果服务端是转发 OpenAI 的流用 Node.js 的 fetch 拿到上游 response.body 之后可以直接 pipe 给客户端const upstream await fetch(https://api.openai.com/v1/chat/completions, { method: POST, headers: { Authorization: Bearer ${process.env.OPENAI_API_KEY}, Content-Type: application/json }, body: JSON.stringify({ model: gpt-4, stream: true, messages }) }); res.writeHead(200, { Content-Type: text/event-stream, Cache-Control: no-cache, Connection: keep-alive, X-Accel-Buffering: no }); const reader upstream.body.getReader(); while (true) { const { done, value } await reader.read(); if (done) break; res.write(value); } res.end();注意这里直接把上游的字节流转发给客户端不做解析让客户端去解析。这样服务端逻辑最简单也最少出错。如果需要在服务端做内容过滤或者统计再考虑加 TransformStream。5.3 超时与心跳SSE 连接如果长时间没有数据中间的网络设备可能会主动断开。OpenAI 的接口在生成过程中会持续推数据一般不会空闲太久。但如果你的服务端在两次 chunk 之间有较长的处理时间建议加心跳const heartbeat setInterval(() { res.write(: heartbeat\n\n); }, 15000); // 流结束时清理 res.on(close, () clearInterval(heartbeat));以冒号开头的行是 SSE 的注释客户端会忽略但能保持连接活跃。热搜词里有一条 “stream disconnected before completion: idle timeout waiting for sse”说的就是空闲超时导致流中断加心跳是标准解法。6. 常见问题与排查实录6.1 流式解析问题速查表现象可能原因排查方向中文乱码TextDecoder 没传 stream: true检查 decode 参数偶尔丢字跨 chunk 边界没缓冲检查 buffer 逻辑假流式一次性出Nginx 缓冲加 X-Accel-Buffering: noJSON 解析报错把 [DONE] 当 JSON 解析单独处理 [DONE]流中途断开空闲超时加心跳或调整超时abort 后报错没区分 AbortErrorcatch 里单独判断多对话串内容全局 buffer 共用每个会话独立封装6.2 几个我踩过的坑第一个坑是JSON.parse的容错。上游返回的 chunk 偶尔会有空行或者格式不标准的行直接 parse 会抛异常。我的做法是 try-catch 包住解析失败就跳过不要让一个坏 chunk 打断整个流。第二个坑是 React 的闭包问题。在 useEffect 里启动流回调里更新 state如果依赖数组没写对会拿到旧的 state。我后来改成用 ref 存流式内容或者用 useReducer 来管理避免闭包陷阱。第三个坑是服务端的 res.write 背压。如果客户端消费慢res.write 返回 false继续写会占用内存。生产环境要监听 drain 事件或者用 pipeline 自动处理背压。提示调试 SSE 的时候用 curl 加 -N 参数可以关闭缓冲直接看到流式输出curl -N -X POST ...。这比在浏览器里看 Network 面板更直观。6.3 关于 OpenAI 兼容接口的注意事项现在很多模型服务都提供 OpenAI 兼容的接口但流式返回的格式可能有细微差异。有的服务返回的 chunk 里delta.content是空字符串有的会在最后一个 chunk 里带finish_reason。解析的时候要兼容这些情况不要假设每个 chunk 都有 content。另外API Key 一定要放在服务端不要暴露在前端代码里这是基本的安全底线。7. 工程化封装的一点思路把上面这些环节串起来一个可复用的流式解析模块应该包含传输层fetch AbortController、解析层TransformStream 管道、渲染层批量更新、状态层会话隔离。我习惯把它封装成一个不依赖框架的核心类然后在上层用 React/Vue 的适配器去对接。这样核心逻辑可以单独测试换框架也不用重写。测试的时候用一个模拟的 ReadableStream 来构造各种边界情况chunk 被切分、中文跨 chunk、[DONE] 标记、空 chunk、异常 chunk。把这些 case 都覆盖到线上出问题的概率会小很多。流式解析这件事协议本身不复杂难的是边界情况的处理。把 buffer 管理、编码处理、中断恢复这几个点做扎实基本就能稳定运行了。