Web Streams 实战：用 ReadableStream 与 TransformStream 处理流数据

为什么一次性读完整个响应不够用

处理一个大文件时，最常见的写法是 await response.json() 或 await response.text()，让浏览器把整个响应体攒齐再返回。这在几 KB 的接口数据上完全没问题，但换成几百 MB 的日志、CSV 导出或 NDJSON 事件流，就会同时撞上三个问题：内存（整份数据要一次性驻留）、首字节延迟（必须等全部下完才能开始处理）、以及没有背压（网络下得多快就得接多快，消费端来不及处理也只能堆着）。

fetch 的 response.body 本身就是一个 ReadableStream，Web Streams API 让我们可以边下边读、边读边转换、边转换边写入。本文从最小读取开始，一路写到管道、背压与工程里常见的坑。

一、最小可用写法：getReader

最底层的消费方式是拿到 reader，循环 read()。每次返回一个 { done, value }，value 是 Uint8Array 而不是字符串，这一点刚上手最容易搞错。

const response = await fetch('/large.json');
const reader = response.body.getReader();   // body 是一个 ReadableStream

let received = 0;
while (true) {
  const { done, value } = await reader.read();
  if (done) break;
  // value 是 Uint8Array，不是字符串
  received += value.length;
  console.log('分片', value.length, '累计', received);
}
console.log('读取完成');

读到 done: true 就结束；中途不想读了可以调 reader.cancel()，它会释放底层连接。注意 一个流只能被消费一次：一旦取了 reader，就不能再对同一个 body 调 json() 或再取第二个 reader，需要两份数据得用后面说的 tee()。

二、带进度的读取

分片读取最直观的收益是能做进度条：累计每个分片的 length，再除以 Content-Length。

async function fetchWithProgress(url, onProgress) {
  const response = await fetch(url);
  const total = Number(response.headers.get('Content-Length')) || 0;
  const reader = response.body.getReader();
  let received = 0;
  const chunks = [];

  while (true) {
    const { done, value } = await reader.read();
    if (done) break;
    chunks.push(value);
    received += value.length;
    if (total) onProgress(received / total);   // 没有长度时只能显示 indeterminate
  }
  return new Blob(chunks);
}

这里有两个必须说清的限制：一是响应被 gzip 压缩或使用 chunked 传输时可能没有 Content-Length（或长度指压缩后大小），这时进度只能降级成不确定态，硬算会得到错误的百分比；二是示例里把所有 chunks 存进数组最后拼 Blob，本质上还是全量驻留内存，只适合中等文件。真正的大文件应该边读边处理，不要累积分片。

三、自己构造 ReadableStream：理解背压

不只是 fetch，任何「按需产出」的数据源都可以包装成流。构造时传入 start / pull / cancel，用 controller.enqueue() 推数据、close() 结束、error() 报错。

let n = 0;
const stream = new ReadableStream({
  start(controller) {
    console.log('开始订阅');
  },
  // pull 只在「消费者要数据、且内部队列有空位」时被调用 —— 这就是背压
  pull(controller) {
    n += 1;
    controller.enqueue(new TextEncoder().encode('line ' + n + '\n'));
    if (n >= 5) controller.close();
  },
  cancel(reason) {
    console.log('消费者取消，清理资源', reason);
  }
});

const reader = stream.getReader();
await reader.read();   // 读取一次才会触发 pull

关键在 pull：它只在消费者需要数据、且内部队列没满时才被调用。浏览器用 highWaterMark 控制队列水位，消费慢了就不继续 pull——这就是背压机制，也是流和普通回调队列的本质区别。cancel() 里应该关闭定时器、WebSocket 或文件句柄，避免消费者提前退出后资源泄漏。

四、TransformStream：逐行解析 NDJSON

真实场景很少只是「读」，更多是「边读边转换」。比如服务端返回 NDJSON（每行一个 JSON 对象）的事件流，分片边界几乎不可能刚好落在换行处，必须自己缓存半行、拼接、再按行解析。TransformStream 的 transform 负责逐片处理，flush 在流结束时收尾。

function ndjsonToObjects() {
  let buffer = '';
  const decoder = new TextDecoder();
  return new TransformStream({
    transform(chunk, controller) {
      // 必须用 { stream: true }，否则跨分片的多字节字符会被截断
      buffer += decoder.decode(chunk, { stream: true });
      const lines = buffer.split('\n');
      buffer = lines.pop();              // 末行可能不完整，留到下一片
      for (const line of lines) {
        if (!line.trim()) continue;
        controller.enqueue(JSON.parse(line));
      }
    },
    flush(controller) {
      buffer += decoder.decode();        // 冲刷解码器里残留的字节
      if (buffer.trim()) controller.enqueue(JSON.parse(buffer));
    }
  });
}

const res = await fetch('/events.ndjson');
const objects = res.body.pipeThrough(ndjsonToObjects());
// 现代浏览器里 ReadableStream 支持异步迭代
for await (const obj of objects) {
  console.log(obj.id, obj.type);
}

三个细节：TextDecoder.decode(chunk, { stream: true }) 的 stream 参数必须加，否则一个 UTF-8 汉字被切在两个分片里就会变成乱码；buffer.split('\n') 之后要 pop() 把可能不完整的末行留到下一片；flush 里补一次 decoder.decode() 冲刷残留字节，再处理最后一行。解析失败时不要吞异常，让它抛出使整条管道失败，否则会静默丢数据。

五、pipeThrough / pipeTo 与写入端

把「读、解码、转换、写入」串成一条管道，用 pipeThrough 接转换器（返回新的 ReadableStream，可继续串联），用 pipeTo 接到 WritableStream 收尾。

const controller = new AbortController();

const writable = new WritableStream({
  write(row) {
    // 返回 Promise：没 resolve 之前不会继续拉取上游，天然形成背压
    return appendToIndexedDB(row);
  },
  close() { console.log('全部写入完成'); },
  abort(reason) { console.log('写入中止', reason); }
});

const res = await fetch('/big.csv', { signal: controller.signal });
await res.body
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(csvToRows())
  .pipeTo(writable, { signal: controller.signal });

// 取消时同时中断网络请求与整条管道
// controller.abort();

WritableStream 的 write() 返回一个 Promise，在它 resolve 之前管道不会继续拉取上游数据——写入数据库慢，读取就自动慢下来，背压贯穿整条链路，不需要自己写节流。pipeTo 的第二个参数可以传 signal，把 AbortController 同时挂在 fetch 和管道上，一次 abort() 就能干净地中断下载与写入；如果只取消 reader 而不 abort fetch，连接其实还在后台继续下载。

六、tee 与几个容易踩的坑

需要把同一份流给两个消费者时用 tee()，它返回两个独立分支：

const [branchA, branchB] = response.body.tee();

// 两个分支各自独立消费，但浏览器要为「慢的那个」缓冲数据
const checksum = computeHash(branchA);      // 快
const preview  = readFirstKB(branchB);      // 很快结束

// 如果慢分支一直不读，缓冲会持续堆积 —— 大文件场景要慎用 tee

但 tee 不是免费的：浏览器必须为较慢的那个分支缓冲数据，如果一边几乎不读，内存会持续堆积。大文件场景更稳妥的做法是重新发起请求，或者在一开始就把数据分流。其余常见坑集中列一下：

• 流只能消费一次：body 取过 reader 就不能再 json() / text()，报错信息通常还不太直观；
• 别把分片全存进数组：那样等于退回全量加载，失去流式处理的意义；
• 多字节字符与跨片换行：{ stream: true } 和 flush 缺一不可；
• 进度条别硬算：压缩 / chunked 下没有可靠总长度，要准备不确定态；
• 取消要成对：reader.cancel() 与 controller.abort() 配合使用，别留下后台下载；
• 兼容性：TextDecoderStream、pipeTo 的 signal、ReadableStream 的异步迭代在老环境需要探测或 polyfill，Node 18+ 才内置全局 Web Streams；
• 错误处理要显式：transform 里抛错会终止整条管道，记得在 pipeTo 外层 catch，并区分「解析失败」和「网络中断」。

小结

Web Streams API 的价值不在语法，而在于它把背压变成了默认行为：生产者按消费者的节奏产出，内存占用与文件大小解耦。从 getReader() 的最小读取，到 TransformStream 逐行解析，再到 pipeTo 串起写入端，一条管道就能覆盖下载、解码、转换、落库全过程。真正需要留心的是边界：流一次性、跨分片字符、tee 的缓冲代价，以及取消时 fetch 与管道要一起收尾。把这些记清楚，处理大文件和实时事件流会从容很多。

本文为原创实践记录，示例代码均可直接运行，欢迎交流更稳妥的大文件流式处理方案。