Node.js v25.0.0 文档

事件：'close'#

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

如果一个可写流是使用 emitClose 选项创建的，它将总是发出 '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 参数。

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

在 'error' 事件之后，除了 'close' 之外，不应该再发出其他事件（包括 '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> 从此可写流取消管道传输的源流

当在可读流上调用 stream.unpipe() 方法，将此可写流从其目标集中移除时，会发出 'unpipe' 事件。

当一个可读流管道传输到此可写流时，如果该可写流发出错误，也会发出此事件。

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() 方法表示不会再有数据写入到可写流。可选的 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() 方法为可写流设置默认的 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。

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' 事件之前调用。

如果接纳 chunk 后，内部缓冲区小于创建流时配置的 highWaterMark，则返回值为 true。如果返回 false，则应停止向流中写入数据的进一步尝试，直到发出 'drain' 事件。

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

在流未排空时写入数据对于 Transform 流尤其成问题，因为 Transform 流默认是暂停的，直到它们被管道连接（pipe）或添加了 '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

处于对象模式（object mode）的 Writable 流将始终忽略 encoding 参数。

事件：'close'#

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

如果 Readable 流创建时带有 emitClose 选项，它将总是触发 'close' 事件。

事件：'data'#

• chunk <Buffer> | <string> | <any> 数据块。对于非对象模式的流，chunk 将是字符串或 Buffer。对于对象模式的流，chunk 可以是除 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>

'error' 事件可以由 Readable 实现随时触发。通常，这可能发生在底层流因内部故障而无法生成数据，或者当流实现尝试推送一个无效的数据块时。

监听器回调函数将接收到一个单独的 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> 目标流，如果它是 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() 方法返回对目标流的引用，从而可以建立管道流链：

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 目标不会自动关闭。如果发生错误，需要手动关闭每个流以防止内存泄漏。

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

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 属性的 getter。encoding 属性可以使用 readable.setEncoding() 方法设置。

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 属性的 getter。

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() 方法将一块数据推回到内部缓冲区。这在某些情况下很有用，例如当一个流被需要“反消费”（un-consume）一些它已从源中乐观拉取的数据的代码消费时，以便这些数据可以传递给其他方。

在 '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。（有关更多信息，请参见与旧版 Node.js 的兼容性。）

当使用一个会触发 '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。

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，那么在将结果传递给结果流之前，会 await 这个 promise。

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 函数，如果它返回一个真值（truthy value），该数据块将被传递到结果流。如果 fn 函数返回一个 promise，那么会 await 这个 promise。

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，那么会 await 这个 promise。

此方法与 for await...of 循环的不同之处在于，它可以选择性地并发处理数据块。此外，forEach 迭代只能通过传递一个 signal 选项并中止相关的 AbortController 来停止，而 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 对至少一个数据块返回了真值，则 promise 的评估值为 true。

此方法类似于 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 调用都返回假值（falsy value），则 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 对所有数据块都返回了真值，则 promise 的评估值为 true。

此方法类似于 Array.prototype.every，它对流中的每个数据块调用 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> 一个在流的每个数据块上调用的归纳函数。
  • previous <any> 上一次调用 fn 获得的值，或者如果指定了 initial 值则是该值，否则是流的第一个数据块。
  • data <any> 来自流的一个数据块。
  • options <Object>
    • signal <AbortSignal> 如果流被销毁，该信号会中止，从而允许尽早中止 fn 调用。
• initial <any> 在归纳中使用的初始值。
• options <Object>
  • signal <AbortSignal> 允许在信号中止时销毁流。
• 返回: <Promise> 一个归纳最终值的 promise。

此方法按顺序对流的每个数据块调用 fn，并将上一个元素计算的结果传递给它。它返回一个归纳最终值的 promise。

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

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

归纳函数逐个元素地迭代流，这意味着没有 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'，除非 emitClose 设置为 false。

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