Node.js stream模块（一）可读流

fs.readFile的问题

如何设计出内存友好的，且人性化的数据生产与消费模式

stream模块

创建可读流对象

_read的作用

push的作用

可读流对象的缓冲区buffer

水位线highWaterMark及实际蓄水量length

进水口设计（_read的调用时机）

出水口设计（read的调用时机）

可读流白盒模型

_read和read的工作逻辑

可读流如何进入流动模式

可读流流动模式下如何自动触发read执行

read如何触发_read执行

push的工作逻辑

可读流如何进入暂停模式

流动模式和暂停模式的联系与区别

data事件和readable事件的触发时机

监听data事件为什么能实现源源不断地消费数据

更新可读流的白盒模型

可读流end事件

read(0)分析

深入理解readable事件触发时机

fs.readFile的问题

之前学习的fs模块在读写文件时，存在一个问题，比如fs.readFile，虽然是分多次读取磁盘文件数据，但是每次读取的数据都被缓存在内存中，当最后一次读取不到磁盘文件数据后，再整合内存中所有的缓存数据为一个完整文件数据，并传入fs.readFile回调函数的data参数中

如果我们将console.log看成是一种消费数据的行为，那么fs.readFile只会在文件数据全部加载缓冲区后，才会触发消费，即一次性消费。

这会产生什么问题呢？

我们想象一个场景，有一个蓄水池，它有一个进水口，一个出水口。进水口连接着水源。出水口连接着消费者。

fs.readFile参与者类比：

水源 → 磁盘中文件数据

蓄水池 → 内存缓冲区

消费者 → console.log

fs.readFile工作逻辑：

蓄水池进水口一直开着，水源不断流入蓄水池，当水源流干了，蓄水池才会打开出水口阀门，把蓄的水流向消费者。

这个工作逻辑有一个大前提：水源的水量不能大于蓄水池的容量，一旦超过，蓄水池就会溢出。

我们再站在消费者的角度来看看：

我是一个快要渴死的消费者，我敲打着蓄水池的阀门，希望喝水，我明明已经听到了蓄水池中水流的声音，蓄水池的墙壁上还有水溢出的迹象，但是蓄水池出水口阀门就是不打开，于是我渴死了，讽刺的是后面蓄水池因为装不了更多的水，爆炸了。

这场悲剧的原因就是：蓄水池太不人性了，明明已经有消费者发出消费诉求了，自己也明明有水，但是就是不打开出水阀门，结果消费者渴死了，蓄水池也被撑爆了。

如何设计出内存友好的，且人性化的数据生产与消费模式

那么你作为蓄水池的设计者，你会如何设计呢？

首先干掉不人性的出水口阀门，换成水龙头，开不开的权利交给消费者。这样消费者就能随时进行消费了。

但是还有一种异常场景：

蓄水池的水快满了，蓄水池工作人员急死了，因为没有消费者开出水口，结果蓄水池再一次撑爆了。

额....好吧，虽然蓄水池已经实现了人性化消费，但是我们不能单纯依靠难以捉摸的消费需求来保证蓄水池不会溢出。

所以我决定给进水口也加一个水龙头，不能让进水如此嚣张了。OK，大功告成，完美解决蓄水池溢出问题。

但是，你是否思考过这样一个场景：

消费者酱爆正在洗头，发现没有水，于是大喊一声：“包租婆，没水啦”。

于是一个新的问题产生了：进水口何时打开，何时关闭？

打开时机不及时，会造成消费者没有水用，关闭时机不及时，会造成蓄水池发生溢出。

所以我给出解决策略：

1、在蓄水池中画一条水位线，当蓄水池水量到达水位线后，就自动关闭进水口，停止进水。

2、每当消费者打开出水口时，蓄水池都会检查自身蓄的水量是否足够，够了，就不打开进水口，不够，就打开进水口。

这样就能保证消费者及时获取到水，也能保证蓄水池不会发生溢出。

stream模块

现在我们再来理解stream的概念，就容易多了，stream翻译过来叫流，虽然听上去是一个动词，但是我们需要把他理解为名词，或者直接用stream会比较好理解。

stream其实就是上面小节中描述的带有两个水龙头的蓄水池。

stream 一端连向数据源，一端连向消费者。stream从数据源读取数据，并会将数据缓存在自身，当消费者需要消费时，再将自身缓存中的数据传输给消费者。

Node.js 提供了stream模块，在stream模块中提供了四种流，分别是：

Readable：可读流

Writeable：可写流

Duplex：双工流（可读可写流）

Transform：转换流（可读可写可变化数据的流）

其中可读流常作为数据生产者，可写流作为数据消费者，双工流和转换流既可以作为生产者也可以作为消费者。

我们这里先讨论可读流Readable。

创建可读流对象

Readable是stream模块下的一个抽象类，在Readable中存在一个抽象方法_read。当我们想要创建可读流对象时，一般通过自定义类继承Readable，并重写_read方法。

const { Readable } = require('stream')class MyReadable extends Readable {constructor() {super()}_read() {}
}

_read的作用

而_read方法的作用是从数据源读取数据，并将数据push到可读流对象的缓冲区中或者 push给消费者。

这里又涉及到两个新概念：push 和可读流对象的缓冲区

push的作用

push方法是Readable类的原型上方法，它的逻辑比较复杂，

简单理解的话就是将_read读取到的数据写入缓冲区缓存或者直接传递给消费者。且如果_read已将将数据源地数据读取完了，则需要显示push(null)来通知可读流对象数据源数据读取完了。

可读流对象的缓冲区buffer

可读流对象的缓冲区，其实就是蓄水池，作用是用来缓存_read读取到的数据的。它是一个单向链表结构。

如图debug所示，可读流对象mr的_readableState.buffer属性就是可读流对象的缓冲区，从它的属性head{data, next}节点设计，我们可以知道它是一个单向链表结构。该缓冲区相当于蓄水池，用于缓存_read读取到的数据。

我们之前了解过蓄水池的设计要求：

1、两个水龙头，一个控制进水，一个控制出水

2、水位线，防止蓄水池溢出

水位线highWaterMark及实际蓄水量length

我们可以将其对照到mr._readableState中的属性

{objectMode: false,highWaterMark: 16384,buffer: {head: null,tail: null,length: 0,},length: 0,pipes: [],flowing: null,ended: false,endEmitted: false,reading: false,constructed: true,sync: true,needReadable: false,emittedReadable: false,readableListening: false,resumeScheduled: false,errorEmitted: false,emitClose: true,autoDestroy: true,destroyed: false,errored: null,closed: false,closeEmitted: false,defaultEncoding: "utf8",awaitDrainWriters: null,multiAwaitDrain: false,readingMore: false,dataEmitted: false,decoder: null,encoding: null,
}

其中highWaterMark就是水位线，可读流的水位线默认是16384 = 16 * 1024B = 16KB，即buffer缓冲区中存储的数据达到16KB时，就有必要停止_read从数据源读取数据了，否则会发生内存溢出。我们需要注意的是highWaterMark是一条警告线，而不是buffer缓冲区的实际容量，我们一般不将水位线设置为buffer实际容量，因为这样水位线就没有意义了，因为刚发出警告，就发生溢出了。

那么如何知道缓冲区buffer中具体蓄了多少数据呢？

mr._readableState.length属性会记录mr._readableState.buffer缓存了多少字节数据。

进水口设计（_read的调用时机）

那么蓄水池的进水口对应哪个属性呢？

其实进水口对应的就是_read方法，当_read方法调用时，就表示打开进水口，从水源读取数据到可读流对象。

那么_read何时调用呢？

我们已经了解过蓄水池的工作机制了，蓄水池默认情况不打开进水口（减少内存占用），只有当消费者打开出水口（read）时，蓄水池才会先检查确认水不够（doRead），然后才打开进水口（调用_read方法）

所以_read的调用和蓄水池出水口是否打开有关系。

出水口设计（read的调用时机）

那么如何打开蓄水池的出水口呢？

在Readable类的原型上有一个方法read，当通过可读流对象调用read时，就表示需要消费数据了。这个被消费的数据可能来缓冲区buffer，也可能来自于没来及存入缓冲区就直接送去消费的数据。

可读流白盒模型

所以我们将蓄水池模型和可读流结合一下

目前我们随时对可读流对象的结构有了一个初步认识，接下来我们需要深入了解read，_read，push的工作逻辑，以及buffer的highWaterMark，length在其中起到的作用

_read和read的工作逻辑

当我们创建了一个可读流对象mr时，不会自动触发_read执行，即可读流对象初始时不会去读取数据源数据。

_read的执行和 read有关。即何时进水，取决于何时需要水。

read执行有两个执行时机：1、自动执行 2、手动执行

read手动执行就是方法调用，没有多复杂的流程。

而read自动执行取决于可读流对象的状态。在mr._readableState中有一个属性叫做flowing，该属性有三个值：null, true, false

flowing = null 表示可读流对象处于初始模式，该模式不会触发read执行

flowing = true 表示可读流对象处于流动模式，该模式会触发read执行

flowing = false 表示可读流对象处于暂停模式，该模式不会触发read执行

那么如何让可读流对象的_readableState.flowing = true，即让可读流对象进入流动模式成为触发read执行的关键。

可读流如何进入流动模式

具体触发可读流进入流动模式的方式很多，我简单列举一下：

1、监听可读流对象的data事件

2、可读流对象调用resume方法

3、可读流对象调用pipe方法

可读流流动模式下如何自动触发read执行

那么流动模式如何实现自动触发read执行的呢？

我们以监听data事件触发流动模式为例说明：

下面debug进入_read后，查看调用栈历史，发现在on之后，_read之前，还经历了flow，以及read。

那么flow方法是干啥的呢？

由于mr.on('data', callback)开启了流动模式，所以state.flowing为true。

所以上面flow方法的意思是：只有当可读流对象处于流动模式，且read读取到null时，flow方法才结束，否则会不停的read。

所以调用栈中，flow后面会不停地调用read。

read如何触发_read执行

我们继续看调用栈中read的逻辑

可以发现，read方法不是必然会调用_read，而是doRead为true时才会调用_read。

那么什么时候doRead为true呢？

下面是Reabable的内置实例方法read的逻辑

// You can override either this method, or the async _read(n) below.
Readable.prototype.read = function(n) {debug('read', n);// Same as parseInt(undefined, 10), however V8 7.3 performance regressed// in this scenario, so we are doing it manually.if (n === undefined) {n = NaN;} else if (!NumberIsInteger(n)) {n = NumberParseInt(n, 10);}const state = this._readableState;const nOrig = n;// If we're asking for more than the current hwm, then raise the hwm.if (n > state.highWaterMark)state.highWaterMark = computeNewHighWaterMark(n);if (n !== 0)state.emittedReadable = false;// If we're doing read(0) to trigger a readable event, but we// already have a bunch of data in the buffer, then just trigger// the 'readable' event and move on.if (n === 0 &&state.needReadable &&((state.highWaterMark !== 0 ?state.length >= state.highWaterMark :state.length > 0) ||state.ended)) {debug('read: emitReadable', state.length, state.ended);if (state.length === 0 && state.ended)endReadable(this);elseemitReadable(this);return null;}n = howMuchToRead(n, state);// If we've ended, and we're now clear, then finish it up.if (n === 0 && state.ended) {if (state.length === 0)endReadable(this);return null;}// All the actual chunk generation logic needs to be// *below* the call to _read.  The reason is that in certain// synthetic stream cases, such as passthrough streams, _read// may be a completely synchronous operation which may change// the state of the read buffer, providing enough data when// before there was *not* enough.//// So, the steps are:// 1. Figure out what the state of things will be after we do// a read from the buffer.//// 2. If that resulting state will trigger a _read, then call _read.// Note that this may be asynchronous, or synchronous.  Yes, it is// deeply ugly to write APIs this way, but that still doesn't mean// that the Readable class should behave improperly, as streams are// designed to be sync/async agnostic.// Take note if the _read call is sync or async (ie, if the read call// has returned yet), so that we know whether or not it's safe to emit// 'readable' etc.//// 3. Actually pull the requested chunks out of the buffer and return.// if we need a readable event, then we need to do some reading.let doRead = state.needReadable;debug('need readable', doRead);// If we currently have less than the highWaterMark, then also read some.if (state.length === 0 || state.length - n < state.highWaterMark) {doRead = true;debug('length less than watermark', doRead);}// However, if we've ended, then there's no point, if we're already// reading, then it's unnecessary, if we're constructing we have to wait,// and if we're destroyed or errored, then it's not allowed,if (state.ended || state.reading || state.destroyed || state.errored ||!state.constructed) {doRead = false;debug('reading, ended or constructing', doRead);} else if (doRead) {debug('do read');state.reading = true;state.sync = true;// If the length is currently zero, then we *need* a readable event.if (state.length === 0)state.needReadable = true;// Call internal read methodthis._read(state.highWaterMark);state.sync = false;// If _read pushed data synchronously, then `reading` will be false,// and we need to re-evaluate how much data we can return to the user.if (!state.reading)n = howMuchToRead(nOrig, state);}let ret;if (n > 0)ret = fromList(n, state);elseret = null;if (ret === null) {state.needReadable = state.length <= state.highWaterMark;n = 0;} else {state.length -= n;if (state.multiAwaitDrain) {state.awaitDrainWriters.clear();} else {state.awaitDrainWriters = null;}}if (state.length === 0) {// If we have nothing in the buffer, then we want to know// as soon as we *do* get something into the buffer.if (!state.ended)state.needReadable = true;// If we tried to read() past the EOF, then emit end on the next tick.if (nOrig !== n && state.ended)endReadable(this);}if (ret !== null) {state.dataEmitted = true;this.emit('data', ret);}return ret;
};

以上逻辑简要概括就是：

当缓冲区数据足够时，doRead为false，即read方法不会触发_read去补充数据，

当缓冲区数据不足消费，doRead为true，所以read方法会触发_read去数据源处读取数据补充到缓冲区。

push的工作逻辑

而_read执行中必然会同步或异步地调用push，我们再来看看push的逻辑

Readable.prototype.push = function(chunk, encoding) {return readableAddChunk(this, chunk, encoding, false);
};function readableAddChunk(stream, chunk, encoding, addToFront) {debug('readableAddChunk', chunk);const state = stream._readableState;let err;if (!state.objectMode) {if (typeof chunk === 'string') {encoding = encoding || state.defaultEncoding;if (state.encoding !== encoding) {if (addToFront && state.encoding) {// When unshifting, if state.encoding is set, we have to save// the string in the BufferList with the state encoding.chunk = Buffer.from(chunk, encoding).toString(state.encoding);} else {chunk = Buffer.from(chunk, encoding);encoding = '';}}} else if (chunk instanceof Buffer) {encoding = '';} else if (Stream._isUint8Array(chunk)) {chunk = Stream._uint8ArrayToBuffer(chunk);encoding = '';} else if (chunk != null) {err = new ERR_INVALID_ARG_TYPE('chunk', ['string', 'Buffer', 'Uint8Array'], chunk);}}if (err) {errorOrDestroy(stream, err);} else if (chunk === null) {state.reading = false;onEofChunk(stream, state);} else if (state.objectMode || (chunk && chunk.length > 0)) {if (addToFront) {if (state.endEmitted)errorOrDestroy(stream, new ERR_STREAM_UNSHIFT_AFTER_END_EVENT());elseaddChunk(stream, state, chunk, true);} else if (state.ended) {errorOrDestroy(stream, new ERR_STREAM_PUSH_AFTER_EOF());} else if (state.destroyed || state.errored) {return false;} else {state.reading = false;if (state.decoder && !encoding) {chunk = state.decoder.write(chunk);if (state.objectMode || chunk.length !== 0)addChunk(stream, state, chunk, false);elsemaybeReadMore(stream, state);} else {addChunk(stream, state, chunk, false);}}} else if (!addToFront) {state.reading = false;maybeReadMore(stream, state);}// We can push more data if we are below the highWaterMark.// Also, if we have no data yet, we can stand some more bytes.// This is to work around cases where hwm=0, such as the repl.return !state.ended &&(state.length < state.highWaterMark || state.length === 0);
}function addChunk(stream, state, chunk, addToFront) {if (state.flowing && state.length === 0 && !state.sync &&stream.listenerCount('data') > 0) {// Use the guard to avoid creating `Set()` repeatedly// when we have multiple pipes.if (state.multiAwaitDrain) {state.awaitDrainWriters.clear();} else {state.awaitDrainWriters = null;}state.dataEmitted = true;stream.emit('data', chunk);} else {// Update the buffer info.state.length += state.objectMode ? 1 : chunk.length;if (addToFront)state.buffer.unshift(chunk);elsestate.buffer.push(chunk);if (state.needReadable)emitReadable(stream);}maybeReadMore(stream, state);
}

可以发现push并不是单纯地将_read从数据源读取到地数据插入缓冲区中，在上面addChunk方法中，当符合如下条件时

if (state.flowing && state.length === 0 && !state.sync && stream.listenerCount('data') > 0)

push方法将数据直接通过data事件，将数据传输给消费者，而不进行缓存

上面条件地意思是：处于流动状态，且缓冲区干了，即当前_read读取的数据就是消费者需要的数据，所以省略了该数据进入缓冲区的步骤，而直接给了消费者，因为最终都是要给消费者，没必要进缓冲区。

如果不符合上面的条件，则_read读取到的数据会被push到缓冲区缓存，然后通过read方法获取。

可读流如何进入暂停模式

前面介绍了如何让可读流进入流动模式，有三种实现方式，那么如何让可读流进入暂停模式呢？

1、可读流对象监听readable事件

需要注意的是read方法调用并不会让暂停模式变为流动模式

2、可读流对象调用pause方法

pause可以让流动模式的可读流变为暂停模式

也可以通过resume方法让被pasue的可读流重新进入流动模式

流动模式和暂停模式的联系与区别

其实可读流对象无论是流动模式，还是暂停模式都是为了消费数据，只是消费数据的方式不同

在流动模式下，可读流对象会在flow方法作用下while循环调用read方法，直到read返回null才停止

在暂停模式下，可读流对象需要手动调用read方法，并自行判断read返回值

所以监听data事件时，其回调函数自动可以获得chunk，

但是监听readable事件时，其回调函数没有chunk参数，而是需要我们手动调用read去获取chunk

data事件和readable事件的触发时机

那么data事件和readable事件的触发时机是什么时候呢？

data	push，read
readable	push，read

检查代码发现，当可读流对象进行push或read时都有可能触发这两个事件

概括一下就是：

当可读流对象的_read方法将数据源数据push到缓冲区后，就会触发readable事件通知消费者消费，或者没有调用_read方法，但是read(0)调用，此时可读流对象会去检查缓冲区存量数据，如果有的话，就会触发readable事件。

当可读流对象的_read方法将数据源数据push给消费者，则会触发data事件通知消费者消费，另外其实每次read方法调用，只要read返回非null数据都会通过data事件将读取的数据给消费者。

所以readable事件触发条件更多关注在可读流对象缓冲区是否有存量数据，有的话就会触发readable事件告知消费者可以消费了。

而data事件关注点会提前一点，即如果缓冲区没有存量数据，则只要_read执行,push执行就会触发data事件。如果缓冲区有存量数据，也会触发data事件。

监听data事件为什么能实现源源不断地消费数据

所以data事件是一个很神奇地设计：

监听data事件，会导致开启流动模式，而流动模式会触发flow方法,flow方法又会引起while循环调用read方法，read方法又会触发data事件，当read方法返回null时，才会停止。

更新可读流的白盒模型

可读流end事件

可读流除了用于消费数据的事件类型data和readable外，还有一个end事件类型，表示可读流已经读取完数据源的数据了，并且可读流的缓冲区数据也被消费完了。

end事件只会触发一次。

read(0)分析

function emitReadable_(stream) {const state = stream._readableState;debug('emitReadable_', state.destroyed, state.length, state.ended);if (!state.destroyed && !state.errored && (state.length || state.ended)) {stream.emit('readable');state.emittedReadable = false;}// The stream needs another readable event if:// 1. It is not flowing, as the flow mechanism will take//    care of it.// 2. It is not ended.// 3. It is below the highWaterMark, so we can schedule//    another readable later.state.needReadable =!state.flowing &&!state.ended &&state.length <= state.highWaterMark;flow(stream);
}

可以发现read(0)会触发_read，并将数据push到缓冲区，当缓冲区有新数据了就会异步地(process.nextTick)触发readable事件

function maybeReadMore_(stream, state) {// Attempt to read more data if we should.//// The conditions for reading more data are (one of):// - Not enough data buffered (state.length < state.highWaterMark). The loop//   is responsible for filling the buffer with enough data if such data//   is available. If highWaterMark is 0 and we are not in the flowing mode//   we should _not_ attempt to buffer any extra data. We'll get more data//   when the stream consumer calls read() instead.// - No data in the buffer, and the stream is in flowing mode. In this mode//   the loop below is responsible for ensuring read() is called. Failing to//   call read here would abort the flow and there's no other mechanism for//   continuing the flow if the stream consumer has just subscribed to the//   'data' event.//// In addition to the above conditions to keep reading data, the following// conditions prevent the data from being read:// - The stream has ended (state.ended).// - There is already a pending 'read' operation (state.reading). This is a//   case where the stream has called the implementation defined _read()//   method, but they are processing the call asynchronously and have _not_//   called push() with new data. In this case we skip performing more//   read()s. The execution ends in this method again after the _read() ends//   up calling push() with more data.while (!state.reading && !state.ended &&(state.length < state.highWaterMark ||(state.flowing && state.length === 0))) {const len = state.length;debug('maybeReadMore read 0');stream.read(0);if (len === state.length)// Didn't get any data, stop spinning.break;}state.readingMore = false;
}

如果read(0)符合条件，还会异步地循环调用read(0)

深入理解readable事件触发时机

当有可从流中读取的数据或已到达流的末尾时，则将触发 'readable' 事件。

1、当有可从流（缓冲区）中读取的数据

2、已到达流（缓冲区）的末尾时

即：

监听readable事件会触发一次read(0)，导致缓冲区被插入数据'a'，缓冲区有数据就会触发readable事件①（缓冲区有数据触发readable事件的情况只有一次）

①：第一次触发地readable事件导致回调中while循环开始，执行mr.read()，导致'b'被插入缓冲区，并且该操作会消费掉缓冲区中16KB的数据，但是此时缓冲区只有两个个字节数据，所以缓冲区被清空了，返回了'ab'，触发了一次readable事件②

①： while循环继续，执行mr.read()，导致'c'被插入缓冲区，并被取出消费，作为mr.read返回值，缓冲区再次被清空，返回了'c'，触发一次readable事件③。

①：while循环继续，执行mr.read()，发现数据源没有数据，则push(null)，结束了读取，而此时缓冲区也空了，触发一次readable事件④，并且mr.read返回null，则跳出循环。打印1。

①readable事件触发完毕，开始触发②readable事件，此时可读流读取结束，read方法不再产生readable事件，所以while立即推出，打印1。

②readable事件触发完毕，开始触发③readable事件，此时可读流读取结束，read方法不再产生readable事件，所以while立即推出，打印1。

③readable事件触发完毕，开始触发④readable事件，此时可读流读取结束，read方法不再产生readable事件，所以while立即推出，打印1。