1. 从能跑到敢上线流式解析为什么必须工程化流式解析这件事第一次跑通的时候特别爽。后端一个接口推过来前端EventSource一挂字一个个往外蹦感觉产品瞬间高级了。但真正把它放进生产环境问题就来了网络断了怎么办用户切到后台再切回来消息丢了一半怎么办服务端推了个半截 JSON前端JSON.parse直接抛异常整个页面白屏怎么办这些问题的共同点是——它们都不是流式本身的问题而是工程化缺失的问题。流式解析的工程化核心就三件事连接的生命周期管理、数据帧的边界处理、异常状态的可恢复性。把这三件事做扎实流式功能才算从 demo 变成了可交付的能力。这篇内容适合两类人看一类是刚把 SSE 或 Web Streams 跑通、准备上线的前端/全栈同学另一类是后端要封装流式接口、需要和前端约定协议边界的工程师。我会围绕SSE、Web Streams API、TransformStream这几个关键词把流式解析从协议层到代码层拆开讲重点放在那些文档里不写、但线上一定会遇到的坑。先给一个整体判断流式解析的难点从来不在怎么读流而在怎么在流断掉、乱序、粘包、超时的情况下让上层业务代码感觉不到这些破事。工程化的目标就是造一层防弹衣把底层的不确定性挡在业务逻辑之外。2. SSE 与 Web Streams两套模型到底该怎么选2.1 SSE 的本质是一个长连接 文本协议很多人把 SSE 当成服务端推送这个理解不够准确。SSE 的全称是 Server-Sent Events它建立在普通 HTTP 之上本质是服务端保持一个 HTTP 响应不关闭持续往响应体里写文本。浏览器端的EventSource只是帮你把这堆文本按协议解析成事件。它的协议格式非常朴素就是纯文本用两个换行分隔一个事件块event: message data: {id:1,text:hello} event: message data: {id:2,text:world}注意几个细节data:后面如果有多行会被拼接成一行event:字段决定事件类型以:开头的行是注释常被用作心跳保活。这些规则看起来简单但前端手写解析时最容易在换行符上翻车——\n和\r\n混用、最后一个事件块没有结尾空行都会导致解析错位。SSE 最大的优势是浏览器原生支持、自动重连、走标准 HTTP 语义对基础设施友好。劣势也很明显只能服务端单向推、只支持文本、连接数受浏览器同域限制HTTP/1.1 下同域大约 6 个。所以它适合通知类、进度类、AI 对话类这种单向、低频、文本为主的场景。2.2 Web Streams API 是更底层的数据管道Web Streams API提供的是ReadableStream、WritableStream、TransformStream这套抽象。它不关心数据是 SSE、是 NDJSON 还是二进制只关心一块一块的数据怎么流动、怎么转换、怎么背压。fetch返回的response.body就是一个ReadableStream。你可以直接读它也可以接一个TransformStream做中间处理。这就是为什么现在很多流式方案不再用EventSource而是用fetch ReadableStream——因为它给了你完全的控制权可以带自定义 header、可以用 POST、可以中途 abort、可以插入任意转换逻辑。TransformStream是这套模型里最值得说的东西。它由一对readable和writable组成你往writable写从readable读中间经过你的transform函数。这天然就是一个解析器的位置上游是原始字节流下游是结构化事件流中间用 TransformStream 做协议解析。2.3 选型对照别为了技术而技术维度EventSource (SSE)fetch Web Streams请求方法仅 GET任意方法支持 POST自定义 Header不支持完全支持自动重连内置需自己实现二进制支持不支持支持解析控制粒度黑盒只能拿事件完全可控中断控制close()AbortController适用场景简单通知、进度AI 对话、复杂协议我的经验是如果只是服务端推个通知用 EventSource 最省事只要涉及 POST 传参、鉴权 header、自定义协议、需要精细控制重连策略一律上 fetch Web Streams。现在主流的 AI 对话类产品几乎都是后者因为对话请求要带上下文、要 POST、要能随时中断生成。3. 手写一个能扛住生产的流式解析器3.1 为什么不能直接response.text()新手最常见的写法是const text await response.text()然后等全部返回再解析。这在流式场景下等于把流式的意义完全抹掉了——用户要等所有内容生成完才看到第一个字。正确做法是拿到response.body这个ReadableStream用getReader()逐块读取。const response await fetch(/api/chat, { method: POST, headers: { Content-Type: application/json }, body: JSON.stringify({ prompt }), signal: controller.signal }); const reader response.body.getReader(); const decoder new TextDecoder(utf-8); while (true) { const { done, value } await reader.read(); if (done) break; const chunk decoder.decode(value, { stream: true }); // 处理 chunk }这里有个极其关键的细节decoder.decode(value, { stream: true })里的stream: true不能省。因为一个 UTF-8 中文字符占 3 个字节网络分块完全可能把这三个字节切到两个 chunk 里。如果不加stream: true每个 chunk 独立解码遇到被切断的多字节字符就会解出乱码经典的锟斤拷就是这么来的。加上这个参数TextDecoder会在内部缓存不完整的字节序列等下一个 chunk 到了再拼起来解码。3.2 用 TransformStream 把字节流变成事件流直接在while循环里写解析逻辑代码会越来越乱。更好的做法是把解析逻辑封装成一个TransformStream让数据管道自己完成转换function createSSEParser() { let buffer ; return new TransformStream({ transform(chunk, controller) { buffer chunk; const blocks buffer.split(\n\n); // 最后一段可能不完整留在 buffer 里 buffer blocks.pop(); for (const block of blocks) { const event parseBlock(block); if (event) controller.enqueue(event); } }, flush(controller) { // 流结束时处理残留的 buffer if (buffer.trim()) { const event parseBlock(buffer); if (event) controller.enqueue(event); } } }); }这段代码里有三个工程化要点值得单独拎出来说第一buffer的存在是为了处理粘包和半包。网络传输不保证一次read()就返回一个完整的事件块。可能一次返回两个半事件也可能一个事件被切成三次返回。所以必须维护一个缓冲区只处理完整的分隔符之前的内容剩下的留在 buffer 里等下一块。blocks.pop()就是干这个的——最后一段永远是不完整的不能处理。第二flush不能忘。流结束的时候buffer 里可能还残留最后一个事件块因为服务端最后一个事件后面可能没有\n\n。如果不在flush里处理最后一个消息就会永远丢失。这个 bug 特别隐蔽因为大部分时候服务端会补上结尾空行只有特定情况下才暴露。第三parseBlock要能容错。一个事件块里可能有event:、data:、id:、retry:多个字段data:还可能有多行。解析时要按行遍历遇到不认识的行直接跳过遇到data:就累加。永远不要假设服务端发来的格式百分百规范防御性解析是流式解析的基本素养。3.3 把管道串起来有了 parser整个链路就清爽了const response await fetch(/api/chat, { /* ... */ }); const eventStream response.body .pipeThrough(new TextDecoderStream()) .pipeThrough(createSSEParser()); const reader eventStream.getReader(); while (true) { const { done, value } await reader.read(); if (done) break; handleEvent(value); // value 已经是结构化对象 }注意这里用了TextDecoderStream它是浏览器内置的、把TextDecoder包装成 TransformStream 的版本效果和前面手写decoder.decode(value, {stream:true})一样但更优雅。整条管道是字节流 → 文本流 → 事件流每一层职责单一测试和替换都很方便。4. 断线、超时、半包线上真正会咬人的地方4.1 stream disconnected before completion 到底在说什么这个报错信息很多人见过字面意思是流在完成前断开了。它可能来自几个完全不同的原因排查时一定要区分服务端主动关闭比如后端生成完了但没发结束标记或者后端进程被重启。中间层超时反向代理、网关、负载均衡对空闲连接有超时限制长时间没有数据流动就被掐断。客户端网络抖动移动端切网络、Wi-Fi 转 4GTCP 连接直接断。空闲超时idle timeout这是最常见的一种连接建立后一段时间内没有任何数据往来被判定为空闲连接强制关闭。区分方法看断开的时间点。如果是固定时长比如 60 秒、120 秒后断开基本就是空闲超时如果是随机时间断开多半是网络或服务端问题。4.2 心跳保活让连接看起来一直在忙对付空闲超时最直接的办法是定期发送心跳。SSE 协议里以:开头的行是注释客户端会忽略但足以让中间层认为连接是活跃的: heartbeat服务端每隔 15~30 秒发一次心跳就能有效避开大多数空闲超时阈值。心跳间隔要小于中间层的最小超时时间这个值需要和运维确认不能拍脑袋。我一般会取超时阈值的 1/3 到 1/2留足余量。客户端这边也要配合如果超过 N 个心跳周期没收到任何数据包括心跳就主动判定连接已死并重连。因为有些网络故障下TCP 连接不会立刻报错而是假死——你以为还连着其实数据早就过不来了。这种半开连接是最坑的只能靠应用层超时来兜底。4.3 重连策略指数退避 断点续传重连不能无脑立刻重试否则服务端一挂所有客户端瞬间发起重连直接把服务打垮惊群效应。标准做法是指数退避function getRetryDelay(attempt) { const base 1000; const max 30000; const delay Math.min(base * Math.pow(2, attempt), max); // 加随机抖动避免所有客户端同时重连 return delay Math.random() * 1000; }Math.random()这个抖动很关键。如果所有客户端都按1s, 2s, 4s, 8s精确重连它们会在同一时刻集体冲击服务端。加上随机抖动重连请求就被打散了。更进阶的是断点续传。SSE 协议支持id:字段和Last-Event-ID请求头服务端给每个事件编号客户端重连时带上最后收到的事件 ID服务端从那个 ID 之后继续推。这样断线期间的消息不会丢。实现这个需要服务端维护一个短期的事件缓冲区成本不低但对消息不能丢的场景比如订单状态、任务进度是刚需。4.4 半包与粘包的完整处理链路前面提了 buffer 的思路这里给一个更完整的排查链路。假设你发现前端偶尔解析出错按这个顺序查先确认是不是编码问题打印原始字节看有没有被切断的多字节字符。如果是检查TextDecoder有没有加stream: true。再确认分隔符把原始文本打出来看事件之间到底是\n\n还是\r\n\r\n。有些服务端框架默认用\r\n前端只按\n\n切就会切不开。然后确认 buffer 逻辑在transform里打印每次的buffer和切出来的blocks看有没有把不完整的块当成完整的处理了。最后确认 flush在流结束时打印残留 buffer看最后一个事件有没有被丢掉。这个顺序是从底层到上层先排除编码再排除协议最后排除逻辑能避免在错误的方向上浪费时间。5. 跨语言协作前后端在流式协议上的约定5.1 后端封装流式接口的通用结构不管后端是 Java、Python 还是 Node封装流式接口的骨架都差不多设置正确的响应头 → 拿到输出流 → 循环写数据 → 主动 flush → 处理客户端断开。以 Java 为例Spring 体系核心是返回一个流式的响应体并确保每次写入后立即 flush否则数据会攒在缓冲区里前端迟迟收不到GetMapping(value /stream, produces MediaType.TEXT_EVENT_STREAM_VALUE) public void stream(HttpServletResponse response) throws IOException { response.setContentType(text/event-stream); response.setCharacterEncoding(UTF-8); response.setHeader(Cache-Control, no-cache); response.setHeader(X-Accel-Buffering, no); // 关键禁用代理缓冲 PrintWriter writer response.getWriter(); for (String msg : generateMessages()) { writer.write(data: msg \n\n); writer.flush(); // 关键每次写完立即 flush } }这里有两个新手必踩的坑坑一X-Accel-Buffering: no。如果前面有 Nginx 之类的反向代理它默认会缓冲响应导致你的流式数据被攒成一大块才发给客户端流式效果完全消失。这个 header 就是告诉代理别缓冲。有些场景还需要在代理配置里显式关闭缓冲。坑二忘记 flush。很多输出流的默认缓冲区是 8KB你不 flush数据就一直躺在缓冲区里。表现就是前端要等很久才一次性收到一大段而不是逐字出现。5.2 前后端必须对齐的协议细节流式接口最容易出问题的地方是前后端对协议的理解不一致。上线前一定要把下面这些点白纸黑字约定清楚约定项建议值说明分隔符\n\n统一用\n别混\r\n数据字段data:多行数据用多个data:结束标记event: done或[DONE]显式告诉前端流正常结束错误传递event: error data业务错误也走流别直接断连心跳: ping注释行客户端忽略编码UTF-8前后端统一结束标记这一条特别重要。如果服务端生成完就直接关连接前端无法区分正常结束和异常断开。加一个显式的结束事件前端收到它就知道是正常完成可以停止重连逻辑没收到就断了才触发重连。这个区分能避免明明生成完了前端还在傻傻重连的尴尬。5.3 错误也要走流一个反直觉但很重要的设计业务错误不要用 HTTP 状态码返回而是用流内的事件返回。因为流一旦开始HTTP 状态码早就发出去了200你没法再改。如果生成到一半出错只能通过流内发一个event: error事件把错误信息传给前端。// 前端处理 if (event.type error) { showError(event.data.message); // 注意这里不要重连因为是业务错误重连也没用 return; }区分可重试错误网络问题和不可重试错误业务错误、参数错误非常关键。对不可重试错误做重连只会浪费资源、放大问题。6. 把流式解析封装成可复用的工程模块6.1 抽象出一个 StreamClient散落在各处的流式代码维护起来是灾难。我的做法是封装一个StreamClient把连接、解析、重连、中断全部收进去业务层只关心收到消息和收到错误两个回调class StreamClient { constructor(url, options {}) { this.url url; this.options options; this.controller null; this.retryCount 0; this.maxRetry options.maxRetry ?? 5; } async start({ onMessage, onError, onDone }) { this.controller new AbortController(); try { const response await fetch(this.url, { ...this.options, signal: this.controller.signal }); if (!response.ok) throw new Error(HTTP ${response.status}); const stream response.body .pipeThrough(new TextDecoderStream()) .pipeThrough(createSSEParser()); const reader stream.getReader(); while (true) { const { done, value } await reader.read(); if (done) break; if (value.type done) { onDone?.(); return; } if (value.type error) { onError?.(value.data); return; } onMessage?.(value.data); } onDone?.(); } catch (err) { if (err.name AbortError) return; // 主动中断不算错误 this.handleRetry({ onMessage, onError, onDone }); } } abort() { this.controller?.abort(); } }这个封装有几个设计取舍值得说AbortError要单独处理。用户主动点停止生成时fetch会抛AbortError。这不是错误不应该触发重连也不应该弹错误提示。很多实现忘了这一条导致用户一停止就弹个报错体验很差。重连要带上状态。如果支持断点续传重连时要带上Last-Event-ID。这个 ID 应该在onMessage里记录重连时作为 header 传回去。重试次数要有上限。无限重连在服务端持续不可用时会让客户端一直空转。设个上限比如 5 次超过就放弃并通知用户。6.2 用状态机管理连接生命周期流式连接的状态其实不少idle、connecting、streaming、reconnecting、done、error。用状态机管理能避免在错误的状态做了错误的操作当前状态触发事件目标状态动作idlestartconnecting发起请求connecting响应成功streaming开始读流streaming收到 donedone关闭连接streaming网络错误reconnecting退避后重连reconnecting重连成功streaming续传reconnecting超过上限error通知用户任意abortidle中断并清理有了这张表代码里的if/else就能收敛成清晰的状态转移每个状态该做什么、不该做什么一目了然。这也是排查问题时的重要工具——出问题时先看当前状态往往就能定位到问题。6.3 内存与性能别让流式拖垮页面流式场景下有两个性能陷阱陷阱一DOM 更新过于频繁。如果每收到一个字就更新一次 DOM高频流式下页面会卡。正确做法是用requestAnimationFrame或定时器做批量更新把短时间内的多个消息合并成一次渲染。陷阱二消息列表无限增长。长对话场景下消息越堆越多内存和渲染压力都会上来。需要做虚拟滚动或者定期裁剪历史消息。这个不是流式独有的问题但流式场景下暴露得更快。还有一个容易忽略的点及时释放 reader 和 stream。流结束后reader应该releaseLock()避免资源泄漏。虽然现代浏览器 GC 会处理大部分情况但显式释放是好习惯。7. 上线前的自检清单与踩坑复盘7.1 一份可以直接抄的自检清单流式功能上线前我会按这个清单过一遍[ ]TextDecoder是否加了stream: true或用了TextDecoderStream[ ] 解析器是否有 buffer 处理半包[ ]flush是否处理了残留数据[ ] 分隔符前后端是否统一\n\nvs\r\n\r\n[ ] 是否有显式的结束事件[ ] 服务端是否每次写入后 flush[ ] 代理层是否禁用了缓冲X-Accel-Buffering: no[ ] 是否有心跳保活[ ] 重连是否用了指数退避 抖动[ ]AbortError是否被正确忽略[ ] 业务错误是否走流内事件而非 HTTP 状态码[ ] 是否有重试次数上限[ ] 高频更新是否做了批量渲染这份清单里的每一条背后都是一个真实踩过的坑。尤其是前三条几乎每个手写流式解析的人都栽过。7.2 几个印象深刻的坑坑一本地好好的上线就断。本地直连后端没有代理流式一切正常。上线后前面挂了网关60 秒空闲就断。原因是网关的空闲超时是 60 秒而我们的心跳间隔设成了 90 秒。心跳间隔必须小于链路中所有中间层的最小超时值这个值要挨个确认不能想当然。坑二中文偶尔乱码。排查了很久最后发现是某个 chunk 恰好把中文字符切开了而解码时没加stream: true。这个 bug 复现概率低但一旦出现就是乱码非常影响观感。凡是处理 UTF-8 文本流stream: true是标配。坑三最后一个消息丢失。服务端生成完最后一个消息后直接关连接没有补结尾空行。前端解析器只处理\n\n之前的内容最后一个消息就留在了 buffer 里永远没被处理。加上flush逻辑后解决。这个坑的隐蔽性在于它只在最后一个消息上出问题前面的都正常很容易被忽略。坑四用户点停止后弹错误。用户主动中断生成结果弹了个网络错误。原因是没区分AbortError和真正的网络错误。主动中断是正常操作不是错误这个区分做不好用户体验会大打折扣。7.3 关于工程化的一点个人体会流式解析这个领域技术门槛其实不高难的是把边界情况想全。我见过太多项目happy path 跑得飞起一遇到网络抖动、服务端重启、用户切后台就各种诡异问题。工程化的价值恰恰体现在这些不 happy的路径上。我的建议是在写第一行流式代码之前先把状态机和错误处理想清楚。哪些错误可重试、哪些不可重试、断线后消息能不能丢、用户中断怎么处理——这些问题想明白了代码自然就稳了。反过来如果一开始只想着怎么把字蹦出来后面补这些逻辑会非常痛苦因为它们是穿插在整个流程里的不是能补上去的。另外流式功能的测试不能只测正常流程。要专门构造断网、慢网、服务端中途关闭、发送畸形数据这些场景。Chrome DevTools 的网络限速和离线模式是很好的工具能模拟出大部分异常情况。有条件的话写一些针对解析器的单元测试把各种半包、粘包的输入喂进去验证输出是否正确。解析器是纯函数式的非常适合单测投入产出比很高。最后说一句关于协议设计的前后端的流式协议越简单越好。不要试图在流里塞太复杂的结构SSE 的文本协议本身就不适合承载复杂语义。把复杂逻辑放在应用层让流只负责搬运这样出问题时排查范围小替换实现也容易。我见过把整个业务状态机塞进流协议的后期维护简直是噩梦。