流（stream）是编程中处理 流式数据 的 抽象 接口。

Node.js中 stream 模块用于构建实现了 流 接口的对象。

例如：HTTP服务器的请求 和 process.stdout 都是流的实例。

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

流的使用方法如下：

const stream = require('stream');

流是用来比喻数据传输的一种形式，数据传输的起点就是流的源头，数据传输的终点就是流的终点。例如在网页发起一个 HTTP 请求，浏览器就是流的源头，服务器就是流的终点。等服务器处理完请求，返回响应时，服务器就变成了流的源头，浏览器变成了流的终点。

数据从一端连续不断地传输到另一端，就像水一样从一端流到另一端，所以用流来比喻数据的传输形式。只不过计算机中的流传输的是数据（字节），而不是水。

在 Node.js 中，stream 模块提供了用于实现流接口的 API。但是很多内置模块都提供了关于流的 API，所以通常不需要显式的调用 stream 模块来使用流。

pipe() 将可写流绑定到可读流，使其自动切换到流动模式并将其所有数据推送到绑定的可写流。 数据流将被自动管理，以便目标可写流不会被更快的可读流漫过。也就是说，pipe() 将数据缓冲限制在可接受的水平，以便不同速度的来源和目标不会压倒可用内存。

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

Readable: 可读流，可以从中读取数据的流（例如，fs.createReadStream()）。

Writable: 可写流，可以写入数据的流（例如，fs.createWriteStream()）。

Duplex: 双工流，Readable 和 Writable 的流（例如，net.Socket）。

Transform: 可以在写入和读取数据时修改或转换数据的 Duplex 流（例如，zlib.createDeflate()）。

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

可能缓冲的数据量取决于传给流的构造函数的 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 端允许将数据写入套接字。 因为数据可能以比接收数据更快或更慢的速度写入套接字，所以每一端都应该独立于另一端进行操作（和缓冲）。

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

Readable 流的示例包括：

  客户端上的 HTTP 响应
  服务器上的 HTTP 请求
  文件系统读取流
  压缩流
  加密流
  TCP 套接字
  子进程的标准输出和标准错误
  process.stdin

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

Readable 流以两种模式之一有效地运行：流动和暂停。在流动模式下，数据会自动从底层系统读取，并通过 EventEmitter 接口使用事件尽快提供给应用程序。在暂停模式下，必须显式调用 stream.read() 方法以从流中读取数据块。

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

  添加 data 事件句柄。
  调用 stream.resume() 方法。
  调用 stream.pipe() 方法将数据发送到 Writable。
  Readable 可以使用以下方法之一切换回暂停模式：

如果没有管道目标，则通过调用 stream.pause() 方法。 如果有管道目标，则删除所有管道目标。 可以通过调用 stream.unpipe() 方法删除多个管道目标。

4种基本的流类型

可读流（Writble): 可写入数据的流，如： fs.createWriteStream()。

可写流（Readable): 可读取数据的流，如： fs.createReadStream()。

双工流（Duplex): 可读又可写的流，如 net.Socket

转换流（Transform): 在读写过程中可以修改或转换数据的Duplex流，如 zlib.createDeflate().

当创建流的时候，可以使用 objectMode 选项把流实例 切换到对象模式，将已存在的流切换到对象模型是不安全的。

可写流 和 可读流都会在内部的缓冲区中存储数据，可以分别使用 writable.writableBuffer 或 readable.readableBuffer 来获取。

可缓冲的数据大小取决于传入流构造函数的 highWaterMark 选项。 对于普通的流， highWaterMark 指定流子节的总数。

对于对象模式的流，highWaterMark 指定流对象的总数。

当调用 stream.push(chunk) 时，数据会被缓冲在可读流中。如果流的 “消费者” 没有调用 stream.read()，则数据会保留在 内部队列 中直到被消费。

一旦内部的可读缓冲的总大小达到 highWaterMark 指定的 阀值时，流会暂时停止从底层资源读取数据，直到当前缓冲的数据被消费（也就是说，流会停止调用内部的用于填充可读缓冲的 readable._read()

当调用 writable.write(chunk) 时，数据会被缓冲在可写流中。

当内部的可写缓冲的总大小 小于 hightWaterMark 设置的阀值时， 调用 writable.write() 会返回 true。一旦内部缓冲的大小达到或超过 highWaterMark 时，则会返回 false。

为了保存内存，某些 Stream API 会限制缓冲区，可以避免读写速度不一致引起的内存的崩溃。

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

Node.js可读流是对提供数据的来源的一种抽象。所有可读流都实现了stream.Readable类定义的接口。可读流常见的例子包括客户端的HTTP响应，服务器的HTTP请求，fs的读取流，zlib流，crypto流，TCP socket，子进程stdout与stderr, process.stdin.

stream.Readable 类定义了如下事件。

1. close事件

close事件在流或其底层资源（如文件描述符）被关闭时触发。表明不会再触发其他事件，也不会再发生操作。

不是所有可读流都会触发close事件。如果使用emitClose选项创建可读流，则它将始终发出close事件。

2. data事件

data事件是在流将数据块传送给“消费者”后触发。对于非对象模式的流，数据块可以是字符串或Buffer。对于对象模式的流，数据块可以是除了null的任何JavaScript值。

当调用 readable.pipe(), readable.resume() 或绑定监听器到 data 事件时，流会转换到流动模式。当调用 readable.read()且有数据块返回时，也会触发data事件。

如果使用 readable.setEncoding() 为流指定 🈯️ 了默认的字符编码，则监听器回调传入的数据为字符串，否则传入的数据为Buffer。

3. end事件

end 事件在流中没有数据可供消费时触发。

end 事件只有在你数据被完全消费掉后才会触发。要想触发该事件，可以将流转换到流动模式，或反复调用 stream.read() 直到数据被消费完。

const readable = getReadableStreamSomehow();

readable.on('data', (chunk) => {
 console.log(`接收到${chunk.length}个字节的数据`);
});

readable.on('end', () => {
 console.log(`已经没有数据`);
});

4. error事件

error事件通常是在当流因底层内部出错而不能产生数据，或推送无效的数据块时触发。

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

5. pause事件

调用 stream.pause() 并且 readsFlowing 不为 false时，会发出pause事件。

6. readable事件

readable事件在当流中有数据可供读取时触发。

const readable = getReadableStreamSomehow();

readable.on('readable', function() {
 // 有数据可读取
 let data;
 while(data = this.read()) {
  console.log(data);
 }
});

当到达流数据的尽头时，readable事件也会触发，但是在end事件之前触发。

readable 事件表明流有新的动态，要么有新的数据，要么到达流的尽头。对于前者，stream.read() 会返回可用的数据。对于后者，stream.read() 会返回 null 。 例如，下面的例子中，foo.txt是一个空文件。

const fs = require('fs');
const rr = fs.createReadStream('data.txt');
rr.on('readable', () => {
 console.log(`读取的数据${rr.read()}`);
});
rr.on('end', () => {
 console.log('结束');
});

通常情况下， readable.pipe() 和 data事件的机制比 readable 事件更容易理解。处理 readable 事件可能造成吞吐量升高。

如果同时使用 readable 事件 和 data 事件，则 readable 事件会优先控制流，也就是说，当调用 stream.read() 时才会触发 data 事件。 readableFlowing 属性会变成 false。

当移除 readable 事件时，如果存在data事件监听器，则流会开始流动，也就是说，无须调用stream.resume() 也会触发data事件。

7. resume事件

调用 stream.resume()并且readsFlowing不为true时，将发出resume事件。

stream.Readable类包含以下常用的方法：

1. destroy

readable.destroy([error]) 方法用于销毁流，并触发 error 事件 和 close 事件. 调用后，可读流将释放所有的内部资源，且忽视后续的push()调用。

实现流时不应该重写这个方法，而是重写readable._destroy()

2. isPaused

readable.isPaused()方法用于返回可读流当前的操纵状态。主要用于 readable.pipe() 底层的机制，大多数情况下无须直接使用该方法

const readable = new stream.Readable();

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

3. pause 与 resume

readable.pause() 方法使流动模式的流停止触发 data 事件，并切换到流动模式。任何可用的数据都会保留在内部缓存中。

相对的, readable.resume() 将被暂停⏸️的可读流恢复触发data事件，并将流切换到流动模式。

const fs = require('fs');

const readable = fs.createReadStream('data.txt');
readable.on('data', (chunk) => {
 console.log(`接收到${chunk.length}字节的数据`);
 // 暂停
 readable.pause();

 console.log('暂停一秒');
 setTimeout(() => {
  console.log('数据重新开始流动');
  // 继续
  readable.resume();
 },1000);
});

readable.on('end', () => {
 console.log('结束');
});

4. pipe

readable.pipe(destination[, options]) 方法用于绑定 可写流 到 可读流，将 可读流 自动切换到流动模式，并将可读流的所有数据推送到绑定的可写流。

数据流会被自动管理，所以即使可读流更快，目标可写流也不会超负荷。

将可读流的所有数据通过管道推送到 write-data.txt 文件

const fs = require('fs')

const readable = fs.createReadStream('data.txt');

const writable = fs.createWriteStream('write-data.txt');

// readable 的所有数据都推送到 write-data.txt
readable.pipe(writable);

可以在单个可读流上绑定多个可写流。

readable.pipe() // 会返回目标流的引用，这样就可以对流进行链式的管道操作。

const fs = require('fs');
const zlib = require('zlib');

const readable = fs.createReadStream('data.txt');
const gzip = zlib.createGzip();
const writable2 = fs.createWriteStream('write-data.txt.gz');

// 在单个可读流上绑定多个可写流
readable.pipe(gzip).pipe(writable2);

默认情况下，当来源可读流触发 end 事件时，目标可写流 也会调用 stream.end() 结束写入。

若要禁用这种默认行为，end选项应设为 false ，这样目标流就回保持打开。

reader.pipe(writer, { end: false });
reader.on('end', () => {
 writer.end('结束');
});

如果可读流发生错误🙅🙅‍♂️，目标可写流不会自动关闭，需要手动关闭所有流以避免内存泄漏。

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

5. read

readable.read([size]) 方法用于从内部缓冲拉取并返回数据。其中，size 指定要读取的数据的字节数。如果没有指定 size 参数，则返回内部缓冲中的所有数据。

该方法如果没有可读的数据，则返回null。默认情况下，readable.read() 返回的数据是 Buffer对象，除非使用 readable.setEncoding() 指定字符编码或流 处于对象模式。

如果可读的数据不足size 个字节，则返回内部缓冲剩余的数据，如果流已经结束则 返回null

readable.read() 应该只对 处于暂停模式的可读流调用。在流动模式中， readable.read() 会自动调用直到内部缓冲的数据完全耗尽。

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

end事件触发后再调用 stream.read([size]) 会返回null，不会抛出错误。

const fs = require('fs');
const readable = fs.createReadStream('data.txt');

// 设置字符编码
readable.setEncoding('utf-8');

// 读取数据
readable.on('readable', () => {
 let chunk;
 while (null !== (chunk = readable.read(10))) {
  console.log(`接收到${chunk.length}字节的数据`);
  console.log(`接收到的数据是：${chunk}`);
 }
})

readable.on('end', () =>{
 console.log('结束')
})

使用 readable.read() 处理数据时，while 循环是必需的。只有在 readable.read() 返回 null 之后，才会发生 readable 事件；readable.setEncoding() 用于设置字符编码。默认情况下没有设置字符编码，流数据返回的是Buffer对象。如果设置了字符编码，则流数据返回指定编码的字符串。

调用 readable.setEncoding('utf-8')会将数据解析为UTF-8数据，并返回字符串。

调用 readable.setEncoding('hex')则会将数据编码成 十六进制字符串。

6. readable.unpipe([destination])

解绑之前使用stream.pipe()绑定的可写流。

如果没有指定目标可写流，则解绑所有管道，如果指定了流目标可写流，但它没有建立管道，则不起作用。

const fs = require('fs');

const readable = fs.createReadStream('data.txt');

const writable = fs.createWriteStream('write-data.txt');

// readable 的所有数据都推送到 write-data.txt
readable.pipe(writable);

setTimeout(() => {
 console.log('停止写入数据');
 readable.unpipe(writable);
 console.log('收到关闭文件流');
 writable.end();
}, 3);

可读流中提供了异步迭代器的使用

const fs = require('fs');

async function print(readable) {
 readable.setEncoding('utf8');
 let data = '';
 // 迭代器
 for await (const k of readable) {
  data += k;
 }
 console.log(data);
}

print(fs.createReadStream('file')).catch(console.log);

如果循环以break 或 throw终止，则流将被销毁。

迭代流将完全销毁流，并以大小等于 highWaterMark选项的块读取流。

如果文件的数据少于64kb，则数据将位于单个块中，因为没有为 fs.createReadStream() 提供 highWaterMark 选项。

可读流 运作于 流动模式 flowing 或 暂停模式 paused 两种模式之一。

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

在暂停模式中，必须显式调用 stream.read() 读取数据块。

所有可读流都开始于暂停模式，可以通过以下方式切换到流动模式。

添加 data 事件句柄

调用 stream.resume()

调用 stream.pipe()

可读流可以通过以下方式 切换回 暂停模式

• 如果没有管道目标，则调用 stream.pause()
• 如果有管道目标，则 移除所有管道目标。 调用 stream.unpipe() 可以移除多个管道目标。

只有提供了消费或忽略数据的机制后，可读流才会产生数据。如果消费的机制被禁用或移除，则可读流会停止产生数据。

为了向后兼容，移除data事件句柄不会自动地暂停流。如果有管道目标，一旦目标变为 drain 状态并请求接收数据时，则调用 stream.pause() 也不能保证流会保持暂停模式。

如果可读流切换到流动模式，且没有可用的“消费者”来处理数据，则数据将会丢失。例如，当调用 readable.resume() 时，没有监听 data 事件 或 data 事件句柄已移除。

添加 readable 事件句柄会使流自动停止流动，并通过 readable.read() 消费数据。 如果 readable 事件句柄被移除，且存在 data 事件句柄，则流会再次开始流动。

可写流是对数据要被写入的目的地的一种抽象。所有可写流都实现了 stream.Writable 类定义的接口。 可写流常见的例子包括客户端的 HTTP 请求、服务器的HTTP响应、fs的写入流、zlib流、 crypto 流，TCP socket、子进程 stdin、process.stdout、 process.stderr。

const myStream = getWritableStreamSomehow();
myStream.write('一些数据');
myStream.write('更多数据');
myStream.end('完成写入数据');

1. close 事件

当流及其任何底层资源已关闭时，将发出close事件。该事件表明不会发出更多事件，也不会进一步计算。

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

2. drain 事件

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

3. error 事件

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

发出 error 事件时，流不会关闭。

4. finish 事件

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

const fs = require('fs');

const writable = fs.createWriteStream('write-data.txt');

for (let i = 0; i<10; i++) {
 writable.write(`写入#${i}!\n`);
}

writable.end('写入结尾\n');
writable.on('finish', () => {
 console.log('写入已完成');
})

5. pipe 事件

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

6. unpipe 事件

当在可读流上调用 stream.unpipe() 时触发。

当可读流通过管道流向可写流发生错误时，也会触发 unpipe 事件。

1. cork

writable.cork() 方法用于强制把所有写入的数据都缓冲到内存中。当调用 stream.uncork() 或 stream.end() 时，缓冲的数据才会被输出。

当写入大量小块数据到流时，内部缓冲可能失败，从而导致性能下降，writable.cork() 主要用于避免这种情况。对于这种情况，实现了 writable._writev() 的流可以用更优的方式对写入的数据进行缓冲。

2. destroy

writable.destroy([error]) 方法用于销毁流。在调用该方法之后，可写流已经结束，随后对 write() 或 end() 的调用都将导致 ERR_STREAM_DESTROYED 错误。

如果数据在关闭之前应该刷新，则应使用 end() 方法而不是 destroy() 方法，或者在销毁流之前等待 drain 事件。实现者不应该重写此方法，而是 实现 writable._destroy()

3. end

调用 writable.end([chunk][, encoding][, callback]) 方法表示不再将数据 写入 Writable。 该方法的参数如下。

• chunk | | | : 要写入的可选数据。对于不在对象模式下运行的流，块必须是字符串、Buffer或Uint8Array。对于对象模式流，块可以是除null之外的任务JavaScript值。
• encoding 如果设置了编码，则chunk是一个字符串。
• callback : 流完成时的可选回调。

调用writable.end()方法表示不再将数据写入 Writable 。可选的块和编码参数允许在关闭流之前立即写入最后一个额外的数据块。如果提供，则附加可选回调函数作为 finish 事件的监听器。示例如下：

const fs = require('fs');

const writable = fs.createWriteStream('write-data.txt');

for(let i = 0; i<10; i++) {
 writable.write(`写入#${i}!\n`);
}

writable.end('写入结尾\n');
writable.on('finish', () => {
 console.log('写入已完成');
})

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

4. setDefaultEncoding

writable.setDefaultEncoding(encoding) 为可写流设置默认的编码。

5. uncork

writable.uncork() 方法用于将调用 stream.cork() 后缓冲的所有数据输出到目标。

当使用 writable.cork() 和 writable.uncork() 来管理流的写入缓冲时，建议使用 process.nextTick() 来延迟调用 writable.uncork()。

通过这种方式，可以对单个 Node.js 事件循环中调用的所有 writable.write() 进行批处理。

stream.cork();
stream.write('一些');
stream.write('数据');
process.nextTick(() => stream.uncork());

如果一个流上多次调用 writable.cork()， 则必须调用同样次数的 writable.uncork() 才能输出缓冲的数据。

writable.write(chunk, [, encoding][, callback]) 写入数据到流，并在数据被完全处理之后调用 callback.

如果发生错误，则 callback 可能被调用也可能不被调用。为了可靠地检测错误，可以为 error 事件添加监听器。该方法的参数如下：

• chunk | | | : 要 写入的数据。对于非对象模式的流， chunk必须时字符串 、 Buffer 或 Unit8Array 。

对于对象模式的流， chunk可以除 null 外的是 任何 JavaScript 值。

• encoding : 如果 chunk 是字符串，则指定 字符编码。

• callback : 当数据块 被输出到目标后的回调函数。

writable.write() 写入数据到流，并在数据被完全处理之后调用 callback。如果发生错误，则 callback 可能被调用也可能不被调用。为了可靠地检测错误，可以为 error事件添加监听器。

在接收了 chunk 后，如果内部的 缓冲小于创建流时配置的 highWaterMark，则返回 true。如果返回 false，则应该停止向流写入数据，直到 drain 事件被触发。

当流还未被排空时，调用 write() 会缓冲 chunk ，并返回 false。一旦所有当前缓冲的数据块都被排空了（被操作系统接收并传输），则触发 drain 事件。建议一旦 write()返回false，则不再写入任何数据块，直到 达到最大内存占用，这时它会无条件中止，甚至在它中止之前，高内存占用将会导致垃圾回收器的性能变差 和 RSS 变高（即使内存不再需要，通常也不会被释放回系统）。

如果远程的另一端没有读取数据，TCP 的 socket 可能永远也不会排空，所以写入到一个不会排空的 socket 可能会导致产生远程可利用的漏洞。

对于 Transform，写入数据到一个不会排空的流尤其成问题，因为 Transform 流默认会被暂停，直到它们被 pipe 或者 添加了data 或 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('完成写入')
});

双工流（Duplex) 是同时实现了 Readable 和 Writable 接口的流。 双工流的例子包括 TCP socket、 zlib 流， crypto 流。

转换流(Transform) 是一种双工流，但它的输出与输入是 相关联的。与双工流一样，转换流也同时实现了 Readable 和 Writable 接口。转换流的例子包括 zlib流和crypto流。