Web TransformStream 怎样体现背压

Web TransformStream 的背压，体现在写入方不能只顾着调用 writer.write()，而要让写入 Promise 和下游队列共同决定下一次写入节奏。数据从 writable 端进入转换器，transform(chunk, controller) 通过 controller.enqueue() 把结果放到 readable 端；当读取方变慢，队列的期望容量下降，继续写入就会逐渐受到等待约束。

可以把它记成一句话：enqueue() 把转换结果交给 readable 队列，writer.ready 则让生产者等待 writable 侧恢复可接受状态。TransformStream 的背压是这两端队列和管道协议共同形成的反馈，不是一个需要单独打开的布尔开关。

官方资料：https://developer.mozilla.org/en-US/docs/Web/API/TransformStream

先分清 writable、transform 和 readable

TransformStream 可以看成一对相连的流：writable 是输入入口，readable 是输出出口。写入端收到一个 chunk 后，转换器负责决定输出什么；输出并不是直接返回给调用者，而是交给 readable 侧的内部队列，随后由读取方消费。

因此，标题里的“背压”至少涉及三个位置：

• 生产者通过 WritableStreamDefaultWriter 写入，写入动作受 writable 侧队列约束。
• 转换器在 transform() 中调用 controller.enqueue()，把结果放入 readable 侧。
• 读取方消费得越慢，readable 侧越接近高水位，管道越需要让上游减慢。

enqueue 体现的是输出队列压力

controller.enqueue(value) 的作用是把一个 chunk 排入关联的 readable 侧。它没有返回“还剩多少空间”的数值，所以不能把 enqueue() 当成同步的流量控制接口；它表达的是“这次转换产生了一个输出”。真正的节奏控制要结合 writable 的写入承诺、readable 的消费速度以及队列策略来理解。

例如下面的转换器把输入文本转成大写。代码中的注释只保留会影响理解的关键点：

const upperCaseStream = new TransformStream({
  transform(chunk, controller) {
    // 每个输入块只生成一个输出块，避免一次转换无界地产生数据。
    const text = String(chunk).toUpperCase();
    controller.enqueue(text);
  },
  flush() {
    // 没有尾部缓存时不补数据，让 TransformStream 按默认流程关闭 readable。
  },
});

const writer = upperCaseStream.writable.getWriter();
const reader = upperCaseStream.readable.getReader();

async function writeWithBackpressure(chunks) {
  for (const chunk of chunks) {
    // 等待 writable 侧恢复可接受状态，再提交下一块输入。
    await writer.ready;
    await writer.write(chunk);
  }
  // 关闭写入端，让 readable 侧在剩余数据消费完后结束。
  await writer.close();
}

async function readAll() {
  let result = "";
  while (true) {
    // read() 的 done 表示转换器已经关闭且队列也没有剩余数据。
    const { value, done } = await reader.read();
    if (done) return result;
    result += value;
  }
}

await Promise.all([
  writeWithBackpressure(["web ", "transform", "stream"]),
  readAll(),
]);

这个例子中的背压关键不在“每次都显式检查某个布尔值”，而在于写入方等待 writer.ready，再等待本次 writer.write() 完成。若下游变慢，写入 Promise 就不会一直以生产者的速度无条件向前推进。

writer.ready 为什么能让生产者放慢

writer.ready 是一个 Promise。当 writable 侧的期望写入量不再处于背压状态时，它才会恢复完成。把它放在循环中，生产者就有了一个自然的“闸门”：队列尚可接受时继续写，队列需要释放空间时等待。

要注意，writer.ready 不是“所有数据已经被最终消费者处理完”的承诺。它更接近“写入端可以继续推进”的信号；而 writer.write() 的完成也表示该写入已经被 writable 管道接受到相应阶段。真正的业务确认仍然取决于下游 sink 和完整管道的错误传播。

desiredSize 应该怎样理解

在转换器内部可以读取 controller.desiredSize，它表示 readable 侧还希望补充多少队列容量。数值为 0 时，可以理解为到达了期望水位；出现负数则表示队列相对期望容量已经超出。它适合帮助你观察或决定是否继续产生额外输出，但不能替代管道本身的 Promise 语义。

例如，一个输入块可能拆成多个输出块，此时更要限制单次转换的产出：

const splitLines = new TransformStream({
  transform(chunk, controller) {
    // 只按换行拆分当前块；真实协议若允许跨块行，需要额外保存尾部缓存。
    for (const line of String(chunk).split("\n")) {
      // 队列已达到期望水位时，不再从一个输入块继续制造更多输出。
      if (controller.desiredSize !== null && controller.desiredSize 

这里的判断只说明“当前转换是否还愿意继续产出”，并不意味着可以随意丢掉剩余业务数据。生产环境中如果不能丢数据，应保存未完成部分，等待下一次输入或把拆分逻辑设计成可暂停的异步转换。

几个容易混淆的边界

如果 transform() 抛出异常，或显式调用 controller.error()，读写两侧会进入错误传播路径。不要在异常后继续复用同一个 writer；应让调用方捕获写入、读取或管道 Promise 的拒绝，并根据业务决定重建流还是终止任务。

把背压写成一句可执行的判断

在大多数 TransformStream 场景中，可以采用下面的简化规则：

1. 转换器一次只为一个输入块生成有限输出。
2. 生产循环在 writer.ready 之后再调用 writer.write()。
3. 需要拆分或扩增数据时观察 controller.desiredSize，但不要用它掩盖丢数据。
4. 关闭写入端后继续消费 readable，直到 read() 返回 done: true。
5. 同时处理写入、读取和管道的异常 Promise，避免只等待其中一端。

所以，Web TransformStream 的背压可以落到一个很具体的关系：下游的 readable 消费速度改变队列压力，队列压力影响 writable 的可推进程度，生产者再通过 writer.ready 把这种压力落实成等待。理解这条关系后，enqueue()、desiredSize 和 writer.ready 就不会再被误认为是三个互不相关的 API。

相关问题

TransformStream 的 writable 和 readable 是同一个流吗？

不是。它们是同一个 TransformStream 暴露出的输入端和输出端，内部由转换算法与队列连接。

只调用 writer.write() 不等待 writer.ready 可以吗？

少量数据通常也能工作，但持续生产时会削弱背压带来的节奏控制，容易让待处理写入快速累积。

desiredSize 小于 0 就必须丢弃数据吗？

不必须。它表示队列压力超过期望值；不能丢数据时，应暂停或缓存未完成的业务片段，而不是静默跳过。