Web Streams backpressure前端流式读取怎样避免消费速度失控为什么结果为空或不符合预期

前端用 fetch() 读取长响应、日志或大文件时，消费端处理不过来，最容易出现两个现象：内存里的待处理 chunk 越积越多，或者读取循环看似正常却得到空结果。处理这类问题的关键不是固定延时，而是让流的队列容量向生产者反馈压力。Web Streams 官方参考地址：https://developer.mozilla.org/en-US/docs/Web/API/Streams_API/Concepts。

先让消费端成为唯一背压阀

ReadableStream 的数据按 chunk 排队，reader 一次取一个。若代码在循环里并发调用大量 read()，再把结果全部塞进数组，生产者仍会持续供给，所谓“异步”反而失去边界。先保持一个 reader、一次只处理一个 chunk，确认处理完成后再进入下一次读取。

流也有锁：reader 建立后，流被锁定给它。另一个函数直接读取同一个流，可能得到锁定错误或拿到空结果。需要换消费者时，先结束或取消当前 reader，再释放锁；不要在多个组件之间共享同一个 reader 引用。

用 desiredSize 判断队列是否接近上限

在自定义流中，队列策略会用 high water mark 减去当前队列总大小，得到 desired size。数值降到 0 或负数，说明暂时不适合继续生成 chunk；生产者应等待下一次 pull()，而不是继续写入一个无界数组。默认的计数策略按 chunk 数量计，按字节控制时要明确使用字节大小策略。

function createBoundedStream(source) {
  return new ReadableStream({
    start(controller) {
      // 只注册一次取消清理，避免 source 在结束后继续产生数据。
      source.onCancel(() => controller.error(new Error("source cancelled")));
    },
    pull(controller) {
      // pull 由队列容量驱动；不要在这里无限循环 enqueue。
      return source.nextChunk().then((chunk) => {
        if (chunk === null) {
          // 没有更多数据时关闭，读取端会收到 done=true。
          controller.close();
          return;
        }
        controller.enqueue(chunk);
        // 这里只观察剩余容量，不能把它当作已发送字节数。
        if (controller.desiredSize !== null && controller.desiredSize 

这段代码是实现示意，source 代表业务数据源。真正接入时要保证 nextChunk() 不会被重入，并在失败时让 promise 拒绝；否则生产者可能已经暂停，消费者却永远等不到下一块。若数据块大小差异很大，按数量设上限并不等于内存有界，应改用按字节计算的 size 函数并设置可解释的高水位。

把读取、取消和异常放在同一生命周期

网络请求通常与页面、组件或用户操作绑定。读取函数应接收 AbortSignal，在取消时停止 fetch；循环结束后再释放 reader 锁。不要只在 UI 上隐藏组件，却让后台循环继续读取和拼接字符串。

async function consumeText(response, signal, handleChunk) {
  const reader = response.body?.getReader();
  if (!reader) throw new Error("响应没有可读取的 body");
  const abort = () => reader.cancel("consumer aborted");
  signal.addEventListener("abort", abort, { once: true });
  try {
    while (true) {
      // 串行等待当前 chunk 处理完成，背压才有机会传回生产端。
      const { value, done } = await reader.read();
      if (done) break;
      await handleChunk(value);
    }
  } finally {
    // 清理监听并释放锁；cancel 的拒绝不能覆盖原始处理错误。
    signal.removeEventListener("abort", abort);
    reader.releaseLock();
  }
}

出现“结果为空”时，先检查是否已有别的 reader 消费过 response.body；流不能像普通字符串一样重复读取。再区分 done 为 true 的正常结束、cancel() 触发的主动终止，以及处理函数抛出的业务异常。三者都应写入可区分的日志字段，不能只打印“读取失败”。

用 pipeThrough 传递处理阶段的压力

如果读取后还要解码、拆行或限速，优先用 TransformStream 组成 pipe chain。后面的 sink 变慢时，背压可以沿链条向前传递。手写“读取—push 到数组—定时清空”的方式很难同时处理异常和取消，也容易把内存上限变成隐含行为。

Web Streams 背压的常见疑问

把 highWaterMark 调大是不是就不会失控？

不是。它只是提高队列偏好容量，可能换来更高内存占用。先让消费者串行、生产者可暂停，再按 chunk 大小和业务延迟选择水位。

desiredSize 为负数时必须立刻丢弃数据吗？

不必。它是背压信号；生产者应暂停或等待下一次 pull，是否丢弃要由协议和业务明确决定。

releaseLock 会把流关闭吗？

不会。它只解除 reader 对流的锁。若仍有未处理数据，另一个合适的 reader 可能继续读取；要终止数据源，应显式 cancel。