Node.js是一個強大的允許開發(fā)人員構建可擴展和高效的應用程序。Node.js的一個關鍵特性是其內(nèi)置對流的支持。流是Node.js中的一個基本概念，它能夠實現(xiàn)高效的數(shù)據(jù)處理，特別是在處理大量信息或實時處理數(shù)據(jù)時。

在本文中，我們將探討Node.js中的流概念，了解可用的不同類型的流（可讀流、可寫流、雙工流和轉換流），并討論有效處理流的最佳實踐。

什么是Node.js流？

流是Node.js應用程序中的一個基本概念，通過按順序讀取或寫入輸入和輸出，實現(xiàn)高效的數(shù)據(jù)處理。它們非常適用于文件操作、網(wǎng)絡通信和其他形式的端到端數(shù)據(jù)交換。

流的獨特之處在于它以小的、連續(xù)的塊來處理數(shù)據(jù)，而不是一次性將整個數(shù)據(jù)集加載到內(nèi)存中。這種方法在處理大量數(shù)據(jù)時非常有益，因為文件大小可能超過可用內(nèi)存。流使得以較小的片段處理數(shù)據(jù)成為可能，從而可以處理更大的文件。

如上圖所示，數(shù)據(jù)通常以塊或連續(xù)流的形式從流中讀取。從流中讀取的數(shù)據(jù)塊可以存儲在緩沖區(qū)中。緩沖區(qū)提供臨時存儲空間，用于保存數(shù)據(jù)塊，直到進一步處理。

為了進一步說明這個概念，考慮一個實時股票市場數(shù)據(jù)源的情景。在金融應用中，實時更新股票價格和市場數(shù)據(jù)對于做出明智的決策至關重要。流式處理使應用程序能夠以較小的連續(xù)塊處理數(shù)據(jù)，而不是獲取和存儲整個數(shù)據(jù)源，這可能是相當龐大和不切實際的。數(shù)據(jù)通過流動，允許應用程序在更新到達時執(zhí)行實時分析、計算和通知。這種流式處理方法節(jié)省了內(nèi)存資源，并確保應用程序能夠迅速響應市場波動，并向交易員和投資者提供最新信息。它消除了在采取行動之前等待整個數(shù)據(jù)源可用的需要。

為什么要使用流？

流提供了與其他數(shù)據(jù)處理方法相比的兩個關鍵優(yōu)勢。

內(nèi)存效率

使用流，處理前不需要將大量數(shù)據(jù)加載到內(nèi)存中。相反，數(shù)據(jù)以較小的可管理塊進行處理，減少了內(nèi)存需求并有效利用了系統(tǒng)資源。

時間效率

流使得數(shù)據(jù)一旦可用就能立即進行處理，而不需要等待整個有效負載的傳輸。這樣可以實現(xiàn)更快的響應時間和改善整體性能。

理解并有效地利用流能夠幫助開發(fā)人員實現(xiàn)最佳的內(nèi)存使用、更快的數(shù)據(jù)處理和增強的代碼模塊化，使其成為Node.js應用程序中強大的功能。然而，不同類型的Node.js流可以用于特定的目的，并在數(shù)據(jù)處理方面提供靈活性。為了在您的Node.js應用程序中有效地使用流，有必要清楚地了解每種流類型。因此，讓我們深入研究一下Node.js中可用的不同流類型。

Node.js流的類型

Node.js 提供了四種主要類型的流，每種流都有特定的用途：

Readable Streams 可讀流

可讀流允許從源（如文件或網(wǎng)絡套接字）讀取數(shù)據(jù)。它們按順序發(fā)出數(shù)據(jù)塊，并可以通過附加監(jiān)聽器到“data”事件來消費。可讀流可以處于流動或暫停狀態(tài)，取決于數(shù)據(jù)的消費方式。

const fs = require('fs');// Create a Readable stream from a file
const readStream = fs.createReadStream('the_princess_bride_input.txt', 'utf8');// Readable stream 'data' event handler
readStream.on('data', (chunk) => {console.log(`Received chunk: ${chunk}`);
});// Readable stream 'end' event handler
readStream.on('end', () => {console.log('Data reading complete.');
});// Readable stream 'error' event handler
readStream.on('error', (err) => {console.error(`Error occurred: ${err}`);
});

如上所示的代碼片段中，我們使用fs模塊使用createReadStream()方法創(chuàng)建一個可讀流。我們將文件路徑
the_princess_bride_input.txt 和編碼 utf8 作為參數(shù)傳遞?？勺x流以小塊方式從文件中讀取數(shù)據(jù)。

我們將事件處理程序附加到可讀流上以處理不同的事件。當數(shù)據(jù)塊可供讀取時，會觸發(fā) data 事件。當可讀流完成從文件中讀取所有數(shù)據(jù)時，會觸發(fā) end 事件。如果在讀取過程中發(fā)生錯誤，則會觸發(fā) error 事件。

通過使用可讀流并監(jiān)聽相應的事件，您可以高效地從源（例如文件）中讀取數(shù)據(jù)，并對接收到的數(shù)據(jù)塊執(zhí)行進一步操作。

Writable Streams 可寫流

可寫流處理將數(shù)據(jù)寫入目標位置，例如文件或網(wǎng)絡套接字。它們提供了像 write() 和 end() 這樣的方法來向流發(fā)送數(shù)據(jù)。可寫流可用于以分塊方式寫入大量數(shù)據(jù)，防止內(nèi)存溢出。

const fs = require('fs');// Create a Writable stream to a file
const writeStream = fs.createWriteStream('the_princess_bride_output.txt');// Writable stream 'finish' event handler
writeStream.on('finish', () => {console.log('Data writing complete.');
});// Writable stream 'error' event handler
writeStream.on('error', (err) => {console.error(`Error occurred: ${err}`);
});// Write a quote from "The  to the Writable stream
writeStream.write('As ');
writeStream.write('You ');
writeStream.write('Wish');
writeStream.end();

在上面的代碼示例中，我們使用fs模塊使用 createWriteStream() 方法創(chuàng)建一個可寫流。我們指定數(shù)據(jù)將被寫入的文件路徑（
the_princess_bride_output.txt ）。

我們將事件處理程序附加到可寫流上，以處理不同的事件。當可寫流完成寫入所有數(shù)據(jù)時，會觸發(fā) finish 事件。如果在寫入過程中發(fā)生錯誤，則會觸發(fā) error 事件。使用 write() 方法將單個數(shù)據(jù)塊寫入可寫流。在這個例子中，我們將字符串'As'、'You'和'Wish'寫入流中。最后，我們調用 end() 來表示數(shù)據(jù)寫入結束。

通過使用可寫流并監(jiān)聽相應的事件，您可以高效地將數(shù)據(jù)寫入目標位置，并在寫入過程完成后執(zhí)行任何必要的清理或后續(xù)操作。

Duplex Streams 雙工流

雙工流代表了可讀和可寫流的組合。它們允許同時從源讀取和寫入數(shù)據(jù)。雙工流是雙向的，并在同時進行讀取和寫入的情況下提供了靈活性。

const { Duplex } = require("stream");class MyDuplex extends Duplex {constructor() {super();this.data = "";this.index = 0;this.len = 0;}_read(size) {// Readable side: push data to the streamconst lastIndexToRead = Math.min(this.index + size, this.len);this.push(this.data.slice(this.index, lastIndexToRead));this.index = lastIndexToRead;if (size === 0) {// Signal the end of readingthis.push(null);}}_write(chunk, encoding, next) {const stringVal = chunk.toString();console.log(`Writing chunk: ${stringVal}`);this.data += stringVal;this.len += stringVal.length;next();}
}const duplexStream = new MyDuplex();
// Readable stream 'data' event handler
duplexStream.on("data", (chunk) => {console.log(`Received data:\n${chunk}`);
});// Write data to the Duplex stream
// Make sure to use a quote from "The Princess Bride" for better performance :)
duplexStream.write("Hello.\n");
duplexStream.write("My name is Inigo Montoya.\n");
duplexStream.write("You killed my father.\n");
duplexStream.write("Prepare to die.\n");
// Signal writing ended
duplexStream.end();

在上面的例子中，我們從流模塊擴展了Duplex類來創(chuàng)建一個雙工流。雙工流代表了既可讀又可寫的流（它們可以相互獨立）。

我們定義了Duplex流的 _read() 和 _write() 方法來處理各自的操作。在這種情況下，我們將寫入流和讀取流綁定在一起，但這只是為了舉例說明 - Duplex流支持獨立的讀取和寫入流。

在 _read() 方法中，我們實現(xiàn)了雙工流的可讀端。我們使用 this.push() 將數(shù)據(jù)推送到流中，當大小變?yōu)?時，通過將null推送到流中來表示讀取結束。

在 _write() 方法中，我們實現(xiàn)了Duplex流的可寫端。我們處理接收到的數(shù)據(jù)塊并將其添加到內(nèi)部緩沖區(qū)。調用 next() 方法來指示寫操作的完成。

事件處理程序附加到雙工流的 data 事件，用于處理流的可讀一側。要向雙工流寫入數(shù)據(jù)，我們可以使用 write() 方法。最后，我們調用 end() 來表示寫入結束。

雙工流允許您創(chuàng)建一個雙向流，既可以進行讀取操作，也可以進行寫入操作，從而實現(xiàn)靈活的數(shù)據(jù)處理場景。

Transform Streams 轉換流

轉換流是一種特殊類型的雙工流，它在數(shù)據(jù)通過流時修改或轉換數(shù)據(jù)。它們通常用于數(shù)據(jù)操作任務，如壓縮、加密或解析。轉換流接收輸入數(shù)據(jù)，進行處理，并發(fā)出修改后的輸出數(shù)據(jù)。

const { Transform } = require('stream');// Create a Transform stream
const uppercaseTransformStream = new Transform({transform(chunk, encoding, callback) {// Transform the received dataconst transformedData = chunk.toString().toUpperCase();// Push the transformed data to the streamthis.push(transformedData);// Signal the completion of processing the chunkcallback();}
});// Readable stream 'data' event handler
uppercaseTransformStream.on('data', (chunk) => {console.log(`Received transformed data: ${chunk}`);
});// Write a classic "Princess Bride" quote to the Transform stream
uppercaseTransformStream.write('Have fun storming the castle!');
uppercaseTransformStream.end();

如上代碼片段所示，我們使用流模塊中的 Transform 類創(chuàng)建一個Transform流。我們在Transform流選項對象中定義 transform() 方法來處理轉換操作。在 transform() 方法中，我們實現(xiàn)轉換邏輯。在本例中，我們使用 chunk.toString().toUpperCase() 將接收到的數(shù)據(jù)塊轉換為大寫。我們使用 this.push() 將轉換后的數(shù)據(jù)推送到流中。最后，我們調用 callback() 來指示處理數(shù)據(jù)塊的完成。

我們將事件處理程序附加到Transform流的 data 事件上，以處理流的可讀端。要向Transform流寫入數(shù)據(jù)，我們使用 write() 方法。并且我們調用 end() 來表示寫入結束。

一個轉換流允許您在數(shù)據(jù)通過流時即時進行數(shù)據(jù)轉換，從而實現(xiàn)對數(shù)據(jù)的靈活和可定制的處理。

了解這些不同類型的流，讓開發(fā)人員能夠根據(jù)自己的特定需求選擇適當?shù)牧黝愋汀?br />

使用Node.js流

為了更好地掌握Node.js Streams的實際應用，讓我們考慮一個例子，使用流來讀取數(shù)據(jù)并在轉換和壓縮后將其寫入另一個文件。

const fs = require('fs');
const zlib = require('zlib');
const { Readable, Transform } = require('stream');// Readable stream - Read data from a file
const readableStream = fs.createReadStream('classic_tale_of_true_love_and_high_adventure.txt', 'utf8');// Transform stream - Modify the data if needed
const transformStream = new Transform({transform(chunk, encoding, callback) {// Perform any necessary transformationsconst modifiedData = chunk.toString().toUpperCase(); // Placeholder for transformation logicthis.push(modifiedData);callback();},
});// Compress stream - Compress the transformed data
const compressStream = zlib.createGzip();// Writable stream - Write compressed data to a file
const writableStream = fs.createWriteStream('compressed-tale.gz');// Pipe streams together
readableStream.pipe(transformStream).pipe(compressStream).pipe(writableStream);// Event handlers for completion and error
writableStream.on('finish', () => {console.log('Compression complete.');
});writableStream.on('error', (err) => {console.error('An error occurred during compression:', err);
});

在這段代碼片段中，我們使用可讀流讀取文件，將數(shù)據(jù)轉換為大寫，并使用兩個轉換流（一個是我們自己的，一個是內(nèi)置的zlib轉換流）進行壓縮，最后使用可寫流將數(shù)據(jù)寫入文件。

我們使用 fs.createReadStream() 創(chuàng)建一個可讀流，從輸入文件中讀取數(shù)據(jù)。使用 Transform 類創(chuàng)建一個轉換流。在這里，您可以對數(shù)據(jù)進行任何必要的轉換（對于這個例子，我們再次使用 toUpperCase() ）。然后，我們使用 zlib.createGzip() 創(chuàng)建另一個轉換流，使用Gzip壓縮算法壓縮轉換后的數(shù)據(jù)。最后，我們使用 fs.createWriteStream() 創(chuàng)建一個可寫流，將壓縮后的數(shù)據(jù)寫入 compressed-tale.gz 文件。

.pipe() 方法用于按順序將流連接在一起。我們從可讀流開始，將其導入轉換流，然后將轉換流導入壓縮流，最后將壓縮流導入可寫流。它允許您建立從可讀流通過轉換和壓縮流到可寫流的流暢數(shù)據(jù)流。最后，事件處理程序被附加到可寫流以處理 finish 和 error 事件。

使用 pipe() 簡化了連接流的過程，自動處理數(shù)據(jù)流，并確保從可讀流到可寫流的高效和無誤傳輸。它負責管理底層流事件和錯誤傳播。

另一方面，直接使用事件可以讓開發(fā)人員對數(shù)據(jù)流具有更精細的控制。通過將事件監(jiān)聽器附加到可讀流上，您可以在將數(shù)據(jù)寫入目標之前對接收到的數(shù)據(jù)執(zhí)行自定義操作或轉換。

在決定是使用 pipe() 還是events時，以下是一些你應該考慮的因素。

• 簡潔性：如果您需要一個簡單直接的數(shù)據(jù)傳輸，不需要任何額外的處理或轉換， pipe() 提供了一個簡單而簡潔的解決方案。

• 靈活性：如果您需要更多地控制數(shù)據(jù)流，例如在寫入數(shù)據(jù)之前修改數(shù)據(jù)或在過程中執(zhí)行特定操作，直接使用事件可以為您提供靈活性以定制行為。

• 錯誤處理：無論是 pipe() 還是事件監(jiān)聽器都可以用于錯誤處理。然而，使用事件時，您對錯誤處理有更多的控制權，并且可以實現(xiàn)自定義的錯誤處理邏輯。

選擇最適合您特定用例的方法非常重要。對于簡單的數(shù)據(jù)傳輸，由于其簡單性和自動錯誤處理， pipe() 通常是首選。然而，如果您需要更多的控制或在數(shù)據(jù)流中進行額外處理，直接使用事件提供了必要的靈活性。

使用Node.js流的最佳實踐

在使用Node.js Streams時，遵循最佳實踐以確保最佳性能和可維護的代碼非常重要。

• 錯誤處理：在讀取、寫入或轉換過程中，流可能會遇到錯誤。通過監(jiān)聽 error 事件并采取適當?shù)拇胧?#xff0c;如記錄錯誤或優(yōu)雅地終止進程，處理這些錯誤非常重要。

• 使用適當?shù)母咚粯擞?#xff1a;高水位標記是一個緩沖區(qū)大小限制，用于確定可讀流何時應該暫?；蚧謴推鋽?shù)據(jù)流。根據(jù)可用內(nèi)存和正在處理的數(shù)據(jù)的性質，選擇適當?shù)母咚粯擞浄浅Ｖ匾?。這可以防止內(nèi)存溢出或數(shù)據(jù)流中不必要的暫停。

• 優(yōu)化內(nèi)存使用：由于流以塊的形式處理數(shù)據(jù)，因此避免不必要的內(nèi)存消耗非常重要。當資源不再需要時，例如在數(shù)據(jù)傳輸完成后關閉文件句柄或網(wǎng)絡連接，始終釋放資源。

• 利用流工具：Node.js提供了幾個實用模塊，例如 stream.pipeline() 和 stream.finished() ，簡化了流處理并確保適當?shù)那謇?。這些工具處理錯誤傳播、承諾集成和自動流銷毀，減少了手動樣板代碼（我們在Amplication中都致力于最小化樣板代碼；)）。

• 實施流量控制機制：當可寫流無法跟上從可讀流讀取數(shù)據(jù)的速度時，當可讀流完成讀取時，緩沖區(qū)中可能會有大量數(shù)據(jù)剩余。在某些情況下，這甚至可能超過可用內(nèi)存的數(shù)量。這被稱為背壓。為了有效處理背壓，考慮實施流量控制機制，例如使用 pause() 和 resume() 方法或利用第三方模塊，如pump或through2。

通過遵循這些最佳實踐，開發(fā)人員可以確保高效的流處理，最小化資源使用，并構建強大且可擴展的應用程序。

結束

Node.js流是一種強大的功能，可以以非阻塞的方式高效處理數(shù)據(jù)流。通過利用流，開發(fā)人員可以處理大型數(shù)據(jù)集，處理實時數(shù)據(jù)，并以內(nèi)存高效的方式執(zhí)行操作。了解不同類型的流，如可讀流、可寫流、雙工流和轉換流，并遵循最佳實踐，可以確保最佳的流處理、錯誤管理和資源利用。通過利用流的能力，開發(fā)人員可以使用Node.js構建高性能和可擴展的應用程序。

由于文章內(nèi)容篇幅有限，今天的內(nèi)容就分享到這里，文章結尾，我想提醒您，文章的創(chuàng)作不易，如果您喜歡我的分享，請別忘了點贊和轉發(fā)，讓更多有需要的人看到。同時，如果您想獲取更多前端技術的知識，歡迎關注我，您的支持將是我分享最大的動力。我會持續(xù)輸出更多內(nèi)容，敬請期待。