1.流是什么？
在node.js官网文档中是这么说的 流（stream）在 Node.js 中是处理流数据的抽象接口（abstract interface）。stream 模块提供了基础的 API 。使用这些 API 可以很容易地来构建实现流接口的对象。 Node.js 提供了多种流对象。 例如， 和 就都是流的实例。
流可以是可读的、可写的，或是可读写的。所有的流都是 的实例。 本人最近研究node，特意记下，分享一下。
言归正传
1. Node.js 中有四种基本的流类型：
· Readable - 可读的流 (例如 fs.createReadStream()). · Writable - 可写的流 (例如 fs.createWriteStream()). · Duplex - 可读写的流 (例如 net.Socket). · Transform - 在读写过程中可以修改和变换数据的 Duplex 流 (例如 zlib.createDeflate()).那么今天的主角就是可读流(createReadStream)，后续会陆续介绍其他3种流
2.可读流(createReadStream)
实现了stream.Readable接口的对象,将对象数据读取为流数据,当监听data事件后,开始发射数据2.1 创建可读流
var rs = fs.createReadStream(path,[options]);ps:如果指定utf8编码highWaterMark要大于3个字节复制代码
2.2 监听data事件
流切换到流动模式,数据会被尽可能快的读出rs.on('data', function (data) { console.log(data);});复制代码 2.3 监听end事件
该事件会在读完数据后被触发rs.on('end', function () { console.log('读取完成');})复制代码 2.4 监听error事件
rs.on('error', function (err) { console.log(err);});复制代码 2.5 监听open事件
rs.on('open', function () { console.log(err);});复制代码 2.6 监听close事件
rs.on('close', function () { console.log(err);});复制代码 2.7 设置编码
与指定{encoding:'utf8'}效果相同，设置编码rs.setEncoding('utf8');复制代码 2.8 暂停和恢复触发data
通过pause()方法和resume()方法rs.on('data', function (data) { rs.pause(); console.log(data);});setTimeout(function () { rs.resume();},2000);复制代码 3. 流中的数据有两种模式,二进制模式和对象模式
· 二进制模式, 每个分块都是buffer或者string对象 · 对象模式, 流内部处理的是一系列普通对象.
4.可读流的两种模式
可读流事实上工作在下面两种模式之一：flowing 和 paused 在 flowing 模式下， 可读流自动从系统底层读取数据，并通过 EventEmitter 接口的事件尽快将数据提供给应用。 在 paused 模式下，必须显式调用 stream.read() 方法来从流中读取数据片段。 所有初始工作模式为 paused 的 Readable 流 如果 Readable 切换到 flowing 模式，且没有消费者处理流中的数据，这些数据将会丢失。 比如， 调用了 readable.resume() 方法却没有监听 'data' 事件，或是取消了 'data' 事件监听，就有可能出现这种情况 下面通过代码简单模拟下实现的逻辑： 如图：
let fs = require('fs');let EventEmitter = require('events');class ReadStream extends EventEmitter {//这里用到的是es6的原型继承 constructor(path, options) { super(path, options); this.path = path;//读取文件的路径 this.highWaterMark = options.highWaterMark || 64 * 1024;//读取的最高水位线 this.buffer = Buffer.alloc(this.highWaterMark);//临时存放数据的容器 this.flags = options.flags || 'r';//打开文件要做的操作,默认为'r' this.encoding = options.encoding;//编码 this.mode = options.mode || 0o666;//读取文件的权限 this.start = options.start || 0;//开始读取的索引位置 this.end = options.end;//结束读取的索引位置(包括结束位置) this.pos = this.start; this.autoClose = options.autoClose || true;当流读完之后自动关闭文件 this.bytesRead = 0;//实际读到的字节数 this.closed = false; this.flowing;初始状态为暂停模式 为true时为流动模式， 为false时为暂停模式 this.needReadable = false;//是否需要发送readable事件 this.length = 0;//缓存区的总长度 this.buffers = [];//这才是真正的缓存区 this.on('end', function () { if (this.autoClose) { this.destroy(); } }); //当给这个实例添加了任意的监听函数时会触发newListener this.on('newListener', (type) => { //如果监听了data事件，流会自动切换的流动模式 //流动模式不缓存，直接发射，然后读取下次的数据。如果你用流动模式而且没有消费，数据就白白丢失了 if (type == 'data') { this.flowing = true; this.read(); } if (type == 'readable') { this.read(0); } }); this.open(); }//打开文件 open() { fs.open(this.path, this.flags, this.mode, (err, fd) => { if (err) { if (this.autoClose) { this.destroy(); return this.emit('error', err); } } this.fd = fd; this.emit('open'); }); } read(n) { if (typeof this.fd != 'number') { return this.once('open', () => this.read()); } n = parseInt(n, 10); if (n != n) { n = this.length; } if (this.length == 0)//重新填充 this.needReadable = true; let ret; if (0 < n < this.length) {//缓存区的数据足够用，并且要读取的字节大于0 ret = Buffer.alloc(n); let b; let index = 0; while (null != (b = this.buffers.shift())) { for (let i = 0; i < b.length; i++) { ret[index++] = b[i]; if (index == ret.length) {//填充完毕 this.length -= n; b = b.slice(i + 1); this.buffers.unshift(b); break; } } } if (this.encoding) ret = ret.toString(this.encoding); } //底层读数据的方法 let _read = () => { let m = this.end ? Math.min(this.end - this.pos + 1, this.highWaterMark) : this.highWaterMark; fs.read(this.fd, this.buffer, 0, m, this.pos, (err, bytesRead) => { if (err) { return } let data; if (bytesRead > 0) {//实际读到的字节数 data = this.buffer.slice(0, bytesRead); this.pos += bytesRead; //让缓存区的数量加实际读到的字节数 this.length += bytesRead; if (this.end && this.pos > this.end) { if (this.needReadable) { this.emit('readable'); } this.emit('end'); } else { this.buffers.push(data); if (this.needReadable) { this.emit('readable'); this.needReadable = false; } } } else { if (this.needReadable) { this.emit('readable'); } return this.emit('end'); } }) } if (this.length == 0 || (this.length < this.highWaterMark)) { _read(0); } return ret; }//销毁流 destroy() { fs.close(this.fd, (err) => { this.emit('close'); }); }//暂停 pause() { this.flowing = false; }//恢复 resume() { this.flowing = true; this.read(); } pipe(dest) { this.on('data', (data) => { let flag = dest.write(data); if (!flag) this.pause(); }); dest.on('drain', () => { this.resume(); }); this.on('end', () => { dest.end(); }); }}module.exports = ReadStream;复制代码 readable时其实会立刻填充缓存区，当你消费掉一个字节后，一旦发现缓冲区的字节数小于最高水位线了，则会立马读到最高水位线highWaterMark(n)个字节填充到缓存区 read(n)是可读流的核心方法，那么方法中传参数跟不传，传几个是有讲究的： n = undefined :即n不传参数,如果处于流动模式，并且缓存区大小不为空，则返回缓存区第一个buffer的长度，否则读取整个缓存 如果读到了数据没有返回值，但是会发射data事件,数据也能取到,也就是用来清空缓存区。也不需要向底层_read()请求数据 n=0：只是填充缓存区，并不真正读取 0 < n < highWaterMark(n)：可读流会从缓存区取出对应大小的数据，并且作为 read 方法返回值返回。同时判断此时缓存区大小是否小于highWaterMark(n)，如果小于，那么将执行底层的 _read 方法，从数据源中读取highWaterMark(n)大小的数据填充到缓存区内。 n>highWaterMark(n):如果缓存中数据不够，便会调用_read方法去底层取数据。
尾声：
碍于篇幅和时间限制，就不再继续介绍下去。本文仅介绍了可读流的基本使用方法，如果觉得有一些地方觉得不太深入，可以去看一下官方文档，官方文档上面有更加全面和详细的介绍。
参考资料：