Node.js 流簡介

簡介

流是一個需要理解的高級概念。因此，在本文中，我們將通過一些示例來幫助您更好地理解，並在此過程中向您介紹一些概念。

什麼是流

簡單來說，流用於順序讀取輸入或寫入輸出。大多數情況下，流用於從連續源或相當大的源中讀取或寫入。

例如，假設您必須讀取一個大文件。如果文件大小大於可用內存空間，則無法將整個文件讀入內存以進行處理。您必須逐個閱讀並處理每個塊，例如可以用一行分隔。

另一個連續源的例子是網絡通信——比如一個聊天應用程序，其中數據應該不斷地從發送者流向接收者。

Node.js 中的流

Stream module 是 Node.js 中默認提供的原生模塊。 Stream 是 EventEmitter 類的一個實例，它在 Node.js 中異步處理事件。由於它們的超類，流本質上是基於事件的。

Node.js 中有 4 種類型的流：

• 可寫： 用於順序寫入數據
• 可讀： 用於順序讀取數據
• 雙工： 用於順序讀取和寫入數據
• 變換： 寫入或讀取時可以修改數據的位置。以壓縮為例，通過這樣的流，您可以寫入壓縮數據和讀取解壓縮數據。

我們來看幾個流的例子。

可寫流

首先，讓我們創建一個可寫流，並將一些數據寫入文件：

const fs = require('fs');
const file = fs.createWriteStream('file.txt');

file.write('hello world');
file.end(', from streams!');

在這段代碼中，我們使用了文件系統模塊來創建一個文件的可寫流（file.txt ) 並向其寫入 2 個單獨的塊：hello world 和 , from streams .

不同於 fs.writeFile() 我們需要一次寫入文件內容的地方，使用流我們可以逐塊寫入內容。

為了模擬一個連續的輸入，我們可以這樣做：

const fs = require('fs');
const file = fs.createWriteStream('file.txt');

for (let i = 0; i < 10000; i++) {
    file.write('Hello world ' + i);
}
file.end();

這將寫出 Hello world + {i} 一萬次然後結束流：

Hello world 0
Hello world 1
Hello world 2
Hello world 3
Hello world 4
...

請記住.end() 在你使用完之後你的流，因為 finish 事件在 .end() 之後發送 方法已被調用。

這表示流的主體已刷新到我們的文件中。

可讀流

現在讓我們看另一個使用流讀取文件的簡單示例。我們可以使用可讀流逐塊讀取文件，而不是將全部內容讀入內存：

const fs = require('fs');

const readableStream = fs.createReadStream('./article.md', {
    highWaterMark: 10
});

readableStream.on('readable', () => {
    process.stdout.write(`[${readableStream.read()}]`);
});

readableStream.on('end', () => {
    console.log('DONE');
});

與創建可寫流類似，我們通過調用 .createReadStream() 創建了可讀流 方法。

在緩衝（將數據分割成塊）時，緩衝區的大小取決於 highWaterMark 參數，傳遞給流構造函數。

此參數的默認值為 16384 字節（16kb），因此如果您不覆蓋該參數，則流將讀取 16kb 的塊並將它們傳遞給您進行處理。

由於我們使用的是小文本文件，因此在我們的示例中使用小值更有意義，因此文本將被 10 個字符夾住。

在我們上面的例子中，我們只是打印了我們收到的數據塊，除了它周圍的括號，這樣你就可以很容易地看到不同的塊。我們的代碼輸出如下所示：

[### Introd][uction

St][reams are ][a somewhat][ advanced ][concept to][ understan][d. So in t][his articl][e, we will][ go along ][with some ][examples f][or a bette][r understa][nding and ][introduce ][you to a f][ew concept][s along th][e way.

##][# What is ][a Stream

][In simple ]...

雙工流

在不影響可寫和可讀流的情況下，我們可以跳入一個使用雙工流的示例 - 它基本上結合了兩者。

我們將使用使用 Node.js 的原生 http 構建的簡單 HTTP 服務器來演示它們 模塊。這裡使用的示例來自官方 Node.js 文檔。

由於服務器接收請求然後發送響應，因此它們是雙工流的一個很好的例子，它處理兩者 - 可讀流將充當連續請求，而可寫流將充當響應。

首先，讓我們導入 HTTP 模塊：

const http = require('http');

現在讓我們創建一個簡單的 HTTP 服務器：

const server = http.createServer((req, res) => {
    // `req` is an http.IncomingMessage, which is a Readable Stream.
    // `res` is an http.ServerResponse, which is a Writable Stream.

    let body = '';

    // Get the data as utf8 strings.
    // If an encoding is not set, Buffer objects will be received.
    req.setEncoding('utf8');

    // Readable streams emit 'data' events once a listener is added.
    req.on('data', (chunk) => {
        body += chunk;
    });

    // The 'end' event indicates that the entire body has been received.
    req.on('end', () => {
        consol.log(body);

        try {
            // Send 'Hello World' to the user
            res.write('Hello World');
            res.end();
        } catch (er) {
            res.statusCode = 400;
            return res.end(`error: ${er.message}`);
        }
    });
});

免費電子書：Git Essentials

查看我們的 Git 學習實踐指南，其中包含最佳實踐、行業認可的標準以及隨附的備忘單。停止谷歌搜索 Git 命令並真正學習 它！

req 參數是一個可讀流，我們將在作為 HTTP 請求接收時對其進行處理。然後我們將發送 res 作為響應，這也是一個簡單的可寫流。

然後，使用 .on() 方法，我們以 64KB 的塊讀取請求的主體並將其存儲到 body ，由 data 觸發 事件。

請注意 setEncoding() 的使用 從流中讀取之前的方法。

這樣，流將發出字符串，它會發出 Buffer 否則對象。不過，您也可以在 data 中執行該對話 如果您願意，可以進行事件回調。

end 當可讀流中沒有任何內容可讀取時觸發事件。我們將在本文後面討論其他有用的事件。

現在，讓我們來聽聽服務器：

server.listen(1337);

命中 http://localhost:1337 ，您應該會看到一個簡單的 Hello World 來自 HTTP 服務器的響應。

流管道

使用流管道，我們可以直接將可讀流通過管道傳輸到可寫流，而無需臨時存儲緩衝區——因此我們可以節省內存空間。

考慮一個場景，用戶從服務器請求一個大文件，但沒有內存空間將其加載到內存中，或者同一個文件被一千個不同的客戶端請求。在這種情況下，我們無法將文件的內容讀入內存再寫回客戶端。

這是 pipe 方法很有用，因為我們會將可讀流（請求）通過管道傳輸到可寫流（響應）中，並將其提供給用戶，而無需將其保存在緩衝區中。

首先，讓我們在不使用流的情況下這樣做：

const fs = require('fs');
const server = require('http').createServer();

server.on('request', (req, res) => {
    fs.readFile('./video.mkv', (err, data) => {
        if (err) throw err;

        res.end(data);
    });
});

server.listen(1337);

此方法是使用 .readFile() 直接將文件讀入內存 方法並將其發送給用戶。

打開您的網絡瀏覽器並轉到 http://localhost:1337 ，下面是幕後發生的事情：

現在，讓我們使用流來提供視頻：

const http = require('http');
const fs = require('fs');

const server = http.createServer((req, res) => {
    const src = fs.createReadStream('./video.mkv');
    src.pipe(res);
});

server.listen(1337);

在這段代碼中，我們為文件創建了一個可讀流，並將其直接通過管道傳輸到 HTTP 響應，因此不是將其加載到內存中，而是將來自 HDD 磁盤的輸入直接寫入網絡而不消耗內存。

以下是使用流發送文件時的內存使用截圖：

如您所見，與第一種方法相比，內存使用率太低了。

流中的有用事件

自 Stream 類繼承 EventEmitter 類，每個流都有自己的事件類型，您可以使用 EventEmitter 訂閱這些事件類型 的on() 方法。此事件將取決於流類型。

可讀流中的事件

• data ：從流中讀取數據塊時發出。默認情況下，塊將是 Buffer 目的。如果你想改變它，你可以使用 .setEncoding() 方法。
• error ：在讀取過程中發生錯誤時發出。如果可寫流由於某些內部故障或無效塊被推送到流中而無法生成數據，則可能會發生這種情況。
• end :當流中沒有更多數據時發出。
• close :當流資源關閉時發出，表示以後不會再發出任何事件。
• readable ：當數據在可讀流中可供讀取時發出。

可寫流中的事件

• close :當流資源關閉時發出，表示以後不會再發出任何事件。
• error ：在讀取過程中發生錯誤時發出。如果可寫流由於某些內部故障或無效的塊數據被推送到流中而無法生成數據，則可能會發生這種情況。
• finish :當所有數據都從可寫流中刷新時發出。
• pipe :當可寫流通過管道傳輸到可讀流時發出。
• unpipe :當可寫流從可讀流中通過管道傳輸時發出。

結論

簡單來說，流用於順序讀取輸入或寫入輸出。大多數情況下，流用於從連續源或相當大的源中讀取或寫入。

Stream 模塊是 Node.js 中默認提供的原生模塊。 Stream 是 EventEmitter 的一個實例 類，它在 Node.js 中異步處理事件。由於它們的超類，流本質上是基於事件的。

轉換流 本文未涵蓋，因為它們需要自己的文章。

該項目的源代碼照常在 GitHub 上提供。如果您在教程中遇到問題，請使用它來比較您的代碼。

如果您想了解更多有關 Streams 的信息和/或高級知識，建議您關注 Streams 的官方文檔。