NodeJS模块研究 - Stream

概述

构建复杂程序的时候，通常会将系统拆解成若干功能，这些功能的之间的接口遵循一定的规范，以实现组合连接，共同完成复杂任务。例如管道运算符 | 。

在 nodejs 中，实现各种功能，总避免不了和“数据”打交道，这些数据可能是 Buffer、字符串、数组等等。但当处理大量数据的时候，如何保证程序的稳健性？如何不让内存爆掉呢？nodejs 提供了 stream 模块，来让开发者更优雅地处理数据。这需要开发者理解“流”的含义，遵循“流”的相关规范，来进行“流”编程。

相较于其他的模块系列，需要理解的概念偏多。本文主要从以下几个方面深入 stream 模块：

• 什么是“流”？
• 流有哪几种类型？
• 内部缓冲的作用？
• 流动模式 vs 暂停模式
• 背压问题
  • 如何产生的？
  • 如何解决背压问题？
• 如何定制流
  • 实现可写流
  • 实现可读流
  • 实现双工和转换流

什么是“流”？

流是数据的集合。但它不一定是一次性全部读入内存的，这和程序中的变量不同。举个例子，一个 100GB 的文件，可以通过 fs.createReadStream() 来将文件二进制一点点读入，每次读入的“一点点二进制”，就是流。

乍一看，这样做好像并没有什么显而易见的好处。一点点读入的流，还要一点点处理。但是，处理难度的提高换来的是程序性能提升。如果通过fs.readFile()一次性将 100GB 大文件读入内存，那么可能会发生以下几个问题：

• 内存错误。本机内存不够，或者超过了 nodejs 的内存限制。
• 程序性能下降。过高的内存占用可能会导致频繁触发 GC，影响其他程序的运行。

借助流，每次仅读入一小部分的数据，待这部分数据被“消费”后，再读入新的数据。转换思路，不一定必须将要用到的数据一次性全部装入内存。

流有哪几种类型？

stream 提供了 4 种基本的流类型：

• Writable：可写入数据流。
• Readable：可读取数据流。
• Duplex：双工流，可读又可写。例如：net.Socket
• Transform：转换流，它是 Duplex 流的一种。它用于在读写过程中，加工数据。例如：zlib

内部缓冲的作用？

在文档开头部分，就用一节专门提到了“缓冲”。可读流和可写流都会在内部缓存器存储数据，Duplex 和 Transform 也在内部维护了缓存器。在开发者基于流开发时，可以通过传递highWaterMark参数，来修改默认缓冲的大小。

理解缓冲的作用，就要先搞明白缓冲的处理流程：

• 可写流中：
  1. 调用 write()向流中写入数据
  2. 数据进入可写流缓冲
  3. 数据等待被消费
• 可读流中：
  1. 调用 read()向流中读取数据
  2. 数据进入可读缓存
  3. 数据等待被消费

在这个过程中，如果可写/可读缓冲中的数据总大小超过了 highWaterMark：

• 可写流的 write()会返回 false，直到缓冲可以继续写入，触发drain事件
• 可读流会停止从底层资源读取数据

有了内部缓冲机制，就可以限制流的读写速度，防止内存被压垮，解决背压问题。

流动模式 vs 暂停模式

这是可读流的两种模式。可读流开始时是处于暂停模式，之后根据监听的事件、调用的 api，来进行两种模式的切换。文档上写的很详细，但是也会让初学者感到困扰。这里直接从编码风格触发，来学习这两种模式。

编码风格一：监听 readable 事件 + read()

如果可读流监听了 readable 事件，那么处于暂停模式。readable 事件回调触发的条件：

• 有新的数据
• 流到达尽头

由于处于暂停模式，因此在事件回调函数中，需要使用 read()来读取数据。

const fs = require("fs");
const rs = fs.createReadStream("./package.json");
rs.on("readable", () => {
    let chunk = null;
    // 在read()时，while循环是必须的
    // 这里的read()可以理解成消费的行为
    while ((chunk = rs.read()) !== null) {
        process.stdout.write(chunk);
    }
})
    .on("end", () => {
        console.log("读取结束");
    })
    .on("error", err => console.error(err));

编码风格二（官方推荐）：监听 data 事件 + pause() + resume()

如果可读流监听 data 事件，那么会自动切换至流动模式，事件回调中可以得到数据。不需要调用 read()，换句话说 read()只在暂停模式下调用。

const fs = require("fs");
const rs = fs.createReadStream("./package.json");
rs.on("data", chunk => {
    process.stdout.write(chunk);
})
    .on("end", () => {
        console.log("读取结束");
    })
    .on("error", err => console.error(err));

如果配合可写流进行更复杂操作，发生了背压问题，没有可用的消费者来处理数据，则数据将会丢失。为了方便理解，可以认为它是自动调用 read()进行消费。此时使用 pause()来切换到暂停模式，待消费者可以处理时，再调用 resume()恢复流动模式。

背压问题

如何产生的？

当处理数据的时候，如果数据生产者产生数据的速度 > 数据消费者处理数据的速度，那么由于速度差异没被消耗完的数据就会持续堆积下来，这个现象就是背压（也称积压）。

它会导致资源过度占用，内存耗尽，也会增加 GC 的负担。

如何解决背压问题？

结合前面对缓冲的讲解，在向可写流写入数据的时候，如果超过可写缓存，应该暂停数据读取，等待缓存中数据被消耗完毕后，再继续流动可读流。

下面是一个基于 stream，复制大文件的函数：

function copyBigFile(src, dest, callback) {
    if (typeof callback !== "function") {
        throw TypeError("Callback should be function");
    }
    const rs = fs.createReadStream(src);
    const ws = fs.createWriteStream(dest);
    rs.on("data", chunk => {
        if (!ws.write(chunk)) {
            rs.pause();
        }
    }).on("end", () => {
        ws.end();
        callback();
    });
    ws.on("drain", () => rs.resume());
}

更好的解决方案是，使用可读流上的pipe()函数，或者 stream 模块的pipeline()函数。

pipe 函数实现了以下几个功能：

• 不断从来源可读流中获得一个指定长度的数据。
• 将获取到的数据写入目标可写流。
• 平衡读取和写入速度，防止读取速度大大超过写入速度时，出现大量滞留数据。

用它来处理背压问题非常简单。前面复制文件的函数改写为：

function copyBigFile(src, dest, callback) {
    if (typeof callback !== "function") {
        throw TypeError("Callback should be function");
    }
    const rs = fs.createReadStream(src);
    const ws = fs.createWriteStream(dest);
    rs.pipe(ws).on("close", callback);
}

pipeline 函数是 pipe 的语法糖，它的参数可以是多个流+一个回调函数：

const { pipeline } = require("stream");
function copyBigFile(src, dest, callback) {
    if (typeof callback !== "function") {
        throw TypeError("Callback should be function");
    }
    const rs = fs.createReadStream(src);
    const ws = fs.createWriteStream(dest);
    pipeline(rs, ws, err => callback(err));
}

如何定制流？

在实现自己的类库的时候，可以借助流来处理大容量数据。nodejs 提供给开发者 API，来定制 4 种类型的流。

实现可写流

继承 Writable 类，需要重写_write()方法。并且在实现中必须调用callback()函数，无论成功失败。

下面实现了一个解码器，并且能将解码结果处理成小写：

const { Writable } = require("stream");
class LowerWritable extends Writable {
    constructor(options) {
        super(options);
        this._data = "";
    }
    // 可以去掉 callback 调用看看效果
    _write(chunk, encoding, callback) {
        try {
            chunk = encoding === "buffer" ? chunk.toString() : chunk;
            this._data += chunk.toLocaleLowerCase();
            callback(null);
        } catch (err) {
            callback(err);
        }
    }
    // 流关闭前被调用
    _final(callback) {
        this._data += "\\nfinish";
        callback(null);
    }
}
const lw = new LowerWritable();
lw.write("HELLO ");
lw.write(Buffer.from("WORLD!"));
lw.end();
lw.on("finish", () => {
    console.log(lw._data);
});

实现可读流

继承 Readable 类，需要重写_read 方法。内部通过 push 方法来推入数据。

为了方便演示，先实现一个产生数据的类，它继承自 EventEmitter ：

const EventEmitter = require("events");
class DataSource extends EventEmitter {
    constructor(str) {
        super();
        this._str = str;
        this._offset = 0;
    }
    read() {
        if (this._offset < this._str.length) {
            // 每次读取一个字符
            this.emit("data", this._str[this._offset]);
            this._offset += 1;
        } else {
            // 读取完成
            this.emit("end");
        }
    }
    stop() {
        this._offset -= 1;
    }
}

最后实现一个可读流，并在读取过程中，将字符转成小写：

const { Readable } = require("stream");
class LowerReadable extends Readable {
    constructor(src) {
        super();
        this._src = src;
        this._src.on("data", chunk => {
            // 一个字符，一个字符的push，而不是一次性push多个字符
            if (!this.push(chunk.toLocaleLowerCase())) {
                this._src.stop();
            }
            // 这里没必要显示地调用 DataSource.read()，否则爆栈
            // 因为 this._read() 会被一直调用，从而触发 data 事件
            // this._src.read();
        });
        this._src.on("end", () => this.push(null));
    }
    _read(size) {
        // 从数据源读取按照读取一个字符长度的数据
        this._src.read();
    }
}
// test code
const dataSrc = new DataSource("HELLO WORLD!");
const lowerStream = new LowerReadable(dataSrc);
lowerStream.pipe(process.stdout);

实现双工和转换流

双工流的实现就是既要重写可写流的_write 方法，也要重写可读流的_read 方法。

转换流的实现是不同的，它需要重写_transform 方法，来接收并产生输出。下面是一段利用转换流来给输出添加标志符的代码：

const { Transform } = require("stream");
const prompt = Buffer.from("<<< ");
class InputLowerTransform extends Transform {
    constructor(options) {
        super(options);
    }
    // 将可写端写入的数据变换后添加到可读端
    _transform(chunk, encode, cb) {
        chunk = Buffer.concat([prompt, chunk], prompt.length + chunk.length);
        // 调用push方法将变换后的数据添加到可读端
        this.push(chunk);
        cb();
    }
}
const transStream = new InputLowerTransform();
process.stdin.pipe(transStream).pipe(process.stdout);

效果如下：

参考连接

• Nodejs 文档
• 数据流中的积压问题
• 通过源码解析 Node.js 中导流（pipe）的实现
• Node.js Stream - 基础篇