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

Love The Way You Lie 2023-10-01 15:39 9阅读 0赞

**目录**

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参数中

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16][]

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

这会产生什么问题呢？

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

fs.readFile参与者类比：

水源 → 磁盘中文件数据

蓄水池 → 内存缓冲区

消费者 → console.log

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

fs.readFile工作逻辑：

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

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

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

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

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

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

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

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

但是还有一种异常场景：

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

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

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

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

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

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

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

所以我给出解决策略：

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

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

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 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读取到的数据的。它是一个单向链表结构。

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 3][]

如图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，也可能来自于没来及存入缓冲区就直接送去消费的数据。

### 可读流白盒模型 ###

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

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 4][]

目前我们随时对可读流对象的结构有了一个初步认识，接下来我们需要深入了解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事件

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 5][]

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

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 6][]

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

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 7][]

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

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

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

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

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 8][]

那么flow方法是干啥的呢？

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 9][]

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

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

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

### read如何触发\_read执行 ###

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

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 10][]

可以发现，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);
        else
          emitReadable(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 method
        this._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);
      else
        ret = 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());
          else
            addChunk(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);
            else
              maybeReadMore(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);
        else
          state.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事件

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 11][]

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

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 12][]

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

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 13][]

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

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 14][]

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

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 15][]

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

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

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

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

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

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

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

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

<table style="width:500px;"> 
 <tbody> 
  <tr> 
   <td>data</td> 
   <td> <p>push，read</p> </td> 
  </tr> 
  <tr> 
   <td>readable</td> 
   <td> <p>push，read</p> </td> 
  </tr> 
 </tbody> 
</table>

检查代码发现，当可读流对象进行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时，才会停止。

### 更新可读流的白盒模型 ###

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 16][]

### 可读流end事件 ###

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

end事件只会触发一次。

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 17][]

### read(0)分析 ###

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 18][]

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事件

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 19][]

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、已到达流（缓冲区）的末尾时

即：

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 20][]

监听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。

[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/e69f75e653c1405ca0757689703774c0.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/dd7a3ac325af470f9b421b9eac08e559.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 2]: https://img-blog.csdnimg.cn/e29357c65216467a988b665f60dfbf14.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 3]: https://img-blog.csdnimg.cn/83bb658ccf2848dcb7d209b65ec05fef.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 4]: https://img-blog.csdnimg.cn/3c84b568f9c94eb1b0cb9e4d655e9660.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 5]: https://img-blog.csdnimg.cn/8d8194a07d8043e891d3034d76f6b64e.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 6]: https://img-blog.csdnimg.cn/03ce959f13464e2ba7481470b4a0e07e.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 7]: https://img-blog.csdnimg.cn/b2d43fa867994e709b4df984a3015e7d.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 8]: https://img-blog.csdnimg.cn/b52f53c5b82d4d10896625efdd14a9a4.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 9]: https://img-blog.csdnimg.cn/7a12eb31793d45e7ab116299fd95ba15.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 10]: https://img-blog.csdnimg.cn/cf682ee8212245f5a90e27c0c99a1b78.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 11]: https://img-blog.csdnimg.cn/2d4a4eb1fee64f34b5669364ca55fede.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 12]: https://img-blog.csdnimg.cn/2504a51bd96b4d279820999b72dc89fd.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 13]: https://img-blog.csdnimg.cn/1677bf39981d4f1fb6ca8ceaccc68e54.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 14]: https://img-blog.csdnimg.cn/95f0268654024564afe0bf1e1caebca3.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 15]: https://img-blog.csdnimg.cn/9984c1eab1be4e738c42979fabfb0eba.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 16]: https://img-blog.csdnimg.cn/caeaa9992f6b4b4f9c14a711eb5d2a01.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 17]: https://img-blog.csdnimg.cn/8668310a89864c6593576978d9aacc4f.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 18]: https://img-blog.csdnimg.cn/5b8db91ad08048a799677426c0d24333.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 19]: https://img-blog.csdnimg.cn/cf285d519d4942c189797d322a37365d.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 20]: https://img-blog.csdnimg.cn/702d791c0cb34575aa90b6c9cf816811.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16