workerd Streams 实现指南：ReadableStream / WritableStream / TransformStream 的双实现架构与源码剖析

workerd Streams 实现指南：ReadableStream / WritableStream / TransformStream 的双实现架构与源码剖析

【免费下载链接】workerdThe JavaScript / Wasm runtime that powers Cloudflare Workers项目地址: https://gitcode.com/GitHub_Trending/wo/workerd

导读：本文基于 docs/streams.md 展开，系统讲解 Cloudflare Workers 运行时 workerd 中 Streams API 的完整实现——同一套ReadableStream/WritableStream/TransformStreamJavaScript 接口背后，同时运行着Internal（kj 后端）与Standard（WHATWG 规范）两套实现。读完本文，你将掌握两种流的内核差异、队列与背压机制、tee()与pipeTo()的底层分流/选路逻辑，以及兼容性标志如何决定new TransformStream()的行为，并能在源码中找到每一处关键实现的落点。

一、总览：一套 API，两套实现

workerd 的 Streams 子系统拥有两套互相独立的实现，但对外暴露的是同一组 JavaScript API：ReadableStream、WritableStream、TransformStream。

1.1 Internal Streams（内部流）

• 定位：workerd 最早期的实现，专为运行时自身需求设计——读取请求体request.body、写入响应体等。
• 本质：对 kj 异步 I/O 原语（kj::AsyncInputStream、kj::AsyncOutputStream）的薄封装。
• 规范贴合度：与 WHATWG Streams 规范仅有表面联系，属于非标准实现。
• 数据形态：仅面向字节，只处理TypedArray与ArrayBuffer。

从源码看，其控制器定义在 internal.h：

class ReadableStreamInternalController: public ReadableStreamController, public kj::PtrTarget { public: using Readable = IoOwn<ReadableStreamSource>; // 状态机：Readable / Closed / Errored 三态 explicit ReadableStreamInternalController(StreamStates::Closed closed) ... explicit ReadableStreamInternalController(StreamStates::Errored errored) ... explicit ReadableStreamInternalController(Readable readable) ...

注释明确说明："Every stream implementation that originates fromwithinthe Workers runtime will use these"（凡源自 Workers 运行时内部的流都使用这套控制器），且其行为"not entirely compliant with the streams specification"。

1.2 Standard Streams（标准流）

• 定位：符合 WHATWG Streams 规范的独立实现，完全由 JavaScript Promise 与用户提供的回调函数驱动。
• 数据形态：既可以字节导向，也可以值导向（可处理任意 JavaScript 值，包括undefined、null）。
• 核心代码：主要在 standard.h 与 standard.c++，其中ReadableStreamJsController与WritableStreamJsController合计约 5400 行，是整个子系统的复杂度中心（见 AGENTS.md）。

1.3 为什么两套实现并存？

因为删除内部实现会破坏向后兼容。runtime 通过兼容性标志系统决定行为归属：

• request.body等各类 API 始终返回Internal 流；
• 带用户回调的new ReadableStream(...)始终创建Standard 流；
• new TransformStream()具体走哪套，由兼容性标志transformstream_enable_standard_constructor决定（详见下文第四节）。

两类实现的完整对照（引用 README.md 的分类矩阵）：

注意：ReadableStream/WritableStream的公开 API不提供任何运行时检查手段来区分一个流是字节导向还是值导向，这也直接影响了后续pipeTo桥接层的设计（见第五节）。

二、核心术语

2.1 控制器（Controllers）

每个ReadableStream和WritableStream都有一个底层控制器，由控制器提供真正的实现。Internal 与 Standard 各有专属控制器类：

控制器模式在 AGENTS.md 中被概括为一条清晰的依赖链：

ReadableStream → ReadableStreamController → 具体实现控制器（Internal / Standard）

2.2 字节导向 vs 值导向

• 字节导向（byte-oriented）：只处理TypedArray和ArrayBuffer数据。
• 值导向（value-oriented）：可处理任意 JavaScript 值，包括undefined与null。一个TypedArray或ArrayBuffer可以穿过值导向流，但会被当作不透明的 JavaScript 值——流不会将其解释为字节数据。

规则很简单：Internal 流永远是字节导向；Standard 流两种皆可。

2.3 BYOB 读取器 vs Default 读取器

Reader从ReadableStream中消费数据，共有两种：

• BYOB Reader（Bring Your Own Buffer）：仅适用于字节导向流。调用方提供一个目标TypedArray，由流来填充数据——数据直接写入调用方的缓冲区，减少拷贝。
• Default Reader：字节导向与值导向流都可用。数据以流自身产生的形态交付给调用方。

三、ReadableStream 工作原理

先看最基础的用法：

const readableStream = getReadableSomehow(); const reader = readableStream.getReader(); const chunk = await reader.read(); console.log(chunk.value); // 读到的数据 console.log(chunk.done); // 流完成时为 true

用户代码反复调用read()直到chunk.done === true。幕后发生了什么，完全取决于这是 Internal 还是 Standard 流。

3.1 Standard ReadableStream：双队列 + 四算法

StandardReadableStream维护两个内部队列：可用数据队列（queue of available data）与pending read 队列（queue of pending reads）。控制器依赖四个回调函数（规范中称为 "algorithms"）：

• start：ReadableStream创建时立即调用，用于初始化；
• pull：请求源提供更多数据；
• cancel：流被显式取消时调用；
• size：计算每个 chunk 的大小，用于背压计算。

创建与填充流程：流创建后 start 算法立即执行；一旦完成，流检查highwater mark（高水位线）——即用 size 算法计算出的、数据队列中应持有的最大数据量。若当前队列大小低于 highwater mark，则调用 pull 算法。

pull 算法向流内推入数据（"pull" 却实际在 push，原文也承认这种反讽）：

• 若没有 pending read、或队列中已有数据，新数据进入队列；
• 若存在 pending read，则立即用新数据满足该读请求；超出该次读取所需的多余部分进入队列。

+----------------+ | pull algorithm | <------------------------------------------+ +----------------+ | | | v | +---------------+ | | enqueue(data) | | +---------------+ | | | v | +--------------------+ +-------------------+ | | has a pending read | ----> | has data in queue | | +--------------------+ yes +-------------------+ | | | no| | no| | v | v | +----------------------+ | +-------------------+ yes | | fulfill pending read | | | add data to queue | <--------+ +----------------------+ | +-------------------+ | | | | | | v | | +-----------------------------+ no | +--> | is queue at highwater mark? | ----------- +-----------------------------+ yes | | v (done)

reader.read()的处理：

• 队列有数据 → 立即满足该读请求；若此举使队列降至 highwater mark 以下，则调用 pull 算法；
• 队列为空 → 将读请求加入 pending read 队列；若队列大小低于 highwater mark，则调用 pull 算法。

+---------------+ | reader.read() | +---------------+ | v +-----------------+ +------------------+ +---------------------+ | queue has data? | ----> | add pending read | ----> | call pull algorithm | +-----------------+ no +------------------+ +---------------------+ | yes | v +--------------+ | fulfill read | +--------------+ | v (done)

重要警告（源自源码与文档的反复强调）：

• 源可以在没有任何活跃 reader、甚至在背压已被触发之后，继续向队列推数据——这些数据会无界地堆积在内存中；
• 单次 push 的数据量可能超过一次 read 能消费的量，多出的部分留在队列里；
• 一旦用户代码拿到控制器引用，就可以在任何时候独立于 pull 算法向队列 enqueue 数据。

标准流背压的核心实现是 queue.h 中的ValueQueue与ByteQueue两类队列（queue.h），它们分别服务于值导向与字节导向的标准流，配合 highwater mark 完成背压计算。

3.2 Internal ReadableStream：单 pending read，无队列

Internal 流的运行方式截然不同，关键差异有三：同时只允许一个 pending read、没有内部数据队列、没有 pull 算法。

const readable = new ReadableStream(); // Standard 流 const reader = readable.getReader(); reader.read(); reader.read(); // 没问题 —— 作为 pending read 排队 const readable = request.body; // Internal 流 const reader = readable.getReader(); reader.read(); reader.read(); // 报错！

Internal 流由ReadableStreamSource支撑，其内部包装了一个kj::AsyncInputStream。调用read()直接转化为对源上的tryRead()调用，返回一个kj::Promise承载数据；同一时间只能有一个 read 在飞行中（in flight）。

+---------------+ | reader.read() | +---------------+ | v +-------------------+ +-----------------------------------+ | has pending read? | ---> | tryRead() on ReadableStreamSource | +-------------------+ no +-----------------------------------+ yes | v +-------+ | error | +-------+

这也意味着：数据只有在存在活跃 reader 时才会流过 Internal 流，且每次 read 返回的数据量绝不超过调用中指定的最大字节数。

3.3 Tee：ReadableStream.tee()

tee()将数据流拆分为两个独立的ReadableStream实例（分支）。

WHATWG 规范中的行为：tee()创建两个分支，共享原流（"trunk"）上的一个 reader。当一个分支 pull 时，数据满足该分支的读取，并复制一份推入另一个分支的队列。这意味着一个分支读得比另一个快，会导致慢分支出现无界的内存增长。

workerd 的修改：不再复制数据，分支之间持有数据的refcounted 引用（kj::Rc<Entry>）。向 trunk 的背压信号基于未消费数据最多的分支来计算。

+----------------+ | pull algorithm | +----------------+ | v .......................................................... +---------------+ . +---------------------+ +-------------------+ | enqueue(data) | ---> | push data to branch | ---> | has pending read? | +---------------+ . +---------------------+ +-------------------+ | . no | yes | | . +-------------------+ | +--------------+ | . | add data to queue | <-----+ | fulfill read | | . +-------------------+ +--------------+ | ............................................................ | . +---------------------+ +-------------------+ +--------> | push data to branch | ---> | has pending read? | . +---------------------+ +-------------------+ . no | yes | . +-------------------+ | +--------------+ . | add data to queue | <-----+ | fulfill read | . +-------------------+ +--------------+ ............................................................

这项优化并不能彻底阻止某个分支读得远慢于另一个分支，但只要底层源尊重背压信号，就能避免原本会发生的内存堆积。对应安全模式见 README.md 中的Rc<Entry>模式：

class Entry: public kj::Refcounted { kj::Rc<Entry> clone(jsg::Lock& js); };

四、WritableStream 工作原理

4.1 Standard WritableStream：五算法 + 双背压通道

StandardWritableStream使用五个算法：

• start：准备接收数据；
• write：每个写入的 chunk 都会调用；
• abort：异常终止时调用；
• close：最后一个 chunk 之后调用；
• size：计算 chunk 大小以服务于 highwater mark。

const writable = getWritableSomehow(); const writer = writable.getWriter(); await writer.write('chunk of data'); await writer.write('another chunk of data');

write 算法是异步的：每次write(data)都会调用它，返回的 promise 在写入完成时 resolve。这个 promise 是主要的背压机制。背压通过两种方式呈现：

• writer.desiredSize——在队列填满之前还可写入的数据量；
• writer.ready——背压解除时 resolve 的 promise；每次触发背压时都会替换成一个新的 promise。

重要：highwater mark 只是建议性的（advisory）。write 算法应当始终做好被调用的准备，无论当前 highwater mark 取值如何——即不能因为背压就假定不会被调用。

4.2 Internal WritableStream：直通 Sink

InternalWritableStream由WritableStreamSink支撑，它是kj::AsyncOutputStream的薄封装。每次write()直接透传给 sink 的write()。此后的行为取决于具体 sink 的实现（例如 HTTP 响应体、TLS 连接等不同 sink 各有其语义）。相关代码见 internal.h 中的WritableStreamInternalController。

五、TransformStream

TransformStream连接一个ReadableStream与一个WritableStream——写入 writable 侧的数据可以被读取于 readable 侧，且可能经过变换。

5.1 IdentityTransformStream（Internal，旧实现）

最初的TransformStream是一个恒等变换：数据从 writable 侧到 readable 侧原样通过。它用一个类同时实现ReadableStreamSource与WritableStreamSink两个接口。

该实现不满足规范。transform.h 的注释记录了这段历史：Workers 中最初的 TransformStream 不过是仅处理字节数据的恒等直通（identity passthrough），不执行任何真正的变换，也不符合流规范；这个旧版本被迁移到了IdentityTransformStream类中。兼容性标志transformstream_enable_standard_constructor未启用时，TransformStream就是IdentityTransformStream的别名；启用后TransformStream才实现标准化行为。旧行为仍可通过new IdentityTransformStream()获得。

const { readable, writable } = new IdentityTransformStream(); const enc = new TextEncoder(); const writer = writable.getWriter(); const reader = readable.getReader(); // 注意：write 的 promise 直到 read() 被调用才会 resolve！ await Promise.all([writer.write(enc.encode('hello')), reader.read()]);

IdentityTransformStream是一个简单的状态机，同一时间只允许一个 in-flight 的 read 或 write。write()的 promise 不会在对应的read()发生前 resolve，反之亦然。它只支持字节数据（ArrayBuffer/TypedArray）。

其状态机（来自 README.md）一目了然：

源码层面，该类定义于 identity-transform-stream.h，并通过newIdentityPipe()（identity-transform-stream.h）构造底层管道。值得一提的还有同文件中的FixedLengthStream——它与恒等流几乎相同，但在 readable 侧带有已知字节长度，目前并不强制校验该上限，其作用是说服 kj-http 层输出Content-Length头（只要数据未被 gzip 等处理）。

5.2 Standard TransformStream（新实现）

StandardTransformStream使用三个算法：

• start：初始化变换；
• transform：接收一个 chunk，修改它，并将结果 enqueue；
• flush：完成变换。

const { writable, readable } = new TransformStream({ transform(chunk, controller) { controller.enqueue(`${chunk}!`.toUpperCase()); }, }); const writer = writable.getWriter(); const reader = readable.getReader(); // write 的 promise 不等待 read。 await writer.write('hello'); await reader.read(); // { value: 'HELLO!', done: false }

两侧都是完整的 Standard 流，各自拥有独立的队列与背压。与IdentityTransformStream不同，write 不会被 read 阻塞（除非背压信号表明队列已满）。任意 JavaScript 值都可以流过。

启用标准构造器的测试配置示例如 htmlrewriter-test.wd-test：

compatibilityFlags = ["nodejs_compat", "experimental", "streams_enable_constructors", "transformstream_enable_standard_constructor", ...]

六、Piping：pipeTo()的四种管道回路

Piping 建立从ReadableStream到WritableStream的数据流动。由目标 writable 决定管道如何实现——经由其控制器的tryPipeFrom()方法完成。该入口定义在 common.h：

// The tryPipeFrom attempts to establish a data pipe where source's data // is delivered to this WritableStreamController as efficiently as possible. virtual kj::Maybe<jsg::Promise<void>> tryPipeFrom( jsg::Lock& js, jsg::Ref<ReadableStream> source, PipeToOptions options) = 0;

根据源与目标的流类型组合，共有四种 pipe loop 变体：

+---------------------------+ | readable.pipeTo(writable) | +---------------------------+ | v +------------------------------------------+ | writableController.tryPipeFrom(readable) | +------------------------------------------+ | v +-----------------------+ +-----------------------+ | is internal writable? | ----> | is internal readable? | +-----------------------+ yes +-----------------------+ | no | yes | no | | | v +----+ +---------------+ +-----------------------+ | | kj-to-kj pipe | | is internal readable? | | | loop | +-----------------------+ | +---------------+ no | yes | | | | | v | | +---------------+ | v | JS-to-JS pipe | | +---------------+ | loop | | | JS-to-kj pipe | +---------------+ | | loop | v +---------------+ +---------------+ | kj-to-JS pipe | | loop | +---------------+

四种组合的对照表（来自 README.md）：

kj-to-kj：最优化路径。kj 直接在AsyncInputStream与AsyncOutputStream之间搬运数据，完全在 JavaScript isolate 锁之外进行，整个过程不运行任何 JavaScript 代码，数据也从不进入 JS 堆。

kj-to-JS / JS-to-kj：在 kj 与 JavaScript Promise 之间搭桥。整个数据流必须在 isolate 锁内进行。数据被限制为字节——因为 Standard 流可能是值导向的，而 API 没有提供运行时检查手段，所以桥接层只能强制按字节处理以保证安全。

JS-to-JS：纯 JavaScript promise 链式调用，概念上等价于：

async function pipe(reader, writer) { for await (const chunk of reader) { await writer.write(chunk); } }

两套控制器的tryPipeFrom分别实现在 standard.c++ 与 internal.c++，注释均明确指出源 "can be either a JavaScript-backed ReadableStream or ReadableStreamSource-backed"（可以是 JS 后端或 ReadableStreamSource 后端）。internal-test.c++中的测试 internal-test.c++ 还验证了：pipe 期间对同一 sink 再次发起tryPipeFrom会被拒绝（"currently being piped to"）。

6.1new Response(standardReadable)：标准流到内部 API 的桥接

将 StandardReadableStream传给new Response()这类 API 是件复杂的事，因为这些 API 是为 Internal 流构建的，内部使用 kj 异步 I/O。

为打通二者，StandardReadableStream可以通过ReadableStreamSourceAPI被消费（与 Internal 流所用的是同一套 API）。当适配器上的pumpTo()被调用时：

1. 获取 isolate 锁；
2. 运行"读 JS 流 → 写 kj 输出"的 promise 循环；
3. 直到数据耗尽或发生错误才结束。

pumpTo的定义见 common.h，其中特别说明：ReadableStreamSource版本的pumpTo()没有amount参数，因为 Streams 规范只定义了"泵出全部数据"；而 common.c++ 中对应实现特意注明不使用KJ_CO_MAGIC BEGIN_DEFERRED_PROXYING，因为底层内存可能与 V8 堆绑定（如ArrayBuffer、Blob 数据）。

七、复杂度预算：为什么这块代码很难改

workerd 的 streams 实现需要同时平衡多个来源的复杂度：

• 同一规范的两套实现（Internal 与 Standard）；
• 两种数据导向（字节与值）；
• 两块内存堆（JavaScript 堆与 kj 堆）；
• 两种异步模型（JavaScript Promise 与 kj Promise）；
• 通过特性标志保持严格向后兼容。

最关键的分界线是 isolate 锁：

• Internal 流的数据流动发生在 isolate 锁之外，由 kj 事件循环驱动；
• Standard 流的数据流动发生在 isolate 锁之内，由 JavaScript Promise 驱动；
• 当两个世界交互时（跨类型 pipe、把标准流传给内部 API），桥接代码必须小心翼翼地同时管理两种异步模型。

7.1 跨请求模型

workerd 中多个请求通过绿色线程（green threads）共享一个 isolate。当某个请求因 I/O 让出（yield）时，另一个请求可能开始执行。SetPromiseCrossContextResolveCallback机制负责把 promise 反应（reactions）延迟调度到正确的请求上下文。streams 代码通过以下手段与之协作（详见 README.md）：

• ioContext.addFunctor()——把 continuation 绑定到正确的IoContext；
• IoOwn<>——确保对象只在正确的上下文中被访问；
• Promise 上下文标记——所有 promise 都标记其来源IoContext；若标记与当前上下文不匹配，反应会被延迟执行。

这也解释了为什么ReadableStreamInternalController中的Readable类型被定义为IoOwn<ReadableStreamSource>（internal.h）。

八、安全模式目录：给维护者的纪律清单

README.md 归纳了一组 When/Why/How 安全模式，是修改这段代码时的强制纪律；AGENTS.md 则将其压缩为 7 条不可违背的不变量（INVARIANTS）。这里择要说明：

1. 永远使用deferControllerStateChange()：当调用可能在 read/write 操作中途触发 JS 回调的代码时。JS 回调可能在操作进行中触发 close/error，状态必须等到操作完成后再变更。实现原理是计数器：beginOperation()递增计数器，endOperation()在计数器归零时应用待定状态转换。

controller.state.beginOperation(); // 递增计数器 auto result = readCallback(); // 可能触发 JS 调用 close() controller.state.endOperation(); // 计数器为 0 时应用待定状态

2. 迭代消费者时永远使用snapshot()：若循环体可能触发 JS，迭代过程中消费者可能被增删，导致迭代器失效。先拷贝再遍历：

auto consumers = ready.consumers.snapshot(); for (auto consumer: consumers) { consumer->push(js, entry->clone(js)); // 可能触发 JS 修改 consumers }

3. StateListener 回调之后永远不要再访问this：回调可能经由owner.doClose()销毁this。

4. lambda continuation 中永远重新检查锁状态：绑定在 promise continuation 上的 lambda 可能在锁释放后才执行，捕获与执行之间引用的对象可能已被销毁。规则是绝不捕获可能悬垂的裸引用，改用addRef()或重新获取。

5. 用户可能持有超过底层对象生命周期的句柄时使用 WeakRef（如ByobRequest），使用前检查存活：

impl.controller->runIfAlive( [](ReadableByteStreamController& controller) { controller.maybeByobRequest = kj::none; });

6. 先转状态、后 resolve promise：continuation 必须看到一致的状态。

void doClose(jsg::Lock& js) { state.transitionTo<StreamStates::Closed>(); // 状态此刻变更 maybeResolvePromise(js, locked.getClosedFulfiller()); // 调度微任务 }

7. 异步写引用 JS 堆数据时用V8Ref持有缓冲：kj 异步写进行期间 GC 可能回收缓冲。

struct Write { jsg::V8Ref<v8::ArrayBuffer> ownBytes; // 阻止 GC kj::ArrayPtr<kj::byte> bytes; // 指向 ownBytes 的裸指针 };

此外还有 Pipe-Lock 生命周期管理：当ReadableStream被pipeTo/pipeThrough时，源的锁状态机进入PipeLocked，目标端的 pipe 机制持有指向该状态的kj::Ptr<PipeController>；只有目标端的 pipe 机制才能释放锁，且必须在丢弃所有kj::Ptr之后通过ReadableStreamController::releasePipeLock()完成——Pipe::releaseSource()与WritableLockImpl::PipeLocked::releaseSource()封装了这一顺序。源侧的状态机依赖 state-machine.h（StateMachine<>、PendingStates<>、ActiveState<>）。

九、兼容性标志速查

十、进一步阅读

• src/workerd/api/streams/README.md——速查参考：分类矩阵、状态机、管道回路选择、安全模式目录
• src/workerd/api/streams/AGENTS.md——文件地图、架构摘要、编码不变量与代码评审规则
• 关键实现文件：internal.h / internal.c++、standard.h / standard.c++、queue.h、identity-transform-stream.h、transform.h、common.h、readable.h、writable.h
• 测试参考：internal-test.c++、standard-test.c++、streams-test.js、src/tests/streams 目录下的 340 个测试文件
• 规范原文：WHATWG Streams spec（streams.spec.whatwg.org）

【免费下载链接】workerdThe JavaScript / Wasm runtime that powers Cloudflare Workers项目地址: https://gitcode.com/GitHub_Trending/wo/workerd