1. 项目概述：从单向推送到双向对话的流式进化

最近在重构一个实时数据大屏项目，前端需要从后端持续获取最新的业务指标。最早我们用的是轮询，后来换成了WebSocket，但总觉得有点“杀鸡用牛刀”——我们只需要服务器单向推送数据，却维护了一个全双工的长连接。直到我开始系统梳理“流式响应”这个技术栈，才发现从EventSource到ReadableStream，再到TransformStream，这三次进化清晰地勾勒出了前端处理流数据的完整路径。这不仅仅是API的迭代，更是开发范式从“接收数据”到“处理数据流”的深刻转变。如果你也在处理实时日志、金融行情、AI生成内容（比如逐字输出的聊天回复）这类场景，理解这套技术组合拳，能让你写出更优雅、更高效、更可控的代码。

简单来说，EventSource（或者说SSE协议）解决了“如何方便地接收服务器推送”的问题；ReadableStream让我们能以标准、高效的方式消费任何来源的流数据；而TransformStream则赋予了我们在数据流动过程中进行实时转换和加工的能力。这三者结合，构成了现代Web应用中处理流式数据的基石。接下来，我就结合自己的踩坑经验，把这“三次进化”背后的设计思路、具体用法和实战技巧掰开揉碎讲清楚。

2. 第一代：EventSource / SSE —— 服务器推送的“优雅解”

EventSource是浏览器内置的、用于接收服务器发送事件（Server-Sent Events, SSE）的客户端接口。它的核心设计哲学是简单和专注：专注于解决服务器向客户端单向、文本流式推送的场景。

2.1 核心原理与协议约定

SSE协议基于普通的HTTP/HTTPS，本质上是一个长连接。服务器通过持有这个连接，可以持续地向客户端发送数据片段。每个数据片段遵循特定的文本格式：

id: 12345 event: message data: This is a line of text data: This is another line of text

• data:： 消息内容，一行或多行。如果多行，最终会拼接成一个字符串，用换行符\n连接。
• id:： 消息ID，用于断线重连时，客户端可以通过Last-Event-ID头告诉服务器“我从哪里开始”。
• event:： 事件类型，默认是message。这允许客户端监听不同的事件。
• 以一个空行表示一个消息的结束。

浏览器端的EventSource对象会帮你处理所有这些协议细节：建立连接、解析数据流、分派事件，甚至在连接断开时自动重试。对于开发者而言，体验近乎“傻瓜式”。

2.2 基础用法与代码示例

前端使用起来非常简单：

// 创建 EventSource 实例，连接到服务器端的 SSE 端点 const eventSource = new EventSource('/api/sse-stream'); // 监听默认的 'message' 事件 eventSource.onmessage = (event) => { console.log('收到数据:', event.data); // 通常 event.data 是字符串，需要根据业务解析（如 JSON.parse） }; // 监听自定义事件（需要服务器发送 `event: update`） eventSource.addEventListener('update', (event) => { console.log('自定义更新事件:', event.data); }); // 监听连接打开事件 eventSource.onopen = () => { console.log('SSE连接已建立'); }; // 监听错误事件 eventSource.onerror = (error) => { console.error('SSE连接错误:', error); // 注意：EventSource 在出错时会自动尝试重连 }; // 在组件卸载或不再需要时，关闭连接 // eventSource.close();

服务器端（以Node.js + Express为例）的实现要点在于设置正确的响应头，并保持连接不关闭，持续写入数据：

app.get('/api/sse-stream', (req, res) => { // 1. 设置SSE必需的响应头 res.writeHead(200, { 'Content-Type': 'text/event-stream', 'Cache-Control': 'no-cache', 'Connection': 'keep-alive', // CORS 相关（如果需要） 'Access-Control-Allow-Origin': '*' }); // 2. 发送一个初始注释（可选，可用来保持连接） res.write(':\n\n'); // 3. 模拟定期发送数据 const intervalId = setInterval(() => { const data = { time: new Date().toISOString(), value: Math.random() * 100 }; // 按照 SSE 格式发送数据 res.write(`data: ${JSON.stringify(data)}\n\n`); // 注意末尾的两个换行 }, 1000); // 4. 当客户端断开连接时清理资源 req.on('close', () => { clearInterval(intervalId); console.log('客户端断开连接'); res.end(); }); });

2.3 优势、局限与适用场景

优势：

1. 极简API：浏览器原生支持，开箱即用，无需额外库。
2. 自动重连：内置重连机制，提高了连接的鲁棒性。
3. 协议轻量：基于HTTP，穿透防火墙和代理更容易，不像WebSocket可能被某些中间件拦截。
4. 与HTTP生态无缝集成：可以方便地使用HTTP认证、Cookie等。

局限：

1. 仅文本：SSE协议规范只支持UTF-8文本。传输二进制数据需要先编码（如Base64），有性能和体积开销。
2. 单向通信：只能服务器向客户端推送。如果需要双向交互，仍需搭配其他HTTP请求（如Fetch）。
3. 连接数限制：浏览器对同一域名下的HTTP连接数有上限（通常6个），大量SSE连接可能受影响。
4. 老浏览器兼容性：虽然主流现代浏览器都支持，但IE全系不支持。

适用场景：

• 实时通知：新闻推送、站内信、订单状态更新。
• 监控仪表盘：服务器监控指标、实时业务图表（股票K线除外，因其对延迟要求极高）。
• 日志流：在运维后台实时查看应用日志输出。
• AI对话流式输出：ChatGPT那种逐字显示的效果，SSE是天然适配的方案。

实操心得一：心跳保活与连接状态管理虽然EventSource有自动重连，但网络环境复杂，有时连接会“假死”（TCP连接还在，但数据不来了）。一个可靠的实践是实现应用层的心跳。服务器定期（比如每15秒）发送一个注释行（:\n\n）或特定格式的心跳事件。前端监听心跳，如果超过一定时间（如30秒）没收到，就主动关闭当前EventSource并新建一个，强制刷新连接。这能有效解决一些中间件（如Nginx）长连接超时配置带来的问题。

3. 第二代：Fetch API 与 ReadableStream —— 拥抱现代流标准

随着Fetch API的普及和Streams API的成熟，我们获得了更底层、更强大的流处理能力。ReadableStream代表了流式数据处理的一种标准化模型，它不局限于SSE，可以来自Fetch响应、本地文件、甚至其他流。

3.1 从Fetch响应中获取流

Fetch API的Response.body属性本身就是一个ReadableStream。这意味着我们可以逐步读取来自网络的响应体，而不是等它全部下载完。

// 使用 Fetch API 获取一个流式响应 const response = await fetch('/api/stream-data'); const reader = response.body.getReader(); // 获取流阅读器 const decoder = new TextDecoder('utf-8'); // 用于将Uint8Array解码为字符串 try { while (true) { const { done, value } = await reader.read(); // 读取一块数据 if (done) { console.log('流读取完毕'); break; } // value 是一个 Uint8Array 类型的块 const chunk = decoder.decode(value, { stream: true }); // 注意 stream: true console.log('收到数据块:', chunk); // 处理 chunk，可能是部分JSON，也可能是SSE格式的数据行 } } catch (error) { console.error('读取流时发生错误:', error); } finally { reader.releaseLock(); // 释放锁 }

这里的关键是reader.read()，它异步返回一个包含done和value的对象。done为true表示流结束；value是一个Uint8Array，代表一块二进制数据。我们需要用TextDecoder将其解码成字符串。注意decode方法的{ stream: true }选项，这非常重要，因为它允许你处理跨块的字符（比如一个多字节UTF-8字符被分在两个数据块里），避免乱码。

3.2 处理类SSE流与流式JSON

很多现代后端API（如OpenAI的Chat Completions）返回的并不是标准的SSE格式（data: ...），而是一个简单的、由换行符分隔的JSON文本流。每行是一个独立的JSON对象。用ReadableStream处理这种流非常灵活：

async function consumeJSONStream(response) { const reader = response.body.getReader(); const decoder = new TextDecoder(); let buffer = ''; // 缓冲区，用于存储未处理完的字符串 try { while (true) { const { done, value } = await reader.read(); if (done) { // 流结束，处理缓冲区剩余内容 if (buffer.trim()) { processLine(buffer); } break; } buffer += decoder.decode(value, { stream: true }); // 按换行符分割缓冲区 const lines = buffer.split('\n'); // 最后一行可能是不完整的，放回缓冲区 buffer = lines.pop() || ''; for (const line of lines) { if (line.trim()) { // 忽略空行 processLine(line); } } } } finally { reader.releaseLock(); } } function processLine(line) { try { // 假设每一行都是一个完整的JSON字符串 const data = JSON.parse(line); console.log('解析出的数据:', data); // 更新UI，例如将AI回复逐字追加到对话框 } catch (e) { console.warn('解析行失败，可能是不完整的数据:', line, e); // 对于SSE格式，可能需要先去掉 `data: ` 前缀 if (line.startsWith('data: ')) { const jsonStr = line.slice(6).trim(); if (jsonStr === '[DONE]') return; // 处理结束标记 if (jsonStr) processLine(jsonStr); // 递归处理 } } }

这个模式是处理流式文本的通用方法：分块读取、解码、缓冲、按分隔符（如\n）切割、处理完整行。它比EventSource更底层，但也更强大，可以处理任意格式的流。

3.3 ReadableStream 的进阶控制

ReadableStream给了我们精细的控制权：

• 取消（Cancellation）：通过reader.cancel(reason)可以主动中断流的读取。这会给源发送一个信号，后端可以据此释放资源。
• 速率控制（Backpressure）：如果处理数据的速度跟不上接收速度，可以暂停读取。ReadableStream内部机制会处理背压，通知数据源慢下来。这是EventSource不具备的。
• 管道（Piping）：可以将一个ReadableStream通过.pipeThrough()或.pipeTo()方法连接到WritableStream或TransformStream，实现流的无缝转换和传递。

实操心得二：错误处理与资源清理使用ReadableStream时，错误处理必须更谨慎。网络错误、解析错误、业务逻辑错误都可能发生。一定要用try...catch...finally包裹核心读取逻辑，并在finally块中调用reader.releaseLock()来释放流上的锁。否则，这个流将无法被其他代码读取。对于React等框架，在组件卸载的清理函数（useEffect的return函数）中，除了调用reader.cancel()，也别忘了releaseLock。

4. 第三代：TransformStream —— 流数据的“中间件”与“加工厂”

如果说ReadableStream让我们能“喝到水”，那么TransformStream就是一套“净水系统”或“分水器”。它位于流管道中间，可以对流经的每一块数据进行转换、过滤、重组，而无需等待所有数据到位。

4.1 理解 TransformStream 的双重身份

一个TransformStream本质上包含一对流：

• 可写端（Writable side）：用于接收输入数据。
• 可读端（Readable side）：用于输出转换后的数据。

其核心是一个transform方法，该方法接收一个chunk（数据块）和一个controller，负责将处理后的chunk通过controller.enqueue()送入可读端，或者选择不处理（过滤）。

4.2 实战案例一：构建通用的 SSE 解析流

虽然我们有EventSource，但有时我们通过Fetch拿到的是一个SSE格式的流，想用ReadableStream的方式消费，又不想手动解析data:前缀和空行。这时可以创建一个SSEParserTransformStream：

class SSEParserTransformStream extends TransformStream { constructor() { let buffer = ''; let eventName = 'message'; let dataBuffer = ''; let lastEventId = ''; super({ transform(chunk, controller) { // chunk 是 Uint8Array，先解码 buffer += new TextDecoder().decode(chunk, { stream: true }); // 按行分割 const lines = buffer.split('\n'); buffer = lines.pop() || ''; // 剩余部分放回缓冲区 for (let line of lines) { line = line.trim(); if (line.startsWith('event:')) { eventName = line.slice(6).trim(); } else if (line.startsWith('data:')) { // data: 后面的内容追加到 dataBuffer dataBuffer += line.slice(5).trim() + '\n'; } else if (line.startsWith('id:')) { lastEventId = line.slice(3).trim(); } else if (line === '') { // 空行表示一个事件结束 if (dataBuffer) { // 移除最后一个多余的换行符 const data = dataBuffer.slice(0, -1); controller.enqueue({ id: lastEventId, event: eventName, data: data }); // 重置状态 dataBuffer = ''; eventName = 'message'; } } // 忽略其他行（如注释 `:`） } }, flush(controller) { // 流结束时，处理缓冲区中可能残留的最后一个不完整事件 if (dataBuffer) { const data = dataBuffer.trim(); controller.enqueue({ id: lastEventId, event: eventName, data: data }); } } }); } } // 使用方式 async function fetchAndParseSSE(url) { const response = await fetch(url); const sseParser = new SSEParserTransformStream(); // 将响应的流，通过解析器转换，然后获取新的可读流 const parsedStream = response.body.pipeThrough(sseParser); const reader = parsedStream.getReader(); while (true) { const { done, value } = await reader.read(); if (done) break; console.log(`事件[${value.event}]:`, value.data); // value.data 已经是解析好的字符串 // 可以直接 JSON.parse(value.data) 如果数据是JSON } }

这个转换流将原始的、带格式的SSE字节流，转换成了一个对象流（{id, event, data}），下游消费起来就干净多了。

4.3 实战案例二：实现流式 JSON 解析与错误恢复

对于流式JSON（每行一个JSON对象），我们可能会遇到行不完整、JSON格式错误等问题。一个健壮的TransformStream可以处理这些边缘情况：

class LineJsonParserTransformStream extends TransformStream { constructor() { let buffer = ''; super({ transform(chunk, controller) { buffer += new TextDecoder().decode(chunk, { stream: true }); const lines = buffer.split('\n'); buffer = lines.pop() || ''; for (const line of lines) { const trimmedLine = line.trim(); if (!trimmedLine) continue; // 跳过空行 try { const obj = JSON.parse(trimmedLine); controller.enqueue(obj); // 成功解析，输出对象 } catch (e) { // 解析失败，可能是不完整的JSON // 策略1：忽略（如果允许丢数据） // console.warn('JSON解析失败，行内容:', trimmedLine); // 策略2：累积到buffer，等待后续数据（更复杂，需考虑内存和超时） // 这里演示简单忽略 } } }, flush(controller) { // 流结束，尝试解析缓冲区剩余内容 if (buffer.trim()) { try { const obj = JSON.parse(buffer.trim()); controller.enqueue(obj); } catch (e) { // 最终仍无法解析，可以记录日志或抛出错误 console.error('流结束仍有未解析数据:', buffer); } } } }); } }

4.4 实战案例三：流数据加工与聚合

TransformStream的威力在于可以串联。假设我们有一个数字流，想要实时计算移动平均：

class MovingAverageTransformStream extends TransformStream { constructor(windowSize = 5) { let window = []; super({ transform(number, controller) { window.push(number); if (window.length > windowSize) { window.shift(); // 保持窗口大小 } const avg = window.reduce((s, n) => s + n, 0) / window.length; controller.enqueue({ raw: number, movingAverage: avg, timestamp: Date.now() }); } }); } } // 使用管道串联多个转换流 async function processSensorData() { // 假设 /api/sensor 返回一个数字流（每行一个数字） const response = await fetch('/api/sensor'); const rawStream = response.body .pipeThrough(new TextDecoderStream()) // 1. 字节流转字符串流 .pipeThrough(new LineJsonParserTransformStream()) // 2. 行解析为数字 .pipeThrough(new MovingAverageTransformStream(10)); // 3. 计算移动平均 const reader = rawStream.getReader(); // ... 消费处理后的数据 }

TextDecoderStream是一个内置的TransformStream，专门用于将Uint8Array流转换为字符串流，非常方便。

实操心得三：转换流的性能与内存考量TransformStream的transform函数是同步执行的。如果里面的操作很耗时（比如复杂的计算或同步的JSON.parse大型对象），会阻塞整个流的管道。对于CPU密集型操作，考虑使用Web Worker，在Worker中创建TransformStream，通过postMessage传递数据。另外，要小心在转换流中累积大量数据（比如为了聚合）。始终设置一个上限，并在flush方法中妥善清理，避免内存泄漏。对于无限流，设计“滑动窗口”或“抽样”逻辑是关键。

5. 组合进化：构建企业级流式处理管道

在实际项目中，我们很少单独使用某一项技术，而是将它们组合起来，形成一个健壮、可维护的流式处理管道。下面以一个“实时日志查看器”为例，展示如何综合运用这三代技术。

5.1 架构设计

目标：一个Web页面，实时显示来自服务器应用容器的日志，支持按关键字过滤、高亮不同日志级别（INFO, WARN, ERROR），并能暂停/恢复日志流。

1. 传输层：服务器通过SSE协议推送日志行。选择SSE是因为日志是单向的文本流，且需要良好的重连能力。
2. 接收与解析层：前端使用Fetch API+ReadableStream接收，而不是EventSource。因为我们需要更底层的控制（如自定义请求头、认证、取消）。
3. 处理层：使用一系列TransformStream进行数据加工：
   • SSEParserTransformStream: 解析SSE格式。
   • LogFilterTransformStream: 根据用户输入的关键字和级别过滤日志行。
   • LogHighlighterTransformStream: 将日志文本转换为带HTML标签和样式的富文本片段。
4. 消费与渲染层：将处理后的流通过reader.read()逐块消费，并追加到页面的虚拟列表或<div>中，实现平滑滚动。

5.2 核心代码实现

// 1. 自定义的SSE解析流（同上，略作调整以输出原始日志字符串） class LogSSEParserTransformStream extends TransformStream { /* ... */ } // 2. 日志过滤流 class LogFilterTransformStream extends TransformStream { constructor(filterKeyword = '', filterLevel = 'ALL') { super({ transform(logEvent, controller) { const logData = logEvent.data; // 假设logEvent是 {id, event, data} const logObj = JSON.parse(logData); // 假设日志是JSON格式 {level: 'INFO', message: '...', timestamp: '...'} let shouldPass = true; if (filterLevel !== 'ALL' && logObj.level !== filterLevel) { shouldPass = false; } if (filterKeyword && !logObj.message.includes(filterKeyword)) { shouldPass = false; } if (shouldPass) { controller.enqueue(logObj); // 输出过滤后的日志对象 } } }); } } // 3. 日志高亮转换流 class LogHighlighterTransformStream extends TransformStream { constructor() { super({ transform(logObj, controller) { let levelClass = 'log-info'; if (logObj.level === 'WARN') levelClass = 'log-warn'; if (logObj.level === 'ERROR') levelClass = 'log-error'; const highlightedHtml = ` <div class="log-line ${levelClass}"> <span class="timestamp">[${logObj.timestamp}]</span> <span class="level">${logObj.level}</span> <span class="message">${this.escapeHtml(logObj.message)}</span> </div> `; controller.enqueue(highlightedHtml); }, escapeHtml(text) { const div = document.createElement('div'); div.textContent = text; return div.innerHTML; } }); } } // 4. 主控制函数 class LogStreamProcessor { constructor(url, containerEl) { this.url = url; this.containerEl = containerEl; this.reader = null; this.filterKeyword = ''; this.filterLevel = 'ALL'; this.isPaused = false; this.pauseBuffer = []; } async start() { try { const response = await fetch(this.url, { headers: { 'Accept': 'text/event-stream' } }); if (!response.ok || !response.body) throw new Error('Stream not available'); // 构建处理管道 const processedStream = response.body .pipeThrough(new TextDecoderStream()) .pipeThrough(new LogSSEParserTransformStream()) .pipeThrough(new LogFilterTransformStream(this.filterKeyword, this.filterLevel)) .pipeThrough(new LogHighlighterTransformStream()); this.reader = processedStream.getReader(); this.consumeStream(); } catch (error) { console.error('启动日志流失败:', error); this.containerEl.innerHTML += `<div class="log-error">连接失败: ${error.message}</div>`; } } async consumeStream() { try { while (true) { if (this.isPaused) { // 如果暂停，等待一小段时间再检查，避免忙等待 await new Promise(resolve => setTimeout(resolve, 100)); continue; } const { done, value } = await this.reader.read(); if (done) { console.log('日志流结束'); break; } // value 现在是高亮后的HTML字符串 this.containerEl.insertAdjacentHTML('beforeend', value); // 自动滚动到底部 this.containerEl.scrollTop = this.containerEl.scrollHeight; } } catch (error) { console.error('消费日志流时出错:', error); this.containerEl.innerHTML += `<div class="log-error">流读取错误: ${error.message}</div>`; } finally { if (this.reader) this.reader.releaseLock(); } } updateFilter(keyword, level) { this.filterKeyword = keyword; this.filterLevel = level; // 注意：直接更新filter不会影响已创建的流。 // 需要重启流，或者设计更复杂的动态过滤机制（例如使用WritableStream + ReadableStream手动控制）。 this.stop(); this.start(); } pause() { this.isPaused = true; } resume() { this.isPaused = false; } async stop() { this.isPaused = true; if (this.reader) { await this.reader.cancel('用户停止'); this.reader.releaseLock(); this.reader = null; } } }

5.3 性能优化与高级技巧

1. 虚拟列表渲染：当日志行数巨大时，直接追加DOM会导致性能急剧下降。应该使用虚拟列表技术（如react-window或自己实现），只渲染可视区域内的日志行。这需要调整消费逻辑，将流数据先存入一个可观察的数据结构（如RxJS Subject或普通数组），再由UI组件根据滚动位置读取。
2. 动态过滤：上述例子中，更新过滤器需要重启整个流，这会造成连接中断和重连。更优的方案是让过滤流（LogFilterTransformStream）支持动态更新条件。这可以通过在转换流内部使用getter/setter或传递一个响应式对象（如Vue的ref、React的useState通过闭包注入）来实现。转换函数每次执行时读取最新的过滤条件。
3. 流量控制与背压：如果日志产生速度远快于UI渲染速度，会导致内存中积压大量未处理的日志对象。完善的管道应该考虑背压。我们可以通过控制reader.read()的调用来实现。例如，只有在UI准备好渲染下一帧时才去读取流中的数据。更优雅的方式是利用ReadableStream的异步迭代：

   for await (const chunk of processedStream) { if (this.isPaused) { // 暂停时，利用for-await的背压机制，流会自动暂停 // 需要配合一个可恢复的信号 await this.waitForResume(); } // 处理chunk }

4. 错误边界与重试策略：实现一个指数退避的重试逻辑。当流因网络错误结束时，不是简单报错，而是等待一段时间（如1s, 2s, 4s, 8s...）后重新调用start()方法。同时，在UI上给用户明确的连接状态提示（“连接中”、“已连接”、“断开重连中...”）。

6. 对比总结与选型指南

经过这三代的演进，我们现在拥有了一个层次清晰、能力强大的流式处理工具箱。下面用一个表格来直观对比：

选型建议：

• 如果你的需求只是“接收服务器发来的文本通知”，且格式是标准的SSE：直接使用EventSource。它简单、稳定、省心。这是绝大多数通知类场景的首选。
• 如果你需要处理非SSE格式的流（如ndjson）、需要携带复杂认证头、或需要主动取消请求：使用Fetch API+ReadableStream。这是与现代Web平台交互的标准方式。
• 如果你需要在数据到达客户端后，进行复杂的实时处理（如过滤、解析、聚合、格式转换）：在Fetch流的基础上，接入一个或多个TransformStream。它们像乐高积木一样，可以组合出强大的数据处理管道。
• 对于全新的项目，尤其是涉及复杂流处理的：建议以Fetch + ReadableStream为基础架构，搭配TransformStream来处理业务逻辑。EventSource可以作为特定场景下的一个简化替代方案。

7. 常见问题与排查技巧实录

在实际集成这些技术时，我踩过不少坑。这里把一些典型问题和解决方法记录下来，希望能帮你节省时间。

问题1：使用Fetch读取SSE流，发现数据接收不完整或延迟很大。

• 排查：首先检查TextDecoder.decode(value, { stream: true })中的{ stream: true }选项是否遗漏。这个选项告诉解码器数据是流式的，可能存在跨块的字符，必须保留中间状态。没有它，遇到跨块的UTF-8字符就会解码错误，导致后续解析失败。
• 排查：检查你的缓冲区切割逻辑。如果数据块不是按行完整到达的，简单的split(‘\n’)可能会把一行拆散。确保你的缓冲区和按行切割逻辑是健壮的，如上面示例所示。
• 排查：网络问题。浏览器开发者工具的Network面板，找到那个Fetch请求，查看“Response”标签页。如果数据是慢慢出现的，说明流是正常的。如果一直不出现，可能是服务器端没有正确刷新缓冲区（res.flush()in Node.js）或者响应头设置不正确。

问题2：TransformStream 似乎没有处理所有数据，或者处理顺序不对。

• 排查：transform方法必须是同步的，或者返回一个Promise。如果你在transform里执行了异步操作（比如fetch），必须确保在enqueue之前await完成，否则数据顺序会乱。
• 排查：检查flush方法。流结束时，缓冲区里可能还有未处理的数据。必须在flush方法中将它们enqueue出去。
• 排查：确保没有在transform或flush中抛出未捕获的异常。这会导致整个流错误终止。

问题3：在React/Vue组件中使用流，组件卸载时流没有正确关闭，导致内存泄漏或错误。

• 解决方案：在组件的清理函数中（React的useEffectcleanup，Vue的onUnmounted）必须执行以下操作：

  // 假设 this.reader 是流阅读器 const cleanup = async () => { if (this.reader) { await this.reader.cancel('组件卸载'); // 1. 取消读取 this.reader.releaseLock(); // 2. 释放锁（非常重要！） this.reader = null; } // 如果有 AbortController 用于 fetch，也调用 abort() if (this.abortController) { this.abortController.abort(); } };

  顺序很重要：先cancel，再releaseLock。

问题4：服务器发送了SSE数据，但EventSource的onmessage事件一直不触发。

• 排查：首先用curl或Postman直接请求SSE端点，看原始数据格式是否正确。必须确保每个事件后面跟的是两个换行符\n\n，而不是一个。这是SSE协议最容易被忽略的地方。
• 排查：检查响应头Content-Type必须是text/event-stream。其他类型如application/json会导致EventSource无法解析。
• 排查：查看浏览器控制台是否有CORS错误。SSE也受同源策略限制。如果需要跨域，服务器必须设置正确的Access-Control-Allow-Origin等头。

问题5：如何模拟一个慢速流或者测试背压机制？

• 技巧：可以在服务器端在每次res.write()后加一个延迟，比如await new Promise(resolve => setTimeout(resolve, 100))。然后在客户端观察数据接收是否平稳。
• 技巧：在客户端的处理逻辑中（比如processLine函数里）加入一个随机延迟，模拟复杂的UI渲染。观察是否会导致内存中数据积压。如果会，就需要引入背压控制逻辑，比如暂停读取直到UI就绪。

流式处理是一个涉及前后端配合的领域，调试时需要两端同时观察。浏览器的开发者工具、服务器的日志以及清晰的协议约定，是快速定位问题的关键。从简单的EventSource入手，逐步深入到ReadableStream和TransformStream，你会发现处理实时数据流的思路越来越清晰，能够应对的场景也越来越复杂。这套技术栈，无疑是开发现代化、响应式Web应用的利器。