流

稳定性：2 - 稳定

流是 Node.js 中用于处理流式数据的抽象接口。 node:stream 模块提供了实现流接口的 API。

Node.js 提供了许多流对象。例如，对 HTTP 服务器的请求 和 process.stdout 都是流实例。

流可以是可读的、可写的，或两者皆是。所有流都是 EventEmitter 的实例。

要访问 node:stream 模块：

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

node:stream 模块对于创建新类型的流实例很有用。通常不需要使用 node:stream 模块来消费流。

本文档包含两个主要部分和一个备注部分。第一部分解释如何在应用程序中使用现有流。第二部分解释如何创建新类型的流。

Node.js 内有四种基本流类型：

• Writable：可以向其写入数据的流（例如，fs.createWriteStream()）。
• Readable：可以从中读取数据的流（例如，fs.createReadStream()）。
• Duplex：既是 Readable 又是 Writable 的流（例如，net.Socket）。
• Transform：可以在写入和读取数据时修改或转换数据的 Duplex 流（例如，zlib.createDeflate()）。

此外，该模块还包括实用函数 stream.duplexPair()、stream.pipeline()、stream.finished()、stream.Readable.from() 和 stream.addAbortSignal()。

stream/promises API 提供了一组替代的流异步实用函数，它们返回 Promise 对象而不是使用回调。该 API 可通过 require('node:stream/promises') 或 require('node:stream').promises 访问。

stream.pipeline(streams, options?): void

stream.pipeline(source, ...transforms?, destination, options?): void

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
  await pipeline(
    fs.createReadStream('archive.tar'),
    zlib.createGzip(),
    fs.createWriteStream('archive.tar.gz'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);

要使用 AbortSignal，将其作为最后一个参数传递给选项对象。 当信号被中止时，底层管道将被调用 destroy，并带有 AbortError。

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');
const zlib = require('node:zlib');

async function run() {
  const ac = new AbortController();
  const signal = ac.signal;

  setImmediate(() => ac.abort());
  await pipeline(
    fs.createReadStream('archive.tar'),
    zlib.createGzip(),
    fs.createWriteStream('archive.tar.gz'),
    { signal },
  );
}

run().catch(console.error); // AbortError

pipeline API 也支持异步生成器：

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');

async function run() {
  await pipeline(
    fs.createReadStream('lowercase.txt'),
    async function* (source, { signal }) {
      source.setEncoding('utf8');  // 使用字符串而不是 `Buffer`。
      for await (const chunk of source) {
        yield await processChunk(chunk, { signal });
      }
    },
    fs.createWriteStream('uppercase.txt'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);

记得处理传递给异步生成器的 signal 参数。 特别是在异步生成器是管道的源（即第一个参数）的情况下，否则管道将永远不会完成。

const { pipeline } = require('node:stream/promises');
const fs = require('node:fs');

async function run() {
  await pipeline(
    async function* ({ signal }) {
      await someLongRunningfn({ signal });
      yield 'asd';
    },
    fs.createWriteStream('uppercase.txt'),
  );
  console.log('Pipeline succeeded.');
}

run().catch(console.error);

pipeline API 提供了 回调版本：

stream.finished(stream, options?): void

const { finished } = require('node:stream/promises');
const fs = require('node:fs');

const rs = fs.createReadStream('archive.tar');

async function run() {
  await finished(rs);
  console.log('Stream is done reading.');
}

run().catch(console.error);
rs.resume(); // 排空流。

finished API 还提供了 回调版本。

stream.finished() 在返回的 promise 被履行或拒绝后，会留下悬空的事件监听器（特别是 'error'、'end'、'finish' 和 'close'）。 这样做的原因是，意外的 'error' 事件（由于不正确的流实现）不会导致意外的崩溃。 如果这是不需要的行为，则应将 options.cleanup 设置为 true：

await finished(rs, { cleanup: true });

由 Node.js API 创建的所有流仅对字符串、{Buffer}、<TypedArray> 和 <DataView> 对象进行操作：

• Strings 和 Buffers 是与流一起使用的最常见类型。
• TypedArray 和 DataView 允许你使用 Int32Array 或 Uint8Array 等类型处理二进制数据。当你将 TypedArray 或 DataView 写入流时，Node.js 会处理原始字节。

然而，流实现有可能与其他类型的 JavaScript 值一起工作（null 除外，它在流中有特殊用途）。 此类流被认为是在“对象模式”下操作。

流实例在创建时使用 objectMode 选项切换到对象模式。尝试将现有流切换到对象模式是不安全的。

Writable 和 Readable 流都会将数据存储在一个内部缓冲区中。

潜在缓冲的数据量取决于传递给流构造函数的 highWaterMark 选项。对于普通流，highWaterMark 选项指定 字节总数。对于以对象模式操作的流，highWaterMark 指定对象总数。对于操作字符串（但不解码）的流，highWaterMark 指定 UTF-16 代码单元总数。

当实现调用 stream.push(chunk) 时，数据会在 Readable 流中缓冲。如果流的消费者不调用 stream.read()，数据将停留在内部队列中直到被消费。

一旦内部读取缓冲区的总大小达到 highWaterMark 指定的阈值，流将暂时停止从底层资源读取数据，直到当前缓冲的数据可以被消费（即，流将停止调用用于填充读取缓冲区的内部 readable._read() 方法）。

当反复调用 writable.write(chunk) 方法时，数据会在 Writable 流中缓冲。当内部写入缓冲区的总大小低于 highWaterMark 设置的阈值时，对 writable.write() 的调用将返回 true。一旦内部缓冲区的大小达到或超过 highWaterMark，将返回 false。

stream API 的一个关键目标，特别是 stream.pipe() 方法，是将数据缓冲限制在可接受的水平，以便不同速度的源和目标不会压倒可用内存。

highWaterMark 选项是一个阈值，而不是限制：它规定了流在停止请求更多数据之前缓冲的数据量。它通常不强制执行严格的内存限制。特定的流实现可以选择执行更严格的限制，但这是可选的。

因为 Duplex 和 Transform 流既是 Readable 又是 Writable，所以它们各自维护 两个 独立的内部缓冲区用于读取和写入，允许每一侧独立操作，同时保持适当且高效的数据流。例如，net.Socket 实例是 Duplex 流，其 Readable 侧允许消费 从 套接字接收的数据，而其 Writable 侧允许写入数据 到 套接字。因为写入套接字的数据速率可能比接收数据的速率快或慢，所以每一侧都应独立操作（和缓冲）。

内部缓冲的机制是内部实现细节，可能随时更改。但是，对于某些高级实现，可以使用 writable.writableBuffer 或 readable.readableBuffer 检索内部缓冲区。不鼓励使用这些未记录的属性。

几乎所有的 Node.js 应用程序，无论多么简单，都以某种方式使用流。以下是在实现 HTTP 服务器的 Node.js 应用程序中使用流的示例：

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

const server = http.createServer((req, res) => {
  // `req` 是一个 http.IncomingMessage，它是一个可读流。
  // `res` 是一个 http.ServerResponse，它是一个可写流。

  let body = '';
  // 将数据获取为 utf8 字符串。
  // 如果未设置编码，将接收到 Buffer 对象。
  req.setEncoding('utf8');

  // 可读流一旦添加了监听器就会发出 'data' 事件。
  req.on('data', (chunk) => {
    body += chunk;
  });

  // 'end' 事件表示整个 body 已接收完毕。
  req.on('end', () => {
    try {
      const data = JSON.parse(body);
      // 向用户写回一些有趣的内容：
      res.write(typeof data);
      res.end();
    } catch (er) {
      //  uh oh! 错误的 json！
      res.statusCode = 400;
      return res.end(`error: ${er.message}`);
    }
  });
});

server.listen(1337);

// $ curl localhost:1337 -d "{}"
// object
// $ curl localhost:1337 -d "\"foo\""
// string
// $ curl localhost:1337 -d "not json"
// error: Unexpected token 'o', "not json" is not valid JSON

Writable 流（例如示例中的 res）暴露了 write() 和 end() 等方法，用于将数据写入流。

Readable 流使用 EventEmitter API 在数据可供从流中读取时通知应用程序代码。可以通过多种方式从流中读取可用数据。

Writable 和 Readable 流都以各种方式使用 EventEmitter API 来通信流的当前状态。

Duplex 和 Transform 流既是 Writable 也是 Readable。

向流写入数据或从流消费数据的应用程序不需要直接实现流接口，并且通常没有理由调用 require('node:stream')。

希望实现新类型流的开发者应参考 [流实现者 API][] 部分。

可写流是对数据写入_目的地_的抽象。

Writable 流的示例包括：

• [HTTP 请求，在客户端][]
• [HTTP 响应，在服务器端][]
• [fs 写入流][]
• zlib 流
• crypto 流
• [TCP 套接字][]
• [子进程 stdin][]
• process.stdout，process.stderr

其中一些示例实际上是实现了 Writable 接口的 Duplex 流。

所有 Writable 流都实现了 stream.Writable 类定义的接口。

虽然 Writable 流的具体实例可能在各方面有所不同，但所有 Writable 流都遵循与以下示例中说明相同的基本使用模式：

const myStream = getWritableStreamSomehow();
myStream.write('some data');
myStream.write('some more data');
myStream.end('done writing data');

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

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

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

// 向提供的可写流写入数据一百万次。
// 注意背压。
function writeOneMillionTimes(writer, data, encoding, callback) {
  let i = 1000000;
  write();
  function write() {
    let ok = true;
    do {
      i--;
      if (i === 0) {
        // 最后一次！
        writer.write(data, encoding, callback);
      } else {
        // 看看我们是应该继续，还是等待。
        // 不要传递回调，因为我们还没完成。
        ok = writer.write(data, encoding);
      }
    } while (i > 0 && ok);
    if (i > 0) {
      // 不得不提前停止！
      // 一旦排空，再写入一些。
      writer.once('drain', write);
    }
  }
}

• 类型：<Error>

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

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

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

在调用 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');

当在可读流上调用 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);

当在 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);

writable.cork(): void

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?): void

销毁流。可选地发出 '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

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

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

• 类型：<boolean>

在发出 'close' 事件后为 true。

• 类型：<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

writable.end(chunk?, encoding?, callback?): void

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

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

// 写入 'hello, ' 然后以 'world!' 结束。
const fs = require('node:fs');
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// 现在不允许再写入！

writable.setDefaultEncoding(encoding): void

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

writable.uncork(): void

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

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

stream.cork();
stream.write('some ');
stream.cork();
stream.write('data ');
process.nextTick(() => {
  stream.uncork();
  // 直到第二次调用 uncork() 数据才会被刷新。
  stream.uncork();
});

另见：writable.cork()。

• 类型：<boolean>

如果安全调用 writable.write() 则为 true，这意味着流未被销毁、出错或结束。

• 类型：<boolean>

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

• 类型：<boolean>

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

• 类型：<integer>

需要调用 writable.uncork() 的次数才能完全打开流。

• 类型：<Error>

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

• 类型：<boolean>

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

• 类型：<number>

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

• 类型：<number>

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

• 类型：<boolean>

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

• 类型：<boolean>

给定 Writable 流的 objectMode 属性的 getter。

writable[Symbol.asyncDispose](): void

使用 AbortError 调用 writable.destroy() 并返回一个在流完成时 fulfilled 的 promise。

writable.write(chunk, encoding?, callback?): void

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 流默认是暂停的，直到它们被管道传输或添加了 'data' 或 'readable' 事件处理程序。

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

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

// 在做任何其他写入之前等待 cb 被调用。
write('hello', () => {
  console.log('Write completed, do more writes now.');
});

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

可读流是对数据消费_源_的抽象。

Readable 流的示例包括：

• HTTP 响应，在客户端
• HTTP 请求，在服务器端
• [fs 读取流][]
• zlib 流
• crypto 流
• [TCP 套接字][]
• [子进程 stdout 和 stderr][]
• process.stdin

所有 Readable 流都实现了 stream.Readable 类定义的接口。

Readable 流有效地在两种模式之一中运行：流动模式和暂停模式。这些模式与 对象模式 分开。 Readable 流可以处于对象模式或不是，无论它是处于流动模式还是暂停模式。

• 在流动模式下，数据从底层系统自动读取，并通过 EventEmitter 接口使用事件尽可能快地提供给应用程序。

• 在暂停模式下，必须显式调用 stream.read() 方法从流中读取数据块。

所有 Readable 流都以暂停模式开始，但可以通过以下方式之一切换到流动模式：

• 添加 'data' 事件处理程序。
• 调用 stream.resume() 方法。
• 调用 stream.pipe() 方法将数据发送到 Writable。

Readable 可以使用以下方式之一切换回暂停模式：

• 如果没有管道目的地，通过调用 stream.pause() 方法。
• 如果有管道目的地，通过移除所有管道目的地。 可以通过调用 stream.unpipe() 方法移除多个管道目的地。

需要记住的重要概念是，Readable 在提供消费或忽略该数据的机制之前不会生成数据。如果消费机制被禁用或移除，Readable 将_尝试_ 停止生成数据。

出于向后兼容性的原因，移除 'data' 事件处理程序不会自动暂停流。此外，如果有管道目的地，则调用 stream.pause() 不能保证流在这些目的地排空并要求更多数据后_保持_ 暂停。

如果 Readable 切换到流动模式并且没有可用的消费者来处理数据，则该数据将丢失。例如，当调用 readable.resume() 方法而没有监听器附加到 'data' 事件时，或者当 'data' 事件处理程序从流中移除时，可能会发生这种情况。

添加 'readable' 事件处理程序会自动使流停止流动，并且必须通过 readable.read() 消费数据。如果 'readable' 事件处理程序被移除，那么如果有 'data' 事件处理程序，流将再次开始流动。

Readable 流的“两种模式”操作是对 Readable 流实现内部发生的更复杂状态管理的简化抽象。

具体来说，在任何给定时间点，每个 Readable 都处于以下三种可能状态之一：

• readable.readableFlowing === null
• readable.readableFlowing === false
• readable.readableFlowing === true

当 readable.readableFlowing 为 null 时，未提供消费流数据的机制。因此，流不会生成数据。在此状态下，附加 'data' 事件的监听器，调用 readable.pipe() 方法，或调用 readable.resume() 方法会将 readable.readableFlowing 切换为 true，导致 Readable 开始在生成数据时主动发出事件。

调用 readable.pause()、readable.unpipe() 或接收背压会导致 readable.readableFlowing 设置为 false，暂时停止事件流动但_不_ 停止数据生成。在此状态下，附加 'data' 事件的监听器不会将 readable.readableFlowing 切换为 true。

const { PassThrough, Writable } = require('node:stream');
const pass = new PassThrough();
const writable = new Writable();

pass.pipe(writable);
pass.unpipe(writable);
// readableFlowing 现在是 false。

pass.on('data', (chunk) => { console.log(chunk.toString()); });
// readableFlowing 仍然是 false。
pass.write('ok');  // 不会发出 'data'。
pass.resume();     // 必须调用以使流发出 'data'。
// readableFlowing 现在是 true。

当 readable.readableFlowing 为 false 时，数据可能会在流的内部缓冲区中积累。

Readable 流 API across 多个 Node.js 版本演变，并提供多种消费流数据的方法。通常，开发者应选择_一种_ 消费数据的方法，并且_绝不应该_ 使用多种方法从单个流消费数据。具体来说，结合使用 on('data')、on('readable')、pipe() 或异步迭代器可能会导致不直观的行为。

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

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

每当流将数据块的所有权 relinquishing 给消费者时，就会发出 '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.`);
});

当流中没有更多数据可供消费时，会发出 '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.');
});

• 类型：<Error>

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

监听器回调将传递单个 Error 对象。

当调用 stream.pause() 且 readableFlowing 不为 false 时，会发出 'pause' 事件。

当有数据可供从流中读取时，会发出 'readable' 事件，直到配置的高水位标记（state.highWaterMark）。实际上，它表示流在缓冲区中有新信息。如果此缓冲区内有可用数据，可以调用 stream.read() 来检索该数据。此外，当到达流末尾时，也可能发出 'readable' 事件。

const readable = getReadableStreamSomehow();
readable.on('readable', function() {
  // 现在有一些数据可供读取。
  let data;

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

如果已到达流末尾，调用 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');
});

运行此脚本的输出是：

$ node test.js
readable: null
end

在某些情况下，附加 'readable' 事件的监听器会导致一些数据被读入内部缓冲区。

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

如果同时使用 'readable' 和 'data'，'readable' 在控制流方面优先，即只有当调用 stream.read() 时才会发出 'data'。 readableFlowing 属性将变为 false。 如果当 'readable' 被移除时有 'data' 监听器，流将开始流动，即无需调用 .resume() 就会发出 'data' 事件。

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

readable.destroy(error?): void

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

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

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

• 类型：<boolean>

在发出 'close' 事件后为 true。

• 类型：<boolean>

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

readable.isPaused(): void

• 返回：<boolean>

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

const readable = new stream.Readable();

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

readable.pause(): void

• 返回：<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);
});

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

readable.pipe(destination, options?): void

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

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

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// 来自 readable 的所有数据都进入 'file.txt'。
readable.pipe(writable);

可以将多个 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);

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

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

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

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

read(size?): void

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

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

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

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

readable.read() 方法应仅在处于暂停模式的 Readable 流上调用。在流动模式下，readable.read() 会自动调用，直到内部缓冲区完全排空。

const readable = getReadableStreamSomehow();

// 当数据被缓冲时，'readable' 可能会被触发多次
readable.on('readable', () => {
  let chunk;
  console.log('Stream is readable (new data received in buffer)');
  // 使用循环确保我们读取所有当前可用数据
  while (null !== (chunk = readable.read())) {
    console.log(`Read ${chunk.length} bytes of data...`);
  }
});

// 当没有更多可用数据时，'end' 将被触发一次
readable.on('end', () => {
  console.log('Reached end of stream.');
});

每次调用 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('');
});

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

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

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

• 类型：<boolean>

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

• 类型：<boolean>

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

• 类型：<boolean>

返回是否已发出 'data'。

• 类型：<null> | <string>

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

• 类型：<boolean>

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

• 类型：<Error>

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

• 类型：<boolean>

此属性反映 [三种状态][] 部分中描述的 Readable 流的当前状态。

• 类型：<number>

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

• 类型：<number>

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

• 类型：<boolean>

给定 Readable 流的 objectMode 属性的 getter。

readable.resume(): void

• 返回：<this>

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

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

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

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

readable.setEncoding(encoding): void

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);
});

readable.unpipe(destination?): void

readable.unpipe() 方法分离之前使用 stream.pipe() 方法附加的 Writable 流。

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

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

const fs = require('node:fs');
const readable = getReadableStreamSomehow();
const writable = fs.createWriteStream('file.txt');
// 来自 readable 的所有数据都进入 'file.txt'，
// 但仅持续第一秒。
readable.pipe(writable);
setTimeout(() => {
  console.log('Stop writing to file.txt.');
  readable.unpipe(writable);
  console.log('Manually close the file stream.');
  writable.end();
}, 1000);

readable.unshift(chunk, encoding?): void

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

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

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

使用 stream.unshift() 的开发者通常应考虑改用 Transform 流。有关更多信息，请参阅 [流实现者 API][] 部分。

// 拉出由 \n\n 分隔的头部。
// 如果我们得到太多，使用 unshift()。
// 使用 (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')) {
        // 找到头部边界。
        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);
        // 在 unshifting 之前移除 'readable' 监听器。
        stream.removeListener('readable', onReadable);
        if (buf.length)
          stream.unshift(buf);
        // 现在可以从流中读取消息正文。
        callback(null, header, stream);
        return;
      }
      // 仍在读取头部。
      header += str;
    }
  }
}

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

readable.wrap(stream): void

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

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

当使用发出 'data' 事件并具有仅为 advisory 的 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(); // 等等。
});

readable[Symbol.asyncIterator](): void

• 返回：<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);

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

readable[Symbol.asyncDispose](): void

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

readable.compose(stream, options?): void

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(['text passed through', 'composed stream']).compose(splitToWords);
const words = await wordsStream.toArray();

console.log(words); // 打印 ['text', 'passed', 'through', 'composed', 'stream']

readable.compose(s) 等同于 stream.compose(readable, s)。

此方法还允许提供 <AbortSignal>，当 abort 时将销毁组合的流。

有关更多信息，请参阅 stream.compose(...streams)。

readable.iterator(options?): void

此方法创建的迭代器让用户可以选择在 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); // 将打印 2 然后是 3
  }

  console.log(readable.destroyed); // true，流已被完全消费
}

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

readable.map(fn, options?): void

稳定性：1 - 实验性

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

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

// 使用同步映射器。
for await (const chunk of Readable.from([1, 2, 3, 4]).map((x) => x * 2)) {
  console.log(chunk); // 2, 4, 6, 8
}
// 使用异步映射器，最多同时进行 2 个查询。
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); // 记录 resolver.resolve4 的 DNS 结果。
}

readable.filter(fn, options?): void

稳定性：1 - 实验性

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

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

// 使用同步谓词。
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
  console.log(chunk); // 3, 4
}
// 使用异步谓词，最多同时进行 2 个查询。
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) {
  // 记录解析后的 dns 记录中 TTL 超过 60 秒的域名。
  console.log(result);
}

readable.forEach(fn, options?): void

稳定性：1 - 实验性

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

此方法与 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';

// 使用同步谓词。
for await (const chunk of Readable.from([1, 2, 3, 4]).filter((x) => x > 2)) {
  console.log(chunk); // 3, 4
}
// 使用异步谓词，最多同时进行 2 个查询。
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) => {
  // 记录结果，类似于 `for await (const result of dnsResults)`
  console.log(result);
});
console.log('done'); // 流已完成

readable.toArray(options?): void

稳定性：1 - 实验性

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

由于此方法将整个流读入内存，它否定了流的好处。它旨在用于互操作性和便利性，而不是作为消费流的主要方式。

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

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

const resolver = new Resolver();

// 使用 .map 并发进行 dns 查询，并使用 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();

readable.some(fn, options?): void

稳定性：1 - 实验性

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

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

// 使用同步谓词。
await Readable.from([1, 2, 3, 4]).some((x) => x > 2); // true
await Readable.from([1, 2, 3, 4]).some((x) => x < 0); // false

// 使用异步谓词，最多同时进行 2 个文件检查。
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); // 如果列表中的任何文件大于 1MB，则为 `true`
console.log('done'); // 流已完成