Node中的Stream是什么

本篇內容主要講解“Node中的Stream是什么”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“Node中的Stream是什么”吧!

stream 是一個抽象的數據接口，它繼承了 EventEmitter，它能夠發送/接受數據，本質就是讓數據流動起來，如下圖：

流不是 Node 中獨有的概念，是操作系統最基本的操作方式，在 Linux 中 | 就是 Stream，只是 Node 層面對其做了封裝，提供了對應的 API

為啥要一點一點？

首先使用下面的代碼創建一個文件，大概在 400MB 左右

當我們使用 readFile 去讀取的時候，如下代碼

正常啟動服務時，占用 10MB 左右的內存

使用curl http://127.0.0.1:8000發起請求時，內存變為了 420MB 左右，和我們創建的文件大小差不多

改為使用使用 stream 的寫法，代碼如下

再次發起請求時，發現內存只占用了 35MB 左右，相比 readFile 大幅減少

如果我們不采用流的模式，等待大文件加載完成在操作，會有如下的問題：

• 內存暫用過多，導致系統崩潰

• CPU 運算速度有限制，且服務于多個程序，大文件加載過大且時間久

總結來說就是，一次性讀取大文件，內存和網絡都吃不消

如何才能一點一點？

我們讀取文件的時候，可以采用讀取完成之后在輸出數據

上述說到 stream 繼承了 EventEmitter 可以是實現監聽數據。首先將讀取數據改為流式讀取，使用 on("data", ()?{}) 接收數據，最后通過 on("end", ()?{}) 最后的結果

有數據傳遞過來的時候就會觸發 data 事件，接收這段數據做處理，最后等待所有的數據全部傳遞完成之后觸發 end 事件。

數據的流轉過程

數據從哪里來—source

數據是從一個地方流向另一個地方，先看看數據的來源。

• http 請求，請求接口來的數據

  

• console 控制臺，標準輸入 stdin

  

• file 文件，讀取文件內容，例如上面的例子

連接的管道—pipe

在 source 和 dest 中有一個連接的管道 pipe，基本語法為 source.pipe(dest) ，source 和 dest 通過 pipe 連接，讓數據從 source 流向 dest

我們不需要向上面的代碼那樣手動監聽 data/end 事件.

pipe 使用時有嚴格的要求，source 必須是一個可讀流，dest 必須是一個可寫流

??? 流動的數據到底是一個什么東西？代碼中的 chunk 是什么？

到哪里去—dest

stream 常見的三種輸出方式

• console 控制臺，標準輸出 stdout

  

• http 請求，接口請求中的 response

  

• file 文件，寫入文件

  

流的種類

可讀流 Readable Streams

可讀流是對提供數據的源頭(source)的抽象

所有的 Readable 都實現了 stream.Readable 類定義的接口

? 讀取文件流創建

fs.createReadStream 創建一個 Readable 對象

讀取模式

可讀流有兩種模式，流動模式(flowing mode)和暫停模式(pause mode)，這個決定了 chunk 數據的流動方式：自動流動和手工流動

在 ReadableStream 中有一個 _readableState 屬性，在其中有一個 flowing 的一個屬性來判斷流的模式，他有三種狀態值：

• ture：表示為流動模式

• false：表示為暫停模式

• null：初始狀態

可以使用熱水器模型來模擬數據的流動。熱水器水箱(buffer 緩存區)存儲著熱水(需要的數據)，當我們打開水龍頭的時候，熱水就會從水箱中不斷流出來，并且自來水也會不斷的流入水箱，這就是流動模式。當我們關閉水龍頭時，水箱會暫停進水，水龍頭則會暫停出水，這就是暫停模式。

流動模式

數據自動地從底層讀取，形成流動現象，并通過事件提供給應用程序。

• 監聽 data 事件即可進入該模式
  當 data 事件被添加后，可寫流中有數據后會將數據推到該事件回調函數中，需要自己去消費數據塊，如果不處理則該數據會丟失

• 調用 stream.pipe 方法將數據發送到 Writeable

• 調用 stream.resume 方法

  

暫停模式

數據會堆積在內部緩沖器中，必須顯式調用 stream.read() 讀取數據塊

• 監聽 readable 事件 可寫流在數據準備好后會觸發該事件回調，此時需要在回調函數中使用 stream.read() 來主動消費數據。readable 事件表明流有新的動態：要么有新的數據，要么流已經讀取所有數據

  

兩種模式之間如何進行轉換呢

• 可讀流在創建完成之后處于初始狀態   //TODO：和網上的分享不一致

• 暫停模式切換到流動模式

  
  

  - 監聽 data 事件
  - 調用 stream.resume 方法
  - 調用 stream.pipe 方法將數據發送到 Writable
  

  
  

  

• 流動模式切換到暫停模式

  
  

  - 移除 data 事件
  - 調用 stream.pause 方法
  - 調用 stream.unpipe 移除管道目標
  

  
  

實現原理

創建可讀流的時候，需要繼承 Readable 對象，并且實現 _read 方法

創建一個自定義可讀流

當我們調用 read 方法時，整體的流程如下：

• doRead

  流中維護了一個緩存，當調用 read 方法的時候來判斷是否需要向底層請求數據

  當緩存區長度為0或者小于 highWaterMark 這個值得時候就會調用 _read 去底層獲取數據 源碼鏈接

  

可寫流 Writeable Stream

可寫流 是對數據寫入目的地的一種抽象，是用來消費上游流過來的數據，通過可寫流把數據寫入設備，常見的寫入流就是本地磁盤的寫入

可寫流的特點

• 通過 write 寫入數據

  

• 通過 end 寫數據并且關閉流，end = write + close

  

• 當寫入數據達到 highWaterMark 的大小時，會觸發 drain 事件

  

  調用 ws.write(chunk) 返回 false，表示當前緩沖區數據大于或等于 highWaterMark 的值，就會觸發 drain 事件。其實是起到一個警示作用，我們依舊可以寫入數據，只是未處理的數據會一直積壓在可寫流的內部緩沖區中，直到積壓沾滿 Node.js 緩沖區后，才會被強行中斷

自定義可寫流

所有的 Writeable 都實現了 stream.Writeable 類定義的接口

只需要實現 _write 方法就能夠將數據寫入底層

• 通過調用調用 writable.write 方法將數據寫入流中，會調用 _write 方法將數據寫入底層

• 當 _write 數據成功后，需要調用 next 方法去處理下一個數據

• 必須調用 writable.end(data) 來結束可寫流，data 是可選的。此后，不能再調用 write 新增數據，否則會報錯

• 在 end 方法調用后，當所有底層的寫操作均完成時，會觸發 finish 事件

雙工流 Duplex Stream

雙工流，既可讀，也可寫。實際上繼承了 Readable 和 Writable 的一種流，那它既可以當做可讀流來用又可以當做可寫流來用

自定義的雙工流需要實現 Readable 的 _read 方法和 Writable 的 _write 方法

net 模塊可以用來創建 socket，socket 在 NodeJS 中是一個典型的 Duplex，看一個 TCP 客戶端的例子

client 就是一個 Duplex，可寫流用于向服務器發送消息，可讀流用于接受服務器消息，兩個流內的數據并沒有直接的關系

轉換流 Transform Stream

上述的例子中，可讀流中的數據(0/1)和可寫流中的數據(’F’,’B’,’B’)是隔離的，兩者并沒有產生關系，但對于 Transform 來說在可寫端寫入的數據經過變換后會自動添加到可讀端。

Transform 繼承于 Duplex，并且已經實現了 _write 和 _read 方法，只需要實現 _tranform 方法即可

gulp 基于 Stream 的自動化構建工具，看一段官網的示例代碼

less → less 轉為 css → 執行 css 壓縮 → 壓縮后的 css

其實 less() 和 minifyCss() 都是對輸入的數據做了一些處理，然后交給了輸出數據

Duplex 和 Transform 的選擇

和上面的示例對比起來，我們發現一個流同時面向生產者和消費者服務的時候我們會選擇 Duplex，當只是對數據做一些轉換工作的時候我們便會選擇使用 Tranform

背壓問題

什么是背壓

背壓問題來源于生產者消費者模式中，消費者處理速度過慢

比如說，我們下載過程，處理速度為3Mb/s，而壓縮過程，處理速度為1Mb/s，這樣的話，很快緩沖區隊列就會形成堆積

要么導致整個過程內存消耗增加，要么導致整個緩沖區慢，部分數據丟失

什么是背壓處理

背壓處理可以理解為一個向上”喊話”的過程

當壓縮處理發現自己的緩沖區數據擠壓超過閾值的時候，就對下載處理“喊話”，我忙不過來了，不要再發了

下載處理收到消息就暫停向下發送數據

如何處理背壓

我們有不同的函數將數據從一個進程傳入另外一個進程。在 Node.js 中，有一個內置函數稱為 .pipe()，同樣地最終，在這個進程的基本層面上我們有二個互不相關的組件：數據的_源頭_，和_消費者_

當 .pipe() 被源調用之后，它通知消費者有數據需要傳輸。管道函數為事件觸發建立了合適的積壓封裝

在數據緩存超出了 highWaterMark 或者寫入的列隊處于繁忙狀態，.write() 會返回 false

當 false 返回之后，積壓系統介入了。它將暫停從任何發送數據的數據流中進入的 Readable。一旦數據流清空了，drain 事件將被觸發，消耗進來的數據流

一旦隊列全部處理完畢，積壓機制將允許數據再次發送。在使用中的內存空間將自我釋放，同時準備接收下一次的批量數據

我們可以看到 pipe 的背壓處理：

• 將數據按照chunk進行劃分，寫入

• 當chunk過大，或者隊列忙碌時，暫停讀取

• 當隊列為空時，繼續讀取數據

到此，相信大家對“Node中的Stream是什么”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！