Web Streams backpressure前端流式读取怎样避免消费速度失控怎样完成一个可复用的最小示例

前端用 fetch() 读取大响应或持续消息时，如果生产端不断塞入数据，而消费端还在解析、渲染或写入本地缓存，内部队列就会越积越多。Web Streams 的 backpressure（背压）解决的正是这个节奏问题：让生产端根据队列剩余空间决定是否继续提供下一块数据，而不是让消费端被动承受无限增长。

可复用的做法是把“生成下一块数据”的动作放进 ReadableStream 的 pull()，每次先观察 controller.desiredSize；消费端用异步迭代逐块处理，并在取消或异常时调用取消逻辑。这样，消费速度会反过来影响生产速度。

适合阅读：正在处理大文件、流式接口、逐块解码或增量渲染的前端开发者。本文的示例是结构演示，代码没有在本文写作环境中执行。

先看清楚：为什么队列会越积越多

把数据流想成两侧：左侧是底层 source，负责产生 chunk；中间是 ReadableStream 的内部队列；右侧是 reader，负责取出 chunk 并处理。生产速度大于消费速度时，队列会暂存更多数据。短时间看只是延迟增加，长时间则可能表现为内存占用上升、渲染卡顿或取消响应变慢。

背压不是简单地“暂停整个程序”，而是把队列容量变成生产端可读取的信号。自定义底层 source 时，pull(controller) 会在流需要更多数据时被调用；生产端可以结合 desiredSize 判断是否有继续供给的空间。

用 pull 和 desiredSize 写出最小生产端

下面的生成器把每个 chunk 看作一小段文本。关键点不是数字本身，而是 pull() 只负责提供有限的一块，并在达到总量后关闭流。highWaterMark 是队列策略的一部分，用来表达希望保留的容量；它不是“允许无限缓存”的开关。

function createNumberStream(total = 20) {
  let next = 1;

  return new ReadableStream({
    start(controller) {
      // start 只保存状态；真正供给数据交给 pull，避免一次性塞满队列。
      controller.enqueue(`chunk-${next}`);
      next += 1;
    },
    pull(controller) {
      // desiredSize 小于等于 0 时，先让消费端腾出空间再继续生产。
      if (controller.desiredSize  total) {
        // 数据已经用完，关闭流，让异步迭代可以自然结束。
        controller.close();
        return;
      }

      // 每次只放入一个 chunk，生产节奏由队列需求驱动。
      controller.enqueue(`chunk-${next}`);
      next += 1;
    },
    cancel(reason) {
      // 真实场景中在这里中止网络请求、释放文件句柄或清理计时器。
      console.info("stream canceled", reason);
    }
  }, {
    // 队列策略只设置期望容量，不能替代消费端的正确处理。
    highWaterMark: 2,
    size() {
      // 每个字符串 chunk 按一个单位计数，便于示例观察背压。
      return 1;
    }
  });
}

这里有一个容易忽略的边界：start() 中预先放入了第一块，后续块由 pull() 产生。如果实际项目的 source 是网络或文件，应把“下一块是否可用”的异步动作放进 pull()，并在失败时调用 controller.error(error)，不要用一个不受控的循环持续 enqueue()。

让消费端真正形成节奏

有了生产端的限制，消费端还需要按块处理。异步迭代会在每次拿到 chunk 后继续执行；如果处理函数包含解码、渲染或持久化，下一次取数自然会等待这一块处理完成。

async function consumeStream(stream, renderChunk) {
  try {
    // for await 每次只拿一个 chunk，处理完成后再继续向流请求数据。
    for await (const chunk of stream) {
      await renderChunk(chunk);
    }
  } catch (error) {
    // 消费端异常时把失败原因记录下来，避免静默吞掉流错误。
    console.error("stream consume failed", error);
    throw error;
  }
}

const stream = createNumberStream(20);
consumeStream(stream, async (chunk) => {
  // 这里替换成解码、增量渲染或写入缓存等真实消费动作。
  await new Promise((resolve) => setTimeout(resolve, 30));
  console.log("consume", chunk);
});

如果消费函数内部又启动了大量并行任务，for await 这一层的节奏仍可能被下游并发抵消。此时应把并发上限放在消费函数内部，并明确“已取出的 chunk”数量，不要把整个流先转换成数组再一次性处理。

结束、取消和异常要分开处理

正常结束用 controller.close()，底层 source 失败用 controller.error()，调用方不再需要数据时由 cancel() 做资源清理。三者语义不同：关闭代表没有更多数据，错误代表流不可继续，取消代表消费者主动放弃。

• 队列接近满：让 pull() 暂停继续供给，不要在外部定时器中无限入队。
• 读取完成：关闭流，消费端结束异步迭代。
• 用户离开页面：调用读取链的取消逻辑，同时中止关联的 fetch 或其它底层资源。
• 解析失败：将错误传递给流，让上层可以统一显示失败状态。

若手动使用 getReader() 而不是异步迭代，记得在 finally 中释放 reader 的锁；当流由 fetch 的 response.body 提供时，也要把 AbortController 与取消动作绑定，避免只停止界面更新却继续下载。

几个容易误判的边界

desiredSize 是“填满内部队列所需的大小”，可能为负数，流已关闭时为 0，流出错时可能为 null。因此不要把它当作精确的剩余字节数，也不要假定每个 chunk 都有相同的物理大小。自定义 size() 时，应让计数单位和你的容量目标一致。

现代浏览器普遍支持 ReadableStream，但项目仍应按目标浏览器检查异步迭代、字节流和相关转换流的支持范围。若需要兼容更旧环境，可将消费逻辑改写为显式 reader.read() 循环，并提供取消和错误分支；不要因为降级而移除背压策略。

相关问题

为什么只在消费端加 setTimeout 还会占用很多内存？

延迟只改变消费速度，没有限制生产端。如果生产端仍不断入队，队列照样增长。应让供给动作受 pull() 和队列需求驱动。

highWaterMark 越大是不是越快？

它只表达期望的排队容量。容量过大可能增加内存和取消延迟，过小则可能让生产端频繁等待，应按 chunk 成本和消费耗时调整。

fetch 的 response.body 需要自己 new ReadableStream 吗？

通常不需要，Response.body 已经是可读流。只有需要插入解码、限速、格式转换或自定义 source 时，才适合用转换流或自定义 ReadableStream 包一层。

回到标题中的问题，避免消费速度失控的核心不是给循环加一个固定延时，而是建立“队列容量—生产动作—消费完成”三者之间的反馈。先用 desiredSize 限制供给，再让消费端逐块推进，并把关闭、错误和取消分开处理，这个最小结构就能迁移到文件读取、流式接口和增量渲染等场景。