Node中的 流

[libin1991]

一股清流

清明时节雨纷纷，果然每逢清明是会下雨的。在这个雨夹雪，不方便外出的日子，宅在家里一起来相互学习分享吧！不然还能怎样呢！哈哈

友情提示：本文可能会涉及到一些Api的内容，会很乏味，很枯燥，很没劲

But我们后面的精彩也会超乎你的想象，因为我们要手写实现一下，不亲自上马怎么知道马跑的有多慢呢，Hurry Up Go Go Go！

用过node的朋友们都知道流的作用非常之厉害，可读可写，无所不能。

相比于fs模块，流更适用于读取一个大文件，一次性读取会占用大量内存，效率很低，而流是将数据分割成段，会一段一段的读取，效率会很高。说了一堆，先上概念，一起看看它是谁

概念

• 流是一组有序的，有起点和终点的字节数据传输手段
• 它不关心文件的整体内容，只关注是否从文件中读到了数据，以及读到数据之后的处理
• 流是一个抽象接口，被 Node 中的很多对象所实现。比如HTTP 服务器request和response对象都是流

node中很多内容都应用到了流，比如http模块的req就是可读流，res是可写流，而socket是可读可写流，看起来屌屌的，那么我们今天就都不讲他们，只来讲一下可读流和可写流这对兄弟

可读流和可写流对文件的操作用的也是fs模块

那么让我们从可读流讲起，先来看下都有哪些方法(Api)

可读流

首先要会用，重点在会用

创建可读流

const fs = require('fs');   // 引入fs核心模块

// fs.createReadStream(path, options)
// 返回的是一个可读流对象
let rs = fs.createReadStream('1.txt', {
    flags: 'r',         // 文件的读取操作，默认是'r':读取
    encoding: 'utf8',   // 设置编码格式，默认是null, null代表的是buffer
    autoClose: true,    // 读取完毕后自动关闭
    highWaterMark: 3,   // 默认是读取64k    64 * 1024字节
    start: 0,
    end: 3              // 文件结束位置索引，和正常的截取slice有所不同，包前又包后(包括自己结束的位置)
});

// 默认情况下，不会将文件中的内容输出
// 内部会先创建一个buffer先读取3字节

// 1.txt文件内容为 123456789

以上代码写了如何创建可读流，看起来要记那么多options项，真是头疼，其实一般情况下，配置项是不用我们写的，这下大家满足了吧

知道了如何创建，我们就看看rs这个可读流对象上有哪些监听事件啊

监听data事件

可读流这种模式它默认情况下是非流动模式(暂停模式)，它什么也不做，就在这等着

大家知道流是基于事件的，所以我们可以去监听事件，监听了data事件的话，就可以将非流动模式转换为流动模式

// 流动模式会疯狂的触发data事件，直到读取完毕
// 根据上面设置的highWaterMark一次读3个字节
rs.on('data', data => { // 非流动模式 -> 流动模式
    console.log(data);  // 触发2次data事件， 分别打出123和4  从0到3共4个(包括末尾)
});

// 题外话：
// 监听data事件的时候，如果没有设置编码格式，data返回的是buffer类型
// so我们可以为data设置encoding为utf8
rs.setEncoding('utf8');     // 等同于options里的encoding: 'utf8'

当我们把想要读取的内容都读完后，还可以监听一个end事件，去判断何时读完

监听end事件

rs.on('end', () => {
    console.log('完毕了'); 
});

// 此时除了会打印data事件里的123, 4之外还会打印 完毕了
// 如下表示:
// 123
// 4
// 完毕了

除了data和end两个事件之外，可读流中还可以监听到error、open以及close事件，由于用处没有前两位大，就委屈一下放在一起写吧

监听error/open/close事件

// error
rs.on('error', err => {
    console.log(err);
});
// open
rs.on('open', () => {
    console.log('文件打开了');
});
// close
rs.on('close', () => {
    console.log('文件关闭了');
});

// 根据上面监听data、end事件，下面打印的内容是
/*
*   文件打开了
    123
    4
    end
    文件关闭了
* */

各类监听事件都知道怎么写了，最后再看两个方法，他们是pause和resume，暂停和恢复触发data

暂停和恢复

// pause
rs.on('data', data => { 
    console.log(data);  // 只会读取一次就暂停了，此时只读到了123
    rs.pause();     // 暂停读取，会暂停data事件触发
});
// resume
setInterval(() => {
    rs.resume();    // 恢复data事件, 继续读取，变为流动模式
                    // 恢复data事件后，还会调用rs.pause，要想再继续触发，把setTimeout换成setInterval持续触发
}, 3000);
// 打印如下:
/*
*   文件打开了
    123
    4   // 隔了3秒后打印
    end
    文件关闭了
* */

说完了可读流的用法，让我们再接再厉(不得不)去看下它的兄弟可写流吧，毕竟对于作为世界第一大群体的程序猿来说，总得有个从入门到精通(放弃)的深层次提升嘛！加了个油的，各位走起。

可写流

废话不多说，上来就是干

创建可写流

const fs = require('fs');
// fs.createWriteStream(path, options);
const ws = fs.createWriteStream('2.txt', {
    flags: 'w',         // 文件的操作, 'w'写入文件，不存在则创建
    mode: 0o666,
    autoClose: true,
    highWaterMark: 3,   // 默认写是16k
    encoding: 'utf8'
});

可写流就有两个方法，分别是write和end方法，直接看下如何使用

write方法

// ws.write(chunk, encoding(可选), callback);
// 写入的chunk数据必须是字符串或者buffer
let flag = ws.write('1', 'utf8', () => {});     // 异步的方法 有返回值

console.log(flag);  // true
flag = ws.write('22', 'utf8', () => {});    
console.log(flag);  // false    超过了highWaterMark的3个字节，不能再写了
flag = ws.write('3', 'utf8', () => {}); 
console.log(flag);  // false

// 2.txt -> 写入了 1223

flag标识符表示的并不是是否写入，而是能否继续写，true为可以继续写入。但是返回false，也不会丢失，还会写到文件内的

接下来再介绍下end方法

end方法

// 可以传入chunk值
ws.end('完毕');   // 当写完后 就不能再写了

// 此时2.txt -> 写入了 1223完毕

讲完了write和end方法，可写流还有一个on监听事件，它可以监听drain(抽干)事件

监听drain事件

// drain方法
// 抽干方法 当都写入后触发drain事件
ws.on('drain', () => {
    console.log('已经抽干了');
});

重头戏来了

• 前面罗里吧嗦都在写如何使用，Api着实让大家看的昏昏欲睡了。
• 但是各位观众，现在才是最最值得高兴的时刻，对于流的操作，我们不仅仅要会用，还应该简单的去实现一下。
• 这样才能满足我们庞大的求知欲并且get到新技能，老样子，直接上代码，从代码中去深入分析一番
• 如果读的疲惫了，那就歇歇吧，当一个佛系青年，看空一切也是一种痛的领悟啊

实现可读流

先来个🌰

// demo.js
const ReadStream = require('./ReadStream'); // 引入实现的可读流

const rs = new ReadStream('1.txt', {
    flags: 'r',
    // encoding: 'utf8',
    autoClose: true,
    highWaterMark: 3,
    start: 0,
    end: 4
});

rs.on('data', data => {
    console.log(data);
    rs.pause();
});

rs.on('end', () => {
   console.log('end');
});

setTimeout(() => {
    rs.resume();
}, 2000);

前方高能，开启敲击模式，如果还不知道node中的buffer和events的话，千万别捉急。大家都是一条船上的人，我会在之后的文章里给大家分享，且先暂且继续看下去啊！坚持住，兄弟姐妹们！

创建ReadStream类

// ReadStream.js
const fs = require('fs');
const EventEmitter = require('events');  // 需要依赖事件发射
// 这里用ES6提供的class写法，大家也一起来看看是怎么写的吧

class ReadStream extends EventEmitter {
    constructor(path, options) {    // 需要传入path和options配置项
        super();    // 继承
        this.path = path;
        // 参照上面new出的实例，我们开始写
        this.flags = options.flags || 'r';  // 文件打开的操作，默认是'r'读取
        this.encoding = options.encoding || null;   // 读取文件编码格式，null为buffer类型
        this.autoClose = options.autoClose || true;
        this.highWaterMark = options.highWaterMark || 64 * 1024;  // 默认是读取64k
        this.start = options.start || 0;
        this.end = options.end;

        this.flowing = null;   // null表示非流动模式
        // 要建立一个buffer,这个buffer就是一次要读多少内容
        // Buffer.alloc(length)  是通过长度来创建buffer，这里每次读取创建highWaterMark个
        this.buffer = Buffer.alloc(this.highWaterMark);  
        this.pos = this.start;  // 记录读取的位置

        this.open();    // 打开文件，获取fd文件描述符

        // 看是否监听了data事件，如果监听了，就变成流动模式
        this.on('newListener', (eventName, callback) => {
            if (eventName === 'data') {   // 相当于用户监听了data事件
                this.flowing = true;  // 此时监听了data会疯狂的触发
                this.read();    // 监听了，就去读，要干脆，别犹豫
            }
        });
    }
}

module.exports = ReadStream;    // 导出

写到这里我们已经创建好了ReadStream类，在该类中我们继承了EventEmitter事件发射的方法

其中我们写了open和read这两个方法，从字面意思就明白了，我们的可读流要想读文件，the first就需要先打开(open)，after我们再去读内容(read)

这就是实习可读流的主要方法，我们接下来先从open方法写起

open方法

class ReadStream extends EventEmitter {
    constructor(path, options) {
        // 省略...
    }

    open() {
        // 用法： fs.open(filename,flags,[mode],callback)
        fs.open(this.path, this.flags, (err, fd) => {   // fd为文件描述符
            // 说实在的我们打开文件，主要就是为了获取fd
            // fd是个从3开始的数字，每打开一次都会累加，4->5->6...
            if (err) {
                if (this.autoClose) {  // 文件打开报错了，是否自动关闭掉
                    this.destory();    // 销毁    
                }
                this.emit('error', err);    // 发射error事件
                return;
            }
            this.fd = fd;   // 如果没有错，保存文件描述符
            this.emit('open');  // 发射open事件
        });
    }

    // 这里用到了一个destory销毁方法，我们也直接实现了吧
    destory() {
        // 先判断有没有fd 有就关闭文件 触发close事件
        if (typeof this.fd === 'number') {
            // 用法： fs.close(fd,[callback])
            fs.close(this.fd, () => {
                this.emit('close'); 
            });
            return;
        }
        this.emit('close');
    }
}

万事开头难，我们把第一步打开文件搞定了，那么就剩下读取了，再接再厉当上王者

read方法

class ReadStream extends EventEmitter {
    constructor(path, options) {
        // 省略...
    }
    // 监听data事件的时候，去读取
    read() {
        console.log(this.fd);   // 直接读fd为undefined，因为open事件是异步的，此时还拿不到fd
        // 此时文件还没打开
        if (typeof this.fd !== 'number') {  // 前面说过fd是个数字
            // 当文件真正打开的时候,会触发open事件
            // 触发事件后再执行read方法,此时fd肯定有了
            return this.once('open', () => this.read());  // once方法只会执行一次
        }
        // 现在有fd了，大声的读出来，不要害羞
        // 用法： fs.read(fd, buffer, offset, length, pos, callback((err, bytesRead)))

        // length就是一次想读几个, 不能大于buffer长度
        // 这里length不能等于highWaterMark,举个🌰
        // 文件内容是12345如果按照highWaterMark:3来读,总共读end:4个，每次读3个字节
        // 分别是123 45空，我们应该知道一共要读几个，总数-读取位置+1得到下一次要读多少个
        // 这里有点绕，大家可以多去试试体会一下
        // 我们根据源码起一个同样的名字
        let howMuchToRead = this.end ? Math.min((this.end-this.pos+1), this.highWaterMark) : this.highWaterMark;

        fs.read(this.fd, this.buffer, 0, howMuchToRead, this.pos, (err, bytesRead) => {
            // bytesRead为读取到的个数，每次读3个，bytesRead就是3
            if (bytesRead > 0) {
                this.pos += bytesRead; // 读到了多少个，累加，下次从该位置继续读

                let buf = this.buffer.slice(0, bytesRead);  // 截取buffer对应的值
                // 其实正常情况下，我们只要把buf当成data传过去即可了
                // 但是考虑到还有编码的问题，所以有可能不是buffer类型的编码
                // 这里需要判断一下是否有encoding
                let data = this.encoding ? buf.toString(this.encoding) : buf.toString(); 

                this.emit('data', data);    // 发射data事件，并把data传过去

                // 如果读取的位置 大于 结束位置 就代表读完了，触发一个end事件
                if (this.pos > this.end) {
                    this.emit('end');
                    this.destory();
                }
                // 流动模式继续触发
                if (this.flowing) {   
                    this.read();
                }
            } else {    // 如果bytesRead没有值了就说明读完了
                this.emit('end');   // 发射end事件，表示文件读完
                this.destory();     // 没有价值了，kill
            }
        });
    }
}

以上就是read方法的主要实现了，其实思路上并不是很难，去除掉注释的话也会更精简些。大家上面也了解了可读流的用法，知道它还有两个方法，那就是pause(暂停)和resumt(恢复)，那么我们择日不如撞日直接写完吧，简单到令人发指的实现，看看不会吃亏的，哈哈

pause和resume方法

class ReadStream extends EventEmitter {
    constructor(path, options) {
        // 省略...
    }
    pause() {
        this.flowing = false;
    }
    resume() {
        this.flowing = true;
        this.read();
    }
}

完事，就是这么so easy，我们实现了自己的可读流了，可喜可贺，可喜可贺。

写在最后

其实说实话，这些基于Api的东西说起来还是很让人枯燥无聊的，大家都是拒绝无聊主义者。但我还是坚持写下来了，也是想让大家和我一起去感受一下大师们是怎么实现怎么思考的过程。

在这里我本来还想把可写流的实现也一起写进文章呢，但是由于篇幅太长，实在是辛苦各位观看了。

不过如果有想研究可写流实现的同学们，可以写进评论里，我也会在继续补充把它们也实现的。谢谢大家的观看了，能坚持下来的都不是折翼的天使啊！