什么是流（steams）

流（stream）是 Node.js 中處理流式數據的抽象接口。

Streams 不是 Node.js 獨有的概念。它們是幾十年前在 Unix 操作系統中引入的。

它們能夠以一種有效的方式來處理文件的讀、寫，網絡通信或任何類型的端到端信息交換。
例如，當你編寫了一段程序用來讀取文件時，傳統的方法是將文件從頭到尾讀入內存，然后再進行處理。而使用流的話，你就可以逐塊讀取它，處理其內容而不將其全部保存在內存中。
以如下代碼為例

const fs = require('fs');const rs = fs.createReadStream('test.md');let data = '';rs.on("data", function (chunk) {data += chunk;});rs.on("end", function() {console.log(data);});

利用 createReadStream 創建一個讀取數據的流，來讀取 test.md 文件的內容，此時監聽 data 事件，它是在當流將數據塊傳送給消費者后觸發。并在對應的 eventHandler 中，拼接 chunk。在 end 事件中，打印到終端上。
之前說流，可以逐塊讀取文件內容，那么這個塊，也就是 chunk 是什么？
一般情況下是 Buffer，修改 data 事件的 eventHandler 來驗證下

rs.on("data", function (chunk) {console.log("chunk", Buffer.isBuffer(chunk)) // log truedata += chunk;});

流的工作方式可以具體的表述為，在內存中準備一段 Buffer，然后在 fs.read() 讀取時逐步從磁盤中將字節復制到 Buffer 中。

為什么要使用 Stream

利用 Stream 來處理數據，主要是因為它的兩個優點：

內存效率：在夠處理數據之前，不需要占用大量內存；

時間效率：處理數據花費的時間更少，因為流是逐塊來處理數據，而不是等到整個數據有效負載才啟動。

首先內存效率，與 fs.readFile 這種會緩沖整個文件相比，流式傳輸充分地利用 Buffer （超過 8kb）不受 V8 內存控制的特點，利用堆外內存完成高效地傳輸。相關驗證可以參考這篇博文，地址。
時間效率，與 fs.FileSync 相比，有些優勢，但是與異步的 fs.readFile 相比，優勢不大。

Node.js 中 Stream 的使用

首先用一張圖來了解下 Node.js 中有哪些內置的 Stream 接口

圖中提供了一些 Node.js 原生的流的示例，有些是可讀、寫的流。 也有一些是可讀寫的流，如 TCP sockets、zlib 以及 crypto。

特別注意： 流的讀、寫與環境是密切相關的。例如 HTTP 響應在客戶端上的可讀流，但它是服務器上的可寫流。同時還需要注意，stdio streams（stdin，stdout，stderr） 在子進程上是相反的流。

使用一個例子來展示流的使用

首先利用如下腳本創建一個比較大的文件（大概 430 MB）

const fs = require('fs');const file = fs.createWriteStream('test.md');for(let i=0; i<= 1e6; i++) {file.write('hello world./n');}file.end();

在當前目錄下，啟動 http 服務

const http = require('http')const fs = require('fs')const server = http.createServer(function (req, res) {fs.readFile(__dirname + '/test.md', (err, data) => {res.end(data)})})server.listen(3000)

得到的結果，如圖

const http = require('http')const fs = require('fs')const server = http.createServer((req, res) => {const stream = fs.createReadStream(__dirname + '/test.md')stream.pipe(res)})server.listen(3000)

時間減少了 2s 多。這可以解釋為，在讀取文件內容，并且不需要改變內容的場景下，流能夠完成只讀取 buffer，然后直接傳輸，不做額外的轉換，避免損耗，提高性能。
上述代碼中，應用了 stream.pipe(...) 。它主要是對流進行鏈式地管道操作，例如

src.pipe(dest1).pipe(dest2)

這樣數據流會被自動管理。

如果可讀流發生錯誤，目標可寫流不會自動關閉，需要手動關閉所有流以避免內存泄漏。

通常，當你使用 pipe 方法時，就不需要使用事件，但如果場景需要以更靈活、自定義的方式使用流，那么就要考慮事件。

Stream events

在上述例子中，我們使用了可讀流的 data 、end 事件來控制文件的讀取，它本質上與 pipe 方法相同，例如

# readable.pipe(writable)readable.on('data', (chunk) => {writable.write(chunk);});readable.on('end', () => {writable.end();});

只不過，使用 event 會更加靈活，可控。

圖中簡單羅列了可讀流、可寫流的相關事件、方法，其中最重要的是

可讀流：

• data 事件：每當流將一大塊數據傳遞時，就會觸發；
• end 事件：當沒有更多數據要從流發出時，就會觸發。

可寫流：

• drain 事件：當可以繼續寫入數據到流時會觸發事件；
• finish 事件：處理完全部數據塊之后觸發。

流的不同類型

除了上面涉及到的可讀、寫流之后，還有 Duplex、Transform 兩類：

• Readable ：可以接收數據，但不能向其發送數據。當你將數據推送到可讀流中時，它會被緩沖，直到消費者開始讀取數據；
• writable ：可以發送數據，但不能從中接收；
• Duplex ：即可讀也可寫；
• Tranform ：與 Duplex 一樣是可寫又可讀的，但它的輸出與輸入是相關聯的。

如何創建一個可讀流

這里只做簡單介紹，具體見 stream module。

const Stream = require('stream')const readableStream = new Stream.Readable()readableStream._read = (size) => {console.log('read', size)}

利用 Stream 模塊初始化一個可讀流，然后向其中發送數據

readableStream.push('hi!')readableStream.push('ho!')

如何創建一個可寫流

為了創建可寫流，需要擴展了基本的 Writable 對象，并實現了它的 _write 方法。

const Stream = require('stream')const writableStream = new Stream.Writable()

實現 _write 方法：

writableStream._write = (chunk, encoding, next) => {console.log(chunk.toString())next()}

結合上述例子實現

利用 readableStream 讀入數據，并輸出到 writableStream

const Stream = require('stream')const readableStream = new Stream.Readable()readableStream._read = (size) => {console.log('read', size)}const writableStream = new Stream.Writable()writableStream._write = (chunk, encoding, next) => {console.log('write', chunk.toString())next()}readableStream.pipe(writableStream)readableStream.push('hi!')readableStream.push('ho!')/* log:read 16384write hi!write ho!*/

以上就是本文的全部內容，希望對大家的學習有所幫助，也希望大家多多支持VeVb武林網。