NodeJS模块研究 - stream

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

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

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

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

🔍 关注公众号“心谭博客” / 👉 查看原文: xxoo521.com / 更多前端与算法的系列文章

什么是“流”?

流是数据的集合。但它不一定是一次性全部读入内存的,这和程序中的变量不同。举个例子,一个 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);

效果如下:

参考连接

扫描下方二维码,回复「博客文章」获取解锁验证码

步骤:打开微信 => 扫描二维码 => 关注「心谭博客」公众号 => 发送「博客文章」即可解锁博客全部文章

输入验证码:
Your browser is out-of-date!

Update your browser to view this website correctly. Update my browser now

×