上次寫Stream pipe細(xì)節(jié)時，在源碼中發(fā)現(xiàn)一段無用邏輯，由此引發(fā)了對Stream data事件觸發(fā)時機(jī)與順序的探索。

無用邏輯

當(dāng)時研究pipe細(xì)節(jié)是基于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();  }}

重點(diǎn)關(guān)注increasedAwaitDrain變量，理解這個變量期望達(dá)到什么目的，然后仔細(xì)閱讀代碼，會發(fā)現(xiàn)if (false === ret && !increasedAwaitDrain)語句中increasedAwaitDrain變量肯定是false，因為前一行才將該變量賦值為false，這樣一來這個變量就變得毫無意義。

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

以上就是關(guān)鍵的三行代碼，因為Node.js是單線程且dest.write(chunk)內(nèi)部沒有修改變量increasedAwaitDrain的值，那么if語句中increasedAwaitDrain的值肯定還是false，即increasedAwaitDrain相關(guān)邏輯沒有達(dá)到所期望的目標(biāo)。

無用代碼出現(xiàn)的原因

前段雖已經(jīng)分析出increasedAwaitDrain沒起到作用，但作者為什么寫了這樣一段邏輯呢？其實在定義increasedAwaitDrain語句的上方，作者說可能存在這樣一種情況：“當(dāng)我們接收到一次上游的ondata事件并嘗試將數(shù)據(jù)寫到下游時，上游可能同時又有一個data事件觸發(fā)，而這兩個ondata的數(shù)據(jù)在寫入下游時可能都返回false，從而導(dǎo)致src._readableState.awaitDrain++執(zhí)行兩次”。

awaitDrain++執(zhí)行兩次是作者不希望看到的情況，因為下游觸發(fā)drain事件時awaitDrain相應(yīng)減1，直到其值為0時才讓上游重新流動，如果awaitDrain++執(zhí)行兩次，下游卻只觸發(fā)一次drain事件，awaitDrain就不會為0，上游不重新流動也就無法繼續(xù)讀取數(shù)據(jù)。

真相的探索過程

雖然從理性上認(rèn)為increasedAwaitDrain沒起到作用，但也無法肯定加絕對，自己嘗試去求助，沒有出現(xiàn)高手指點(diǎn)出問題所在，但一個同事聽我描述后，說可能這就是個BUG，雖心中覺得可能性不大，但還是抱著試試看的心態(tài)切換到master分支上去瞅瞅，隨即發(fā)現(xiàn)最新的代碼里并沒有與increasedAwaitDrain類似的邏輯，間接說明v8.11.1分支上increasedAwaitDrain相關(guān)邏輯的確無用。

雖然比較肯定這里存在一段無用代碼，但應(yīng)該如何理解作者在increasedAwaitDrain上方的注釋呢？為了進(jìn)一步揭露真相，自己繼續(xù)花時間去看了看stream.Readable相關(guān)代碼，想知道data事件的觸發(fā)時機(jī)與順序是如何決定的。

readable流的簡單原理

在進(jìn)一步解釋data事件的觸發(fā)順序前，簡單講一下readable流的實現(xiàn)原理，如果需要自己實現(xiàn)一個readable流，可以使用new stream.Readable(options)方法，其中options可包含四個屬性：highWaterMark、encoding、objectMode、read。最主要的是read屬性，當(dāng)流的使用者需要數(shù)據(jù)時，read方法被用來從數(shù)據(jù)源獲取數(shù)據(jù)，然后通過this.push(chunk)將數(shù)據(jù)傳遞給使用者，如果沒有更多數(shù)據(jù)可供讀取時使用this.push(null)表示讀取結(jié)束。

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雖然沒有明顯調(diào)用read方法，但內(nèi)部依舊是通過調(diào)用read方法結(jié)合this.push輸出數(shù)據(jù)，并且在源代碼內(nèi)部可以發(fā)現(xiàn)通過參數(shù)傳遞的read方法實際上被賦值給this._read，然后在Readable.prototype.read中調(diào)用this._read獲取數(shù)據(jù)。

靈魂代碼

為了進(jìn)一步說明stream.Readable的data事件觸發(fā)順序與場景，將有關(guān)官方源碼經(jīng)過修改和刪減成如下：

function Readable(options) {  this._read = options.read; // 將參數(shù)傳遞的read函數(shù)賦值到this._read}// 使用者通過調(diào)用read方法獲取數(shù)據(jù)Readable.prototype.read = function (size) {  var state = this._readableState;  // 模擬鎖，一次_read如果沒有返回（this.push），后續(xù)read不會繼續(xù)調(diào)用_read讀取數(shù)據(jù)  if (!state.reading) {    state.reading = true;    state.sync = true; // sync用于在push方法中指示_read內(nèi)部是否同步調(diào)用了push    this._read(size);    state.sync = false;      }  // _read內(nèi)部如果是同步調(diào)用push，數(shù)據(jù)會放入緩沖區(qū)  // _read內(nèi)部如果是異步調(diào)用push且緩沖區(qū)沒有內(nèi)容，數(shù)據(jù)可能emit data返回  // 嘗試從緩沖區(qū)（state.buffer）中獲取大小為size的數(shù)據(jù)，如果獲取成功則觸發(fā)data事件  if (ret)     this.emit('data', ret);  return ret;};// 在this._read執(zhí)行過程中通過this.push輸出數(shù)據(jù)Readable.prototype.push = function (chunk, encoding) {  var state = this._readableState;  // 本次_read獲取到數(shù)據(jù)，打開鎖  state.reading = false;  // 流動模式 & 緩沖區(qū)沒有數(shù)據(jù) & 非同步返回，則直接觸發(fā)data事件  if (state.flowing && state.length === 0 && !state.sync) {    stream.emit('data', chunk);    stream.read(0); // 觸發(fā)下一次讀取，_read異步push的話還是會到這里，類似flow中的保持流出于流動  }  else {    // 將數(shù)據(jù)放入緩沖區(qū)    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事件的觸發(fā)時機(jī)與順序

時機(jī)

data的觸發(fā)只有兩處：

• 流如果處于流動模式 & 緩沖區(qū)沒有數(shù)據(jù) & 異步調(diào)用push，此時數(shù)據(jù)不經(jīng)過緩沖區(qū)，直接觸發(fā)data事件
• 不滿足上述情況時，push的數(shù)據(jù)會被放入緩沖區(qū)，然后再嘗試從緩沖區(qū)讀取指定size的數(shù)據(jù)并觸發(fā)data事件

順序

關(guān)于data的觸發(fā)順序，實際是由emit順序決定，為討論原始問題：“increasedAwaitDrain相關(guān)邏輯為什么可以被刪除？”，將代碼簡化：

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

當(dāng)監(jiān)聽流的data事件時，流最終會通過resume并調(diào)用flow函數(shù)進(jìn)入流動模式模式，即不斷的調(diào)用read方法讀取數(shù)據(jù)。接下來分析以下幾種場景，當(dāng)dest.write(chunk)返回false時++count會執(zhí)行幾次，注意結(jié)合前文的靈魂代碼。

• 場景一：每次_read同步push一次數(shù)據(jù)

當(dāng)發(fā)生第一次讀取，數(shù)據(jù)同步push到緩沖區(qū)，緊接著從緩沖區(qū)中讀取數(shù)據(jù)并通過emit data的方式傳遞到ondata中，如果此時dest.write(chunk)返回false，count++將執(zhí)行一次，接著由于調(diào)用了stream.pause()，while條件state.flowing為false導(dǎo)致stream.read不再被調(diào)用，在流重新流動前，count的值不會繼續(xù)增加。

• 場景二：每次_read異步push一次數(shù)據(jù)

當(dāng)發(fā)生第一次讀取，異步push的數(shù)據(jù)將直接通過emit data傳遞到ondata中，而read函數(shù)中的emit由于無法從緩沖區(qū)讀取數(shù)據(jù)從而不會觸發(fā)，同時read返回null導(dǎo)致while循環(huán)也相應(yīng)停止，此種情況下異步push觸發(fā)data事件后，緊接著的stream.read(0)會繼續(xù)保持流的流動，當(dāng)dest.write(chunk)返回false，count++執(zhí)行一次并將流暫停，緊接著會繼續(xù)調(diào)用一次read，但這次數(shù)據(jù)將被放入緩沖區(qū)且不觸發(fā)data事件，count++依舊只執(zhí)行一次。

場景二流暫停一次后再次流動時，數(shù)據(jù)消耗模式與之前會有所差異，會優(yōu)先消耗緩沖區(qū)數(shù)據(jù)直至為空時回到之前的模式，但這同樣不會導(dǎo)致count++執(zhí)行多次。

• 場景三：每次_read多次同步push數(shù)據(jù)

與場景一類似，只是每次_read會多次往緩沖區(qū)寫入數(shù)據(jù)，最終data事件還是依靠從緩沖區(qū)讀數(shù)據(jù)后觸發(fā)。

• 場景四：每次_read多次異步push數(shù)據(jù)

同場景二類似，假設(shè)在一次_read中有兩次異步push，當(dāng)?shù)谝粋€異步push執(zhí)行時，data事件觸發(fā)且其中的dest.write(chunk)返回false，導(dǎo)致count++同時流被暫停，等第二個異步push執(zhí)行時，由于流已經(jīng)暫停，數(shù)據(jù)將寫入緩沖區(qū)而不是觸發(fā)data事件，所以count++只執(zhí)行一次。

• 場景五：_read操作可能同步或異步push

不管是同步或者異步push，當(dāng)一次ondata內(nèi)部將流設(shè)置為暫停模式后，flow函數(shù)中while條件state.flowing為false將導(dǎo)致stream.read不再調(diào)用，異步的push的emit data判斷條件同樣不再滿足，即目前階段內(nèi)部不會再有data事件觸發(fā)直到外部再次間接或直接調(diào)用read方法。

以上五個場景是為了分析該問題而模擬的，實際只要能理解第五個場景就能明白所有。

小結(jié)

文章最終寫出來的內(nèi)容與我最開始的初衷所偏離，而且自己不知道如何評價這篇文章的好壞，但為了寫這文章花了兩天業(yè)余時間去深入理解stream.Readable卻是非常有收獲的一件事情，更堅定自己在寫文章的路途上可以走的更遠(yuǎn)。

PS：猜測為什么有爛電影的存在，可能是因為導(dǎo)演長時間投入的創(chuàng)作會讓他迷失在內(nèi)部而無法發(fā)現(xiàn)問題，寫文章也是，難以通過閱讀去優(yōu)化費(fèi)心思寫的文章。

PS：下圖是美團(tuán)博客的，也許我寫了這么多卻抵不上這張圖，說明方式很重要。

總結(jié)

以上就是這篇文章的全部內(nèi)容了，希望本文的內(nèi)容對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值，謝謝大家對VeVb武林網(wǎng)的支持。