各位同仁，各位技术爱好者，大家好！
今天，我们将深入探讨一个在现代Web应用中至关重要的主题：ReadableStream 与 WritableStream 的管道流控，特别是如何实现一个端到端的流量拉取模型（Pull Model）。Web Streams API 为我们处理数据流提供了强大而灵活的工具，但要驾驭这些工具，实现精细的流控，需要我们对底层的机制有深刻的理解和巧妙的设计。
1. Web Streams API 基础回顾
在深入拉取模型之前，我们先快速回顾一下Web Streams API的核心概念。它提供了一套标准接口，用于以流式方式处理数据，无论是来自网络、文件系统还是其他计算密集型任务。
1.1 ReadableStream：数据的生产者
ReadableStream 代表一个可读的数据源。数据以块（chunks）的形式从流中流出。它的核心思想是“惰性生产”：只有当消费者请求时，它才会尝试生成或获取数据。
一个 ReadableStream 由一个底层源（UnderlyingSource）驱动，该源定义了如何生成或获取数据块。关键方法包括：
start(controller): 流开始时调用一次，用于初始化。pull(controller): 当流的内部队列需要更多数据时（即controller.desiredSize大于0），会被反复调用。这是流向其底层源请求数据的钩子。cancel(reason): 当流被消费者取消时调用。
消费者通过获取 ReadableStreamDefaultReader 来读取数据，使用 reader.read() 方法来获取数据块。
1.2 WritableStream：数据的消费者
WritableStream 代表一个可写的数据目的地。它接收数据块，并将其处理或存储。
一个 WritableStream 由一个底层汇（UnderlyingSink）驱动，该汇定义了如何处理数据块。关键方法包括：
start(controller): 流开始时调用一次，用于初始化。write(chunk, controller): 当有数据块写入流时，会被调用。close(): 当所有数据都已写入且流被关闭时调用。abort(reason): 当流因错误而异常关闭时调用。
生产者通过获取 WritableStreamDefaultWriter 来写入数据，使用 writer.write(chunk) 方法来写入数据块。writer.ready 属性是一个 Promise，它在流准备好接受更多数据时解析，是实现反压（backpressure）的关键。
1.3 TransformStream：数据转换器
TransformStream 结合了 ReadableStream 和 WritableStream 的特性，它既可以消费数据（作为 WritableStream），又可以生产数据（作为 ReadableStream）。它通常用于在数据流经管道时对其进行转换。虽然不是本文核心，但在更复杂的流处理场景中非常有用。
1.4 管道连接：pipeTo
Web Streams API 提供了一个方便的 pipeTo() 方法，可以将一个 ReadableStream 直接连接到一个 WritableStream。
readableStream.pipeTo(writableStream)
.then(() => console.log('Pipe finished successfully'))
.catch(error => console.error('Pipe failed:', error));pipeTo() 实现了内置的反压机制。当 writableStream 的内部队列满载时，它会向 readableStream 发送信号，使其暂停数据生产，直到 writableStream 再次准备好接受数据。这是一种推模型（Push Model）的实现：数据由源主动推送，并通过反压机制进行流量控制。
2. 流控的挑战：推模型与拉模型
尽管 pipeTo() 提供了自动的反压机制，但在某些高级或特定的场景下，我们可能需要更精细、更主动的流控。这就是推模型和拉模型的核心区别。
2.1 推模型（Push Model）及其局限性
在推模型中，数据源（ReadableStream）会主动地生成或获取数据，并尝试将其推送到消费者（WritableStream）。当消费者处理速度跟不上时，它会通过反压机制（例如，WritableStream 的 writer.ready Promise 悬挂）通知生产者暂停。
优点：
- 实现简单，
pipeTo()提供了开箱即用的解决方案。 - 通常效率高，数据可以连续流动。
局限性：
- 源端主动性： 源端总是尝试生成数据，即使消费者可能不需要那么多或处理不过来。如果反压信号传递有延迟，或者源端无法立即停止生产，可能导致不必要的缓冲。
- 资源浪费： 在消费者处理速度远低于生产者的情况下，生产者可能会在等待反压信号时，在内部缓冲区中累积大量数据，占用不必要的内存或计算资源。
- 控制粒度： 生产者是主导者，消费者只能被动地“刹车”。对于需要消费者精确控制何时、何地获取数据的场景，推模型显得力不从心。
2.2 拉模型（Pull Model）的目标
拉模型的核心思想是：数据的消费者（WritableStream）主动向生产者（ReadableStream）请求数据，而不是生产者主动推送。 只有当消费者准备好并明确请求时，生产者才会生成或提供数据。这使得消费者成为数据流的主导者，从而实现更精细的流量控制。
优点：
- 精准控制： 消费者完全控制数据获取的速度和时机。
- 最小化缓冲： 避免不必要的预取和缓冲，节省资源。
- 资源高效： 特别适用于资源受限的环境（如嵌入式设备、低内存浏览器标签页）或对延迟敏感的应用（如实时音视频、游戏渲染）。
- 需求驱动： 数据生成与消费需求严格同步，避免浪费。
适用场景：
- UI渲染： 只有当UI准备好渲染下一帧数据时才请求数据。
- 有限内存环境： 严格控制内存使用，只在需要时加载数据。
- 特定网络协议： 实现一些需要客户端主动请求数据的协议。
- 数据处理流水线： 当下游处理步骤的瓶颈变化较大时，通过拉模型可以更好地适应。
实现端到端的拉模型，意味着我们要打破 pipeTo() 的默认行为，构建一个自定义的协调器，让 WritableStream 的“准备就绪”状态直接驱动 ReadableStream 的“数据生成”过程。
3. 设计端到端的拉模型
要实现一个端到端的拉模型，我们需要三个核心组件：
- 受控的
ReadableStream(The "Managed Source")： 这个ReadableStream不会自己主动从其底层源拉取数据。相反，它提供一个接口，允许外部协调器显式地将数据块注入其中。它的pull()方法将主要用于报告其内部状态，而不是主动生成数据。 - 拉取式
WritableStream(The "Pulling Sink")： 这个WritableStream在处理完一个数据块并准备好接受下一个数据块时，会通过其writer.readyPromise 明确地发出“我已准备就绪”的信号。 - 协调器 (The "Pull-Driver Orchestrator")： 这是整个系统的核心。它将监听
Pulling Sink的“准备就绪”信号。一旦接收到信号，它就会主动从原始数据源获取一个数据块，然后将其注入到Managed Source中，再从Managed Source中读取出来并写入Pulling Sink。这个过程确保了数据的生成严格受控于消费者的需求。
下面我们将逐一实现这些组件。
4. 实现细节与代码示例
4.1 原始数据源模拟
首先，我们模拟一个异步数据源，它在每次生成数据块时会有一个延迟，以模拟I/O操作或计算耗时。这个源是基于 AsyncGenerator 实现的，非常适合按需生成数据。
/**
* 模拟一个异步数据源，以块为单位生成数据。
* @param numChunks 要生成的总块数。
* @param chunkSize 每个数据块的大小（字节）。
* @param delayMs 每生成一个块的延迟时间（毫秒）。
*/
async function* simulatedDataSource(
numChunks: number,
chunkSize: number,
delayMs: number
): AsyncGenerator<Uint8Array> {
for (let i = 0; i < numChunks; i++) {
const chunk = new Uint8Array(chunkSize).fill(i % 256); // 填充一些数据
await new Promise(resolve => setTimeout(resolve, delayMs)); // 模拟I/O或计算延迟
console.log(`[Source] Generated chunk ${i + 1}/${numChunks} of size ${chunk.length} bytes.`);
yield chunk;
}
console.log("[Source] All chunks generated. Source exhausted.");
}4.2 受控的 ReadableStream (ManagedReadableSource)
这个 ReadableStream 的底层源（UnderlyingSource）不会主动从外部获取数据。它提供一个 supplyChunk 方法，允许我们的协调器将数据块推入其内部队列。它的 pull 方法被调用时，会解析一个内部 Promise，用于通知协调器它已经处理完内部需求（例如，desiredSize 允许更多数据），但并不会主动去生成数据。
/**
* 一个受外部控制的 ReadableStream UnderlyingSource。
* 数据通过 `supplyChunk` 方法注入，而不是通过其 `pull` 方法主动获取。
* 它的 `pull` 方法用于向协调器发出信号，表明其内部队列已准备好接收数据。
*/
class ManagedReadableSource implements UnderlyingSource<Uint8Array> {
private controller!: ReadableStreamDefaultController<Uint8Array>;
private pullResolver: (() => void) | null = null;
private currentPullPromise: Promise<void> | null = null;
private isClosed = false;
constructor() {
// 初始化时创建一个 Promise，确保首次 pull 能够被等待
this.currentPullPromise = new Promise(resolve => this.pullResolver = resolve);
}
start(controller: ReadableStreamDefaultController<Uint8Array>) {
this.controller = controller;
console.log("[ManagedReadableSource] Started.");
}
async pull(): Promise<void> {
// Stream 内部调用 pull() 时，我们不主动生成数据。
// 相反，我们解析当前的 pullPromise，告诉外部协调器，流的内部机制已准备好处理更多数据。
// 然后创建一个新的 Promise，等待下一次 pull。
if (this.pullResolver) {
this.pullResolver(); // 解决之前的 pull Promise
this.pullResolver = null;
}
// 创建一个新的 Promise，供下一次 pull 调用等待
this.currentPullPromise = new Promise(resolve => {
this.pullResolver = resolve;
});
console.log(`[ManagedReadableSource] Pull method called. Desired size: ${this.controller.desiredSize}`);
return this.currentPullPromise;
}
cancel(reason: any) {
this.isClosed = true;
if (this.pullResolver) {
this.pullResolver(); // 如果有挂起的 pull Promise，立即解决
this.pullResolver = null;
}
console.log(`[ManagedReadableSource] Cancelled: ${reason}`);
}
/**
* 由外部协调器调用，将数据块推入流中。
* @param chunk 要推入的数据块。
*/
supplyChunk(chunk: Uint8Array) {
if (this.isClosed) {
console.warn("[ManagedReadableSource] Attempted to supply chunk to a closed stream.");
return;
}
this.controller.enqueue(chunk);
console.log(`[ManagedReadableSource] Enqueued chunk of size ${chunk.length} bytes. Desired size: ${this.controller.desiredSize}`);
}
/**
* 由外部协调器调用，通知流数据已全部提供。
*/
close() {
if (!this.isClosed) {
this.controller.close();
this.isClosed = true;
if (this.pullResolver) {
this.pullResolver(); // 确保任何挂起的 pull Promise 都能解决
this.pullResolver = null;
}
console.log("[ManagedReadableSource] Closed.");
}
}
/**
* 获取流的期望大小，表示它还需要多少数据。
*/
get desiredSize(): number {
return this.controller.desiredSize;
}
}4.3 拉取式 WritableStream (PullingSink)
这个 WritableStream 的底层汇（UnderlyingSink）在处理完一个数据块后，会通过解析 ready Promise 来明确地通知外部它已准备好接受更多数据。这是实现拉模型中“消费者请求”的关键。
/**
* 一个拉取式的 WritableStream UnderlyingSink。
* 它在处理完一个数据块后，会显式地通过 `ready` Promise 信号告知外部已准备好接受更多数据。
*/
class PullingSink implements UnderlyingSink<Uint8Array> {
private processChunk: (chunk: Uint8Array) => Promise<void>;
private isClosed = false;
private resolveReady: (() => void) | null = null;
private readyPromise: Promise<void> | null = null;
constructor(processChunk: (chunk: Uint8Array) => Promise<void>) {
this.processChunk = processChunk;
// 初始状态为准备就绪
this.readyPromise = Promise.resolve();
}
start(controller: WritableStreamDefaultController) {
console.log("[PullingSink] Started.");
}
async write(chunk: Uint8Array, controller: WritableStreamDefaultController): Promise<void> {
if (this.isClosed) {
throw new Error("[PullingSink] Stream is closed, cannot write.");
}
console.log(`[PullingSink] Received chunk of size ${chunk.length} bytes for processing.`);
// 在处理当前块之前，创建一个新的 readyPromise，以便后续的 `writer.ready` 调用等待。
// 这模拟了 sink 内部的容量管理。当它忙于处理时，它不是 ready 的。
const currentProcessingPromise = new Promise<void>(resolve => {
this.resolveReady = resolve;
});
this.readyPromise = currentProcessingPromise;
// 模拟数据处理时间
await this.processChunk(chunk);
console.log("[PullingSink] Finished processing chunk. Signalling readiness.");
// 处理完成后，解决 readyPromise，表示可以接受下一个块了。
if (this.resolveReady) {
this.resolveReady();
this.resolveReady = null;
}
}
close(): Promise<void> {
this.isClosed = true;
if (this.resolveReady) {
this.resolveReady(); // 确保任何挂起的 ready Promise 都能解决
this.resolveReady = null;
}
console.log("[PullingSink] Closed.");
return Promise.resolve();
}
abort(reason: any): Promise<void> {
this.isClosed = true;
if (this.resolveReady) {
this.resolveReady(); // 确保任何挂起的 ready Promise 都能解决
this.resolveReady = null;
}
console.error(`[PullingSink] Aborted: ${reason}`);
return Promise.reject(reason);
}
/**
* 获取 WritableStream 的 `ready` Promise，外部协调器将监听此 Promise。
*/
get ready(): Promise<void> {
return this.readyPromise!;
}
}4.4 协调器 (PullModelOrchestrator)
协调器是连接 ManagedReadableSource 和 PullingSink 的核心。它等待 PullingSink 准备就绪，然后向原始数据源请求一个数据块，将其传递给 ManagedReadableSource，再从 ManagedReadableSource 读取并写入 PullingSink。这个循环确保了数据流严格按照消费者的需求进行。
/**
* 拉模型协调器：连接原始数据源、受控的 ReadableStream 和拉取式 WritableStream，
* 实现端到端的流量拉取控制。
*/
class PullModelOrchestrator {
private sourceGenerator: AsyncGenerator<Uint8Array>;
private managedSource: ManagedReadableSource;
private pullingSink: PullingSink;
private reader: ReadableStreamDefaultReader<Uint8Array>;
private writer: WritableStreamDefaultWriter<Uint8Array>;
private isRunning = false;
private stopReason: any = null;
constructor(
sourceGenerator: AsyncGenerator<Uint8Array>,
processSinkChunk: (chunk: Uint8Array) => Promise<void>
) {
this.sourceGenerator = sourceGenerator;
// 创建我们的自定义 ReadableStream，它由协调器控制
this.managedSource = new ManagedReadableSource();
this.reader = new ReadableStream(this.managedSource).getReader();
// 创建我们的自定义 WritableStream，它会信号告知准备就绪
this.pullingSink = new PullingSink(processSinkChunk);
this.writer = new WritableStream(this.pullingSink).getWriter();
}
/**
* 启动拉模型的数据传输循环。
*/
async start(): Promise<void> {
if (this.isRunning) {
console.warn("[Orchestrator] Already running.");
return;
}
this.isRunning = true;
console.log("[Orchestrator] Started: Initiating pull model data transfer.");
try {
while (this.isRunning) {
// 1. 等待 WritableStream 准备好接受数据。
// 这是拉模型的核心：sink 决定了节奏。
console.log("[Orchestrator] Waiting for sink readiness...");
await this.writer.ready;
console.log("[Orchestrator] Sink is ready. Requesting data from source.");
// 2. 现在 sink 已准备好，我们从原始数据源拉取一个数据块。
// 这一步有效地“按需生成”数据。
const { value: sourceChunk, done: sourceDone } = await this.sourceGenerator.next();
if (sourceDone) {
console.log("[Orchestrator] Source exhausted. Signalling end of data.");
this.managedSource.close(); // 通知受控 ReadableStream 数据已结束
await this.writer.close(); // 关闭 WritableStream
this.isRunning = false;
break;
}
// 3. 将从原始源获取的块提供给我们的 ManagedReadableSource。
// 这将使 ManagedReadableSource 的内部队列填充。
this.managedSource.supplyChunk(sourceChunk!);
console.log(`[Orchestrator] Supplied chunk of size ${sourceChunk!.length} bytes to ManagedReadableSource.`);
// 4. 从 ManagedReadableSource 读取块，并写入 WritableStream。
// `reader.read()` 会消费我们刚刚提供的块。
// 因为我们已经等待了 `writer.ready`，所以我们知道 writer 可以立即接受它。
const { value: chunkToPipe, done: readerDone } = await this.reader.read();
if (readerDone) {
console.log("[Orchestrator] Managed ReadableStream closed unexpectedly during read.");
await this.writer.close();
this.isRunning = false;
break;
}
console.log(`[Orchestrator] Reading chunk of size ${chunkToPipe!.length} from internal buffer and writing to sink.`);
await this.writer.write(chunkToPipe!);
console.log("[Orchestrator] Chunk successfully written to sink. Loop continues.");
}
} catch (error) {
console.error("[Orchestrator] Encountered an error:", error);
this.stopReason = error;
await this.stop(error); // 停止整个流程
} finally {
if (this.reader) this.reader.releaseLock();
if (this.writer) this.writer.releaseLock();
console.log("[Orchestrator] Stopped.");
}
}
/**
* 停止数据传输流程。
* @param reason 停止的原因。
*/
async stop(reason?: any): Promise<void> {
if (!this.isRunning) return;
this.isRunning = false;
console.log("[Orchestrator] Stopping...");
try {
this.managedSource.cancel(reason); // 取消 ReadableStream
await this.writer.abort(reason); // 中止 WritableStream
} catch (e) {
console.error("[Orchestrator] Error during stop cleanup:", e);
}
}
}4.5 完整示例与运行
现在，我们将所有组件整合起来，创建一个演示程序。我们将模拟一个源生成数据速度快于消费者处理速度的场景，以清楚地展示拉模型如何有效地控制流量。
// -------------------- 完整示例代码 --------------------
// 模拟数据源 (与前面相同)
async function* simulatedDataSource(
numChunks: number,
chunkSize: number,
delayMs: number
): AsyncGenerator<Uint8Array> {
for (let i = 0; i < numChunks; i++) {
const chunk = new Uint8Array(chunkSize).fill(i % 256);
await new Promise(resolve => setTimeout(resolve, delayMs));
console.log(`[Source] Generated chunk ${i + 1}/${numChunks} of size ${chunk.length} bytes.`);
yield chunk;
}
console.log("[Source] All chunks generated. Source exhausted.");
}
// 受控的 ReadableStream (与前面相同)
class ManagedReadableSource implements UnderlyingSource<Uint8Array> {
private controller!: ReadableStreamDefaultController<Uint8Array>;
private pullResolver: (() => void) | null = null;
private currentPullPromise: Promise<void> | null = null;
private isClosed = false;
constructor() {
this.currentPullPromise = new Promise(resolve => this.pullResolver = resolve);
}
start(controller: ReadableStreamDefaultController<Uint8Array>) {
this.controller = controller;
console.log("[ManagedReadableSource] Started.");
}
async pull(): Promise<void> {
if (this.pullResolver) {
this.pullResolver();
this.pullResolver = null;
}
this.currentPullPromise = new Promise(resolve => {
this.pullResolver = resolve;
});
console.log(`[ManagedReadableSource] Pull method called. Desired size: ${this.controller.desiredSize}`);
return this.currentPullPromise;
}
cancel(reason: any) {
this.isClosed = true;
if (this.pullResolver) {
this.pullResolver();
this.pullResolver = null;
}
console.log(`[ManagedReadableSource] Cancelled: ${reason}`);
}
supplyChunk(chunk: Uint8Array) {
if (this.isClosed) {
console.warn("[ManagedReadableSource] Attempted to supply chunk to a closed stream.");
return;
}
this.controller.enqueue(chunk);
console.log(`[ManagedReadableSource] Enqueued chunk of size ${chunk.length} bytes. Desired size: ${this.controller.desiredSize}`);
}
close() {
if (!this.isClosed) {
this.controller.close();
this.isClosed = true;
if (this.pullResolver) {
this.pullResolver();
this.pullResolver = null;
}
console.log("[ManagedReadableSource] Closed.");
}
}
get desiredSize(): number {
return this.controller.desiredSize;
}
}
// 拉取式 WritableStream (与前面相同)
class PullingSink implements UnderlyingSink<Uint8Array> {
private processChunk: (chunk: Uint8Array) => Promise<void>;
private isClosed = false;
private resolveReady: (() => void) | null = null;
private readyPromise: Promise<void> | null = null;
constructor(processChunk: (chunk: Uint8Array) => Promise<void>) {
this.processChunk = processChunk;
this.readyPromise = Promise.resolve();
}
start(controller: WritableStreamDefaultController) {
console.log("[PullingSink] Started.");
}
async write(chunk: Uint8Array, controller: WritableStreamDefaultController): Promise<void> {
if (this.isClosed) {
throw new Error("[PullingSink] Stream is closed, cannot write.");
}
console.log(`[PullingSink] Received chunk of size ${chunk.length} bytes for processing.`);
const currentProcessingPromise = new Promise<void>(resolve => {
this.resolveReady = resolve;
});
this.readyPromise = currentProcessingPromise;
await this.processChunk(chunk);
console.log("[PullingSink] Finished processing chunk. Signalling readiness.");
if (this.resolveReady) {
this.resolveReady();
this.resolveReady = null;
}
}
close(): Promise<void> {
this.isClosed = true;
if (this.resolveReady) {
this.resolveReady();
this.resolveReady = null;
}
console.log("[PullingSink] Closed.");
return Promise.resolve();
}
abort(reason: any): Promise<void> {
this.isClosed = true;
if (this.resolveReady) {
this.resolveReady();
this.resolveReady = null;
}
console.error(`[PullingSink] Aborted: ${reason}`);
return Promise.reject(reason);
}
get ready(): Promise<void> {
return this.readyPromise!;
}
}
// 协调器 (与前面相同)
class PullModelOrchestrator {
private sourceGenerator: AsyncGenerator<Uint8Array>;
private managedSource: ManagedReadableSource;
private pullingSink: PullingSink;
private reader: ReadableStreamDefaultReader<Uint8Array>;
private writer: WritableStreamDefaultWriter<Uint8Array>;
private isRunning = false;
private stopReason: any = null;
constructor(
sourceGenerator: AsyncGenerator<Uint8Array>,
processSinkChunk: (chunk: Uint8Array) => Promise<void>
) {
this.sourceGenerator = sourceGenerator;
this.managedSource = new ManagedReadableSource();
this.reader = new ReadableStream(this.managedSource).getReader();
this.pullingSink = new PullingSink(processSinkChunk);
this.writer = new WritableStream(this.pullingSink).getWriter();
}
async start(): Promise<void> {
if (this.isRunning) {
console.warn("[Orchestrator] Already running.");
return;
}
this.isRunning = true;
console.log("[Orchestrator] Started: Initiating pull model data transfer.");
try {
while (this.isRunning) {
console.log("[Orchestrator] Waiting for sink readiness...");
await this.writer.ready;
console.log("[Orchestrator] Sink is ready. Requesting data from source.");
const { value: sourceChunk, done: sourceDone } = await this.sourceGenerator.next();
if (sourceDone) {
console.log("[Orchestrator] Source exhausted. Signalling end of data.");
this.managedSource.close();
await this.writer.close();
this.isRunning = false;
break;
}
this.managedSource.supplyChunk(sourceChunk!);
console.log(`[Orchestrator] Supplied chunk of size ${sourceChunk!.length} bytes to ManagedReadableSource.`);
const { value: chunkToPipe, done: readerDone } = await this.reader.read();
if (readerDone) {
console.log("[Orchestrator] Managed ReadableStream closed unexpectedly during read.");
await this.writer.close();
this.isRunning = false;
break;
}
console.log(`[Orchestrator] Reading chunk of size ${chunkToPipe!.length} from internal buffer and writing to sink.`);
await this.writer.write(chunkToPipe!);
console.log("[Orchestrator] Chunk successfully written to sink. Loop continues.");
}
} catch (error) {
console.error("[Orchestrator] Encountered an error:", error);
this.stopReason = error;
await this.stop(error);
} finally {
if (this.reader) this.reader.releaseLock();
if (this.writer) this.writer.releaseLock();
console.log("[Orchestrator] Stopped.");
}
}
async stop(reason?: any): Promise<void> {
if (!this.isRunning) return;
this.isRunning = false;
console.log("[Orchestrator] Stopping...");
try {
this.managedSource.cancel(reason);
await this.writer.abort(reason);
} catch (e) {
console.error("[Orchestrator] Error during stop cleanup:", e);
}
}
}
// -------------------- 运行演示 --------------------
async function runDemo() {
const NUM_CHUNKS = 5;
const CHUNK_SIZE = 1024; // 1KB
const SOURCE_GENERATION_DELAY_MS = 100; // 源每100ms生成一个块
const SINK_PROCESSING_DELAY_MS = 500; // 消费端每500ms处理一个块
console.log("--- Starting Pull Model Demonstration ---");
console.log(`Total chunks to process: ${NUM_CHUNKS}`);
console.log(`Source generation delay: ${SOURCE_GENERATION_DELAY_MS}ms`);
console.log(`Sink processing delay: ${SINK_PROCESSING_DELAY_MS}ms`);
console.log("---------------------------------------");
// 实例化原始数据源
const myDataSource = simulatedDataSource(NUM_CHUNKS, CHUNK_SIZE, SOURCE_GENERATION_DELAY_MS);
// 实例化消费端的处理逻辑
const mySinkProcessor = async (chunk: Uint8Array) => {
console.log(`[SinkProcessor] Processing chunk of size ${chunk.length} bytes.`);
await new Promise(resolve => setTimeout(resolve, SINK_PROCESSING_DELAY_MS)); // 模拟耗时处理
console.log("[SinkProcessor] Finished processing chunk.");
};
// 实例化并启动协调器
const orchestrator = new PullModelOrchestrator(myDataSource, mySinkProcessor);
await orchestrator.start();
console.log("--- Pull Model Demonstration Finished ---");
}
runDemo().catch(console.error);运行结果分析：
当你运行上述代码时，你会观察到以下日志输出模式：
[Orchestrator] Waiting for sink readiness...[Orchestrator] Sink is ready. Requesting data from source.[Source] Generated chunk 1/5 of size 1024 bytes.(源生成数据)[ManagedReadableSource] Enqueued chunk...[Orchestrator] Reading chunk...and writing to sink.[PullingSink] Received chunk...for processing.[SinkProcessor] Processing chunk...- (500ms 延迟)
[SinkProcessor] Finished processing chunk.[PullingSink] Finished processing chunk. Signalling readiness.[Orchestrator] Chunk successfully written to sink. Loop continues.[Orchestrator] Waiting for sink readiness...(循环回到第1步)
这个模式清晰地展示了拉模型：只有在 PullingSink 完成了当前块的处理并发出“准备就绪”信号后（第10步），Orchestrator 才会指示 simulatedDataSource 生成下一个数据块（第3步）。即使 simulatedDataSource 的生成速度非常快（100ms），它也不会提前生成所有数据，而是被 PullingSink 的处理速度（500ms）所限制。这完美地实现了端到端的流量拉取控制。
5. 拉模型与推模型的对比
| 特性 | 推模型 (pipeTo) |
拉模型 (自定义协调器) |
|---|---|---|
| 数据流向 | 源 -> 消费者 (源主动推送) | 消费者 -> 源 (消费者主动请求) |
| 控制主体 | 源端主导，消费者通过反压机制“刹车”。 | 消费端主导，源端按需提供数据。 |
| 反压机制 | 隐式，通过 WritableStream 的 desiredSize 和 writer.ready Promise 暂停源的 pull()。 |
显式，协调器等待 writer.ready 后才触发源数据生成。 |
| 缓冲策略 | 源端可能在反压生效前，或在反压信号传递过程中，预先缓冲数据。 | 最小化缓冲。数据仅在消费者准备好时才被请求和传递。 |
| 资源利用 | 源端可能因预取导致不必要的资源（CPU、内存）占用。 | 更高效，只在需要时分配资源，避免浪费。 |
| 实现复杂度 | 简单，使用 pipeTo() 即可。 |
复杂，需要自定义 UnderlyingSource 和 UnderlyingSink，并实现一个协调器。 |
| 适用场景 | 大多数通用流处理，如文件下载、网络数据传输。 | 对延迟敏感、资源受限、需要精细控制数据生成时机的场景，如UI渲染、实时数据处理。 |
6. 高级考量与边缘情况
6.1 错误处理与取消
在我们的协调器中，try...catch 块用于捕获在数据传输过程中可能发生的错误。当捕获到错误时，stop() 方法会被调用，它会尝试取消 ReadableStream 并中止 WritableStream。这确保了资源能够被及时释放，并向其他部分传播错误状态。
ManagedReadableSource.cancel(): 当协调器决定停止或发生错误时，会调用此方法，允许流进行清理。PullingSink.abort(reason): 当协调器决定停止或发生错误时，会调用此方法，允许汇进行清理。
6.2 零拷贝传输
虽然我们的拉模型主要关注流控，但在实际应用中，性能优化也是关键。Web Streams API 鼓励零拷贝（zero-copy）传输，即数据块在流经管道时尽量避免不必要的数据复制。TransformStream 在某些情况下可以实现零拷贝转换。在我们的拉模型中，数据从 sourceGenerator 到 ManagedReadableSource，再到 PullingSink，如果中间没有进行数据修改，理论上可以避免深拷贝。
6.3 灵活性与可扩展性
我们实现的拉模型提供了一个强大的框架。你可以根据具体需求进行扩展：
- 多阶段处理： 在
ManagedReadableSource和PullingSink之间插入TransformStream来实现数据转换。 - 动态调整速度：
SINK_PROCESSING_DELAY_MS可以根据系统负载、网络状况等因素动态调整，从而实现自适应的流量控制。 - 多种数据源/汇：
simulatedDataSource和mySinkProcessor可以替换为真实的网络请求、文件操作、Web Workers 计算等。
6.4 内部缓冲与反压
即使在拉模型中，ManagedReadableSource 内部仍然有一个 controller.desiredSize 机制。如果协调器在 PullingSink 准备好后，一次性 supplyChunk 太多数据（比如，它从 sourceGenerator 连续拉取了多个块），那么 ManagedReadableSource 可能会在内部缓冲这些块。然而，我们目前的实现是每次 writer.ready 解决，才从 sourceGenerator 拉取一个块，并立即通过 reader.read() 消费并写入 writer.write()。这确保了 ManagedReadableSource 的内部缓冲区通常只会包含一个或很少的块，因为数据被迅速地传递了。因此，真正的反压是直接作用在 sourceGenerator.next() 上的，它只有在 PullingSink 准备好时才会被调用。
7. 结语
Web Streams API 为现代Web应用中的数据处理提供了无与伦比的灵活性和控制力。虽然 pipeTo() 提供的推模型在大多数场景下足够高效，但当我们需要对数据流进行极致的精细控制，特别是在资源受限或需要消费者驱动的场景中，实现一个端到端的拉模型就显得尤为重要。通过自定义 ReadableStream 和 WritableStream 的底层行为，并引入一个智能协调器，我们能够构建出高度可控、高效且响应迅速的数据处理管道。这不仅加深了我们对流机制的理解，也为构建更健壮、更优化的Web应用开辟了新的道路。