Web Streams 怎样处理背压和取消

Web Streams 处理“生产快、消费慢”时，正确做法是让管道自己传播背压，而不是在生产端无条件调用 enqueue()。把数据源放进 ReadableStream 的 pull()，让 TransformStream 只做转换，再用 pipeTo() 连接慢速 sink；取消时统一调用 AbortController.abort()，并在 source 的 cancel() 中关闭连接、定时器或文件句柄。

参考入口：https://developer.mozilla.org/en-US/docs/Web/API/Streams_API

背压解决“什么时候继续生产”，取消解决“什么时候彻底停止并释放资源”。前者看队列需求，后者看生命周期信号，两件事要同时设计。

先划清 source、transform 和 sink 的责任

可以把一条流想成三段：source 负责从网络、文件或业务生成数据；transform 负责解码、过滤或分片；sink 负责写入页面、缓存或另一个接口。队列属于流的内部状态，业务代码只应控制边界，不要把所有数据先收进数组。

一个实用的判断是：source 只回答“现在是否有需求”，transform 只回答“这个块如何变化”，sink 只回答“这个块何时真正写完”。职责混在一起时，最容易出现重复消费、无法停止和内存持续增长。

用 pull 与 desiredSize 让生产端尊重背压

可读流的队列会根据 queuing strategy 和 high water mark 计算需求。desiredSize 越小，表示可继续接收的空间越少；当它小于或等于零，生产端就不应继续无条件塞入新块。实际生产代码通常把读取动作放在 pull()，让运行时在消费者需要时再调用。

function makeStream(getChunk) {
  let closed = false;

  return new ReadableStream({
    async pull(controller) {
      // pull 由下游需求触发；避免用无限循环提前填满队列。
      if (closed || controller.desiredSize 

这里的 desiredSize 是保护线，不是业务吞吐量指标。若每个块大小差异很大，应使用 ByteLengthQueuingStrategy 或自定义 size()，不要只按块数估算内存。

用 pipeThrough 和 pipeTo 连接转换与落点

pipeThrough() 把可读侧接到转换流，返回转换后的可读侧；pipeTo() 把最终结果交给 writable sink，并返回一个表示整条管道完成或失败的 Promise。这样，转换阶段不用自行维护“暂停/恢复”协议，队列压力会沿着管道传播。

const upperCase = new TransformStream({
  transform(chunk, controller) {
    // 转换只处理当前块；enqueue 后由下游队列继续施加压力。
    controller.enqueue(chunk.toUpperCase());
  },
});

const sink = new WritableStream({
  async write(chunk) {
    // 用异步写入模拟慢速落点；Promise 完成前不会确认本次写入。
    await saveChunk(chunk);
  },
  close() {
    // 正常结束时提交或关闭落点。
    console.log("sink closed");
  },
  abort(reason) {
    // 异常中止时撤销未完成的落点操作。
    console.warn("sink aborted:", reason);
  },
});

const pipePromise = source
  .pipeThrough(upperCase)
  .pipeTo(sink);

pipePromise.catch((error) => {
  // 统一处理 source、transform 或 sink 的失败。
  console.error("pipeline failed:", error);
});

不要在 sink 的 write() 里启动无法等待的后台任务，否则表面上写入已完成，实际队列仍在增长。要让背压有效，写入完成必须由返回的 Promise 表达。

用 AbortController 贯穿用户取消与超时

取消是“我已经不再需要这些数据”，不是“把流正常读完”。给 pipeTo() 传入 AbortSignal 后，页面离开、用户点击停止或超时都可以走同一条路径；source 的 cancel() 负责释放底层资源，sink 的 abort() 负责处理未完成写入。

const controller = new AbortController();
const timeoutId = setTimeout(() => {
  // 超时与用户主动停止共享同一个取消信号。
  controller.abort(new Error("stream timeout"));
}, 8000);

source
  .pipeThrough(upperCase)
  .pipeTo(sink, { signal: controller.signal })
  .then(() => {
    // 正常完成才走这里；取消通常会让 Promise rejected。
    console.log("pipeline completed");
  })
  .catch((error) => {
    // 区分取消和其他错误，避免把用户主动停止记成系统故障。
    if (controller.signal.aborted) {
      console.log("pipeline canceled:", controller.signal.reason);
    } else {
      console.error("pipeline failed:", error);
    }
  })
  .finally(() => {
    // 无论成功、失败还是取消，都不能遗留超时定时器。
    clearTimeout(timeoutId);
  });

stopButton.onclick = () => {
  // 用户不再需要结果时，停止整条 pipe。
  controller.abort("user stopped");
};

close() 表示正常结束，cancel() 表示可读侧放弃继续读取，abort() 表示 writable 侧被突然终止。三者都可能需要清理资源，但语义和错误传播不同，不能用一个“结束”回调替代。

按队列、关闭和取消边界做复查

上线前可以用下面这张速查表检查管道：

最后记住两个边界：队列高水位只限制“愿意缓存多少”，不能替代业务限流；AbortController 只发出取消信号，真正的连接关闭和资源释放仍要由 underlying source、transform 或 sink 实现。

常见问题

背压是不是只需要手动判断 desiredSize？

不一定。标准 pipe 链路会管理队列和传播需求，手动判断主要用于自定义 source 的生产策略；关键是不要绕开 pull 和写入 Promise。

调用 cancel 后还能继续读队列里的数据吗？

不能把它当作暂停。cancel 表示消费者放弃流，已经排队但未读的数据会被丢弃；如果只是暂时处理不过来，应让下游自然施加背压，而不是调用 cancel。