Node.js v24.0.0 文档

事件：'close'#

当流及其任何底层资源（例如文件描述符）已关闭时，会发出 'close' 事件。 该事件表示不再发出任何事件，并且不会发生进一步的计算。

如果使用 emitClose 选项创建 Writable 流，则它将始终发出 'close' 事件。

事件：'drain'#

如果调用 stream.write(chunk) 返回 false，则在适合恢复向流写入数据时，将发出 'drain' 事件。

// Write the data to the supplied writable stream one million times.
// Be attentive to back-pressure.
function writeOneMillionTimes(writer, data, encoding, callback) {
  let i = 1000000;
  write();
  function write() {
    let ok = true;
    do {
      i--;
      if (i === 0) {
        // Last time!
        writer.write(data, encoding, callback);
      } else {
        // See if we should continue, or wait.
        // Don't pass the callback, because we're not done yet.
        ok = writer.write(data, encoding);
      }
    } while (i > 0 && ok);
    if (i > 0) {
      // Had to stop early!
      // Write some more once it drains.
      writer.once('drain', write);
    }
  }
} copy

事件：'error'#

• <Error>

如果在写入或管道传输数据时发生错误，则会发出 'error' 事件。 调用侦听器回调时，会传递一个 Error 参数。

当发出 'error' 事件时，除非在创建流时将 autoDestroy 选项设置为 false，否则流将被关闭。

在 'error' 之后，应该不再发出其他事件（包括 'error' 事件）。

事件：'finish'#

在调用 stream.end() 方法之后，并且所有数据都已刷新到底层系统之后，将发出 'finish' 事件。

const writer = getWritableStreamSomehow();
for (let i = 0; i < 100; i++) {
  writer.write(`hello, #${i}!\n`);
}
writer.on('finish', () => {
  console.log('All writes are now complete.');
});
writer.end('This is the end\n'); copy

事件：'pipe'#

• src <stream.Readable> 管道传输到此可写流的源流

当在可读流上调用 stream.pipe() 方法，并将此可写流添加到其目标集合时，会触发 'pipe' 事件。

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('pipe', (src) => {
  console.log('Something is piping into the writer.');
  assert.equal(src, reader);
});
reader.pipe(writer); copy

事件：'unpipe'#

• src <stream.Readable> 取消管道连接到此可写流的源流。

当在 Readable 流上调用 stream.unpipe() 方法，并从此可读流的目标集合中删除此 Writable 时，会触发 'unpipe' 事件。

如果此 Writable 流在 Readable 流通过管道连接到它时发出错误，也会触发此事件。

const writer = getWritableStreamSomehow();
const reader = getReadableStreamSomehow();
writer.on('unpipe', (src) => {
  console.log('Something has stopped piping into the writer.');
  assert.equal(src, reader);
});
reader.pipe(writer);
reader.unpipe(writer); copy

writable.cork()#

writable.cork() 方法强制将所有写入的数据缓冲在内存中。当调用 stream.uncork() 或 stream.end() 方法时，将刷新缓冲的数据。

writable.cork() 的主要目的是为了适应在快速连续地将多个小块写入流的情况。writable.cork() 不会立即将它们转发到底层目标，而是缓冲所有块，直到调用 writable.uncork()，这将把它们全部传递给 writable._writev()（如果存在）。这可以防止队首阻塞的情况，即在等待处理第一个小块时对数据进行缓冲。但是，在不实现 writable._writev() 的情况下使用 writable.cork() 可能会对吞吐量产生不利影响。

另请参阅：writable.uncork()，writable._writev()。

writable.destroy([error])#

• error <Error> 可选，与 'error' 事件一起发出的错误。
• 返回：<this>

销毁流。 可选择发出一个 'error' 事件，并发出一个 'close' 事件（除非将 emitClose 设置为 false）。 在此调用之后，可写流已结束，随后对 write() 或 end() 的调用将导致 ERR_STREAM_DESTROYED 错误。 这是一种破坏性且直接销毁流的方法。 之前对 write() 的调用可能没有耗尽，并可能触发 ERR_STREAM_DESTROYED 错误。 如果在关闭之前应刷新数据，请使用 end() 而不是 destroy，或在销毁流之前等待 'drain' 事件。

const { Writable } = require('node:stream');

const myStream = new Writable();

const fooErr = new Error('foo error');
myStream.destroy(fooErr);
myStream.on('error', (fooErr) => console.error(fooErr.message)); // foo error copy

const { Writable } = require('node:stream');

const myStream = new Writable();

myStream.destroy();
myStream.on('error', function wontHappen() {}); copy

const { Writable } = require('node:stream');

const myStream = new Writable();
myStream.destroy();

myStream.write('foo', (error) => console.error(error.code));
// ERR_STREAM_DESTROYED copy

一旦调用了 destroy()，任何进一步的调用都将是空操作，并且除了来自 _destroy() 的错误之外，不会再发出 'error'。

实现者不应覆盖此方法，而应实现 writable._destroy()。

writable.closed#

• <boolean>

在发出 'close' 之后为 true。

writable.destroyed#

• <boolean>

在调用 writable.destroy() 之后为 true。

const { Writable } = require('node:stream');

const myStream = new Writable();

console.log(myStream.destroyed); // false
myStream.destroy();
console.log(myStream.destroyed); // true copy

writable.end([chunk[, encoding]][, callback])#

• chunk <string> | <Buffer> | <TypedArray> | <DataView> | <any> 可选的要写入的数据。 对于不在对象模式下运行的流，chunk 必须是 <string>、<Buffer>、<TypedArray> 或 <DataView>。 对于对象模式流，chunk 可以是除 null 之外的任何 JavaScript 值。
• encoding <string> 如果 chunk 是字符串，则为编码
• callback <Function> 流完成时的回调。
• 返回：<this>

调用 writable.end() 方法表示不再有数据写入 Writable。 可选的 chunk 和 encoding 参数允许在关闭流之前立即写入最后一个额外的数据块。

在调用 stream.end() 之后调用 stream.write() 方法会引发错误。

// Write 'hello, ' and then end with 'world!'.
const fs = require('node:fs');
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// Writing more now is not allowed! copy

writable.setDefaultEncoding(encoding)#

• encoding <string> 新的默认编码
• 返回：<this>

writable.setDefaultEncoding() 方法为 Writable 流设置默认的 encoding。

writable.uncork()#

writable.uncork() 方法刷新自调用 stream.cork() 以来缓冲的所有数据。

当使用 writable.cork() 和 writable.uncork() 来管理写入流的缓冲时，请使用 process.nextTick() 来推迟对 writable.uncork() 的调用。 这样做允许对给定 Node.js 事件循环阶段中发生的所有 writable.write() 调用进行批处理。

stream.cork();
stream.write('some ');
stream.write('data ');
process.nextTick(() => stream.uncork()); copy

如果在流上多次调用 writable.cork() 方法，则必须调用相同数量的 writable.uncork() 才能刷新缓冲的数据。

stream.cork();
stream.write('some ');
stream.cork();
stream.write('data ');
process.nextTick(() => {
  stream.uncork();
  // The data will not be flushed until uncork() is called a second time.
  stream.uncork();
}); copy

另请参阅：writable.cork()。

writable.writable#

• <boolean>

如果可以安全地调用 writable.write()，则为 true，这意味着流尚未被销毁、发生错误或结束。

writable.writableAborted#

• <boolean>

返回流是否在发出 'finish' 之前被销毁或发生错误。

writable.writableEnded#

• <boolean>

在调用 writable.end() 之后为 true。 此属性不指示数据是否已刷新，为此请使用 writable.writableFinished。

writable.writableCorked#

• <integer>

为了完全解除流的封锁，需要调用 writable.uncork() 的次数。

writable.errored#

• <Error>

如果流已因错误而被销毁，则返回错误。

writable.writableFinished#

• <boolean>

在发出 'finish' 事件之前立即设置为 true。

writable.writableHighWaterMark#

• <number>

返回创建此 Writable 时传递的 highWaterMark 的值。

writable.writableLength#

• <number>

此属性包含队列中准备好写入的字节数（或对象）。 该值提供有关 highWaterMark 状态的内省数据。

writable.writableNeedDrain#

• <boolean>

如果流的缓冲区已满且流将发出 'drain'，则为 true。

writable.writableObjectMode#

• <boolean>

给定 Writable 流的属性 objectMode 的 getter。

writable[Symbol.asyncDispose]()#

使用 AbortError 调用 writable.destroy() 并返回一个 promise，该 promise 在流完成时实现。

writable.write(chunk[, encoding][, callback])#

• chunk <string> | <Buffer> | <TypedArray> | <DataView> | <any> 可选的要写入的数据。 对于不在对象模式下运行的流，chunk 必须是 <string>、<Buffer>、<TypedArray> 或 <DataView>。 对于对象模式流，chunk 可以是除 null 之外的任何 JavaScript 值。
• encoding <string> | <null> 编码，如果 chunk 是字符串。默认值： 'utf8'
• callback <Function> 当此数据块刷新时调用的回调函数。
• 返回值： <boolean> 如果流希望调用代码在继续写入其他数据之前等待 'drain' 事件被触发，则返回 false；否则返回 true。

writable.write() 方法将一些数据写入流，并在数据被完全处理后调用提供的 callback。 如果发生错误，则将使用该错误作为其第一个参数来调用 callback。 callback 是异步调用的，并且在 'error' 被触发之前调用。

如果内部缓冲区小于在创建流时配置的 highWaterMark，并且允许写入 chunk，则返回值为 true。 如果返回 false，则应停止进一步尝试将数据写入流，直到 'drain' 事件被触发。

当流未排空时，调用 write() 将缓冲 chunk，并返回 false。 一旦所有当前缓冲的块都被排空（被操作系统接受进行传递），'drain' 事件将被触发。 一旦 write() 返回 false，在 'drain' 事件被触发之前，不要写入更多的块。 虽然允许在未排空的流上调用 write()，但 Node.js 将缓冲所有写入的块，直到达到最大内存使用量，此时它将无条件中止。 即使在中止之前，高内存使用量也会导致糟糕的垃圾收集器性能和高 RSS（即使在不再需要内存之后，通常也不会将其释放回系统）。 由于如果远程对等方不读取数据，TCP 套接字可能永远不会排空，因此写入未排空的套接字可能导致可远程利用的漏洞。

当流未排空时写入数据对于 Transform 来说尤其成问题，因为 Transform 流默认情况下处于暂停状态，直到它们被管道连接或添加了 'data' 或 'readable' 事件处理程序。

如果要写入的数据可以按需生成或获取，建议将逻辑封装到 Readable 中，并使用 stream.pipe()。 但是，如果首选调用 write()，则可以使用 'drain' 事件来尊重背压并避免内存问题

function write(data, cb) {
  if (!stream.write(data)) {
    stream.once('drain', cb);
  } else {
    process.nextTick(cb);
  }
}

// Wait for cb to be called before doing any other write.
write('hello', () => {
  console.log('Write completed, do more writes now.');
}); copy

对象模式下的 Writable 流将始终忽略 encoding 参数。

事件：'close'#

当流及其任何底层资源（例如文件描述符）已关闭时，会发出 'close' 事件。 该事件表示不再发出任何事件，并且不会发生进一步的计算。

如果使用 emitClose 选项创建 Readable 流，则它将始终触发 'close' 事件。

事件：'data'#

• chunk <Buffer> | <string> | <any> 数据块。 对于不在对象模式下运行的流，该块将是字符串或 Buffer。 对于处于对象模式的流，该块可以是除 null 之外的任何 JavaScript 值。

每当流将数据块的所有权放弃给消费者时，都会触发 'data' 事件。 通过调用 readable.pipe()、readable.resume() 或通过将侦听器回调附加到 'data' 事件，将流切换到流动模式时，可能会发生这种情况。 每当调用 readable.read() 方法并且有数据块可供返回时，也会触发 'data' 事件。

将 'data' 事件侦听器附加到未显式暂停的流会将该流切换到流动模式。 然后将尽快传递数据。

如果使用 readable.setEncoding() 方法为流指定了默认编码，则侦听器回调会将数据块作为字符串传递；否则，数据将作为 Buffer 传递。

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
}); copy

事件：'end'#

当没有更多数据可以从流中使用时，将触发 'end' 事件。

除非完全使用数据，否则不会触发 'end' 事件。 这可以通过将流切换到流动模式，或通过重复调用 stream.read() 直到所有数据都被使用来完成。

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
});
readable.on('end', () => {
  console.log('There will be no more data.');
}); copy

事件：'error'#

• <Error>

Readable 实现可能随时触发 'error' 事件。 通常，如果底层流由于底层内部故障而无法生成数据，或者流实现尝试推送无效的数据块，则可能会发生这种情况。

侦听器回调将传递一个 Error 对象。

事件：'pause'#

当调用 stream.pause() 且 readableFlowing 不是 false 时，将触发 'pause' 事件。

事件：'readable'#

当流中有数据可供读取时，就会触发 'readable' 事件，数据量最大可达配置的高水位线 (state.highWaterMark)。 实际上，它表明流的缓冲区中有新信息。 如果此缓冲区中有数据，则可以调用 stream.read() 来检索该数据。 此外，当流已到达末尾时，也可能会触发 'readable' 事件。

const readable = getReadableStreamSomehow();
readable.on('readable', function() {
  // There is some data to read now.
  let data;

  while ((data = this.read()) !== null) {
    console.log(data);
  }
}); copy

如果流已到达末尾，则调用 stream.read() 将返回 null 并触发 'end' 事件。 如果从未有任何数据可读取，情况也是如此。 例如，在以下示例中，foo.txt 是一个空文件。

const fs = require('node:fs');
const rr = fs.createReadStream('foo.txt');
rr.on('readable', () => {
  console.log(`readable: ${rr.read()}`);
});
rr.on('end', () => {
  console.log('end');
}); copy

运行此脚本的输出是

$ node test.js
readable: null
end copy

在某些情况下，附加 'readable' 事件的监听器会导致将一定量的数据读取到内部缓冲区中。

通常，readable.pipe() 和 'data' 事件机制比 'readable' 事件更容易理解。 但是，处理 'readable' 可能会提高吞吐量。

如果同时使用 'readable' 和 'data'，则 'readable' 在控制流中具有优先权，即只有在调用 stream.read() 时才会触发 'data'。 readableFlowing 属性将变为 false。 如果在删除 'readable' 时存在 'data' 监听器，则流将开始流动，即无需调用 .resume() 即可触发 'data' 事件。

事件：'resume'#

当调用 stream.resume() 且 readableFlowing 不为 true 时，会触发 'resume' 事件。

readable.destroy([error])#

• error <Error> 将在 'error' 事件中作为有效负载传递的错误
• 返回：<this>

销毁流。 可选地触发一个 'error' 事件，并触发一个 'close' 事件（除非将 emitClose 设置为 false）。 调用此方法后，可读流将释放任何内部资源，并且后续对 push() 的调用将被忽略。

一旦调用了 destroy()，任何进一步的调用都将是空操作，并且除了来自 _destroy() 的错误之外，不会再发出 'error'。

实现者不应覆盖此方法，而应实现 readable._destroy()。

readable.closed#

• <boolean>

在发出 'close' 之后为 true。

readable.destroyed#

• <boolean>

在调用 readable.destroy() 之后为 true。

readable.isPaused()#

• 返回：<boolean>

readable.isPaused() 方法返回 Readable 的当前操作状态。 这主要由 readable.pipe() 方法的基础机制使用。 在大多数典型情况下，没有理由直接使用此方法。

const readable = new stream.Readable();

readable.isPaused(); // === false
readable.pause();
readable.isPaused(); // === true
readable.resume();
readable.isPaused(); // === false copy

readable.pause()#

• 返回：<this>

readable.pause() 方法将导致处于流动模式的流停止触发 'data' 事件，从而退出流动模式。 任何变得可用的数据将保留在内部缓冲区中。

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
  readable.pause();
  console.log('There will be no additional data for 1 second.');
  setTimeout(() => {
    console.log('Now data will start flowing again.');
    readable.resume();
  }, 1000);
}); copy

如果存在 'readable' 事件监听器，则 readable.pause() 方法无效。

readable.pipe(destination[, options])#

• destination <stream.Writable> 用于写入数据的目标
• options <Object> 管道选项
  • end <boolean> 在读取器结束时结束写入器。 默认值： true。
• 返回：<stream.Writable> *destination*，如果它是 Duplex 或 Transform 流，则允许管道链

readable.pipe() 方法将 Writable 流附加到 readable，使其自动切换到流动模式，并将所有数据推送到附加的 Writable。 将自动管理数据流，以便目标 Writable 流不会被更快的 Readable 流淹没。

以下示例将所有数据从 readable 管道传输到名为 file.txt 的文件中。

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt'.
readable.pipe(writable); copy

可以将多个 Writable 流附加到单个 Readable 流。

readable.pipe() 方法返回对 *destination* 流的引用，从而可以设置管道流的链。

const fs = require('node:fs');
const zlib = require('node:zlib');
const r = fs.createReadStream('file.txt');
const z = zlib.createGzip();
const w = fs.createWriteStream('file.txt.gz');
r.pipe(z).pipe(w); copy

默认情况下，当源 Readable 流触发 'end' 时，会在目标 Writable 流上调用 stream.end()，以便目标不再可写。 要禁用此默认行为，可以将 end 选项作为 false 传递，导致目标流保持打开状态。

reader.pipe(writer, { end: false });
reader.on('end', () => {
  writer.end('Goodbye\n');
}); copy

一个重要的注意事项是，如果在处理过程中 Readable 流触发错误，则 Writable 目标不会自动关闭。 如果发生错误，则必须手动关闭每个流，以防止内存泄漏。

在 Node.js 进程退出之前，永远不会关闭 process.stderr 和 process.stdout Writable 流，无论指定的选项如何。

readable.read([size])#

• size <number> 可选参数，用于指定要读取的数据量。
• 返回：<string> | <Buffer> | <null> | <any>

readable.read() 方法从内部缓冲区读取数据并返回。 如果没有数据可读取，则返回 null。 默认情况下，数据作为 Buffer 对象返回，除非已使用 readable.setEncoding() 方法指定了编码，或者流以对象模式运行。

可选的 size 参数指定要读取的特定字节数。 如果没有 size 个字节可读取，则将返回 null，除非流已结束，在这种情况下，将返回内部缓冲区中剩余的所有数据。

如果未指定 size 参数，则将返回内部缓冲区中包含的所有数据。

size 参数必须小于或等于 1 GiB。

readable.read() 方法应仅在以暂停模式运行的 Readable 流上调用。 在流动模式下，会自动调用 readable.read()，直到内部缓冲区完全耗尽。

const readable = getReadableStreamSomehow();

// 'readable' may be triggered multiple times as data is buffered in
readable.on('readable', () => {
  let chunk;
  console.log('Stream is readable (new data received in buffer)');
  // Use a loop to make sure we read all currently available data
  while (null !== (chunk = readable.read())) {
    console.log(`Read ${chunk.length} bytes of data...`);
  }
});

// 'end' will be triggered once when there is no more data available
readable.on('end', () => {
  console.log('Reached end of stream.');
}); copy

每次调用 readable.read() 都会返回一个数据块或 null，表示当前没有更多数据可读取。 这些块不会自动连接。 由于单个 read() 调用不会返回所有数据，因此可能需要使用 while 循环来持续读取块，直到检索到所有数据。 读取大文件时，.read() 可能会暂时返回 null，表明它已消耗所有缓冲内容，但可能还有更多数据要缓冲。 在这种情况下，一旦缓冲区中有更多数据，就会触发新的 'readable' 事件，并且 'end' 事件表示数据传输的结束。

因此，要从 readable 读取文件的全部内容，需要跨多个 'readable' 事件收集块。

const chunks = [];

readable.on('readable', () => {
  let chunk;
  while (null !== (chunk = readable.read())) {
    chunks.push(chunk);
  }
});

readable.on('end', () => {
  const content = chunks.join('');
}); copy

对象模式下的 Readable 流将始终从对 readable.read(size) 的调用返回单个项目，而不管 size 参数的值如何。

如果 readable.read() 方法返回一个数据块，也会触发一个 'data' 事件。

在触发 'end' 事件后调用 stream.read([size]) 将返回 null。 不会引发运行时错误。

readable.readable#

• <boolean>

如果可以安全地调用 readable.read()，则为 true，这意味着流尚未被销毁或触发 'error' 或 'end'。

readable.readableAborted#

• <boolean>

返回流是否在触发 'end' 之前被销毁或发生错误。

readable.readableDidRead#

• <boolean>

返回是否已触发 'data'。

readable.readableEncoding#

• <null> | <string>

获取给定 Readable 流的 encoding 属性。可以使用 readable.setEncoding() 方法设置 encoding 属性。

readable.readableEnded#

• <boolean>

当 'end' 事件发出时变为 true。

readable.errored#

• <Error>

如果流已因错误而被销毁，则返回错误。

readable.readableFlowing#

• <boolean>

此属性反映了 Readable 流的当前状态，如 三种状态 部分所述。

readable.readableHighWaterMark#

• <number>

返回创建此 Readable 时传递的 highWaterMark 的值。

readable.readableLength#

• <number>

此属性包含队列中准备读取的字节数（或对象数）。 该值提供了有关 highWaterMark 状态的内省数据。

readable.readableObjectMode#

• <boolean>

获取给定 Readable 流的 objectMode 属性。

readable.resume()#

• 返回：<this>

readable.resume() 方法使显式暂停的 Readable 流恢复发出 'data' 事件，从而将流切换到流动模式。

readable.resume() 方法可用于完全消耗流中的数据，而无需实际处理任何数据

getReadableStreamSomehow()
  .resume()
  .on('end', () => {
    console.log('Reached the end, but did not read anything.');
  }); copy

如果存在 'readable' 事件监听器，则 readable.resume() 方法无效。

readable.setEncoding(encoding)#

• encoding <string> 要使用的编码。
• 返回：<this>

readable.setEncoding() 方法设置从 Readable 流读取的数据的字符编码。

默认情况下，不分配任何编码，并且流数据将作为 Buffer 对象返回。 设置编码会导致流数据作为指定编码的字符串返回，而不是作为 Buffer 对象返回。 例如，调用 readable.setEncoding('utf8') 将导致输出数据被解释为 UTF-8 数据，并作为字符串传递。 调用 readable.setEncoding('hex') 将导致数据以十六进制字符串格式编码。

Readable 流将正确处理通过流传递的多字节字符，否则如果只是从流中提取为 Buffer 对象，这些字符将无法正确解码。

const readable = getReadableStreamSomehow();
readable.setEncoding('utf8');
readable.on('data', (chunk) => {
  assert.equal(typeof chunk, 'string');
  console.log('Got %d characters of string data:', chunk.length);
}); copy

readable.unpipe([destination])#

• destination <stream.Writable> 可选的要取消管道的特定流
• 返回：<this>

readable.unpipe() 方法分离先前使用 stream.pipe() 方法连接的 Writable 流。

如果未指定 destination，则分离所有管道。

如果指定了 destination，但未为其设置任何管道，则该方法不执行任何操作。

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// All the data from readable goes into 'file.txt',
// but only for the first second.
readable.pipe(writable);
setTimeout(() => {
  console.log('Stop writing to file.txt.');
  readable.unpipe(writable);
  console.log('Manually close the file stream.');
  writable.end();
}, 1000); copy

readable.unshift(chunk[, encoding])#

• chunk <Buffer> | <TypedArray> | <DataView> | <string> | <null> | <any> 要取消移入读取队列的数据块。 对于不在对象模式下运行的流，chunk 必须是 <string>、<Buffer>、<TypedArray>、<DataView> 或 null。 对于对象模式流，chunk 可以是任何 JavaScript 值。
• encoding <string> 字符串块的编码。 必须是有效的 Buffer 编码，例如 'utf8' 或 'ascii'。

将 chunk 作为 null 传递表示流的结束 (EOF)，其行为与 readable.push(null) 相同，之后无法写入更多数据。 EOF 信号放在缓冲区的末尾，任何缓冲的数据仍将被刷新。

readable.unshift() 方法将数据块推回内部缓冲区。 这在某些情况下很有用，在这些情况下，流正被需要“取消消耗”从源中乐观地提取的一些数据的代码使用，以便可以将数据传递给其他方。

在发出 'end' 事件后无法调用 stream.unshift(chunk) 方法，否则将抛出运行时错误。

使用 stream.unshift() 的开发人员通常应考虑切换到使用 Transform 流。 有关更多信息，请参见 流实现者的 API 部分。

// Pull off a header delimited by \n\n.
// Use unshift() if we get too much.
// Call the callback with (error, header, stream).
const { StringDecoder } = require('node:string_decoder');
function parseHeader(stream, callback) {
  stream.on('error', callback);
  stream.on('readable', onReadable);
  const decoder = new StringDecoder('utf8');
  let header = '';
  function onReadable() {
    let chunk;
    while (null !== (chunk = stream.read())) {
      const str = decoder.write(chunk);
      if (str.includes('\n\n')) {
        // Found the header boundary.
        const split = str.split(/\n\n/);
        header += split.shift();
        const remaining = split.join('\n\n');
        const buf = Buffer.from(remaining, 'utf8');
        stream.removeListener('error', callback);
        // Remove the 'readable' listener before unshifting.
        stream.removeListener('readable', onReadable);
        if (buf.length)
          stream.unshift(buf);
        // Now the body of the message can be read from the stream.
        callback(null, header, stream);
        return;
      }
      // Still reading the header.
      header += str;
    }
  }
} copy

与 stream.push(chunk) 不同，stream.unshift(chunk) 不会通过重置流的内部读取状态来结束读取过程。 如果在读取期间（即从自定义流上的 stream._read() 实现中）调用 readable.unshift()，这可能会导致意外的结果。 在调用 readable.unshift() 后立即调用 stream.push('') 将适当地重置读取状态，但最好只是避免在执行读取的过程中调用 readable.unshift()。

readable.wrap(stream)#

• stream <Stream> 一个“旧式”可读流
• 返回：<this>

在 Node.js 0.10 之前，流没有实现当前定义的整个 node:stream 模块 API。（有关更多信息，请参见兼容性。）

当使用发出 'data' 事件并且具有 stream.pause() 方法（仅提供建议）的旧 Node.js 库时，可以使用 readable.wrap() 方法来创建 Readable 流，该流使用旧流作为其数据源。

很少需要使用 readable.wrap()，但该方法已作为与旧 Node.js 应用程序和库交互的便利方法提供。

const { OldReader } = require('./old-api-module.js');
const { Readable } = require('node:stream');
const oreader = new OldReader();
const myReader = new Readable().wrap(oreader);

myReader.on('readable', () => {
  myReader.read(); // etc.
}); copy

readable[Symbol.asyncIterator]()#

• 返回: <AsyncIterator> 以完全使用流。

const fs = require('node:fs');

async function print(readable) {
  readable.setEncoding('utf8');
  let data = '';
  for await (const chunk of readable) {
    data += chunk;
  }
  console.log(data);
}

print(fs.createReadStream('file')).catch(console.error); copy

如果循环以 break、return 或 throw 终止，则该流将被销毁。 换句话说，迭代一个流将完全消耗该流。 流将以等于 highWaterMark 选项大小的块读取。 在上面的代码示例中，如果文件的数据少于 64 KiB，数据将位于单个块中，因为没有为 fs.createReadStream() 提供 highWaterMark 选项。

readable[Symbol.asyncDispose]()#

使用 AbortError 调用 readable.destroy() 并返回一个 promise，该 promise 在流完成时完成。

readable.compose(stream[, options])#

• stream <Stream> | <Iterable> | <AsyncIterable> | <Function>
• options <Object>
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Duplex> 一个与流 stream 组成的流。

import { Readable } from 'node:stream';

async function* splitToWords(source) {
  for await (const chunk of source) {
    const words = String(chunk).split(' ');

    for (const word of words) {
      yield word;
    }
  }
}

const wordsStream = Readable.from(['this is', 'compose as operator']).compose(splitToWords);
const words = await wordsStream.toArray();

console.log(words); // prints ['this', 'is', 'compose', 'as', 'operator'] copy

有关更多信息，请参见 stream.compose。

readable.iterator([options])#

• options <Object>
  • destroyOnReturn <boolean> 设置为 false 时，在异步迭代器上调用 return，或使用 break、return 或 throw 退出 for await...of 迭代将不会销毁流。 默认: true。
• 返回: <AsyncIterator> 以使用流。

通过此方法创建的迭代器使用户可以选择取消流的销毁，如果 for await...of 循环通过 return、break 或 throw 退出，或者如果流在迭代期间发出错误，则迭代器应销毁流。

const { Readable } = require('node:stream');

async function printIterator(readable) {
  for await (const chunk of readable.iterator({ destroyOnReturn: false })) {
    console.log(chunk); // 1
    break;
  }

  console.log(readable.destroyed); // false

  for await (const chunk of readable.iterator({ destroyOnReturn: false })) {
    console.log(chunk); // Will print 2 and then 3
  }

  console.log(readable.destroyed); // True, stream was totally consumed
}

async function printSymbolAsyncIterator(readable) {
  for await (const chunk of readable) {
    console.log(chunk); // 1
    break;
  }

  console.log(readable.destroyed); // true
}

async function showBoth() {
  await printIterator(Readable.from([1, 2, 3]));
  await printSymbolAsyncIterator(Readable.from([1, 2, 3]));
}

showBoth(); copy

readable.map(fn[, options])#

• fn <Function> | <AsyncFunction> 一个函数，用于映射流中的每个块。
  • data <any> 来自流的数据块。
  • options <Object>
    • signal <AbortSignal> 如果流被销毁，则中止，允许提前中止 fn 调用。
• options <Object>
  • concurrency <number> fn 在流上同时调用的最大并发数。 默认值: 1。
  • highWaterMark <number> 等待用户使用映射项时要缓冲多少项。 默认值: concurrency * 2 - 1。
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Readable> 使用函数 fn 映射的流。

此方法允许映射流。 对于流中的每个块，都将调用 fn 函数。 如果 fn 函数返回一个 Promise - 该 Promise 将在传递给结果流之前被 await。

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous mapper.
for await (const chunk of Readable.from([1, 2, 3, 4]).map((x) => x * 2)) {
  console.log(chunk); // 2, 4, 6, 8
}
// With an asynchronous mapper, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map((domain) => resolver.resolve4(domain), { concurrency: 2 });
for await (const result of dnsResults) {
  console.log(result); // Logs the DNS result of resolver.resolve4.
} copy

readable.filter(fn[, options])#

• fn <Function> | <AsyncFunction> 用于从流中筛选块的函数。
  • data <any> 来自流的数据块。
  • options <Object>
    • signal <AbortSignal> 如果流被销毁，则中止，允许提前中止 fn 调用。
• options <Object>
  • concurrency <number> fn 在流上同时调用的最大并发数。 默认值: 1。
  • highWaterMark <number> 等待用户使用筛选后的项目时要缓冲多少项。 默认值: concurrency * 2 - 1。
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Readable> 使用谓词 fn 筛选的流。

此方法允许筛选流。 对于流中的每个块，都将调用 fn 函数，如果它返回真值，则该块将被传递到结果流。 如果 fn 函数返回一个 Promise - 该 Promise 将被 await。

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous predicate.
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
  console.log(chunk); // 3, 4
}
// With an asynchronous predicate, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).filter(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address.ttl > 60;
}, { concurrency: 2 });
for await (const result of dnsResults) {
  // Logs domains with more than 60 seconds on the resolved dns record.
  console.log(result);
} copy

readable.forEach(fn[, options])#

• fn <Function> | <AsyncFunction> 要在流的每个块上调用的函数。
  • data <any> 来自流的数据块。
  • options <Object>
    • signal <AbortSignal> 如果流被销毁，则中止，允许提前中止 fn 调用。
• options <Object>
  • concurrency <number> fn 在流上同时调用的最大并发数。 默认值: 1。
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Promise> 流完成时的 Promise。

此方法允许迭代流。 对于流中的每个块，都将调用 fn 函数。 如果 fn 函数返回一个 Promise - 该 Promise 将被 await。

此方法与 for await...of 循环不同，因为它可以选择性地并发处理块。 此外，只能通过传递 signal 选项并在中止相关的 AbortController 时停止 forEach 迭代，而 for await...of 可以使用 break 或 return 停止。 在任何一种情况下，流都将被销毁。

此方法与侦听 'data' 事件不同，因为它在底层机制中使用 readable 事件，并且可以限制并发 fn 调用的数量。

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

// With a synchronous predicate.
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
  console.log(chunk); // 3, 4
}
// With an asynchronous predicate, making at most 2 queries at a time.
const resolver = new Resolver();
const dnsResults = Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address;
}, { concurrency: 2 });
await dnsResults.forEach((result) => {
  // Logs result, similar to `for await (const result of dnsResults)`
  console.log(result);
});
console.log('done'); // Stream has finished copy

readable.toArray([options])#

• options <Object>
  • signal <AbortSignal> 如果信号被中止，则允许取消 toArray 操作。
• 返回: <Promise> 一个包含流内容的数组的 Promise。

此方法允许轻松获取流的内容。

由于此方法将整个流读入内存，因此它抵消了流的优势。 它旨在用于互操作性和便利性，而不是作为使用流的主要方式。

import { Readable } from 'node:stream';
import { Resolver } from 'node:dns/promises';

await Readable.from([1, 2, 3, 4]).toArray(); // [1, 2, 3, 4]

// Make dns queries concurrently using .map and collect
// the results into an array using toArray
const dnsResults = await Readable.from([
  'nodejs.org',
  'openjsf.org',
  'www.linuxfoundation.org',
]).map(async (domain) => {
  const { address } = await resolver.resolve4(domain, { ttl: true });
  return address;
}, { concurrency: 2 }).toArray(); copy

readable.some(fn[, options])#

• fn <Function> | <AsyncFunction> 要在流的每个块上调用的函数。
  • data <any> 来自流的数据块。
  • options <Object>
    • signal <AbortSignal> 如果流被销毁，则中止，允许提前中止 fn 调用。
• options <Object>
  • concurrency <number> fn 在流上同时调用的最大并发数。 默认值: 1。
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Promise> 如果 fn 为至少一个块返回一个真值，则解析为 true 的 Promise。

此方法类似于 Array.prototype.some，并在流中的每个块上调用 fn，直到等待的返回值是 true（或任何真值）。 一旦对块的 fn 调用等待的返回值是真值，流将被销毁，并且 Promise 将被 true 履行。 如果对块的 fn 调用都没有返回真值，则 Promise 将被 false 履行。

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).some((x) => x > 2); // true
await Readable.from([1, 2, 3, 4]).some((x) => x < 0); // false

// With an asynchronous predicate, making at most 2 file checks at a time.
const anyBigFile = await Readable.from([
  'file1',
  'file2',
  'file3',
]).some(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
console.log(anyBigFile); // `true` if any file in the list is bigger than 1MB
console.log('done'); // Stream has finished copy

readable.find(fn[, options])#

• fn <Function> | <AsyncFunction> 要在流的每个块上调用的函数。
  • data <any> 来自流的数据块。
  • options <Object>
    • signal <AbortSignal> 如果流被销毁，则中止，允许提前中止 fn 调用。
• options <Object>
  • concurrency <number> fn 在流上同时调用的最大并发数。 默认值: 1。
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Promise> 一个 Promise，解析为 fn 计算结果为真值的第一个块，如果未找到元素，则解析为 undefined。

此方法类似于 Array.prototype.find，并在流中的每个块上调用 fn，以查找 fn 的真值的块。 一旦 fn 调用的等待返回值是真值，流将被销毁，并且 Promise 将被 fn 返回真值的 值履行。 如果所有对块的 fn 调用都返回一个假值，则 Promise 将被 undefined 履行。

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).find((x) => x > 2); // 3
await Readable.from([1, 2, 3, 4]).find((x) => x > 0); // 1
await Readable.from([1, 2, 3, 4]).find((x) => x > 10); // undefined

// With an asynchronous predicate, making at most 2 file checks at a time.
const foundBigFile = await Readable.from([
  'file1',
  'file2',
  'file3',
]).find(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
console.log(foundBigFile); // File name of large file, if any file in the list is bigger than 1MB
console.log('done'); // Stream has finished copy

readable.every(fn[, options])#

• fn <Function> | <AsyncFunction> 要在流的每个块上调用的函数。
  • data <any> 来自流的数据块。
  • options <Object>
    • signal <AbortSignal> 如果流被销毁，则中止，允许提前中止 fn 调用。
• options <Object>
  • concurrency <number> fn 在流上同时调用的最大并发数。 默认值: 1。
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Promise> 如果 fn 为所有块返回一个真值，则解析为 true 的 Promise。

此方法类似于 Array.prototype.every，并在流中的每个块上调用 fn，以检查所有等待的返回值是否是 fn 的真值。 一旦对块的 fn 调用等待的返回值是假值，流将被销毁，并且 Promise 将被 false 履行。 如果所有对块的 fn 调用都返回一个真值，则 Promise 将被 true 履行。

import { Readable } from 'node:stream';
import { stat } from 'node:fs/promises';

// With a synchronous predicate.
await Readable.from([1, 2, 3, 4]).every((x) => x > 2); // false
await Readable.from([1, 2, 3, 4]).every((x) => x > 0); // true

// With an asynchronous predicate, making at most 2 file checks at a time.
const allBigFiles = await Readable.from([
  'file1',
  'file2',
  'file3',
]).every(async (fileName) => {
  const stats = await stat(fileName);
  return stats.size > 1024 * 1024;
}, { concurrency: 2 });
// `true` if all files in the list are bigger than 1MiB
console.log(allBigFiles);
console.log('done'); // Stream has finished copy

readable.flatMap(fn[, options])#

• fn <Function> | <AsyncGeneratorFunction> | <AsyncFunction> 一个函数，用于映射流中的每个块。
  • data <any> 来自流的数据块。
  • options <Object>
    • signal <AbortSignal> 如果流被销毁，则中止，允许提前中止 fn 调用。
• options <Object>
  • concurrency <number> fn 在流上同时调用的最大并发数。 默认值: 1。
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Readable> 使用函数 fn 平面映射的流。

此方法通过将给定的回调应用于流的每个块，然后展平结果来返回一个新的流。

可以从 fn 返回一个流或另一个可迭代对象或异步可迭代对象，并且结果流将被合并（展平）到返回的流中。

import { Readable } from 'node:stream';
import { createReadStream } from 'node:fs';

// With a synchronous mapper.
for await (const chunk of Readable.from([1, 2, 3, 4]).flatMap((x) => [x, x])) {
  console.log(chunk); // 1, 1, 2, 2, 3, 3, 4, 4
}
// With an asynchronous mapper, combine the contents of 4 files
const concatResult = Readable.from([
  './1.mjs',
  './2.mjs',
  './3.mjs',
  './4.mjs',
]).flatMap((fileName) => createReadStream(fileName));
for await (const result of concatResult) {
  // This will contain the contents (all chunks) of all 4 files
  console.log(result);
} copy

readable.drop(limit[, options])#

• limit <number> 从可读对象中删除的块数。
• options <Object>
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Readable> 一个删除了 limit 个块的流。

此方法返回一个删除了前 limit 个块的新流。

import { Readable } from 'node:stream';

await Readable.from([1, 2, 3, 4]).drop(2).toArray(); // [3, 4] copy

readable.take(limit[, options])#

• limit <number> 从可读对象中获取的块数。
• options <Object>
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Readable> 一个获取了 limit 个块的流。

此方法返回一个包含前 limit 个块的新流。

import { Readable } from 'node:stream';

await Readable.from([1, 2, 3, 4]).take(2).toArray(); // [1, 2] copy

readable.reduce(fn[, initial[, options]])#

• fn <Function> | <AsyncFunction> 一个 reducer 函数，用于在流中的每个块上调用。
  • previous <any> 从上次调用 fn 获得的值，或者如果指定了 initial 值，否则为流的第一个块。
  • data <any> 来自流的数据块。
  • options <Object>
    • signal <AbortSignal> 如果流被销毁，则中止，允许提前中止 fn 调用。
• initial <any> 在缩减中使用的初始值。
• options <Object>
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Promise> 缩减的最终值的 Promise。

此方法按顺序在流的每个块上调用 fn，并将其传递给先前元素计算的结果。 它返回缩减的最终值的 Promise。

如果没有提供 initial 值，则流的第一个块将用作初始值。 如果流为空，则 Promise 将被 TypeError 拒绝，并带有 ERR_INVALID_ARGS 代码属性。

import { Readable } from 'node:stream';
import { readdir, stat } from 'node:fs/promises';
import { join } from 'node:path';

const directoryPath = './src';
const filesInDir = await readdir(directoryPath);

const folderSize = await Readable.from(filesInDir)
  .reduce(async (totalSize, file) => {
    const { size } = await stat(join(directoryPath, file));
    return totalSize + size;
  }, 0);

console.log(folderSize); copy

reducer 函数逐元素地迭代流，这意味着没有 concurrency 参数或并行性。 要并发执行 reduce，您可以将异步函数提取到 readable.map 方法。

import { Readable } from 'node:stream';
import { readdir, stat } from 'node:fs/promises';
import { join } from 'node:path';

const directoryPath = './src';
const filesInDir = await readdir(directoryPath);

const folderSize = await Readable.from(filesInDir)
  .map((file) => stat(join(directoryPath, file)), { concurrency: 2 })
  .reduce((totalSize, { size }) => totalSize + size, 0);

console.log(folderSize); copy

duplex.allowHalfOpen#

• <boolean>

如果 false，则当可读端结束时，流将自动结束可写端。 最初由 allowHalfOpen 构造函数选项设置，默认为 true。

可以手动更改此设置以更改现有 Duplex 流实例的半开放行为，但必须在发出 'end' 事件之前更改。

transform.destroy([error])#

• error <Error>
• 返回：<this>

销毁流，并选择性地发出 'error' 事件。 在此调用之后，转换流将释放任何内部资源。 实现者不应覆盖此方法，而应实现 readable._destroy()。 Transform 的 _destroy() 的默认实现还会发出 'close'，除非在 false 中设置了 emitClose。

一旦调用了 destroy()，任何进一步的调用都将是空操作，并且不会发出任何进一步的错误，除非来自 _destroy() 可能作为 'error' 发出。