2.8　Stream

Stream模塊為Node操作流式數據提供了支持。

Stream的思想最早見于早期的UNIX，在UNIX中使用“|”符號會創建一個匿名管道，其本質上也是一個Stream，用于兩個程序（或者是設備）之間的數據傳輸。

2.8.1　Stream的種類

要使用Node的stream模塊，需要增加引用：

在Nodejs中，一共有四種基礎的stream類型：

• Readable：可讀流(for example fs.createReadStream())。
• Writable：可寫流(for example fs.createWriteStream())。
• Duplex：既可讀，又可寫(for example net.Socket)。
• Transform：操作寫入的數據，然后讀取結果，通常用于輸入數據和輸出數據不要求匹配的場景，例如zlib.createDeflate()。

我們重點介紹Readable和Writable這兩種stream。

1．Readable Stream

Readable Stream定義的方法和事件如下所示：

• Event: 'close'
• Event: 'data'
• Event: 'end'
• Event: 'error'
• Event: 'readable'
• readable.isPaused()
• readable.pause()
• readable.pipe(destination[, options])
• readable.read([size])
• readable.resume()
• readable.setEncoding(encoding)
• readable.unpipe([destination])
• readable.unshift(chunk)
• readable.wrap(stream)

代碼2.21　Readable stream的例子

2．Writeable Stream

Writeable Stream主要使用write方法來寫入數據，API列表如下文所示：

write方法同樣是異步的，假設我們創建一個可讀流讀取一個較大的文件，再調用pipe方法將數據通過一個可寫流寫入另一個位置。如果讀取的速度大于寫入的速度，那么Node將會在內存中緩存這些數據。

當然緩沖區也是有大小限制的（state.highWatermark），當達到閾值后，write方法會返回false，可讀流也進入暫停狀態，當writeable stream將緩沖區清空之后，會觸發drain事件，上游的readable重新開始讀取數據。

另一個比較重要的是pipe方法，其聲明如下：

pipe方法相當于在可讀流和可寫流之間架起了橋梁，使得數據可以通過管道由可讀流進入可寫流。下面是使用pipe方法改寫的靜態文件服務器。

代碼2.22　使用pipe改寫的靜態文件服務器

pipe方法接收一個writable對象，當readable對象調用pipe方法時，會在內部調用writable對象的write方法進行寫入。

2.8.2　ReadLine

ReadLine是一個Node原生模塊，該模塊比較不起眼，提供了按行讀取Stream中數據的功能。

下面是ReadLine模塊的監聽事件及方法：

• Event : 'close'
• Event : 'line'
• Event : 'pause'
• Event : 'resume'
• Event : 'SIGCONT'
• Event : 'SIGINT'
• Event : 'SIGTSTP'
• rl.close()
• rl.pause()
• rl.prompt([preserveCursor])
• rl.question(query, callback)
• rl.resume()
• rl.setPrompt(prompt)
• rl.write(data[, key])

該模塊通常用來和stream搭配使用，但因為在實際項目中通常會定制自己的stream或者自定義讀取方法，導致該模塊的地位有些尷尬。下面是readLine的一個例子。

代碼2.23　使用readLine模塊讀取文件

readLine并沒有提供形如new readline()形式的構造方法，而是使用createInterface方法初始化了一個rl對象。

想象下有如下場景，一個可讀流中包含了很多條獨立的信息需要逐條處理，這可能是一個消息隊列，這時使用readline模塊就比較方便。

2.8.3　自定義Stream

在實際開發中，如果想要使用流式API，而原生的Stream又不能滿足需求時，可以考慮實現自己的Stream類，常用的方法是繼承原生的Stream類，然后做一些擴展。

下面我們拿Readable Stream為例來說明如何實現一個自定義的Stream。

上面的代碼實現了名為MyReadable的類，它繼承自Readable類，并且接受一個數組作為參數。

想要繼承Readable類，就要在自定義的類內部實現_read方法，該方法內部使用push方法往可讀流添加數據。

當我們給可讀流對象注冊data事件后，可讀流會在nextTick中調用_read方法，并觸發第一次data事件（讀者可能會認為可讀流開始讀取是在調用構造函數之后，但此時data事件還未注冊，可能會捕獲不到最初的事件，因此可讀流開始產生數據的操作是放在nextTick中的）。

當有消費者從readable中取數據時會自動調用該方法。在上面的例子里我們在_read方法里調用了push方法，該方法用來向可讀流中填充數據，下面是一個消費者的例子：

每次觸發data事件時都會得到相應的數組元素，當數組為空時，_read方法會被調用。即：

如果end事件被觸發，則代表讀取完畢。