Fetch Streams 分段读取大响应体的背压控制

读取几十 MB 甚至更大的接口响应时，直接调用 response.json() 或 response.text() 会等到完整内容到齐后再交给业务层。更稳妥的做法是使用 response.body，把字节流逐段解码、转换并交给慢速消费者。只要管道末端没有盲目把所有块塞进数组，pipeTo() 就能把消费速度向前传递，形成背压。

官方文档：https://developer.mozilla.org/en-US/docs/Web/API/Streams_API/Using_readable_streams

Fetch Streams 的正确分层：字节、文本、业务三段管道

Response.body 是一个可读字节流。它的上游是网络响应，下游可以接文本解码器、行拆分器或业务写入端。每个块都可能停在一个多字节字符中间，所以不能直接对单个块调用 new TextDecoder().decode(chunk) 后就当成完整文本；应让 TextDecoderStream 保留解码状态。

下面的示例按换行拆分日志记录。TransformStream 只负责转换，业务写入交给末端，这样每层职责清楚，也方便替换成 CSV、NDJSON 或增量渲染。

async function consumeLines(url, onLine, signal) {
  // 先判断 HTTP 状态；fetch 对 404/500 仍可能正常 resolve。
  const response = await fetch(url, { signal });
  if (!response.ok) throw new Error(`HTTP ${response.status}`);
  if (!response.body) throw new Error("ReadableStream 不可用");

  let pending = "";
  const splitLines = new TransformStream({
    transform(chunk, controller) {
      // 保留末尾半行，避免一个记录跨越两个网络块时被截断。
      pending += chunk;
      const lines = pending.split(/\r?\n/);
      pending = lines.pop() ?? "";
      for (const line of lines) if (line) controller.enqueue(line);
    },
    flush(controller) {
      // 响应末尾没有换行时，仍然交付最后一条记录。
      if (pending) controller.enqueue(pending);
    }
  });

  const sink = new WritableStream({
    async write(line) {
      // 慢速处理会让 pipeTo 等待，避免业务队列无限增长。
      await onLine(line);
    }
  });

  await response.body
    .pipeThrough(new TextDecoderStream())
    .pipeThrough(splitLines)
    .pipeTo(sink, { signal });
}

pipeTo 如何把慢速消费变成背压

背压不是给 fetch() 增加一个“限速参数”，而是管道末端暂时不能接收时，流系统让前面的队列停止继续填充。pipeTo() 会连接可读端和可写端；当 write() 返回的 Promise 尚未完成，后续数据就不会无条件地继续进入业务处理。

取消、错误和资源边界要一起设计

超时和页面卸载时，应让同一个 AbortSignal 贯穿 fetch() 与 pipeTo()。管道发生异常时，pipeTo() 返回的 Promise 会拒绝；调用方要捕获 AbortError，把它和服务器错误区分开。若自行使用 getReader()，也要在 finally 中释放 reader；使用管道时则不要同时给同一个流再加一个 reader，因为 pipe 操作会锁定流。

const controller = new AbortController();
const timer = setTimeout(() => controller.abort(), 15_000);

try {
  // 同一个 signal 同时控制请求和管道，超时后两处都能收尾。
  await consumeLines("/api/large-log", handleLine, controller.signal);
} catch (error) {
  // 用户取消不是服务端失败，日志和提示应分开处理。
  if (error.name === "AbortError") console.info("流式读取已取消");
  else throw error;
} finally {
  // 无论成功、失败还是取消，都清理定时器。
  clearTimeout(timer);
}

常见问题

ReadableStream 能直接替代 response.json() 吗？

不能简单替代。流式读取适合增量处理；如果业务必须等待完整 JSON 并进行一次性校验，仍可使用 json()，但要接受完整响应进入内存的边界。

设置更大的 highWaterMark 就一定更快吗？

不一定。更大的队列可能减少短暂抖动，却也会提高内存占用和取消延迟。先让末端写入真正 await，再根据数据形态调整队列策略。

为什么读取到的中文偶尔乱码？

常见原因是把每个网络块独立解码。UTF-8 字符可能跨块，使用 TextDecoderStream 或保留 decoder 状态的解码方式即可避免这种截断。

把 Fetch Streams 看成“源、转换、消费”三段边界，重点就从手动拼接字符串转向控制队列、错误和取消。大响应体真正需要的是可持续的消费速度，而不是把缓冲区调到最大。