node中的Stream－Readable和Writeable解讀

在node中，只要涉及到文件IO的場景一般都會涉及到一個類－Stream。Stream是對IO設備的抽象表示，其在JAVA中也有涉及，主要體現在四個類－InputStream、Reader、OutputStream、Writer，其中InputStream和OutputStream類針對字節數據進行讀寫；Reader和Writer針對字符數據讀寫。同時Java中有多種針對這四種類型的擴展類，如節點流、緩沖流和轉換流等。比較而言，node中Stream類型也和Java中的類似，同樣提供了支持字節和字符讀寫的Readable和Writeable類，也存在轉換流Transform類，本文主要分析node中Readable和Writeable的實現機制，從底層的角度更好的理解Readable和Writeable實現機制，解讀在讀寫過程中發生的一些重要事件。

Readable類

Readable對應於Java中的InputStream和Reader兩個類，針對Readable設置encode編碼可完成內部數據由Buffer到字符的轉換。Readable Stream有兩種模式，即flowing和paused模式。這兩種模式對於用戶而言區別在於是否需要手動調用Readable.prototype.read(n)，讀取緩沖區的數據。查詢node API文檔可知觸發flowing模式有三種方式：

• 偵聽data事件
• readable.resume()
• readable.pipe()
  而觸發paused模式同樣有幾種方式：
• 移除data事件
• readable.pause()
• readable.unpipe()
  可能這樣講解大家仍不明白Readable Stream這兩種模式的區別，那么下文從更深層次分析兩種模式的機制。

深入Readable的實現

Readable繼承EventEmitter，大家也都知道。但是相信大家應該不怎么熟悉Readable的實例屬性_readableState。該屬性是一個ReadableState類型的對象，保存了Readable實例的重要信息，如讀取模式（是否為對象模式）、highWaterMark（緩沖區存放的最大字節數）、緩沖區、flowing模式等。在Readable的實現中，處處使用ReadableState對象記錄當前讀取狀態，並設置緩沖區保證讀操作的順利進行。

首先需要針對Readable.prototype.read方法進行特別解讀：

  if (n === 0 &&
      state.needReadable &&
      (state.length >= state.highWaterMark || state.ended)) {
    debug('read: emitReadable', state.length, state.ended);
    if (state.length === 0 && state.ended)
      endReadable(this);
    else
      emitReadable(this);
    return null;
  }

當讀入的數據為0時，執行emitReadable操作。這意味着，針對Readable Stream執行read(0)方法會觸發readable事件，但是不會讀當前緩沖區。因此使用read(0)可以完成一些比較巧妙的事情，如在readable處理函數中可以使用read(0)觸發下一次readable事件，可選的操作讀緩沖區。

繼續分析代碼，如果讀入的數據並不是0，則計算讀取緩沖區的具體字節數，

n = howMuchToRead(n, state);

function howMuchToRead(n, state) {
  if (state.length === 0 && state.ended)
    return 0;

  if (state.objectMode)
    return n === 0 ? 0 : 1;

  if (n === null || isNaN(n)) {
    // only flow one buffer at a time
    if (state.flowing && state.buffer.length)
      return state.buffer[0].length;
    // 若是paused狀態，則讀全部的緩沖區
    else
      return state.length;
  }

  if (n <= 0)
    return 0;

  if (n > state.highWaterMark)
    state.highWaterMark = computeNewHighWaterMark(n);

  // don't have that much.  return null, unless we've ended.
  if (n > state.length) {
    if (!state.ended) {
      state.needReadable = true;
      return 0;
    } else {
      return state.length;
    }
  }

  return n;
}

針對對象模式的讀取，每次只讀一個；對於處在flowing模式下的讀取，每次只讀緩沖區中第一個buffer的長度；在paused模式下則讀取全部緩沖區的長度；若讀取的字節數大於設置的緩沖區最大值，則適當擴大緩沖區的大小（默認為16k，最大為8m）；若讀取的長度大於當前緩沖區的大小，設置needReadable屬性並准備數據等待下一次讀取。

接下來，判斷是否需要准備數據。在這里，依賴於needReadable的值，

var doRead = state.needReadable;
  debug('need readable', doRead);

  if (state.length === 0 || state.length - n < state.highWaterMark) {
    doRead = true;
    debug('length less than watermark', doRead);
  }

  // reading, then it's unnecessary.
  if (state.ended || state.reading) {
    doRead = false;
    debug('reading or ended', doRead);
  }

如果當前緩沖區為空，或者緩沖區並未超出我們設定的最大值，那么就可以繼續准備數據；如果此時正在准備數據或者已經結束讀取，那么就放棄准備數據。一旦doRead為true，那么進入准備數據階段，

  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
    // 默認Readable未實現_read，拋出Error
    // 針對自定義的Readable子類，_read可修改state.buffer的數量，進行預處理，
    // 然后由下面的fromList讀出去緩存中的相關數據
    this._read(state.highWaterMark);
    state.sync = false;
  }

接下來設置相關的標志位，進行_read處理。針對這個私有方法_read，文檔上有特殊說明，自定義的Readable實現類需要實現這個方法，在該方法中手動添加數據到Readable對象的讀緩沖區，然后進行Readable的讀取。可以理解為_read函數為讀取數據前的准備工作（准備數據），針對的是流的實現者而言。

  if (doRead && !state.reading)
    n = howMuchToRead(nOrig, state);
    
  var ret;
  if (n > 0)
    ret = fromList(n, state);
  else
    ret = null;

  if (ret === null) {
    state.needReadable = true;
    n = 0;
  }

  state.length -= n;

  if (state.length === 0 && !state.ended)
    state.needReadable = true;

  if (nOrig !== n && state.ended && state.length === 0)
    endReadable(this);

  // flowing模式下的數據讀取依賴於 read函數
  // data事件觸發的次數，依賴於howMuchToRead計算的次數
  if (ret !== null)
    this.emit('data', ret);

一旦在_read中更新了緩沖區，那么我們需要重新計算（消費者，即可寫流）讀取的字節數。fromList方法完成了讀緩沖區的slice，如果是objectMode下的讀，則只讀緩沖區的第一個對象；針對未傳參數的read方法而言，默認讀取全部緩沖區等等。從讀緩沖區讀取完數據之后設置相關flag，如needReadable，最終，觸發data事件，結束！

上節提到，設置data事件的執行函數會進入flowing模式的讀，而上文看到正是read方法觸發了data事件，而默認條件下Readable處於paused狀態，因此在paused狀態讀取數據需要手動執行read函數，每次read讀取完畢觸發一次data事件。從這點看出，flowing和paused狀態區別在於是否需要手動執行read()來獲取數據。flowing狀態下，我們無需執行read，僅需要設置data事件處理函數或者設定導流目標pipe；而在paused狀態下，不僅僅是簡單的執行read方法，因為讀緩沖區的內容時刻在改變，一旦讀緩沖區又有新數據，簡單執行read()就沒法滿足需求(因為我們無法知道是否又有新數據到來)，因此需要偵聽讀緩沖區的相關事件，即readable事件，在該事件處理函數中進行read相關數據。

那么，什么情況下會觸發readable事件呢？在實現_read私有方法中，我們使用stream.push(chunk)或stream.unshift(chunk)方法注入數據到讀緩沖區，那么push和unshift方法都實現了下面的邏輯，

if (state.flowing && state.length === 0 && !state.sync) {
  stream.emit('data', chunk);
  stream.read(0);
} 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);
}

function emitReadable(stream) {
  var state = stream._readableState;
  state.needReadable = false;
  if (!state.emittedReadable) {
    debug('emitReadable', state.flowing);
    state.emittedReadable = true;
    if (state.sync)
      process.nextTick(emitReadable_, stream);
    else
      emitReadable_(stream);
  }
}

function emitReadable_(stream) {
  debug('emit readable');
  stream.emit('readable');
  flow(stream);
}
// 在flowing狀態下，自動讀取流（替代paused狀態下手動read）
function flow(stream) {
  var state = stream._readableState;
  debug('flow', state.flowing);
  if (state.flowing) {
    do {
      var chunk = stream.read();
    } while (null !== chunk && state.flowing);
  }
}

一旦處於flowing模式並且當前緩沖區沒有數據，那么就立即將預處理的push（unshift）數據傳遞給data事件處理函數，並執行stream.read(0)。前文已經交代過，read(0)僅僅用來觸發readable事件，並不讀取緩沖區，這就是觸發readable的第一種情況。

第二種則是第一種情況之外的所有情景，即根據操作（push、unshift）的不同將數據插入讀緩沖區的不同位置。最后執行emitReadable函數，觸發readable事件。針對emitReadable函數，它的作用就是異步觸發readable事件，並執行flow函數。flow函數則針對flowing狀態的Readable做自適應讀取，免去了手動執行read函數和何時執行read函數的苦惱。

這樣，對於Readable的實現者，一旦在_read函數插入有效數據到讀緩沖區，都會觸發readable事件，在paused狀態下，設置readable事件處理函數並手動執行read函數，便可完成數據的讀取；而在flowing狀態下，通過設置data事件處理函數或者定義pipe目標流同樣可以實現讀取。

既然pipe同樣可以觸發Readable進入flowing狀態，那么pipe方法具體做了什么呢？其實pipe針對Readable和Writeable做了限流，首先針對Readable的data事件進行偵聽，並執行Writeable的write函數，當Writeable的寫緩沖區大於一個臨界值（highWaterMark），導致write函數返回false（此時意味着Writeable無法匹配Readable的速度，Writeable的寫緩沖區已經滿了），此時，pipe修改了Readable模式，執行pause方法，進入paused模式，停止讀取讀緩沖區。而同時Writeable開始刷新寫緩沖區，刷新完畢后異步觸發drain事件，在該事件處理函數中，設置Readable為flowing狀態，並繼續執行flow函數不停的刷新讀緩沖區，這樣就完成了pipe限流。需要注意的是，Readable和Writeable各自維護了一個緩沖區，在實現的上有區別：Readable的緩沖區是一個數組，存放Buffer、String和Object類型；而Writeable則是一個有向鏈表，依次存放需要寫入的數據。

Writeable解讀

Writeable對應Java的OutputStream和Writer類，實現字節和字符數據的寫。與Readable類似，Writeable的實例對象同樣維護了一個狀態對象－WriteableState，記錄了當前輸出流的狀態信息，如寫緩沖區的最大值（hightWaterMark）、緩沖區（有向鏈表）和緩沖區長度等信息。在本節中，主要分析輸出流的關鍵方法write和事件drain，並解析輸出流的實現者需要實現的方法_write和write的關系。

function write
----------------------------
if (state.ended)
    writeAfterEnd(this, cb);
  else if (validChunk(this, state, chunk, cb)) {
    state.pendingcb++;
    ret = writeOrBuffer(this, state, chunk, encoding, cb);
  }

  return ret;

在write方法中，判斷寫入數據的格式並執行writeOrBuffer函數，並返回執行結果，該返回值標示當前寫緩沖區是否已滿。真正執行寫入邏輯的是writeOrBuffer函數，該函數的作用在於刷新或者更新寫緩沖區，下面看看主要做了什么，

function writeOrBuffer(stream, state, chunk, encoding, cb) {
  chunk = decodeChunk(state, chunk, encoding);

  if (chunk instanceof Buffer)
    encoding = 'buffer';
  var len = state.objectMode ? 1 : chunk.length;

  state.length += len;

  // 如果緩存的長度大於highWaterMark，需要刷新緩沖，所以設置needDrain標志
  var ret = state.length < state.highWaterMark;
  // we must ensure that previous needDrain will not be reset to false.
  if (!ret)
    state.needDrain = true;

  // 緩存未處理的寫請求，在clearBuffer中執行緩存
  // 由此看出，Readable和Writeable都有緩存，Readable 中緩存的方式是數組（項為Buffer,字符串或對象），Writeable的
  // 緩存則是對象鏈表
  if (state.writing || state.corked) {
    var last = state.lastBufferedRequest;
    state.lastBufferedRequest = new WriteReq(chunk, encoding, cb);
    if (last) {
      last.next = state.lastBufferedRequest;
    } else {
      state.bufferedRequest = state.lastBufferedRequest;
    }
    state.bufferedRequestCount += 1;
  } else {
    doWrite(stream, state, false, len, chunk, encoding, cb);
  }

  return ret;
}

writeOrBuffer首先針對數據進行編碼，字符串轉換成Buffer類型，如果設置了Writeable的ObjectMode模式則仍為Object類型；接下來更新寫緩沖區的長度，並判斷寫緩沖區長度是否超過設定的Writeable的最大值（默認16k），如果超過超過則ret＝false並更新WriteableState的屬性needDrain＝true。ret的結果其實就是write方法返回值，因此一旦write返回值為false，意味着當前寫緩沖區已滿，需要停止繼續寫入數據。

在Readable的pipe方法中，涉及到了Writeable的drain事件。該事件的觸發意味着寫緩沖區已可以繼續緩存數據，可見drain事件與寫緩沖區嚴格相關。繼續分析writeOrBuffer函數，若當前輸出流正在寫數據，那么則當前數據緩存至寫緩沖區（創建WriteReq對象）；否則執行doWrite函數，刷新緩沖區。

function doWrite(stream, state, writev, len, chunk, encoding, cb) {
  state.writelen = len;
  state.writecb = cb;
  state.writing = true;
  state.sync = true;
  if (writev)
    stream._writev(chunk, state.onwrite);
  else
    stream._write(chunk, encoding, state.onwrite);
  state.sync = false;
}

doWrite函數設置了需要寫入數據的長度、寫入狀態等信息，並執行輸出流實現者需要實現的_write函數。在_write函數中，針對數據流向做最后的處理，這里分析_write函數的具體實現。_write函數有三個參數，分別為chunk,encoding和state.onwrite回調函數，對該回調函數稍后分析，先着重講解_write函數的實現。在node的fs模塊中，可以通過fs.createWriteStream創建Writeable實例，通過執行

var writeStream = fs.createWriteStream('./output',{decodeStrings: false});
console.log(writeStream._write.toString());

-----------------輸出-----------------

function (data, encoding, cb) {
  if (!(data instanceof Buffer))
    return this.emit('error', new Error('Invalid data'));

  if (typeof this.fd !== 'number')
    return this.once('open', function() {
      this._write(data, encoding, cb);
    });

  var self = this;
  fs.write(this.fd, data, 0, data.length, this.pos, function(er, bytes) {
    if (er) {
      self.destroy();
      return cb(er);
    }
    self.bytesWritten += bytes;
    cb();
  });

  if (this.pos !== undefined)
    this.pos += data.length;
}

看出，在_write實現中，只接受Buffer類型的數據，接着執行fs.write操作，寫入到對應文件描述符fd對應的文件中，寫入成功或失敗后執行回調函數，即state.onwrite函數。

function onwrite(stream, er) {
  var state = stream._writableState;
  var sync = state.sync;
  var cb = state.writecb;

  onwriteStateUpdate(state);

  // 默認未重寫_write方法，會收到er值
  if (er)
    onwriteError(stream, state, sync, er, cb);
  else {
    // Check if we're actually ready to finish, but don't emit yet
    var finished = needFinish(state);

    // 寫緩存的數據
    if (!finished &&
        !state.corked &&
        !state.bufferProcessing &&
        state.bufferedRequest) {
      clearBuffer(stream, state);
    }

    // 異步觸發drain事件
    if (sync) {
      process.nextTick(afterWrite, stream, state, finished, cb);
    } else {
      afterWrite(stream, state, finished, cb);
    }
  }
}

在state.onwrite函數中主要工作有兩個：

• 寫緩沖區的數據
• 寫完緩沖區的數據后，異步觸發drain事件

第一步，在clearBuffer函數中，就是取出寫緩沖區（有向鏈表）的第一個WriteReq對象，執行doWrite函數，寫入緩沖區的第一個數據；這樣循環往復最終清空寫緩沖區，重置一些標志位。

第二步，異步執行afterWrite函數，觸發drain事件，並判斷是否寫操作完畢觸發“finish”事件。這里之所以強調異步觸發drain事件，是因為為了保證先獲得write()返回值為false，給用戶綁定drain處理函數的時隙，然后再觸發drain事件。

至此，Writeable的重要流程已全部走通。可以看出來，在核心的write()中，判斷寫緩沖區是否已滿並返回該值，在適當條件下緩存數據或調用_write()寫數據，在Writeable實現者需要實現的** _write() 中，主要任務是數據寫入方向控制，完成最基本的任務**。

總結

對比Readable的read()和_read(),我總結了下這四個函數在“讀寫過程”中的執行順序與關系，如下圖所示：