深入理解 Node Stream 内部机制

文章来自公众号：小胡子哥

相信很多人对 Node 的 Stream 已经不陌生了，不论是请求流、响应流、文件流还是 socket 流，这些流的底层都是使用 stream 模块封装的，甚至我们平时用的最多的 console.log 打印日志也使用了它，不信你打开 Node runtime 的源码，看看 lib/console.js：

function write(ignoreErrors, stream, string, errorhandler) {
  // ...
  stream.once('error', noop);
  stream.write(string, errorhandler);
  //...
}
Console.prototype.log = function log(...args) {
  write(this._ignoreErrors,
        this._stdout,
        `${util.format.apply(null, args)}\n`,
        this._stdoutErrorHandler);
};

Stream 模块做了很多事情，了解了 Stream，那么 Node 中其他很多模块理解起来就顺畅多了。

本文代码和图片可以在这里取用：https://github.com/barretlee/dive-into-node-stream。

stream 模块

如果你了解 生产者和消费者问题 的解法，那理解 stream 就基本没有压力了，它不仅仅是资料的起点和落点，还包含了一系列状态控制，可以说一个 stream 就是一个状态管理单元。了解内部机制的最佳方式除了看 Node 官方文档，还可以去看看 Node 的 源码：

• lib/module.js

• lib/_stream_readable.js

• lib/_stream_writable.js

• lib/_stream_tranform.js

• lib/_stream_duplex.js

把 Readable 和 Writable 看明白，Tranform 和 Duplex 就不难理解了。

Readable Stream

Readable Stream 存在两种模式，一种是叫做 FlowingMode，流动模式，在 Stream 上绑定 ondata 方法就会自动触发这个模式，比如：

const readable = getReadableStreamSomehow();
readable.on('data', (chunk) => {
  console.log(`Received ${chunk.length} bytes of data.`);
});

这个模式的流程图如下：

资源的数据流并不是直接流向消费者，而是先 push 到缓存池，缓存池有一个水位标记 highWatermark，超过这个标记阈值，push 的时候会返回 false，什么场景下会出现这种情况呢？

• 消费者主动执行了 .pause()

• 消费速度比数据 push 到缓存池的生产速度慢

有个专有名词来形成这种情况，叫做「背压」，Writable Stream 也存在类似的情况。

流动模式，这个名词还是很形象的，缓存池就像一个水桶，消费者通过管口接水，同时，资源池就像一个水泵，不断地往水桶中泵水，而 highWaterMark 是水桶的浮标，达到阈值就停止蓄水。下面是一个简单的 Demo：

const Readable = require('stream').Readable;
// Stream 实现
class MyReadable extends Readable {
  constructor(dataSource, options) {
    super(options);
    this.dataSource = dataSource;
  }
  // 继承了 Readable 的类必须实现这个函数
  // 触发系统底层对流的读取
  _read() {
    const data = this.dataSource.<
                        
声明：所有来源为“聚合数据”的内容信息，未经本网许可，不得转载！如对内容有异议或投诉，请与我们联系。邮箱：marketing@think-land.com