Node.js v24.0.0 文档

new ReadableStream([underlyingSource [, strategy]])#

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

readableStream.locked#

• 类型: <boolean> 如果此 <ReadableStream> 存在活动的读取器，则设置为 true。

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

readableStream.cancel([reason])#

• reason <any>
• 返回: 取消完成后，会兑现一个 undefined 的 Promise。

readableStream.getReader([options])#

• options <Object>
  • mode <string> 'byob' 或 undefined
• 返回: <ReadableStreamDefaultReader> | <ReadableStreamBYOBReader>

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

const stream = new ReadableStream();

const reader = stream.getReader();

console.log(await reader.read());const { ReadableStream } = require('node:stream/web');

const stream = new ReadableStream();

const reader = stream.getReader();

reader.read().then(console.log);copy

导致 readableStream.locked 为 true。

readableStream.pipeThrough(transform[, options])#

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

将此 <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);
  // Prints: Aconst {
  ReadableStream,
  TransformStream,
} = require('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);

(async () => {
  for await (const chunk of transformedStream)
    console.log(chunk);
    // Prints: A
})();copy

readableStream.pipeTo(destination[, options])#

• destination <WritableStream> 此 ReadableStream 的数据将被写入的 <WritableStream>。
• options <Object>
  • preventAbort <boolean> 当为 true 时，此 ReadableStream 中的错误不会导致 destination 被中止。
  • preventCancel <boolean> 当为 true 时，destination 中的错误不会导致此 ReadableStream 被取消。
  • preventClose <boolean> 当为 true 时，关闭此 ReadableStream 不会导致 destination 被关闭。
  • signal <AbortSignal> 允许使用 <AbortController> 取消数据传输。
• 返回: 一个 Promise，用 undefined 兑现

当管道操作处于活动状态时，导致 readableStream.locked 为 true。

readableStream.tee()#

• 返回: <ReadableStream[]>

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

导致 readableStream.locked 为 true。

readableStream.values([options])#

• 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()); copy

异步迭代#

<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()); copy

异步迭代器将消耗 <ReadableStream> 直到它终止。

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

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

使用 postMessage() 传输#

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

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]); copy

new ReadableStreamDefaultReader(stream)#

• stream <ReadableStream>

创建一个新的 <ReadableStreamDefaultReader>，它被锁定到给定的 <ReadableStream>。

readableStreamDefaultReader.cancel([reason])#

• reason <any>
• 返回：使用 undefined 兑现的 Promise。

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

readableStreamDefaultReader.closed#

• 类型: <Promise> 当关联的 <ReadableStream> 关闭时，用 undefined 兑现；如果流出错或在流完成关闭之前释放了读取器的锁，则拒绝该 Promise。

readableStreamDefaultReader.read()#

• 返回: 一个 Promise，用一个对象兑现
  • value <any>
  • done <boolean>

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

readableStreamDefaultReader.releaseLock()#

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

new ReadableStreamBYOBReader(stream)#

• stream <ReadableStream>

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

readableStreamBYOBReader.cancel([reason])#

• reason <any>
• 返回：使用 undefined 兑现的 Promise。

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

readableStreamBYOBReader.closed#

• 类型: <Promise> 当关联的 <ReadableStream> 关闭时，用 undefined 兑现；如果流出错或在流完成关闭之前释放了读取器的锁，则拒绝该 Promise。

readableStreamBYOBReader.read(view[, options])#

• view <Buffer> | <TypedArray> | <DataView>
• options <Object>
  • min <number> 如果设置了此选项，则返回的 Promise 只会在 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 上的所有现有视图失效。 这可能会给您的应用程序带来灾难性的后果。

readableStreamBYOBReader.releaseLock()#

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

readableStreamDefaultController.close()#

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

readableStreamDefaultController.desiredSize#

• 类型: <number>

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

readableStreamDefaultController.enqueue([chunk])#

• chunk <any>

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

readableStreamDefaultController.error([error])#

• error <any>

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

readableByteStreamController.byobRequest#

• 类型: <ReadableStreamBYOBRequest>

readableByteStreamController.close()#

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

readableByteStreamController.desiredSize#

• 类型: <number>

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

readableByteStreamController.enqueue(chunk)#

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

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

readableByteStreamController.error([error])#

• error <any>

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

readableStreamBYOBRequest.respond(bytesWritten)#

• bytesWritten <number>

发出信号，表示已将 bytesWritten 个字节写入 readableStreamBYOBRequest.view。

readableStreamBYOBRequest.respondWithNewView(view)#

• view <Buffer> | <TypedArray> | <DataView>

发出信号，表示已使用写入新 Buffer、TypedArray 或 DataView 的字节来满足请求。

readableStreamBYOBRequest.view#

• 类型: <Buffer> | <TypedArray> | <DataView>

new WritableStream([underlyingSink[, strategy]])#

• underlyingSink <Object>
  • start <Function> 用户定义的函数，在创建 WritableStream 时立即调用。
    • controller <WritableStreamDefaultController>
    • 返回：undefined 或使用 undefined 兑现的 Promise。
  • write <Function> 用户定义的函数，当数据块写入 WritableStream 时调用。
    • chunk <any>
    • controller <WritableStreamDefaultController>
    • 返回：使用 undefined 兑现的 Promise。
  • close <Function> 用户定义的函数，在 WritableStream 关闭时调用。
    • 返回：使用 undefined 兑现的 Promise。
  • abort <Function> 用户定义的函数，用于突然关闭 WritableStream。
    • reason <any>
    • 返回：使用 undefined 兑现的 Promise。
  • type <any> type 选项保留供将来使用，并且必须未定义。
• strategy <Object>
  • highWaterMark <number> 应用背压之前的最大内部队列大小。
  • size <Function> 用于标识每个数据块大小的用户定义的函数。
    • chunk <any>
    • 返回：<number>

writableStream.abort([reason])#

• reason <any>
• 返回：使用 undefined 兑现的 Promise。

突然终止 WritableStream。 所有排队的写入都将被取消，并且与其关联的 Promise 将被拒绝。

writableStream.close()#

• 返回：使用 undefined 兑现的 Promise。

当不再需要其他写入时，关闭 WritableStream。

writableStream.getWriter()#

• 返回: <WritableStreamDefaultWriter>

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

writableStream.locked#

• 类型: <boolean>

默认情况下，writableStream.locked 属性为 false，并且在有活动的写入器附加到此 WritableStream 时，它会切换为 true。

使用 postMessage() 传输#

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

const stream = new WritableStream(getWritableSinkSomehow());

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

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

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

new WritableStreamDefaultWriter(stream)#

• stream <WritableStream>

创建一个新的 WritableStreamDefaultWriter，该写入器已锁定到给定的 WritableStream。

writableStreamDefaultWriter.abort([reason])#

• reason <any>
• 返回：使用 undefined 兑现的 Promise。

突然终止 WritableStream。 所有排队的写入都将被取消，并且与其关联的 Promise 将被拒绝。

writableStreamDefaultWriter.close()#

• 返回：使用 undefined 兑现的 Promise。

当不再需要其他写入时，关闭 WritableStream。

writableStreamDefaultWriter.closed#

• 类型: <Promise> 当关联的 <WritableStream> 关闭时，会用 undefined 履行；如果流出错或写入器的锁在流完成关闭之前被释放，则会被拒绝。

writableStreamDefaultWriter.desiredSize#

• 类型: <number>

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

writableStreamDefaultWriter.ready#

• 类型: <Promise> 当写入器准备好使用时，会用 undefined 履行。

writableStreamDefaultWriter.releaseLock()#

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

writableStreamDefaultWriter.write([chunk])#

• chunk: <any>
• 返回：使用 undefined 兑现的 Promise。

将一个新的数据块追加到 <WritableStream> 的队列中。

writableStreamDefaultController.error([error])#

• error <any>

由用户代码调用，以指示在处理 WritableStream 数据时发生了错误。 调用后，<WritableStream> 将被中止，并且当前正在进行的写入将被取消。

writableStreamDefaultController.signal#

• 类型：<AbortSignal> 一个 AbortSignal，当 <WritableStream> 被中止时，可用于取消挂起的写入或关闭操作。

new TransformStream([transformer[, writableStrategy[, readableStrategy]]])#

• transformer <Object>
  • start <Function> 一个用户定义的函数，在创建 TransformStream 时立即调用。
    • controller <TransformStreamDefaultController>
    • 返回：undefined 或一个以 undefined 履行的 promise
  • transform <Function> 一个用户定义的函数，它接收并可能修改写入 transformStream.writable 的数据块，然后再将其转发到 transformStream.readable。
    • chunk <any>
    • controller <TransformStreamDefaultController>
    • 返回：使用 undefined 兑现的 Promise。
  • flush <Function> 一个用户定义的函数，在 TransformStream 的可写端关闭之前立即调用，表示转换过程的结束。
    • controller <TransformStreamDefaultController>
    • 返回：使用 undefined 兑现的 Promise。
  • readableType <any> readableType 选项保留供将来使用，并且*必须*是 undefined。
  • writableType <any> writableType 选项保留供将来使用，并且*必须*是 undefined。
• writableStrategy <Object>
  • highWaterMark <number> 应用背压之前的最大内部队列大小。
  • size <Function> 用于标识每个数据块大小的用户定义的函数。
    • chunk <any>
    • 返回：<number>
• readableStrategy <Object>
  • highWaterMark <number> 应用背压之前的最大内部队列大小。
  • size <Function> 用于标识每个数据块大小的用户定义的函数。
    • chunk <any>
    • 返回：<number>

transformStream.readable#

• 类型：<ReadableStream>

transformStream.writable#

• 类型：<WritableStream>

使用 postMessage() 传输#

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

const stream = new TransformStream();

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

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

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

transformStreamDefaultController.desiredSize#

• 类型: <number>

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

transformStreamDefaultController.enqueue([chunk])#

• chunk <any>

将数据块追加到可读端的队列。

transformStreamDefaultController.error([reason])#

• reason <any>

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

transformStreamDefaultController.terminate()#

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

new ByteLengthQueuingStrategy(init)#

• init <Object>
  • highWaterMark <number>

byteLengthQueuingStrategy.highWaterMark#

• 类型: <number>

byteLengthQueuingStrategy.size#

• 类型：<Function>
  • chunk <any>
  • 返回：<number>

new CountQueuingStrategy(init)#

• init <Object>
  • highWaterMark <number>

countQueuingStrategy.highWaterMark#

• 类型: <number>

countQueuingStrategy.size#

• 类型：<Function>
  • chunk <any>
  • 返回：<number>

new TextEncoderStream()#

创建一个新的 TextEncoderStream 实例。

textEncoderStream.encoding#

• 类型：<string>

TextEncoderStream 实例支持的编码。

textEncoderStream.readable#

• 类型：<ReadableStream>

textEncoderStream.writable#

• 类型：<WritableStream>

new TextDecoderStream([encoding[, options]])#

• 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.encoding#

• 类型：<string>

TextDecoderStream 实例支持的编码。

textDecoderStream.fatal#

• 类型: <boolean>

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

textDecoderStream.ignoreBOM#

• 类型: <boolean>

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

textDecoderStream.readable#

• 类型：<ReadableStream>

textDecoderStream.writable#

• 类型：<WritableStream>

new CompressionStream(format)#

• format <string> 'deflate', 'deflate-raw', 或 'gzip' 之一。

compressionStream.readable#

• 类型：<ReadableStream>

compressionStream.writable#

• 类型：<WritableStream>

new DecompressionStream(format)#

• format <string> 'deflate', 'deflate-raw', 或 'gzip' 之一。

decompressionStream.readable#

• 类型：<ReadableStream>

decompressionStream.writable#

• 类型：<WritableStream>

streamConsumers.arrayBuffer(stream)#

• stream <ReadableStream> | <stream.Readable> | <AsyncIterator>
• 返回: <Promise> 使用包含流完整内容的 ArrayBuffer 完成。

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(`from readable: ${data.byteLength}`);
// Prints: from readable: 76const { arrayBuffer } = require('node:stream/consumers');
const { Readable } = require('node:stream');
const { TextEncoder } = require('node:util');

const encoder = new TextEncoder();
const dataArray = encoder.encode('hello world from consumers!');
const readable = Readable.from(dataArray);
arrayBuffer(readable).then((data) => {
  console.log(`from readable: ${data.byteLength}`);
  // Prints: from readable: 76
});copy

streamConsumers.blob(stream)#

• stream <ReadableStream> | <stream.Readable> | <AsyncIterator>
• 返回: <Promise> 使用包含流完整内容的 <Blob> 完成。

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(`from readable: ${data.size}`);
// Prints: from readable: 27const { blob } = require('node:stream/consumers');

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

const readable = dataBlob.stream();
blob(readable).then((data) => {
  console.log(`from readable: ${data.size}`);
  // Prints: from readable: 27
});copy

streamConsumers.buffer(stream)#

• stream <ReadableStream> | <stream.Readable> | <AsyncIterator>
• 返回: <Promise> 使用包含流完整内容的 <Buffer> 完成。

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(`from readable: ${data.length}`);
// Prints: from readable: 27const { buffer } = require('node:stream/consumers');
const { Readable } = require('node:stream');
const { Buffer } = require('node:buffer');

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

const readable = Readable.from(dataBuffer);
buffer(readable).then((data) => {
  console.log(`from readable: ${data.length}`);
  // Prints: from readable: 27
});copy

streamConsumers.json(stream)#

• stream <ReadableStream> | <stream.Readable> | <AsyncIterator>
• 返回: <Promise> 使用流的内容完成，该内容被解析为 UTF-8 编码的字符串，然后通过 JSON.parse() 传递。

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(`from readable: ${data.length}`);
// Prints: from readable: 100const { json } = require('node:stream/consumers');
const { Readable } = require('node:stream');

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

const readable = Readable.from(JSON.stringify(items));
json(readable).then((data) => {
  console.log(`from readable: ${data.length}`);
  // Prints: from readable: 100
});copy

streamConsumers.text(stream)#

• stream <ReadableStream> | <stream.Readable> | <AsyncIterator>
• 返回: <Promise> 使用流的内容完成，该内容被解析为 UTF-8 编码的字符串。

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(`from readable: ${data.length}`);
// Prints: from readable: 27const { text } = require('node:stream/consumers');
const { Readable } = require('node:stream');

const readable = Readable.from('Hello world from consumers!');
text(readable).then((data) => {
  console.log(`from readable: ${data.length}`);
  // Prints: from readable: 27
});copy