Nodejs中stream流模塊怎么樣

這篇文章將為大家詳細講解有關Nodejs中stream流模塊怎么樣，小編覺得挺實用的，因此分享給大家做個參考，希望大家閱讀完這篇文章后可以有所收獲。

stream流模塊，是Node中非常核心的一個模塊，其它模塊如fs、http等都基于流stream模塊的實例。

而對于大多前端小白在剛入門Node的學習過程中，對于流的概念及使用還是不太好清晰的理解，因為在前端的工作中似乎很少有過關于"流"處理相關的應用。

1. 流，是什么？

單純“流”這個字，我們很容易產生水流，流動等的概念。

官方定義：流，是用于在 Node.js 中處理流數據的抽象接口

從官方的定義中，我們可以看出：

• 流，是Node提供的一種處理數據的工具

• 流，是Node中的一種抽象接口

準確的理解，流，可以理解為數據流，它是一種用來傳輸數據的手段，在一個應用程序中，流，是一種有序的，有起點和終點的數據流。

造成我們對stream流不太好的理解的主要原因就是，它是一種抽象的概念。

2. 流，的具體使用場景

為了讓我們能夠清楚的理解stream模塊，我們首先來以具體的應用場景來說明stream模塊有哪些實際應用之處。

stream流，在Node中主要應用在大量數據處理的需求上，如fs對大文件的讀取和寫入、http請求響應、文件的壓縮、數據的加密/解密等應用。

我們以上面的圖片說明流的使用，水桶可以理解為數據源，水池可以理解為數據目標，中間連接的管道，我們可以理解為數據流，通過數據流管道，數據從數據源流向數據目標。

3. 流的分類

在Node中，流被分為4類：可讀流，可寫流，雙工流，轉換流。

• Writable: 可以寫入數據的流

• Readable: 可以從中讀取數據的流

• Duplex: Readable 和 Writable 的流

• Transform: 可以在寫入和讀取數據時修改或轉換數據的 Duplex 流

所有的流都是 EventEmitter 的實例。即我們可以通過事件機制監聽數據流的變化。

4. 數據模式和緩存區

在深入學習4類流的具體使用之前，我們需要理解兩個概念數據模式和緩存區，有助于我們在接下來流的學習中更好的理解。

4.1 數據模式

Node.js API 創建的所有流都只對字符串和 Buffer（或 Uint8Array）對象進行操作。

4.2 緩存區

Writable和 Readable 流都將數據存儲在內部緩沖區(buffer)中。

可緩沖的數據量取決于傳給流的構造函數的 highWaterMark 選項, 對于普通的流，highWaterMark 選項指定字節的總數;對于在對象模式下操作的流,highWaterMark選項指定對象的總數。

highWaterMark 選項是閾值，而不是限制：它規定了流在停止請求更多數據之前緩沖的數據量。

當實現調用 stream.push(chunk) 時，數據緩存在 Readable 流中。 如果流的消費者沒有調用 stream.read()，則數據會一直駐留在內部隊列中，直到被消費。

一旦內部讀取緩沖區的總大小達到 highWaterMark 指定的閾值，則流將暫時停止從底層資源讀取數據，直到可以消費當前緩沖的數據

當重復調用 writable.write(chunk) 方法時，數據會緩存在 Writable 流中。

5. 可讀流

5.1 流讀取的流動與暫停

Readable 流以兩種模式之一有效地運行：流動和暫停。

• 流動模式：從系統底層讀取數據并push()到緩存區，達到highWaterMark后 push() 會返回 false，資源停止流向緩存區，并觸發data事件消費數據。

• 暫停模式：所有的Readable流都是以Paused暫停模式開始，必須顯式調用stream.read()方法來從流中讀取數據。每一次數據達到緩存區都會觸發一次 readable 事件，也就是每一次 push() 都會觸發 readable。

• 暫停模式切換到流動模式的方式：

• 添加data事件句柄

• 調用stream.resume()方法

• 調用stream.pipe()方法將數據發送到 Writable

• 流動模式切換到暫停模式的方式：

• 如果沒有管道目標，則通過調用 stream.pause() 方法。

• 如果有管道目標，則刪除所有管道目標。 可以通過調用 stream.unpipe()方法刪除多個管道目標。

5.2 可讀流常用示例

import path from 'path';
import fs, { read } from 'fs';

const filePath = path.join(path.resolve(), 'files', 'text.txt');

const readable = fs.createReadStream(filePath);
// 如果使用 readable.setEncoding() 方法為流指定了默認編碼，則監聽器回調將把數據塊作為字符串傳入；否則數據將作為 Buffer 傳入。
readable.setEncoding('utf8');
let str = '';

readable.on('open', (fd) => {
  console.log('開始讀取文件')
})
// 每當流將數據塊的所有權移交給消費者時，則會觸發 'data' 事件
readable.on('data', (data) => {
  str += data;
  console.log('讀取到數據')
})
// 方法將導致處于流動模式的流停止觸發 'data' 事件，切換到暫停模式。 任何可用的數據都將保留在內部緩沖區中。
readable.pause();
// 方法使被顯式暫停的 Readable 流恢復觸發 'data' 事件，將流切換到流動模式。
readable.resume();
// 當調用 stream.pause() 并且 readableFlowing 不是 false 時，則會觸發 'pause' 事件。
readable.on('pause', () => {
  console.log('讀取暫停')
})
// 當調用 stream.resume() 并且 readableFlowing 不是 true 時，則會觸發 'resume' 事件。
readable.on('resume', () => {
  console.log('重新流動')
})
// 當流中沒有更多數據可供消費時，則會觸發 'end' 事件。
readable.on('end', () => {
  console.log('文件讀取完畢');
})
// 當流及其任何底層資源（例如文件描述符）已關閉時，則會觸發 'close' 事件。
readable.on('close', () => {
  console.log('關閉文件讀取')
})
// 將 destWritable 流綁定到 readable，使其自動切換到流動模式并將其所有數據推送到綁定的 Writable。 數據流將被自動管理
readable.pipe(destWriteable)
// 如果底層流由于底層內部故障而無法生成數據，或者當流實現嘗試推送無效數據塊時，可能會發生這種情況。
readable.on('error', (err) => {
  console.log(err)
  console.log('文件讀取發生錯誤')
})

6. 可寫流

6.1 可寫流的流動與暫停

writeable流 與 readable流 是比較相似的，數據流過來的時候，會直接寫入到緩存區，當寫入速度比較緩慢或者寫入暫停時，數據流會在緩存區緩存起來；

當生產者寫入速度過快，把隊列池裝滿了之后，就會出現「背壓」，這個時候是需要告訴生產者暫停生產的，當隊列釋放之后，writable流 會給生產者發送一個 drain 消息，讓它恢復生產。

6.2 可寫流示例

import path from 'path';
import fs, { read } from 'fs';

const filePath = path.join(path.resolve(), 'files', 'text.txt');
const copyFile = path.join(path.resolve(), 'files', 'copy.txt');

let str = '';
// 創建可讀流
const readable = fs.createReadStream(filePath);
// 如果使用 readable.setEncoding() 方法為流指定了默認編碼
readable.setEncoding('utf8');

// 創建可寫流
const wirteable = fs.createWriteStream(copyFile);
// 編碼
wirteable.setDefaultEncoding('utf8');

readable.on('open', (fd) => {
  console.log('開始讀取文件')
})
// 每當流將數據塊的所有權移交給消費者時，則會觸發 'data' 事件
readable.on('data', (data) => {
  str += data;
  console.log('讀取到數據');

  // 寫入
  wirteable.write(data, 'utf8');
})

wirteable.on('open', () => {
  console.log('開始寫入數據')
})
// 如果對 stream.write(chunk) 的調用返回 false，則 'drain' 事件將在適合繼續將數據寫入流時觸發。
// 即生產數據的速度大于寫入速度，緩存區裝滿之后，會暫停生產著從底層讀取數據
// writeable緩存區釋放之后，會發送一個drain事件讓生產者繼續讀取
wirteable.on('drain', () => {
  console.log('繼續寫入')
})
// 在調用 stream.end() 方法之后，并且所有數據都已刷新到底層系統，則觸發 'finish' 事件。
wirteable.on('finish', () => {
  console.log('數據寫入完畢')
})

readable.on('end', () => {
  // 數據讀取完畢通知可寫流
  wirteable.end()
})
// 當在可讀流上調用 stream.pipe() 方法將此可寫流添加到其目標集時，則觸發 'pipe' 事件。
// readable.pipe(destWriteable)
wirteable.on('pipe', () => {
  console.log('管道流創建')
})

wirteable.on('error', () => {
  console.log('數據寫入發生錯誤')
})

關于“Nodejs中stream流模塊怎么樣”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，使各位可以學到更多知識，如果覺得文章不錯，請把它分享出去讓更多的人看到。