[[384117]]

通過(guò)流我們可以將一大塊數(shù)據(jù)拆分為一小部分一點(diǎn)一點(diǎn)的流動(dòng)起來(lái)，而無(wú)需一次性全部讀入，在 Linux 下我們可以通過(guò) | 符號(hào)實(shí)現(xiàn)，類似的在 Nodejs 的 Stream 模塊中同樣也為我們提供了 pipe() 方法來(lái)實(shí)現(xiàn)。

1. Stream pipe 基本示例

選擇 Koa 來(lái)實(shí)現(xiàn)這個(gè)簡(jiǎn)單的 Demo，因?yàn)橹坝腥嗽?“Nodejs技術(shù)棧” 交流群?jiǎn)栠^(guò)一個(gè)問(wèn)題，怎么在 Koa 中返回一個(gè) Stream，順便在下文借此機(jī)會(huì)提下。

1.1 未使用 Stream pipe 情況

在 Nodejs 中 I/O 操作都是異步的，先用 util 模塊的 promisify 方法將 fs.readFile 的 callback 形式轉(zhuǎn)為 Promise 形式，這塊代碼看似沒(méi)問(wèn)題，但是它的體驗(yàn)不是很好，因?yàn)樗菍?shù)據(jù)一次性讀入內(nèi)存再進(jìn)行的返回，當(dāng)數(shù)據(jù)文件很大的時(shí)候也是對(duì)內(nèi)存的一種消耗，類似內(nèi)存泄漏這種問(wèn)題也很容易出現(xiàn)，因此不推薦它。

const Koa = require('koa'); 
const fs = require('fs'); 
const app = new Koa(); 
const { promisify } = require('util'); 
const { resolve } = require('path'); 
const readFile = promisify(fs.readFile); 
 
app.use(async ctx => { 
  try { 
    ctx.body = await readFile(resolve(__dirname, 'test.json')); 
  } catch(err) { ctx.body = err }; 
}); 
 
app.listen(3000); 

1.2 使用 Stream pipe 情況

下面，再看看怎么通過(guò) Stream 的方式在 Koa 框架中響應(yīng)數(shù)據(jù)

... 
app.use(async ctx => { 
  try { 
    const readable = fs.createReadStream(resolve(__dirname, 'test.json')); 
    ctx.body = readable; 
  } catch(err) { ctx.body = err }; 
}); 

以上在 Koa 中直接創(chuàng)建一個(gè)可讀流賦值給 ctx.body 就可以了，你可能疑惑了為什么沒(méi)有 pipe 方法，因?yàn)榭蚣芙o你封裝好了，不要被表象所迷惑了，看下相關(guān)源碼：

// https://github.com/koajs/koa/blob/master/lib/application.js#L256 
function respond(ctx) { 
  ... 
  let body = ctx.body; 
  if (body instanceof Stream) return body.pipe(res); 
  ... 
} 

沒(méi)有神奇之處，框架在返回的時(shí)候做了層判斷，因?yàn)?res 是一個(gè)可寫流對(duì)象，如果 body 也是一個(gè) Stream 對(duì)象（此時(shí)的 Body 是一個(gè)可讀流），則使用 body.pipe(res) 以流的方式進(jìn)行響應(yīng)。

1.3 使用 Stream VS 不使用 Stream

看到一個(gè)圖片，不得不說(shuō)畫的實(shí)在太萌了，來(lái)源 https://www.cnblogs.com/vajoy/p/6349817.html

2. pipe 的調(diào)用過(guò)程與實(shí)現(xiàn)原理

以上最后以流的方式響應(yīng)數(shù)據(jù)最核心的實(shí)現(xiàn)就是使用 pipe 方法來(lái)實(shí)現(xiàn)的輸入、輸出，本節(jié)的重點(diǎn)也是研究 pipe 的實(shí)現(xiàn)，最好的打開(kāi)方式通過(guò)閱讀源碼一起來(lái)看看吧。

2.1 順藤摸瓜

在應(yīng)用層我們調(diào)用了 fs.createReadStream() 這個(gè)方法，順藤摸瓜找到這個(gè)方法創(chuàng)建的可讀流對(duì)象的 pipe 方法實(shí)現(xiàn)，以下僅列舉核心代碼實(shí)現(xiàn)，基于 Nodejs v12.x 源碼。

2.1.1 /lib/fs.js

導(dǎo)出一個(gè) createReadStream 方法，在這個(gè)方法里面創(chuàng)建了一個(gè) ReadStream 可讀流對(duì)象，且 ReadStream 來(lái)自 internal/fs/streams 文件，繼續(xù)向下找。

// https://github.com/nodejs/node/blob/v12.x/lib/fs.js 
// 懶加載，主要在用到的時(shí)候用來(lái)實(shí)例化 ReadStream、WriteStream ... 等對(duì)象 
function lazyLoadStreams() { 
  if (!ReadStream) { 
    ({ ReadStream, WriteStream } = require('internal/fs/streams')); 
    [ FileReadStream, FileWriteStream ] = [ ReadStream, WriteStream ]; 
  } 
} 
 
function createReadStream(path, options) { 
  lazyLoadStreams(); 
  return new ReadStream(path, options); // 創(chuàng)建一個(gè)可讀流 
} 
 
module.exports = fs = { 
  createReadStream, // 導(dǎo)出 createReadStream 方法 
  ... 
} 

2.1.2 /lib/internal/fs/streams.js

這個(gè)方法里定義了構(gòu)造函數(shù) ReadStream，且在原型上定義了 open、_read、_destroy 等方法，并沒(méi)有我們要找的 pipe 方法。

但是呢通過(guò) ObjectSetPrototypeOf 方法實(shí)現(xiàn)了繼承，ReadStream 繼承了 Readable 在原型中定義的函數(shù)，接下來(lái)繼續(xù)查找 Readable 的實(shí)現(xiàn)。

// https://github.com/nodejs/node/blob/v12.x/lib/internal/fs/streams.js 
const { Readable, Writable } = require('stream'); 
 
function ReadStream(path, options) { 
  if (!(this instanceof ReadStream)) 
    return new ReadStream(path, options); 
 
  ... 
  Readable.call(this, options); 
  ... 
} 
ObjectSetPrototypeOf(ReadStream.prototype, Readable.prototype); 
ObjectSetPrototypeOf(ReadStream, Readable); 
 
ReadStream.prototype.open = function() { ... }; 
 
ReadStream.prototype._read = function(n) { ... };; 
 
ReadStream.prototype._destroy = function(err, cb) { ... }; 
... 
 
module.exports = { 
  ReadStream, 
  WriteStream 
}; 

2.1.3 /lib/stream.js

在 stream.js 的實(shí)現(xiàn)中，有條注釋：在 Readable/Writable/Duplex/... 之前導(dǎo)入 Stream，原因是為了避免 cross-reference(require)，為什么會(huì)這樣？

第一步 stream.js 這里將 require('internal/streams/legacy') 導(dǎo)出復(fù)制給了 Stream。

在之后的 _stream_readable、Writable、Duplex ... 模塊也會(huì)反過(guò)來(lái)引用 stream.js 文件，具體實(shí)現(xiàn)下面會(huì)看到。

Stream 導(dǎo)入了 internal/streams/legacy

上面 /lib/internal/fs/streams.js 文件從 stream 模塊獲取了一個(gè) Readable 對(duì)象，就是下面的 Stream.Readable 的定義。

// https://github.com/nodejs/node/blob/v12.x/lib/stream.js 
// Note: export Stream before Readable/Writable/Duplex/... 
// to avoid a cross-reference(require) issues 
const Stream = module.exports = require('internal/streams/legacy'); 
 
Stream.Readable = require('_stream_readable'); 
Stream.Writable = require('_stream_writable'); 
Stream.Duplex = require('_stream_duplex'); 
Stream.Transform = require('_stream_transform'); 
Stream.PassThrough = require('_stream_passthrough'); 
... 

2.1.4 /lib/internal/streams/legacy.js

上面的 Stream 等于 internal/streams/legacy，首先繼承了 Events 模塊，之后呢在原型上定義了 pipe 方法，剛開(kāi)始看到這里的時(shí)候以為實(shí)現(xiàn)是在這里了，但后來(lái)看 _stream_readable 的實(shí)現(xiàn)之后，發(fā)現(xiàn) _stream_readable 繼承了 Stream 之后自己又重新實(shí)現(xiàn)了 pipe 方法，那么疑問(wèn)來(lái)了這個(gè)模塊的 pipe 方法是干嘛的？什么時(shí)候會(huì)被用？翻譯文件名 “legacy=遺留”？有點(diǎn)沒(méi)太理解，難道是遺留了？有清楚的大佬可以指點(diǎn)下，也歡迎在公眾號(hào) “Nodejs技術(shù)棧” 后臺(tái)加我微信一塊討論下！

// https://github.com/nodejs/node/blob/v12.x/lib/internal/streams/legacy.js 
const { 
  ObjectSetPrototypeOf, 
} = primordials; 
const EE = require('events'); 
function Stream(opts) { 
  EE.call(this, opts); 
} 
ObjectSetPrototypeOf(Stream.prototype, EE.prototype); 
ObjectSetPrototypeOf(Stream, EE); 
 
Stream.prototype.pipe = function(dest, options) { 
  ... 
}; 
 
module.exports = Stream; 

2.1.5 /lib/_stream_readable.js

在 _stream_readable.js 的實(shí)現(xiàn)里面定義了 Readable 構(gòu)造函數(shù)，且繼承于 Stream，這個(gè) Stream 正是我們上面提到的 /lib/stream.js 文件，而在 /lib/stream.js 文件里加載了 internal/streams/legacy 文件且重寫了里面定義的 pipe 方法。

經(jīng)過(guò)上面一系列的分析，終于找到可讀流的 pipe 在哪里，同時(shí)也更進(jìn)一步的認(rèn)識(shí)到了在創(chuàng)建一個(gè)可讀流時(shí)的執(zhí)行調(diào)用過(guò)程，下面將重點(diǎn)來(lái)看這個(gè)方法的實(shí)現(xiàn)。

module.exports = Readable; 
Readable.ReadableState = ReadableState; 
 
const EE = require('events'); 
const Stream = require('stream'); 
 
ObjectSetPrototypeOf(Readable.prototype, Stream.prototype); 
ObjectSetPrototypeOf(Readable, Stream); 
 
function Readable(options) { 
  if (!(this instanceof Readable)) 
    return new Readable(options); 
 
  ... 
  Stream.call(this, options); // 繼承自 Stream 構(gòu)造函數(shù)的定義 
} 
... 

2.2 _stream_readable 實(shí)現(xiàn)分析

2.2.1 聲明構(gòu)造函數(shù) Readable

聲明構(gòu)造函數(shù) Readable 繼承 Stream 的構(gòu)造函數(shù)和原型。

Stream 是 /lib/stream.js 文件，上面分析了，這個(gè)文件繼承了 events 事件，此時(shí)也就擁有了 events 在原型中定義的屬性，例如 on、emit 等方法。

const Stream = require('stream'); 
ObjectSetPrototypeOf(Readable.prototype, Stream.prototype); 
ObjectSetPrototypeOf(Readable, Stream); 
 
function Readable(options) { 
  if (!(this instanceof Readable)) 
    return new Readable(options); 
 
  ... 
 
  Stream.call(this, options); 
} 

2.2.2 聲明 pipe 方法，訂閱 data 事件

在 Stream 的原型上聲明 pipe 方法，訂閱 data 事件，src 為可讀流對(duì)象，dest 為可寫流對(duì)象。

我們?cè)谑褂?pipe 方法的時(shí)候也是監(jiān)聽(tīng)的 data 事件，一邊讀取數(shù)據(jù)一邊寫入數(shù)據(jù)。

看下 ondata() 方法里的幾個(gè)核心實(shí)現(xiàn)：

• dest.write(chunk)：接收 chunk 寫入數(shù)據(jù)，如果內(nèi)部的緩沖小于創(chuàng)建流時(shí)配置的 highWaterMark，則返回 true，否則返回 false 時(shí)應(yīng)該停止向流寫入數(shù)據(jù)，直到 'drain' 事件被觸發(fā)。
• src.pause()：可讀流會(huì)停止 data 事件，意味著此時(shí)暫停數(shù)據(jù)寫入了。

之所以調(diào)用 src.pause() 是為了防止讀入數(shù)據(jù)過(guò)快來(lái)不及寫入，什么時(shí)候知道來(lái)不及寫入呢，要看 dest.write(chunk) 什么時(shí)候返回 false，是根據(jù)創(chuàng)建流時(shí)傳的 highWaterMark 屬性，默認(rèn)為 16384 (16KB)，對(duì)象模式的流默認(rèn)為 16。

注意：是 16KB 不是 16Kb，也是之前犯的一個(gè)錯(cuò)誤，大寫的 B 和小寫的 b 在這里是有區(qū)別的。計(jì)算機(jī)中所有數(shù)據(jù)都以 0 和 1 表示，其中 0 或 1 稱作一個(gè)位（bit），用小寫的 b 表示。大寫的 B 表示字節(jié)（byte），1byte = 8bit，大寫 K 表示千，所以是千個(gè)位（Kb）和千個(gè)字節(jié)（KB），一般都是使用 KB 表示一個(gè)文件的大小。

Readable.prototype.pipe = function(dest, options) { 
  const src = this; 
  src.on('data', ondata); 
  function ondata(chunk) { 
    const ret = dest.write(chunk); 
    if (ret === false) { 
      ... 
      src.pause(); 
    } 
  } 
  ... 
}; 

2.2.3 訂閱 drain 事件，繼續(xù)流動(dòng)數(shù)據(jù)

上面提到在 data 事件里，如果調(diào)用 dest.write(chunk) 返回 false，就會(huì)調(diào)用 src.pause() 停止數(shù)據(jù)流動(dòng)，什么時(shí)候再次開(kāi)啟呢？

如果說(shuō)可以繼續(xù)寫入事件到流時(shí)會(huì)觸發(fā) drain 事件，也是在 dest.write(chunk) 等于 false 時(shí)，如果 ondrain 不存在則注冊(cè) drain 事件。

Readable.prototype.pipe = function(dest, options) { 
  const src = this; 
  src.on('data', ondata); 
  function ondata(chunk) { 
    const ret = dest.write(chunk); 
    if (ret === false) { 
      ... 
      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.on('drain', ondrain); 
      } 
      src.pause(); 
    } 
  } 
  ... 
}; 
 
// 當(dāng)可寫入流 dest 耗盡時(shí)，它將會(huì)在可讀流對(duì)象 source 上減少 awaitDrain 計(jì)數(shù)器 
// 為了確保所有需要緩沖的寫入都完成，即 state.awaitDrain === 0 和 src 可讀流上的 data 事件存在，切換流到流動(dòng)模式 
function pipeOnDrain(src) { 
  return function pipeOnDrainFunctionResult() { 
    const state = src._readableState; 
    debug('pipeOnDrain', state.awaitDrain); 
    if (state.awaitDrain) 
      state.awaitDrain--; 
    if (state.awaitDrain === 0 && EE.listenerCount(src, 'data')) { 
      state.flowing = true; 
      flow(src); 
    } 
  }; 
} 
 
// stream.read() 從內(nèi)部緩沖拉取并返回?cái)?shù)據(jù)。如果沒(méi)有可讀的數(shù)據(jù)，則返回 null。在可讀流上 src 還有一個(gè) readable 屬性，如果可以安全地調(diào)用 readable.read()，則為 true 
function flow(stream) { 
  const state = stream._readableState; 
  debug('flow', state.flowing); 
  while (state.flowing && stream.read() !== null); 
} 

2.2.4 觸發(fā) data 事件

調(diào)用 readable 的 resume() 方法，觸發(fā)可讀流的 'data' 事件，進(jìn)入流動(dòng)模式。

Readable.prototype.pipe = function(dest, options) { 
  const src = this; 
  // Start the flow if it hasn't been started already. 
  if (!state.flowing) { 
    debug('pipe resume'); 
    src.resume(); 
  } 
  ... 

然后實(shí)例上的 resume（Readable 原型上定義的）會(huì)在調(diào)用 resume() 方法，在該方法內(nèi)部又調(diào)用了 resume_()，最終執(zhí)行了 stream.read(0) 讀取了一次空數(shù)據(jù)（size 設(shè)置的為 0），將會(huì)觸發(fā)實(shí)例上的 _read() 方法，之后會(huì)在觸發(fā) data 事件。

function resume(stream, state) { 
  ... 
  process.nextTick(resume_, stream, state); 
} 
 
function resume_(stream, state) { 
  debug('resume', state.reading); 
  if (!state.reading) { 
    stream.read(0); 
  } 
 
  ... 
} 

2.2.5 訂閱 end 事件

end 事件：當(dāng)可讀流中沒(méi)有數(shù)據(jù)可供消費(fèi)時(shí)觸發(fā)，調(diào)用 onend 函數(shù)，執(zhí)行 dest.end() 方法，表明已沒(méi)有數(shù)據(jù)要被寫入可寫流，進(jìn)行關(guān)閉（關(guān)閉可寫流的 fd），之后再調(diào)用 stream.write() 會(huì)導(dǎo)致錯(cuò)誤。

Readable.prototype.pipe = function(dest, options) { 
  ... 
  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 onend() { 
    debug('onend'); 
    dest.end(); 
  } 
} 

2.2.6 觸發(fā) pipe 事件

在 pipe 方法里面最后還會(huì)觸發(fā)一個(gè) pipe 事件，傳入可讀流對(duì)象

Readable.prototype.pipe = function(dest, options) { 
  ... 
  const source = this; 
  dest.emit('pipe', src); 
  ... 
}; 

在應(yīng)用層使用的時(shí)候可以在可寫流上訂閱 pipe 事件，做一些判斷，具體可參考官網(wǎng)給的這個(gè)示例 stream_event_pipe

2.2.7 支持鏈?zhǔn)秸{(diào)用

最后返回 dest，支持類似 unix 的用法：A.pipe(B).pipe(C)

Stream.prototype.pipe = function(dest, options) { 
  return dest; 
}; 

3. 總結(jié)

本文總體分為兩部分：

• 第一部分相對(duì)較基礎(chǔ)，講解了 Nodejs Stream 的 pipe 方法在 Koa2 中是怎么去應(yīng)用的。
• 第二部分仍找它的實(shí)現(xiàn)，以及對(duì)源碼的一個(gè)簡(jiǎn)單分析，其實(shí) pipe 方法核心還是要去監(jiān)聽(tīng) data 事件，向可寫流寫入數(shù)據(jù)，如果內(nèi)部緩沖大于創(chuàng)建流時(shí)配置的 highWaterMark，則要停止數(shù)據(jù)流動(dòng)，直到 drain 事件觸發(fā)或者結(jié)束，當(dāng)然還要監(jiān)聽(tīng) end、error 等事件做一些處理。

4. Reference

• nodejs.cn/api/stream.html
• cnodejs.org/topic/56ba030271204e03637a3870
• github.com/nodejs/node/blob/master/lib/_stream_readable.js

本文轉(zhuǎn)載自微信公眾號(hào)「Nodejs技術(shù)?！?，可以通過(guò)以下二維碼關(guān)注。轉(zhuǎn)載本文請(qǐng)聯(lián)系Nodejs技術(shù)棧公眾號(hào)。