登录
推荐 文章 Go 技术 课程 下载 专题 AI
首页 >  文章 >  前端

Web Streams TransformStream 如何处理背压:ReadableStream 到 WritableStream 的队列边界

来源:17golang原创

时间:2026-08-28 06:25:10 428浏览 收藏

前端把大文件切成小块上传、导出日志或做压缩时,真正容易失控的不是 transform() 里那几行代码,而是生产速度和消费速度不一致。TransformStream 正好把这段链路拆成可观察的边界:数据从 ReadableStream 进入,经过转换队列,再交给 WritableStream 消费。

把写入端的容量和消费速度交给流的背压机制,生产者才不会因为一时写得快而把内存队列推高。

要点速览

  • TransformStream 的 readable 和 writable 是两条相连但有边界的流。
  • 背压要从 writable.getWriter().desiredSize 观察,而不是靠固定延时猜测。
  • 转换器的 flush() 负责收尾,abort() 负责处理下游中断。
  • 测试时要故意让消费端变慢,才能看见队列边界是否真的生效。

先把三段流的职责分开

ReadableStream 负责提供输入,TransformStream 负责把输入块变成输出块,WritableStream 负责最终消费。浏览器提供的 pipeThrough() 会把前一段的 readable 接到后一段的 writable,调用者不需要手动搬运每个 chunk。

这里有一个容易忽略的事实:转换器两边都有自己的策略。输入侧决定什么时候继续拉取,输出侧的队列策略决定下游变慢时能积压多少。这个从消费端向生产端传播的压力,在 Web Streams 语境里通常称为 backpressure;代码里只写一个转换函数,并不等于已经设计好了背压。

ReadableStream 进入 TransformStream,再由 WritableStream 消费的背压数据路径
ReadableStream、TransformStream 与 WritableStream 的数据路径。

用 TransformStream 把队列边界写出来

下面的例子把文本块转成大写,并把消费端故意放慢。writableStrategy.highWaterMark 不是“最多处理多少条”的业务限制,它更像一个让流判断是否该继续供给的队列水位。

const upperCase = new TransformStream(
  {
    transform(chunk, controller) {
      controller.enqueue(chunk.toUpperCase());
    },
    flush(controller) {
      controller.enqueue("[done]");
    }
  },
  { highWaterMark: 1 },
  { highWaterMark: 1 }
);

const writer = upperCase.writable.getWriter();
console.log(writer.desiredSize); // 观察输出队列余量

await writer.write("a");
await writer.write("b");
writer.releaseLock();

const reader = upperCase.readable.getReader();
console.log(await reader.read());
console.log(await reader.read());
console.log(await reader.read()); // flush 的 [done]
reader.releaseLock();

示例中的 ReadableStream 不是直接参与写入的对象,写入发生在 writer.write(),输出由 TransformStreamcontroller.enqueue() 放进 readable 一侧。生产代码更常用 pipeThrough(),但先手动拿到 writer 有助于看清 desiredSize 变化。

TransformStream 输出队列达到 highWaterMark 后由 WritableStream 消费的边界示意
当 WritableStream 消费变慢时,TransformStream 输出队列的水位成为背压边界。

为什么固定 sleep 不能替代背压

给循环加 await new Promise(resolve => setTimeout(resolve, 20)) 只能让生产者平均变慢,它不知道下游当前还有多少空间。网络抖动、磁盘写入和压缩耗时变化时,同一个 20 毫秒可能太快,也可能白白浪费时间。

更稳妥的做法是让流连接自己传播压力,并在需要手动写入时观察 writer 的 promise:

const writer = writable.getWriter();
await writer.write(chunk);
if (writer.desiredSize 

writer.ready 解决的是“什么时候可以继续写”,不是“这次写入是否已经落到磁盘”。如果业务需要确认文件已完成,还要把真正的落盘确认放在 WritableStream 的写入实现里。

flush 和 abort 要分别验收

正常结束时,flush() 只执行一次,适合写入尾标记、补齐最后一个缓冲区或释放转换器内部状态。下游出错或主动终止时走 abort(),这时不应该再追加正常结束标记。

测试至少覆盖三种状态:所有 chunk 正常读完且出现 [done];转换函数抛错后 readable 进入错误状态;消费端中断后不再继续向 WritableStream 写入。只测“两个字符串能转大写”,测不出背压设计的价值。

常见问题

highWaterMark 是字节数吗?

不一定。默认策略按 chunk 计数;如果使用按字节计量的策略,水位才按 size 算。不要把它直接当成文件大小限制。

什么时候优先使用 pipeThrough?

当输入、转换和输出可以组成稳定链路时优先使用 pipeThrough(),它会保留流之间的关闭与错误传播关系;需要逐块插入业务判定时再手动持有 reader 或 writer。

为什么读取不到 flush 的内容?

通常是没有等待 writable 端关闭,或者 reader 只读取了业务 chunk。先等待写入完成并关闭 writer,再继续读取直到 done。

收尾检查

检查 Web Streams 时,可以沿着 ReadableStreamTransformStreamWritableStream 三个节点逐段问:输入是否持续供给、转换队列是否越过水位、消费端是否真正完成。把这三个问题写进测试,比给生产循环补一个固定延时更接近真实运行状态。

声明:本文转载于:17golang原创 如有侵犯,请联系study_golang@163.com删除
相关阅读
更多>
最新阅读
更多>
课程推荐
更多>