On this page

稳定性：2 - 稳定

WHATWG 流标准 的实现。

WHATWG 流标准（或“web 流”）定义了一个用于处理流式数据的 API。它类似于 Node.js 流 API，但出现得较晚，并已成为许多 JavaScript 环境中流式数据的“标准”API。

主要有三种类型的对象：

  • ReadableStream - 代表流式数据的来源。
  • WritableStream - 代表流式数据的目的地。
  • TransformStream - 代表转换流式数据的算法。
import {
  ReadableStream,
} from 'node:stream/web';

import {
  setInterval as every,
} from 'node:timers/promises';

import {
  performance,
} from 'node:perf_hooks';

const SECOND = 1000;

const stream = new ReadableStream({
  async start(controller) {
    for await (const _ of every(SECOND))
      controller.enqueue(performance.now());
  },
});

for await (const value of stream)
  console.log(value);
JavaScript

有关更多详细信息，请参阅相关文档：

Attributes
stream:<ReadableStream>
cloneForBranch2:<boolean>
当为  true 时，排入第二个分支的块会从排入第一个分支的块中克隆。 默认： false 。

对 stream 执行 WHATWG ReadableStreamTee 抽象操作。

这与 readableStream.tee() 的区别仅在于 cloneForBranch2 为 true 时。tee() 方法始终传入 false，而其他 Web 平台规范（例如 Fetch 的主体克隆）则传入 true，以便第二个分支接收克隆的块，并且对一个分支的消费不会修改另一个分支看到的块。

Attributes
underlyingSource:<Object>
start:<Function>
用户定义的函数，在创建  ReadableStream 时立即调用。
controller:<ReadableStreamDefaultController> | <ReadableByteStreamController>
pull:<Function>
用户定义的函数，当  ReadableStream 内部队列未满时重复调用。操作可以是同步或异步的。如果是异步的，则在先前返回的 promise 被兑现之前，不会再次调用该函数。
controller:<ReadableStreamDefaultController> | <ReadableByteStreamController>
cancel:<Function>
用户定义的函数，在  ReadableStream 被取消时调用。
reason:<any>
type:<string>
必须是  'bytes' 或 undefined 。
autoAllocateChunkSize:<number>
仅当  type 等于 'bytes' 时使用。当设置为非零值时，视图缓冲区会自动分配给 ReadableByteStreamController.byobRequest 。当未设置时，必须使用流的内部队列通过默认读取器 ReadableStreamDefaultReader 传输数据。
strategy:<Object>
highWaterMark:<number>
应用背压之前的最大内部队列大小。
size:<Function>
用户定义的函数，用于识别每个数据块的大小。
chunk:<any>

readableStream.locked 属性默认为 false，当存在活动读取器消费流的数据时切换为 true。

Attributes
reason:<any>
Attributes
options:<Object>
mode:<string>
'byob' 或 undefined
import { ReadableStream } from 'node:stream/web';

const stream = new ReadableStream();

const reader = stream.getReader();

console.log(await reader.read());
JavaScript

导致 readableStream.locked 为 true。

Attributes
transform:<Object>
readable:<ReadableStream>
ReadableStream ， transform.writable 将从此 ReadableStream 接收的潜在修改后的数据推送到此流。
writable:<WritableStream>
WritableStream ，此 ReadableStream 的数据将写入此流。
options:<Object>
preventAbort:<boolean>
当为  true 时，此 ReadableStream 中的错误不会导致 transform.writable 被中止。
preventCancel:<boolean>
当为  true 时，目标 transform.writable 中的错误不会导致此 ReadableStream 被取消。
preventClose:<boolean>
当为  true 时，关闭此 ReadableStream 不会导致 transform.writable 被关闭。
signal:<AbortSignal>
允许使用 <AbortController> 取消数据传输。

将此 <ReadableStream> 连接到 transform 参数中提供的 <ReadableStream> 和 <WritableStream> 对，使得来自此 <ReadableStream> 的数据被写入 transform.writable，可能被转换，然后推送到 transform.readable。一旦管道配置完成，将返回 transform.readable。

在管道操作活跃时，导致 readableStream.locked 为 true。

import {
  ReadableStream,
  TransformStream,
} from 'node:stream/web';

const stream = new ReadableStream({
  start(controller) {
    controller.enqueue('a');
  },
});

const transform = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
});

const transformedStream = stream.pipeThrough(transform);

for await (const chunk of transformedStream)
  console.log(chunk);
  // 输出：A
JavaScript
Attributes
destination:<WritableStream>
一个 <WritableStream> ，此  ReadableStream 的数据将写入此流。
options:<Object>
preventAbort:<boolean>
当为  true 时，此 ReadableStream 中的错误不会导致 destination 被中止。
preventCancel:<boolean>
当为  true 时， destination 中的错误不会导致此 ReadableStream 被取消。
preventClose:<boolean>
当为  true 时，关闭此 ReadableStream 不会导致 destination 被关闭。
signal:<AbortSignal>
允许使用 <AbortController> 取消数据传输。

在管道操作活跃时，导致 readableStream.locked 为 true。

返回一对新的 <ReadableStream> 实例，此 ReadableStream 的数据将转发到这两个实例。每个实例将接收相同的数据。

导致 readableStream.locked 为 true。

Attributes
options:<Object>
preventCancel:<boolean>
当为  true  时，防止 <ReadableStream> 在异步迭代器突然终止时被关闭。  默认值 ： false 。

创建并返回一个可用于消费此 <ReadableStream> 数据的异步迭代器。

在异步迭代器活跃时，导致 readableStream.locked 为 true。

import { Buffer } from 'node:buffer';

const stream = new ReadableStream(getSomeSource());

for await (const chunk of stream.values({ preventCancel: true }))
  console.log(Buffer.from(chunk).toString());
JavaScript

<ReadableStream> 对象支持使用 for await 语法的异步迭代器协议。

import { Buffer } from 'node:buffer';

const stream = new ReadableStream(getSomeSource());

for await (const chunk of stream)
  console.log(Buffer.from(chunk).toString());
JavaScript

异步迭代器将消费 <ReadableStream> 直到其终止。

默认情况下，如果异步迭代器提前退出（通过 break、return 或 throw），<ReadableStream> 将被关闭。为了防止 <ReadableStream> 自动关闭，请使用 readableStream.values() 方法获取异步迭代器并将 preventCancel 选项设置为 true。

<ReadableStream> 不得被锁定（即，它不得有现有的活动读取器）。在异步迭代期间，<ReadableStream> 将被锁定。

<ReadableStream> 实例可以使用 <MessagePort> 进行传输。

const stream = new ReadableStream(getReadableSourceSomehow());

const { port1, port2 } = new MessageChannel();

port1.onmessage = ({ data }) => {
  data.getReader().read().then((chunk) => {
    console.log(chunk);
  });
};

port2.postMessage(stream, [stream]);
JavaScript

一个实用方法，从可迭代对象创建一个新的 <ReadableStream>。

import { ReadableStream } from 'node:stream/web';

async function* asyncIterableGenerator() {
  yield 'a';
  yield 'b';
  yield 'c';
}

const stream = ReadableStream.from(asyncIterableGenerator());

for await (const chunk of stream)
  console.log(chunk); // 打印：'a', 'b', 'c'
JavaScript

要将生成的 <ReadableStream> 管道传输到 <WritableStream>，<Iterable> 应产生一系列 <Buffer>、<TypedArray> 或 <DataView> 对象。

import { ReadableStream } from 'node:stream/web';
import { Buffer } from 'node:buffer';

async function* asyncIterableGenerator() {
  yield Buffer.from('a');
  yield Buffer.from('b');
  yield Buffer.from('c');
}

const stream = ReadableStream.from(asyncIterableGenerator());

await stream.pipeTo(createWritableStreamSomehow());
JavaScript
Attributes
stream:<ReadableStream>

创建一个新的 <ReadableStreamDefaultReader>，并将其锁定到给定的 <ReadableStream>。

Attributes
reason:<any>

取消 <ReadableStream>，并返回一个在底层流被取消时完成的 promise。

  • 类型：<Promise> 当关联的 <ReadableStream> 关闭时以 undefined 完成；如果流出错，或者在流完成关闭之前释放了读取器的锁，则拒绝。
  • 返回：一个以如下对象完成的 promise：
    Attributes
    value:<any>
    done:<boolean>

请求底层 <ReadableStream> 的下一个数据块，并返回一个在数据可用后完成的 promise。

释放此读取器持有的底层 <ReadableStream> 上的锁。

BYOB 是"自带缓冲区"（bring your own buffer）的缩写。这是一种模式，允许更有效地读取面向字节的数据，避免额外的复制。

import {
  open,
} from 'node:fs/promises';

import {
  ReadableStream,
} from 'node:stream/web';

import { Buffer } from 'node:buffer';

class Source {
  type = 'bytes';
  autoAllocateChunkSize = 1024;

  async start(controller) {
    this.file = await open(new URL(import.meta.url));
    this.controller = controller;
  }

  async pull(controller) {
    const view = controller.byobRequest?.view;
    const {
      bytesRead,
    } = await this.file.read({
      buffer: view,
      offset: view.byteOffset,
      length: view.byteLength,
    });

    if (bytesRead === 0) {
      await this.file.close();
      this.controller.close();
    }
    controller.byobRequest.respond(bytesRead);
  }
}

const stream = new ReadableStream(new Source());

async function read(stream) {
  const reader = stream.getReader({ mode: 'byob' });

  const chunks = [];
  let result;
  do {
    result = await reader.read(Buffer.alloc(100));
    if (result.value !== undefined)
      chunks.push(Buffer.from(result.value));
  } while (!result.done);

  return Buffer.concat(chunks);
}

const data = await read(stream);
console.log(Buffer.from(data).toString());
JavaScript
Attributes
stream:<ReadableStream>

创建一个新的 ReadableStreamBYOBReader，锁定到给定的 <ReadableStream>。

Attributes
reason:<any>

取消 <ReadableStream> 并返回一个 promise，当底层流被取消时该 promise 被兑现。

  • 类型：<Promise> 当关联的 <ReadableStream> 关闭时兑现为 undefined，如果流出错或读取器的锁在流完成关闭之前被释放，则被拒绝。
Attributes
view:<Buffer> | <TypedArray> | <DataView>
options:<Object>
min:<number>
设置时，仅当至少有  min 个元素可用时，返回的 promise 才会被兑现。 未设置时，当至少有一个元素可用时，promise 被兑现。
value:<TypedArray> | <DataView>
done:<boolean>

从底层 <ReadableStream> 请求下一个数据块，并返回一个 promise，一旦数据可用，该 promise 将被兑现。

不要将池化的 <Buffer> 对象实例传递到此方法。 池化的 Buffer 对象是使用 Buffer.allocUnsafe() 或 Buffer.from() 创建的，或者通常由各种 node:fs 模块回调返回。这些类型的 Buffer 使用共享的底层 <ArrayBuffer> 对象，该对象包含所有池化 Buffer 实例的所有数据。当将 Buffer、<TypedArray> 或 <DataView> 传递到 readableStreamBYOBReader.read() 时，视图的底层 ArrayBuffer 被_分离_，使该 ArrayBuffer 上可能存在的所有现有视图无效。这可能会给您的应用程序带来灾难性后果。

释放此读取器对底层 <ReadableStream> 的锁。

关闭与此控制器关联的 <ReadableStream>。

返回填充 <ReadableStream> 队列所需的剩余数据量。

Attributes
chunk:<any>

将新的数据块附加到 <ReadableStream> 的队列。

Attributes
error:<any>

发出一个错误信号，导致 <ReadableStream> 出错并关闭。

关闭与此控制器关联的 <ReadableStream>。

返回填充 <ReadableStream> 队列所需的剩余数据量。

Attributes
chunk:<Buffer> | <TypedArray> | <DataView>

将新的数据块附加到 <ReadableStream> 的队列中。

Attributes
error:<any>

发出一个错误信号，导致 <ReadableStream> 出错并关闭。

Attributes
bytesWritten:<number>

表示已向 readableStreamBYOBRequest.view 写入了 bytesWritten 个字节。

Attributes
view:<Buffer> | <TypedArray> | <DataView>

表示请求已完成，并且字节已写入新的 Buffer、TypedArray 或 DataView。

import {
  WritableStream,
} from 'node:stream/web';

const stream = new WritableStream({
  write(chunk) {
    console.log(chunk);
  },
});

await stream.getWriter().write('Hello World');
JavaScript
Attributes
underlyingSink:<Object>
start:<Function>
用户定义的函数，在创建  WritableStream 时立即调用。
controller:<WritableStreamDefaultController>
write:<Function>
用户定义的函数，当数据块写入  WritableStream 时调用。
chunk:<any>
controller:<WritableStreamDefaultController>
close:<Function>
用户定义的函数，在  WritableStream 关闭时调用。
abort:<Function>
用户定义的函数，用于突然关闭  WritableStream 。
reason:<any>
type:<any>
type 选项保留供将来使用， 必须 为 undefined。
strategy:<Object>
highWaterMark:<number>
应用背压之前的最大内部队列大小。
size:<Function>
用户定义的函数，用于识别每个数据块的大小。
chunk:<any>
Attributes
reason:<any>

突然终止 WritableStream。所有排队的写入将被取消，其关联的 promise 将被拒绝。

  • 返回：一个兑现值为 undefined 的 promise。

当不再期望额外写入时，关闭 WritableStream。

创建并返回一个新的写入器实例，可用于将数据写入 WritableStream。

writableStream.locked 属性默认为 false，当有活动写入器附加到此 WritableStream 时切换为 true。

<WritableStream> 实例可以使用 <MessagePort> 进行传输。

const stream = new WritableStream(getWritableSinkSomehow());

const { port1, port2 } = new MessageChannel();

port1.onmessage = ({ data }) => {
  data.getWriter().write('hello');
};

port2.postMessage(stream, [stream]);
JavaScript
Attributes
stream:<WritableStream>

创建一个新的 WritableStreamDefaultWriter，锁定到给定的 WritableStream。

Attributes
reason:<any>

突然终止 WritableStream。所有排队的写入将被取消，其关联的 promise 将被拒绝。

  • 返回：一个兑现值为 undefined 的 promise。

当不再期望额外写入时，关闭 WritableStream。

  • 类型：<Promise> 当关联的 <WritableStream> 关闭时兑现为 undefined，如果流出错或写入器的锁在流完成关闭之前被释放，则被拒绝。

填充 <WritableStream> 队列所需的数据量。

  • 类型：<Promise> 当写入器准备好使用时兑现为 undefined。

释放此写入器对底层 <WritableStream> 的锁。

Attributes
chunk:<any>

将新的数据块附加到 <WritableStream> 的队列。

Attributes
error:<any>

由用户代码调用，用于发出信号，表示在处理 WritableStream 数据时发生了错误。调用时，<WritableStream> 将被中止，当前待处理的写入将被取消。

import {
  TransformStream,
} from 'node:stream/web';

const transform = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
});

await Promise.all([
  transform.writable.getWriter().write('A'),
  transform.readable.getReader().read(),
]);
JavaScript
Attributes
transformer:<Object>
start:<Function>
用户定义的函数，在创建  TransformStream 时立即调用。
controller:<TransformStreamDefaultController>
transform:<Function>
用户定义的函数，接收并可能修改写入  transformStream.writable 的数据块，然后将其转发到 transformStream.readable 。
chunk:<any>
controller:<TransformStreamDefaultController>
flush:<Function>
用户定义的函数，在  TransformStream 的可写侧关闭之前立即调用，信号表明转换过程结束。
controller:<TransformStreamDefaultController>
cancel:<Function>
用户定义的函数，当  TransformStream 的可读侧被取消或可写侧被中止时调用。
reason:<any>
readableType:<any>
readableType 选项保留供将来使用， 必须 为 undefined 。
writableType:<any>
writableType 选项保留供将来使用， 必须 为 undefined 。
writableStrategy:<Object>
highWaterMark:<number>
应用背压之前的最大内部队列大小。
size:<Function>
用户定义的函数，用于识别每个数据块的大小。
chunk:<any>
readableStrategy:<Object>
highWaterMark:<number>
应用背压之前的最大内部队列大小。
size:<Function>
用户定义的函数，用于识别每个数据块的大小。
chunk:<any>

<TransformStream> 实例可以使用 <MessagePort> 进行传输。

const stream = new TransformStream();

const { port1, port2 } = new MessageChannel();

port1.onmessage = ({ data }) => {
  const { writable, readable } = data;
  // ...
};

port2.postMessage(stream, [stream]);
JavaScript

填充可读侧队列所需的数据量。

Attributes
chunk:<any>

将数据块附加到可读侧的队列。

Attributes
reason:<any>

向可读侧和可写侧发出信号，表明在处理转换数据时发生了错误，导致两侧突然关闭。

关闭传输的可读侧，并导致可写侧因错误而突然关闭。

Attributes
init:<Object>
highWaterMark:<number>
Attributes
init:<Object>
highWaterMark:<number>

创建一个新的 TextEncoderStream 实例。

TextEncoderStream 实例支持的编码。

Attributes
encoding:<string>
标识此  TextDecoder 实例支持的 encoding 。 默认值： 'utf-8' 。
options:<Object>
fatal:<boolean>
如果解码失败是致命的，则为  true 。
ignoreBOM:<boolean>
当为  true 时， TextDecoderStream 将在解码结果中包含字节顺序标记。当为 false 时，字节顺序标记将从输出中移除。此选项仅在 encoding 为 'utf-8' 、 'utf-16be' 或 'utf-16le' 时使用。 默认值： false 。

创建一个新的 TextDecoderStream 实例。

TextDecoderStream 实例支持的编码。

如果解码错误导致抛出 TypeError，则值将为 true。

如果解码结果将包含字节顺序标记，则值将为 true。

Attributes
format:<string>
'deflate' 、 'deflate-raw' 、 'gzip' 或 'brotli' 之一。
Attributes
format:<string>
'deflate' 、 'deflate-raw' 、 'gzip' 或 'brotli' 之一。

它们通过以下方式访问：

import {
  arrayBuffer,
  blob,
  buffer,
  json,
  text,
} from 'node:stream/consumers';
JavaScript
Attributes
stream:<ReadableStream> | <stream.Readable> | <AsyncIterator>
import { arrayBuffer } from 'node:stream/consumers';
import { Readable } from 'node:stream';
import { TextEncoder } from 'node:util';

const encoder = new TextEncoder();
const dataArray = encoder.encode('hello world from consumers!');

const readable = Readable.from(dataArray);
const data = await arrayBuffer(readable);
console.log(`来自可读流：${data.byteLength}`);
// 打印：来自可读流：76
JavaScript
Attributes
stream:<ReadableStream> | <stream.Readable> | <AsyncIterator>
import { blob } from 'node:stream/consumers';

const dataBlob = new Blob(['hello world from consumers!']);

const readable = dataBlob.stream();
const data = await blob(readable);
console.log(`来自可读流：${data.size}`);
// 打印：来自可读流：27
JavaScript
Attributes
stream:<ReadableStream> | <stream.Readable> | <AsyncIterator>
import { buffer } from 'node:stream/consumers';
import { Readable } from 'node:stream';
import { Buffer } from 'node:buffer';

const dataBuffer = Buffer.from('hello world from consumers!');

const readable = Readable.from(dataBuffer);
const data = await buffer(readable);
console.log(`来自可读流：${data.length}`);
// 打印：来自可读流：27
JavaScript
Attributes
stream:<ReadableStream> | <stream.Readable> | <AsyncIterator>
import { bytes } from 'node:stream/consumers';
import { Readable } from 'node:stream';
import { Buffer } from 'node:buffer';

const dataBuffer = Buffer.from('hello world from consumers!');

const readable = Readable.from(dataBuffer);
const data = await bytes(readable);
console.log(`来自可读流：${data.length}`);
// 打印：来自可读流：27
JavaScript
Attributes
stream:<ReadableStream> | <stream.Readable> | <AsyncIterator>
import { json } from 'node:stream/consumers';
import { Readable } from 'node:stream';

const items = Array.from(
  {
    length: 100,
  },
  () => ({
    message: 'hello world from consumers!',
  }),
);

const readable = Readable.from(JSON.stringify(items));
const data = await json(readable);
console.log(`来自可读流：${data.length}`);
// 打印：来自可读流：100
JavaScript
Attributes
stream:<ReadableStream> | <stream.Readable> | <AsyncIterator>
import { text } from 'node:stream/consumers';
import { Readable } from 'node:stream';

const readable = Readable.from('Hello world from consumers!');
const data = await text(readable);
console.log(`来自可读流：${data.length}`);
// 打印：来自可读流：27
JavaScript