Node.js stream模块（三）背压机制

蔚落 2023-10-01 16:14 1阅读 0赞

我们知道 可读流是作为数据生产者，而可写流作为数据消费者。

那么二者必然是可以结合使用的。即可读流生产出来的数据给可写流消费。

我们这里使用文件可读流和文件可写流来模拟这种情况：

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16][]

实现很简单，可读流对象通过data事件可以进入流动模式，还是源源不断地生产数据，而可写流对象通过write方法可以写入数据并消费。所以二者结合就实现了生产与消费地结合。

但是我们注意到：文件可读流的缓冲区水位线默认是64KB，而文件可写流的缓冲区水位线默认是16KB。为何官方要设计出这种差异呢？

其实这和磁盘文件的读取和写入速度有关，一般来说，从磁盘中读取文件的速度 要快于 向磁盘中写入文件。

举个例子：抄作业很快，写作业很慢。

而文件可读流其实就是从磁盘中读取文件数据，类似于抄作业，

文件可写流就是向磁盘中写入文件数据，类似于写作业，所以文件可读流速度快于文件可写流。

我再站在生产者，消费者角度思考：

即 生产者的生产速度，快于，消费者的消费速度。

而生产速度和消费速度的不匹配，就会产生一个严重的问题：供大于求。

在经济学上，这是要发生美国奶农倾倒牛奶的事件的。而这是人文社会所不愿意看到的，所以出现了政府干预市场，比如中国一直推行的减排政策，归根揭底还是减少不必要的生产活动。

而在流的世界里，也模拟出了这种减排政策，即背压机制。

即当消费者的消费速度赶不上生产者的生产速度时，就让生产者暂停生产，让消费者能缓一口气去消费。而等消费者消费完后，再通知生产者继续生产。

const fs = require('fs')
    const path = require('path')
    
    
    const rs = fs.createReadStream(path.join(__dirname, 'test.txt'), {
      highWaterMark: 4  // 可读流缓冲区水位线默认64KB
    })
    
    const ws = fs.createWriteStream(path.join(__dirname, 'test2.txt'), {
      highWaterMark: 1 // 可写流缓冲区水位线默认16KB
    })
    
    rs.on('data', (chunk) => {
      let flag = ws.write(chunk)
      if (!flag) {
        rs.pause()
      }
    })
    
    ws.on('drain', () => {
      rs.resume()
    })

监听可读流的data事件，会触发可读流的流动模式，会源源不断地生产数据，生产出来的chunk，会被交给可写流write消费，但是如果write返回false，表示数据量已经超过了缓冲区水位线，如果继续生产下去，可能会造成写流缓冲区发生内存溢出，所以此时需要暂停生产，即rs.pause()，让可读流进入暂停模式。

而当可写流将缓冲区数据消费完了，就会触发drain事件，即表示生产者可以继续生产了，即rs.resume()。

这就是背压机制的基本逻辑。

以上对于背压机制的实现有点公式化，所以针对可读流内置了一个pipe方法，

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

其内在实现就是背压机制。

Readable.prototype.pipe = function(dest, pipeOpts) {
      const src = this;
      const state = this._readableState;
    
      if (state.pipes.length === 1) {
        if (!state.multiAwaitDrain) {
          state.multiAwaitDrain = true;
          state.awaitDrainWriters = new SafeSet(
            state.awaitDrainWriters ? [state.awaitDrainWriters] : []
          );
        }
      }
    
      state.pipes.push(dest);
      debug('pipe count=%d opts=%j', state.pipes.length, pipeOpts);
    
      const doEnd = (!pipeOpts || pipeOpts.end !== false) &&
                  dest !== process.stdout &&
                  dest !== process.stderr;
    
      const endFn = doEnd ? onend : unpipe;
      if (state.endEmitted)
        process.nextTick(endFn);
      else
        src.once('end', endFn);
    
      dest.on('unpipe', onunpipe);
      function onunpipe(readable, unpipeInfo) {
        debug('onunpipe');
        if (readable === src) {
          if (unpipeInfo && unpipeInfo.hasUnpiped === false) {
            unpipeInfo.hasUnpiped = true;
            cleanup();
          }
        }
      }
    
      function onend() {
        debug('onend');
        dest.end();
      }
    
      let ondrain;
    
      let cleanedUp = false;
      function cleanup() {
        debug('cleanup');
        // Cleanup event handlers once the pipe is broken.
        dest.removeListener('close', onclose);
        dest.removeListener('finish', onfinish);
        if (ondrain) {
          dest.removeListener('drain', ondrain);
        }
        dest.removeListener('error', onerror);
        dest.removeListener('unpipe', onunpipe);
        src.removeListener('end', onend);
        src.removeListener('end', unpipe);
        src.removeListener('data', ondata);
    
        cleanedUp = true;
    
        // If the reader is waiting for a drain event from this
        // specific writer, then it would cause it to never start
        // flowing again.
        // So, if this is awaiting a drain, then we just call it now.
        // If we don't know, then assume that we are waiting for one.
        if (ondrain && state.awaitDrainWriters &&
            (!dest._writableState || dest._writableState.needDrain))
          ondrain();
      }
    
      function pause() {
        // If the user unpiped during `dest.write()`, it is possible
        // to get stuck in a permanently paused state if that write
        // also returned false.
        // => Check whether `dest` is still a piping destination.
        if (!cleanedUp) {
          if (state.pipes.length === 1 && state.pipes[0] === dest) {
            debug('false write response, pause', 0);
            state.awaitDrainWriters = dest;
            state.multiAwaitDrain = false;
          } else if (state.pipes.length > 1 && state.pipes.includes(dest)) {
            debug('false write response, pause', state.awaitDrainWriters.size);
            state.awaitDrainWriters.add(dest);
          }
          src.pause();
        }
        if (!ondrain) {
          // When the dest drains, it reduces the awaitDrain counter
          // on the source.  This would be more elegant with a .once()
          // handler in flow(), but adding and removing repeatedly is
          // too slow.
          ondrain = pipeOnDrain(src, dest);
          dest.on('drain', ondrain);
        }
      }
    
      src.on('data', ondata);
      function ondata(chunk) {
        debug('ondata');
        const ret = dest.write(chunk);
        debug('dest.write', ret);
        if (ret === false) {
          pause();
        }
      }
    
      // If the dest has an error, then stop piping into it.
      // However, don't suppress the throwing behavior for this.
      function onerror(er) {
        debug('onerror', er);
        unpipe();
        dest.removeListener('error', onerror);
        if (EE.listenerCount(dest, 'error') === 0) {
          const s = dest._writableState || dest._readableState;
          if (s && !s.errorEmitted) {
            // User incorrectly emitted 'error' directly on the stream.
            errorOrDestroy(dest, er);
          } else {
            dest.emit('error', er);
          }
        }
      }
    
      // Make sure our error handler is attached before userland ones.
      prependListener(dest, 'error', onerror);
    
      // Both close and finish should trigger unpipe, but only once.
      function onclose() {
        dest.removeListener('finish', onfinish);
        unpipe();
      }
      dest.once('close', onclose);
      function onfinish() {
        debug('onfinish');
        dest.removeListener('close', onclose);
        unpipe();
      }
      dest.once('finish', onfinish);
    
      function unpipe() {
        debug('unpipe');
        src.unpipe(dest);
      }
    
      // Tell the dest that it's being piped to.
      dest.emit('pipe', src);
    
      // Start the flow if it hasn't been started already.
    
      if (dest.writableNeedDrain === true) {
        if (state.flowing) {
          pause();
        }
      } else if (!state.flowing) {
        debug('pipe resume');
        src.resume();
      }
    
      return dest;
    };

pipe方法可以实现协调数据生成和消费。我们通常将pipe形象的成为管道。

pipe方法的调用者一般是可读流对象，而pipe方法参数一般是可写流对象，当pipe方法调用时，就会触发可写流对象的pipe事件。

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 2][]

pipe事件的回调函数会收到调用pipe事件的可读流对象

![watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16 3][]

pipe方法的返回值是pipe方法参数对应的可写流对象。

pipe的本质作用还是将上游数据传递到下游，pipe就是一个传输管道。

我们也可以通过unpipe方法来切断传输管道，即切断传输。这样就能终端pipe传输数据。

[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA5LyP5Z-O5LmL5aSW_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/ccbce8d0c5a04574a375bb96ab0bcecf.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/73fb87187c0c480cb903d691a2b1872f.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/89144bb6265549bba484eadda598fa16.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/ed755f68903843ea8f57efaf7cafc0c9.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA5LyP5Z-O5LmL5aSW,size_20,color_FFFFFF,t_70,g_se,x_16