nodejs的流总结

Posted 2021-02-20 mengff

流是可读的、可写的，或可读可写的。所有的流都是EventEmitter的实例。
stream 的基本概念，即 source -> 管道 -> dest 这个模型图。

流的常见的来源方式主要有三种：

1. 从控制台输入
2. http 请求中的 request
3. 读取文件

流的常见输出方式主要有三种：

1. 输出到控制台
2. http 请求中的 response
3. 写入文件

1. 流的缓冲(buffer)

可写流和可读流都有自己的缓冲，缓冲大小由构造参数highWaterMark决定。

当调用 stream.push(chunk) 时，数据会被缓冲在可读流中。 如果流的消费者没有调用 stream.read()，则数据会保留在内部队列中直到被消费。
一旦内部的可读缓冲的总大小达到 highWaterMark 指定的阈值时，流会暂时停止从底层资源读取数据，直到当前缓冲的数据被消费 （也就是说，流会停止调用内部的用于填充可读缓冲的 readable._read()）。

当调用 writable.write(chunk) 时，数据会被缓冲在可写流中。 当内部的可写缓冲的总大小小于 highWaterMark 设置的阈值时，调用 writable.write() 会返回 true。
一旦内部缓冲的大小达到或超过 highWaterMark 时，则会返回 false。

而 Duplex 和 Transform 都是可读又可写的，他们都有两个buffer，读取和写入的独立的。

2. 流的对象模式

nodejs的流默认运作在字符串、Buffer或Unit8Array上。若流使用了其他的javascript值，这些流会以“对象模式”运行。
创建流时，用objectMod哦可以把流切换到对象模式。

3. nodejs流的种类

nodejs共有4种流

1. 可读流, readableStream
2. 可写流，writeableStream
3. 双工流，duplexStream
4. 转换流，transformStream

3.1. 可读流(readableStream)

可读流有两种模式：

1. 流动模式(flowing) 自动读取数据
2. 暂停模式(pasused) 显示调用stream.read读取数据

可读流默认处于暂停模式，可以通过以下方法切换为流动模式：

1. 添加data事件，on(‘data‘,functoin(){...})
2. 调用stream.resume()方法
3. 调用stream.pipe()方法

可读流的例子：

客户端的 HTTP 响应
服务器的 HTTP 请求
fs 的读取流
zlib 流
crypto 流
TCP socket
子进程 stdout 与 stderr
process.stdin

3.2 可写流(writeableStream)

可写流有背压问题：当写入速度过快，超过其buffer的阈值，只能暂停写入，或者将写入丢弃。

可写流的 stream.wirte 方法，在背压时候，会返回false，这个时候，不能再写入。

可写流的 drain 事件，在背压解除时触发，在 drain 事件中，可以恢复写入。

可写流的例子：

客户端的 HTTP 请求
服务器的 HTTP 响应
fs 的写入流
zlib 流
crypto 流
TCP socket
子进程 stdin
process.stdout、process.stderr

3.3 双工流(Duplex)

双工流（Duplex）是同时实现了 Readable 和 Writable 接口的流。

Duplex 流的例子包括：

TCP socket

3.4 转换流(Transform)

转换流（Transform）是一种 Duplex 流，但它的输出与输入是相关联的。 与 Duplex 流一样， Transform 流也同时实现了 Readable 和 Writable 接口。

Transform 流的例子包括：

zlib 流
crypto 流

4. 流的管道(pipe)方法

管道方法，有效的解决了背压问题，并且简化了流的调用，十分实用。

readable.pipe(writable);

readable 通过 pipe（管道）传输给 writable

pipe的源码如下：

Readable.prototype.pipe = function(writable, options) {
  this.on(‘data‘, (chunk) => {
    let ok = writable.write(chunk);
    if(!ok) this.pause();// 背压，暂停
    
  });
  writable.on(‘drain‘, () => {
    // 恢复
    this.resume();
  });
  // 告诉 writable 有流要导入
  writable.emit(‘pipe‘, this);
  // 支持链式调用
  return writable;
};