上次写Stream pipe细节时,在源码中发现一段无用逻辑,由此引发了对Stream data事件触发时机与顺序的探索。

无用逻辑

当时研究pipe细节是基于Node.js v8.11.1的源码,其中针对上游的ondata事件处理有如下一段代码:

// If the user pushes more data while we're writing to dest then we'll end up
// in ondata again. However, we only want to increase awaitDrain once because
// dest will only emit one 'drain' event for the multiple writes.
// => Introduce a guard on increasing awaitDrain.
var increasedAwaitDrain = false;
src.on('data', ondata);
function ondata(chunk) {
  debug('ondata');
  increasedAwaitDrain = false;
  var ret = dest.write(chunk);
  if (false === ret && !increasedAwaitDrain) {
    if (((state.pipesCount === 1 && state.pipes === dest) ||
        (state.pipesCount > 1 && state.pipes.indexOf(dest) !== -1)) &&
      !cleanedUp) {
      debug('false write response, pause', src._readableState.awaitDrain);
      src._readableState.awaitDrain  ;
      increasedAwaitDrain = true;
    }
    src.pause();
  }
}

重点关注increasedAwaitDrain变量,理解这个变量期望达到什么目的,然后仔细阅读代码,会发现if (false === ret && !increasedAwaitDrain)语句中increasedAwaitDrain变量肯定是false,因为前一行才将该变量赋值为false,这样一来这个变量就变得毫无意义。

increasedAwaitDrain = false; 
var ret = dest.write(chunk); 
if (false === ret && !increasedAwaitDrain) {}

以上就是关键的三行代码,因为Node.js是单线程且dest.write(chunk)内部没有修改变量increasedAwaitDrain的值,那么if语句中increasedAwaitDrain的值肯定还是false,即increasedAwaitDrain相关逻辑没有达到所期望的目标。

无用代码出现的原因

前段虽已经分析出increasedAwaitDrain没起到作用,但作者为什么写了这样一段逻辑呢?其实在定义increasedAwaitDrain语句的上方,作者说可能存在这样一种情况:“当我们接收到一次上游的ondata事件并尝试将数据写到下游时,上游可能同时又有一个data事件触发,而这两个ondata的数据在写入下游时可能都返回false,从而导致src._readableState.awaitDrain 执行两次”。

awaitDrain 执行两次是作者不希望看到的情况,因为下游触发drain事件时awaitDrain相应减1,直到其值为0时才让上游重新流动,如果awaitDrain 执行两次,下游却只触发一次drain事件,awaitDrain就不会为0,上游不重新流动也就无法继续读取数据。

真相的探索过程

虽然从理性上认为increasedAwaitDrain没起到作用,但也无法肯定加绝对,自己尝试去求助,没有出现高手指点出问题所在,但一个同事听我描述后,说可能这就是个BUG,虽心中觉得可能性不大,但还是抱着试试看的心态切换到master分支上去瞅瞅,随即发现最新的代码里并没有与increasedAwaitDrain类似的逻辑,间接说明v8.11.1分支上increasedAwaitDrain相关逻辑的确无用。

虽然比较肯定这里存在一段无用代码,但应该如何理解作者在increasedAwaitDrain上方的注释呢?为了进一步揭露真相,自己继续花时间去看了看stream.Readable相关代码,想知道data事件的触发时机与顺序是如何决定的。

readable流的简单原理

在进一步解释data事件的触发顺序前,简单讲一下readable流的实现原理,如果需要自己实现一个readable流,可以使用new stream.Readable(options)方法,其中options可包含四个属性:highWaterMark、encoding、objectMode、read。最主要的是read属性,当流的使用者需要数据时,read方法被用来从数据源获取数据,然后通过this.push(chunk)将数据传递给使用者,如果没有更多数据可供读取时使用this.push(null)表示读取结束。

const Readable = require('stream').Readable;
let letter = 'ABCDEFG'.split('');
let index = 0;
const rs = new Readable({
  read(size) {
    this.push(letter[index  ] || null);
  }
});
rs.on('data', chunk => {
  console.log(chunk.toString());
});
// 输出
// A
// B
// C
// ...

这里ondata虽然没有明显调用read方法,但内部依旧是通过调用read方法结合this.push输出数据,并且在源代码内部可以发现通过参数传递的read方法实际上被赋值给this._read,然后在Readable.prototype.read中调用this._read获取数据。

灵魂代码

为了进一步说明stream.Readable的data事件触发顺序与场景,将有关官方源码经过修改和删减成如下:

function Readable(options) {
  this._read = options.read; // 将参数传递的read函数赋值到this._read
}
// 使用者通过调用read方法获取数据
Readable.prototype.read = function (size) {
  var state = this._readableState;
  // 模拟锁,一次_read如果没有返回(this.push),后续read不会继续调用_read读取数据
  if (!state.reading) {
    state.reading = true;
    state.sync = true; // sync用于在push方法中指示_read内部是否同步调用了push
    this._read(size);
    state.sync = false;    
  }
  // _read内部如果是同步调用push,数据会放入缓冲区
  // _read内部如果是异步调用push且缓冲区没有内容,数据可能emit data返回
  // 尝试从缓冲区(state.buffer)中获取大小为size的数据,如果获取成功则触发data事件
  if (ret) 
    this.emit('data', ret);
  return ret;
};
// 在this._read执行过程中通过this.push输出数据
Readable.prototype.push = function (chunk, encoding) {
  var state = this._readableState;
  // 本次_read获取到数据,打开锁
  state.reading = false;
  // 流动模式 & 缓冲区没有数据 & 非同步返回,则直接触发data事件
  if (state.flowing && state.length === 0 && !state.sync) {
    stream.emit('data', chunk);
    stream.read(0); // 触发下一次读取,_read异步push的话还是会到这里,类似flow中的保持流出于流动
  }
  else {
    // 将数据放入缓冲区
    state.length  = chunk.length;
    state.buffer.push(chunk);
  }
};
// 暂停流动
Readable.prototype.pause = function() {
  if (this._readableState.flowing !== false) {
    this._readableState.flowing = false;
    this.emit('pause');
  }
  return this;
};
function flow(stream) {
  const state = stream._readableState;
  while (state.flowing && stream.read() !== null);
}

data事件的触发时机与顺序

时机

data的触发只有两处:

  • 流如果处于流动模式 & 缓冲区没有数据 & 异步调用push,此时数据不经过缓冲区,直接触发data事件
  • 不满足上述情况时,push的数据会被放入缓冲区,然后再尝试从缓冲区读取指定size的数据并触发data事件

顺序

关于data的触发顺序,实际是由emit顺序决定,为讨论原始问题:“increasedAwaitDrain相关逻辑为什么可以被删除?”,将代码简化:

let count = 0;
src.on('data', chunk => {
  let ret = dest.write(chunk);
  if (!ret) {
    count  ;
    src.pause();
  }
});

当监听流的data事件时,流最终会通过resume并调用flow函数进入流动模式模式,即不断的调用read方法读取数据。接下来分析以下几种场景,当dest.write(chunk)返回false时 count会执行几次,注意结合前文的灵魂代码。

  • 场景一:每次_read同步push一次数据

当发生第一次读取,数据同步push到缓冲区,紧接着从缓冲区中读取数据并通过emit data的方式传递到ondata中,如果此时dest.write(chunk)返回false,count 将执行一次,接着由于调用了stream.pause(),while条件state.flowing为false导致stream.read不再被调用,在流重新流动前,count的值不会继续增加。

  • 场景二:每次_read异步push一次数据

当发生第一次读取,异步push的数据将直接通过emit data传递到ondata中,而read函数中的emit由于无法从缓冲区读取数据从而不会触发,同时read返回null导致while循环也相应停止,此种情况下异步push触发data事件后,紧接着的stream.read(0)会继续保持流的流动,当dest.write(chunk)返回false,count 执行一次并将流暂停,紧接着会继续调用一次read,但这次数据将被放入缓冲区且不触发data事件,count 依旧只执行一次。

场景二流暂停一次后再次流动时,数据消耗模式与之前会有所差异,会优先消耗缓冲区数据直至为空时回到之前的模式,但这同样不会导致count 执行多次。

  • 场景三:每次_read多次同步push数据

与场景一类似,只是每次_read会多次往缓冲区写入数据,最终data事件还是依靠从缓冲区读数据后触发。

  • 场景四:每次_read多次异步push数据

同场景二类似,假设在一次_read中有两次异步push,当第一个异步push执行时,data事件触发且其中的dest.write(chunk)返回false,导致count 同时流被暂停,等第二个异步push执行时,由于流已经暂停,数据将写入缓冲区而不是触发data事件,所以count 只执行一次。

  • 场景五:_read操作可能同步或异步push

不管是同步或者异步push,当一次ondata内部将流设置为暂停模式后,flow函数中while条件state.flowing为false将导致stream.read不再调用,异步的push的emit data判断条件同样不再满足,即目前阶段内部不会再有data事件触发直到外部再次间接或直接调用read方法。

以上五个场景是为了分析该问题而模拟的,实际只要能理解第五个场景就能明白所有。

小结

文章最终写出来的内容与我最开始的初衷所偏离,而且自己不知道如何评价这篇文章的好坏,但为了写这文章花了两天业余时间去深入理解stream.Readable却是非常有收获的一件事情,更坚定自己在写文章的路途上可以走的更远。

PS:猜测为什么有烂电影的存在,可能是因为导演长时间投入的创作会让他迷失在内部而无法发现问题,写文章也是,难以通过阅读去优化费心思写的文章。

PS:下图是美团博客的,也许我写了这么多却抵不上这张图,说明方式很重要。

总结

以上就是这篇文章的全部内容了,希望本文的内容对大家的学习或者工作具有一定的参考学习价值,谢谢大家对Devmax的支持。如果你想了解更多相关内容请查看下面相关链接

Node.js Stream ondata触发时机与顺序的探索的更多相关文章

  1. CentOS 8.2服务器上安装最新版Node.js的方法

    这篇文章主要介绍了CentOS 8.2服务器上安装最新版Node.js的方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下

  2. node.js三个步骤实现一个服务器及Express包使用

    这篇文章主要介绍了node.js三个步骤实现一个服务器及Express包使用,文章通过新建一个文件展开全文内容,具有一定的参考价值,需要的小伙伴可以参考一下

  3. Node.js调试技术总结分享

    Node.js是一个可以快速构建网络服务及应用的平台。该平台的构建是基于Chrome's JavaScript runtime,也就是说,实际上它是对Google V8引擎(应用于Google Chrome浏览器)进行了封装。 今天介绍Node.js调式目前有几种技术,需要的朋友可以参考下。

  4. node.js实现http服务器与浏览器之间的内容缓存操作示例

    这篇文章主要介绍了node.js实现http服务器与浏览器之间的内容缓存操作,结合实例形式分析了node.js http服务器与浏览器之间的内容缓存原理与具体实现技巧,需要的朋友可以参考下

  5. 教你如何使用node.js制作代理服务器

    本文介绍了如何使用node.js制作代理服务器,图文并茂,十分的详细,代码很简洁易懂,这里推荐给大家。

  6. node.js中的fs.openSync方法使用说明

    这篇文章主要介绍了node.js中的fs.openSync方法使用说明,本文介绍了fs.openSync方法说明、语法、接收参数、使用实例和实现源码,需要的朋友可以参考下

  7. Node.js+ELK日志规范的实现

    这篇文章主要介绍了Node.js+ELK日志规范的实现,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧

  8. node.js爬虫框架node-crawler初体验

    这篇文章主要介绍了node.js爬虫框架node-crawler的相关资料,帮助大家利用node.js进行爬虫,感兴趣的朋友可以了解下

  9. node.js中的fs.existsSync方法使用说明

    这篇文章主要介绍了node.js中的fs.existsSync方法使用说明,本文介绍了fs.existsSync方法说明、语法、接收参数、使用实例和实现源码,需要的朋友可以参考下

  10. 说说如何利用 Node.js 代理解决跨域问题

    这篇文章主要介绍了Node.js代理解决跨域问题,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,需要的朋友们下面随着小编来一起学习学习吧

随机推荐

  1. Error: Cannot find module ‘node:util‘问题解决

    控制台 安装 Vue-Cli 最后一步出现 Error: Cannot find module 'node:util' 问题解决方案1.问题C:\Windows\System32>cnpm install -g @vue/cli@4.0.3internal/modules/cjs/loader.js:638 throw err; &nbs

  2. yarn的安装和使用(全网最详细)

    一、yarn的简介:Yarn是facebook发布的一款取代npm的包管理工具。二、yarn的特点:速度超快。Yarn 缓存了每个下载过的包,所以再次使用时无需重复下载。 同时利用并行下载以最大化资源利用率,因此安装速度更快。超级安全。在执行代码之前,Yarn 会通过算法校验每个安装包的完整性。超级可靠。使用详细、简洁的锁文件格式和明确的安装算法,Yarn 能够保证在不同系统上无差异的工作。三、y

  3. 前端环境 本机可切换node多版本 问题源头是node使用的高版本

    前言投降投降 重头再来 重装环境 也就分分钟的事 偏要折腾 这下好了1天了 还没折腾出来问题的源头是node 使用的高版本 方案那就用 本机可切换多版本最终问题是因为nodejs的版本太高,导致的node-sass不兼容问题,我的node是v16.14.0的版本,项目中用了"node-sass": "^4.7.2"版本,无法匹配当前的node版本根据文章的提

  4. nodejs模块学习之connect解析

    这篇文章主要介绍了nodejs模块学习之connect解析,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧

  5. nodejs npm package.json中文文档

    这篇文章主要介绍了nodejs npm package.json中文文档,本文档中描述的很多行为都受npm-config(7)的影响,需要的朋友可以参考下

  6. 详解koa2学习中使用 async 、await、promise解决异步的问题

    这篇文章主要介绍了详解koa2学习中使用 async 、await、promise解决异步的问题,小编觉得挺不错的,现在分享给大家,也给大家做个参考。一起跟随小编过来看看吧

  7. Node.js编写爬虫的基本思路及抓取百度图片的实例分享

    这篇文章主要介绍了Node.js编写爬虫的基本思路及抓取百度图片的实例分享,其中作者提到了需要特别注意GBK转码的转码问题,需要的朋友可以参考下

  8. CentOS 8.2服务器上安装最新版Node.js的方法

    这篇文章主要介绍了CentOS 8.2服务器上安装最新版Node.js的方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友可以参考下

  9. node.js三个步骤实现一个服务器及Express包使用

    这篇文章主要介绍了node.js三个步骤实现一个服务器及Express包使用,文章通过新建一个文件展开全文内容,具有一定的参考价值,需要的小伙伴可以参考一下

  10. node下使用UglifyJS压缩合并JS文件的方法

    下面小编就为大家分享一篇node下使用UglifyJS压缩合并JS文件的方法,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧

返回
顶部