ReadableStream逐块读取响应并处理背压的实现方法

如果你要把 fetch 的响应边读边显示，关键不是把所有内容先读进内存，而是让每一次 reader.read() 都等待当前 chunk 被消费。这样消费者变慢时不会无界预读；用户停止、命中目标或处理失败时，再用 reader.cancel() 结束不再需要的流。

官方地址：https://developer.mozilla.org/en-US/docs/Web/API/ReadableStream

最小可靠组合是：response.body.getReader() 获取锁定的 reader，循环等待 read()，用 TextDecoder.decode(..., {stream:true}) 解码，并在提前结束时等待 reader.cancel(reason)，最后释放锁。

本文只讨论默认 reader、逐块读取和 cancel；如果要同时处理多条转换链，应另看 TransformStream。

获取 reader 并确认流的锁定边界

response.body 是可读流，也可能为空。调用 getReader() 后，流会被锁定，其他代码不能再取得第二个 reader。因此应把 reader 的所有权放在一个函数里，避免组件卸载、重复请求和事件回调同时读取。

async function getReaderFromResponse(response) {
  // 没有 body 时不能继续读取，先把异常变成调用方可处理的结果。
  if (!response.body) {
    throw new Error("响应没有可读取的 body");
  }

  // 默认 reader 按 chunk 读取字节流，并锁定这条流的读取权。
  return response.body.getReader();
}

不要把同一个 response.body 交给多个读取函数。需要分流时要明确选择 tee() 或其他架构，但那已经超出本文的 reader 与 cancel 边界。

用 read 循环逐块解码并控制消费节奏

read() 返回 { done, value }。文本响应不能简单地对每个 Uint8Array 单独调用解码器，否则一个汉字跨 chunk 时可能出现替换字符。应复用一个 TextDecoder，并在结束时补一次不带 stream:true 的 flush。

async function readTextStream(response, onChunk, signal) {
  const reader = await getReaderFromResponse(response);
  const decoder = new TextDecoder();
  let text = "";

  try {
    while (true) {
      // 每次只取一个 chunk；处理完当前内容后才发起下一次读取。
      const { done, value } = await reader.read();
      if (done) {
        // flush 处理最后一个不完整的多字节字符。
        text += decoder.decode();
        return text;
      }

      // stream:true 让跨 chunk 的 UTF-8 字符留给下一次解码。
      text += decoder.decode(value, { stream: true });
      onChunk(text);

      // 外部停止信号到来时，结束未完成的消费，不继续预读。
      if (signal?.aborted) {
        await reader.cancel("用户停止读取");
        return text;
      }
    }
  } catch (error) {
    // 处理函数或网络异常时通知底层 source，并把原始错误交给调用方。
    await reader.cancel("读取或处理失败").catch(() => {});
    throw error;
  } finally {
    // 无论完成、取消还是异常，都释放 body 的读取锁。
    reader.releaseLock();
  }
}

示例中的 onChunk 可以更新文本节点、追加解析结果或交给状态管理器。不要在循环里再启动一个不受控的后台读取任务；当前 chunk 没有处理完，下一次 read() 就不应提前排队。

把背压理解为消费端的等待关系

对响应体来说，最实用的背压处理是让消费端按自己的速度调用 read()：渲染、解析或落盘耗时，就自然延迟下一次读取。reader 本身没有一个让你随意设置的“速度参数”，也不要把每次 chunk 的大小误认为业务层可以固定控制。

如果你自己用 new ReadableStream() 创建底层 source，才会在 controller 上看到 desiredSize；它表示内部队列相对高水位的期望容量，可能为负，属于 source 的入流控制信号。本文的 fetch 示例只负责正确消费和取消，不能据此声称已经控制了服务器发送速度。

在提前结束和异常时调用 cancel

cancel() 表示这条流后续的数据都不要了，返回一个 Promise。它和“暂时没有处理完”不同：处理慢时只需延后下一次 read()，只有确认不再需要剩余内容时才取消。取消后流不可继续读取，原因字符串可用于底层 source 的记录。

命中搜索词、用户点击停止、组件销毁或解析器报错，都可以进入同一条取消路径。若还需要保留已收到的前缀，应先保存业务状态，再等待 cancel；不要把取消后的流当成可恢复断点。

释放锁并复查三种结果

正常结束时，done 为 true，应完成 decoder flush；主动取消时，保留已接收文本并把结束原因交给界面；异常时，先取消底层流，再把错误抛给重试或提示逻辑。三种结果都要经过 finally 的 releaseLock()。

排查“读取卡住”时，先看是否有一个循环已经锁住 body，再看消费回调是否永远没有返回；排查“文字乱码”时，确认是否复用了 TextDecoder 和 stream:true；排查“停止后仍在增长”时，确认是否等待了 cancel()，而不是只把页面按钮设为 disabled。

相关问题

ReadableStream 的 reader 能重复创建吗？同一时刻不能。已有 reader 会锁定流，先释放锁或让当前函数完整结束。

处理慢时要马上 cancel 吗？不需要。慢只是背压信号，延迟下一次 read()；只有业务确定不再需要剩余数据时才 cancel。

cancel 后还能继续读吗？不能。若只是暂时让出读取权，应区分业务暂停和真正放弃剩余数据的取消语义。