源代码: lib/stream.js
流是用于在 Node.js 中处理流数据的抽象接口。
stream 模块提供了用于实现流接口的 API。
Node.js 提供了许多流对象。
例如，对 HTTP 服务器的请求和 process.stdout 都是流的实例。
流可以是可读的、可写的、或两者兼而有之。
所有的流都是 EventEmitter 的实例。
要访问 stream 模块：
const stream = require('stream');stream 模块对于创建新类型的流实例很有用。
通常不需要使用 stream 模块来消费流。
本文档包含两个主要章节和第三章节的注意事项。 第一章节描述了如何在应用程序中使用现有的流。 第二章节描述了如何创建新类型的流。
Node.js 中有四种基本的流类型：
Writable: 可以写入数据的流（例如，fs.createWriteStream()）。Readable: 可以从中读取数据的流（例如，fs.createReadStream()）。Duplex: Readable 和 Writable 的流（例如，net.Socket）。Transform: 可以在写入和读取数据时修改或转换数据的 Duplex 流（例如，zlib.createDeflate()）。此外，此模块还包括实用函数 stream.pipeline()、stream.finished()、stream.Readable.from() 和 stream.addAbortSignal()。
stream/promises API 为返回 Promise 对象（而不是使用回调）的流提供了一组替代的异步实用函数。
API 可通过 require('stream/promises') 或 require('stream').promises 访问。
Node.js API 创建的所有流都只对字符串和 Buffer（或 Uint8Array）对象进行操作。
但是，流的实现可以使用其他类型的 JavaScript 值（除了 null，它在流中具有特殊用途）。
这样的流被认为是在"对象模式"下运行的。
流的实例在创建流时使用 objectMode 选项切换到对象模式。
尝试将现有的流切换到对象模式是不安全的。
Writable 和 Readable 流都将数据存储在内部缓冲区中。
可能缓冲的数据量取决于传给流的构造函数的 highWaterMark 选项。
对于普通的流，highWaterMark 选项指定字节的总数。
对于在对象模式下操作的流，highWaterMark 指定对象的总数。
当实现调用 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('http');
const server = http.createServer((req, res) => {
// `req` 是 http.IncomingMessage，它是可读流。
// `res` 是 http.ServerResponse，它是可写流。
let body = '';
// 以 utf8 字符串形式获取数据。
// 如果未设置编码，则将接收缓冲区对象。
req.setEncoding('utf8');
// 一旦添加了监听器，则可读流就会触发 'data' 事件。
req.on('data', (chunk) => {
body += chunk;
});
// 'end' 事件表示已经接收到整个正文。
req.on('end', () => {
try {
const data = JSON.parse(body);
// 给用户回写一些有趣的东西：
res.write(typeof data);
res.end();
} catch (er) {
// 哦哦！糟糕的 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 in JSON at position 1Writable 流（例如示例中的 res）暴露了用于将数据写入流的方法，例如 write() 和 end()。
当数据可从流中读取时，Readable 流使用 EventEmitter API 来通知应用程序代码。
可以通过多种方式从流中读取可用数据。
Writable 和 Readable 流都以各种方式使用 EventEmitter API 来传达流的当前状态。
Duplex 和 Transform 流都是 Writable 和 Readable。
向流中写入数据或从流中消费数据的应用程序不需要直接实现流的接口，并且通常没有理由调用 require('stream')。
希望实现新类型的流的开发者应参考流实现者的 API 章节。
可写流是数据写入目标的抽象。
Writable 流的示例包括：
其中一些示例实际上是实现 Writable 接口的 Duplex 流。
所有的 Writable 流都实现了 stream.Writable 类定义的接口。
虽然 Writable 流的特定实例可能以各种方式不同，但所有的 Writable 流都遵循相同的基本使用模式，如下例所示：
const myStream = getWritableStreamSomehow();
myStream.write('some data');
myStream.write('some more data');
myStream.end('done writing data');stream.Writable 类#'close' 事件#当流及其任何底层资源（例如文件描述符）已关闭时，则会触发 'close' 事件。
该事件表明将不再触发更多事件，并且不会发生进一步的计算。
如果 Writable 流是使用 emitClose 选项创建的，则始终会触发 'close' 事件。
'drain' 事件#如果对 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' 事件）。
'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');'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);'unpipe' 事件#src <stream.Readable> 取消管道此可写流的源流当在 Readable 流上调用 stream.unpipe() 方法时，则会触发 'unpipe' 事件，从其目标集合中删除此 Writable。
当 Readable 流管道进入它时，如果此 Writable 流触发错误，则这也会触发。
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()#writable.cork() 方法强制所有写入的数据都缓存在内存中。
当调用 stream.uncork() 或 stream.end() 方法时，缓冲的数据将被刷新。
writable.cork() 的主要目的是适应将几个小块快速连续写入流的情况。
writable.cork() 不是立即将它们转发到底层目标，而是缓冲所有块，直到 writable.uncork() 被调用，如果存在，writable.uncork() 会将它们全部传给 writable._writev()。
这可以防止在等待处理第一个小块时正在缓冲数据的行头阻塞情况。
但是，在不实现 writable._writev() 的情况下使用 writable.cork() 可能会对吞吐量产生不利影响。
另请参阅：writable.uncork()、writable._writev()。
writable.destroy([error])#销毁流
可选地触发 'error' 事件，并且触发 'close' 事件（除非 emitClose 设置为 false）。
在此调用之后，则可写流已结束，随后对 write() 或 end() 的调用将导致 ERR_STREAM_DESTROYED 错误。
这是销毁流的破坏性和直接的方式。
先前对 write() 的调用可能没有排空，并且可能触发 ERR_STREAM_DESTROYED 错误。
如果数据应该在关闭之前刷新，或者在销毁流之前等待 'drain' 事件，则使用 end() 而不是销毁。
const { Writable } = require('stream');
const myStream = new Writable();
const fooErr = new Error('foo error');
myStream.destroy(fooErr);
myStream.on('error', (fooErr) => console.error(fooErr.message)); // foo errorconst { Writable } = require('stream');
const myStream = new Writable();
myStream.destroy();
myStream.on('error', function wontHappen() {});const { Writable } = require('stream');
const myStream = new Writable();
myStream.destroy();
myStream.write('foo', (error) => console.error(error.code));
// ERR_STREAM_DESTROYED一旦 destroy() 被调用，任何进一步的调用都将是空操作，除了来自 _destroy() 的其他错误可能不会作为 'error' 触发。
实现者不应覆盖此方法，而应实现 writable._destroy()。
writable.destroyed#在调用 writable.destroy() 之后是 true。
const { Writable } = require('stream');
const myStream = new Writable();
console.log(myStream.destroyed); // false
myStream.destroy();
console.log(myStream.destroyed); // truewritable.end([chunk[, encoding]][, callback])#chunk <string> | <Buffer> | <Uint8Array> | <any> 可选的要写入的数据。
对于不在对象模式下操作的流，chunk 必须是字符串、Buffer 或 Uint8Array。
对于对象模式的流，chunk 可以是除 null 之外的任何 JavaScript 值。encoding <string> chunk 为字符串时的编码callback <Function> 流结束时的回调。调用 writable.end() 方法表示不再有数据写入 Writable。
可选的 chunk 和 encoding 参数允许在关闭流之前立即写入最后一个额外的数据块。
在调用 stream.end() 之后调用 stream.write() 方法将引发错误。
// 写入 'hello, ' 然后以 'world!' 结尾。
const fs = require('fs');
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// 现在不允许写入更多！writable.setDefaultEncoding(encoding)#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());如果在一个流上多次调用 writable.cork() 方法，则必须调用相同数量的 writable.uncork() 调用来刷新缓冲的数据。
stream.cork();
stream.write('some ');
stream.cork();
stream.write('data ');
process.nextTick(() => {
stream.uncork();
// 在第二次调用 uncork() 之前不会刷新数据。
stream.uncork();
});另见: writable.cork()。
writable.writable#如果调用 writable.write() 是安全的，则为 true，这意味着流没有被销毁、出错或结束。
writable.writableEnded#在调用 writable.end() 之后是 true。
此属性不指示数据是否已刷新，为此则使用 writable.writableFinished 代替。
writable.writableCorked#需要调用 writable.uncork() 以完全解开流的次数。
writable.writableFinished#在触发 'finish' 事件之前立即设置为 true。
writable.writableHighWaterMark#返回创建此 Writable 时传入的 highWaterMark 的值。
writable.writableLength#此属性包含队列中准备写入的字节数（或对象数）。
该值提供有关 highWaterMark 状态的内省数据。
writable.writableNeedDrain#如果流的缓冲区已满并且流将触发 'drain'，则为 true。
writable.writableObjectMode#给定 Writable 流的属性 objectMode 的获取器。
writable.write(chunk[, encoding][, callback])#chunk <string> | <Buffer> | <Uint8Array> | <any> 可选的要写入的数据。
对于不在对象模式下操作的流，chunk 必须是字符串、Buffer 或 Uint8Array。
对于对象模式的流，chunk 可以是除 null 之外的任何 JavaScript 值。encoding <string> | <null> 如果 chunk 为字符串，则为编码。 默认值: 'utf8'callback <Function> 当刷新此数据块时的回调。'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 流是默认暂停，直到它们被管道传输、或添加 'data' 或 'readable' 事件句柄。
如果要写入的数据可以按需生成或获取，则建议将逻辑封装成 Readable 并且使用 stream.pipe()。
但是，如果首选调用 write()，则可以使用 'drain' 事件遵守背压并避免内存问题：
function write(data, cb) {
if (!stream.write(data)) {
stream.once('drain', cb);
} else {
process.nextTick(cb);
}
}
// 在执行任何其他写入之前等待回调被调用。
write('hello', () => {
console.log('Write completed, do more writes now.');
});对象模式下的 Writable 流将始终忽略 encoding 参数。
可读流是对被消费的数据的来源的抽象。
Readable 流的示例包括：
所有的 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 === nullreadable.readableFlowing === falsereadable.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('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()); });
pass.write('ok'); // 不会触发 'data'。
pass.resume(); // 必须调用才能使流触发 'data'。虽然 readable.readableFlowing 是 false，但数据可能会在流的内部缓冲区中累积。
Readable 流的 API 跨越多个 Node.js 版本的演进，并提供了多种消费流数据的方法。
一般情况下，开发者应该选择其中一种消费数据的方式，切忌使用多种方式消费单一流中的数据。
具体来说，使用 on('data')、on('readable')、pipe() 或异步迭代器的组合可能会导致不直观的行为。
建议大多数用户使用 readable.pipe() 方法，因为它已被实施以提供使用流数据的最简单方法。
需要对数据传输和生成进行更细粒度控制的开发者可以使用 EventEmitter 和 readable.on('readable')/readable.read() 或 readable.pause()/readable.resume() API。
stream.Readable 类#'close' 事件#当流及其任何底层资源（例如文件描述符）已关闭时，则会触发 'close' 事件。
该事件表明将不再触发更多事件，并且不会发生进一步的计算。
如果 Readable 流是使用 emitClose 选项创建的，则始终会触发 'close' 事件。
'data' 事件#chunk <Buffer> | <string> | <any> 数据块。
对于不在对象模式下操作的流，块将是字符串或 Buffer。
对于处于对象模式的流，块可以是除 null 之外的任何 JavaScript 值。每当流将数据块的所有权移交给消费者时，则会触发 'data' 事件。
每当通过调用 readable.pipe()、readable.resume()、或通过将监听器回调绑定到 'dat