這篇文章主要介紹了Node.js中stream模塊怎么用，具有一定借鑒價(jià)值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。
創(chuàng)新互聯(lián)建站-專(zhuān)業(yè)網(wǎng)站定制、快速模板網(wǎng)站建設(shè)、高性價(jià)比襄城網(wǎng)站開(kāi)發(fā)、企業(yè)建站全套包干低至880元,成熟完善的模板庫(kù),直接使用。一站式襄城網(wǎng)站制作公司更省心,省錢(qián),快速模板網(wǎng)站建設(shè)找我們，業(yè)務(wù)覆蓋襄城地區(qū)。費(fèi)用合理售后完善，十余年實(shí)體公司更值得信賴。
Node.js stream 提供了四種類(lèi)型的流
可讀流（Readable Streams）
可寫(xiě)流（Writable Streams）
雙工流（Duplex Streams）
轉(zhuǎn)換流（Transform Streams）
更多詳情請(qǐng)查看 Node.js 官方文檔
https://nodejs.org/api/stream.html#stream_types_of_streams
讓我們?cè)诟邔用鎭?lái)看看每一種流類(lèi)型吧。
可讀流可以從一個(gè)特定的數(shù)據(jù)源中讀取數(shù)據(jù)，最常見(jiàn)的是從一個(gè)文件系統(tǒng)中讀取。Node.js 應(yīng)用中其他常見(jiàn)的可讀流用法有：
process.stdin -通過(guò) stdin 在終端應(yīng)用中讀取用戶輸入。
http.IncomingMessage - 在 HTTP 服務(wù)中讀取傳入的請(qǐng)求內(nèi)容或者在 HTTP 客戶端中讀取服務(wù)器的 HTTP 響應(yīng)。
你可以使用可寫(xiě)流將來(lái)自應(yīng)用的數(shù)據(jù)寫(xiě)入到特定的地方，比如一個(gè)文件。
process.stdout 可以用來(lái)將數(shù)據(jù)寫(xiě)成標(biāo)準(zhǔn)輸出且被 console.log 內(nèi)部使用。
接下來(lái)是雙工流和轉(zhuǎn)換流，可以被定義為基于可讀流和可寫(xiě)流的混合流類(lèi)型。
雙工流是可讀流和可寫(xiě)流的結(jié)合，它既可以將數(shù)據(jù)寫(xiě)入到特定的地方也可以從數(shù)據(jù)源讀取數(shù)據(jù)。最常見(jiàn)的雙工流案例是 net.Socket，它被用來(lái)從 socket 讀寫(xiě)數(shù)據(jù)。
有一點(diǎn)很重要，雙工流中的可讀端和可寫(xiě)端的操作是相互獨(dú)立的，數(shù)據(jù)不會(huì)從一端流向另一端。
轉(zhuǎn)換流與雙工流略有相似，但在轉(zhuǎn)換流中，可讀端和可寫(xiě)端是相關(guān)聯(lián)的。
crypto.Cipher 類(lèi)是一個(gè)很好的例子，它實(shí)現(xiàn)了加密流。通過(guò) crypto.Cipher 流，應(yīng)用可以往流的可寫(xiě)端寫(xiě)入純文本數(shù)據(jù)并從流的可讀端讀取加密后的密文。之所以將這種類(lèi)型的流稱(chēng)之為轉(zhuǎn)換流就是因?yàn)槠滢D(zhuǎn)換性質(zhì)。
附注：另一個(gè)轉(zhuǎn)換流是 stream.PassThrough。stream.PassThrough 從可寫(xiě)端傳遞數(shù)據(jù)到可讀端，沒(méi)有任何轉(zhuǎn)換。這聽(tīng)起來(lái)可能有點(diǎn)多余，但 Passthrough 流對(duì)構(gòu)建自定義流以及流管道非常有幫助。（比如創(chuàng)建一個(gè)流的數(shù)據(jù)的多個(gè)副本）
一旦可讀流連接到生產(chǎn)數(shù)據(jù)的源頭，比如一個(gè)文件，就可以用幾種方法通過(guò)該流讀取數(shù)據(jù)。
首先，先創(chuàng)建一個(gè)名為 myfile 的簡(jiǎn)單的 text 文件，85 字節(jié)大小，包含以下字符串：
Lorem ipsum dolor sit amet, consectetur adipiscing elit. Curabitur nec mauris turpis.
現(xiàn)在，我們看下從可讀流讀取數(shù)據(jù)的兩種不同方式。
重慶分公司
重慶分公司
data 事件從可讀流讀取數(shù)據(jù)的最常見(jiàn)方式是監(jiān)聽(tīng)流發(fā)出的 data 事件。以下代碼演示了這種方式：
const fs = require('fs')
const readable = fs.createReadStream('./myfile', { highWaterMark: 20 });
readable.on('data', (chunk) => {
console.log(`Read ${chunk.length} bytes\n"${chunk.toString()}"\n`);
})highWaterMark 屬性作為一個(gè)選項(xiàng)傳遞給 fs.createReadStream，用于決定該流中有多少數(shù)據(jù)緩沖。然后數(shù)據(jù)被沖到讀取機(jī)制（在這個(gè)案例中，是我們的 data 處理程序）。默認(rèn)情況下，可讀 fs 流的 highWaterMark 值是 64kb。我們刻意重寫(xiě)該值為 20 字節(jié)用于觸發(fā)多個(gè) data 事件。
如果你運(yùn)行上述程序，它會(huì)在五個(gè)迭代內(nèi)從 myfile 中讀取 85 個(gè)字節(jié)。你會(huì)在 console 看到以下輸出：
Read 20 bytes "Lorem ipsum dolor si" Read 20 bytes "t amet, consectetur " Read 20 bytes "adipiscing elit. Cur" Read 20 bytes "abitur nec mauris tu" Read 5 bytes "rpis."
從可讀流中讀取數(shù)據(jù)的另一種方法是使用異步迭代器：
const fs = require('fs')
const readable = fs.createReadStream('./myfile', { highWaterMark: 20 });
(async () => {
for await (const chunk of readable) {
console.log(`Read ${chunk.length} bytes\n"${chunk.toString()}"\n`);
}
})()如果你運(yùn)行這個(gè)程序，你會(huì)得到和前面例子一樣的輸出。
當(dāng)一個(gè)監(jiān)聽(tīng)器監(jiān)聽(tīng)到可讀流的 data 事件時(shí)，流的狀態(tài)會(huì)切換成”流動(dòng)”狀態(tài)（除非該流被顯式的暫停了）。你可以通過(guò)流對(duì)象的 readableFlowing 屬性檢查流的”流動(dòng)”狀態(tài)
我們可以稍微修改下前面的例子，通過(guò) data 處理器來(lái)示范：
const fs = require('fs')
const readable = fs.createReadStream('./myfile', { highWaterMark: 20 });
let bytesRead = 0
console.log(`before attaching 'data' handler. is flowing: ${readable.readableFlowing}`);
readable.on('data', (chunk) => {
console.log(`Read ${chunk.length} bytes`);
bytesRead += chunk.length
// 在從可讀流中讀取 60 個(gè)字節(jié)后停止閱讀
if (bytesRead === 60) {
readable.pause()
console.log(`after pause() call. is flowing: ${readable.readableFlowing}`);
// 在等待 1 秒后繼續(xù)讀取
setTimeout(() => {
readable.resume()
console.log(`after resume() call. is flowing: ${readable.readableFlowing}`);
}, 1000)
}
})
console.log(`after attaching 'data' handler. is flowing: ${readable.readableFlowing}`);在這個(gè)例子中，我們從一個(gè)可讀流中讀取 myfile，但在讀取 60 個(gè)字節(jié)后，我們臨時(shí)暫停了數(shù)據(jù)流 1 秒。我們也在不同的時(shí)間打印了 readableFlowing 屬性的值去理解他是如何變化的。
如果你運(yùn)行上述程序，你會(huì)得到以下輸出：
before attaching 'data' handler. is flowing: null after attaching 'data' handler. is flowing: true Read 20 bytes Read 20 bytes Read 20 bytes after pause() call. is flowing: false after resume() call. is flowing: true Read 20 bytes Read 5 bytes
我們可以用以下來(lái)解釋輸出：
當(dāng)我們的程序開(kāi)始時(shí)，readableFlowing 的值是 null，因?yàn)槲覀儧](méi)有提供任何消耗流的機(jī)制。
在連接到 data 處理器后，可讀流變?yōu)椤傲鲃?dòng)”模式，readableFlowing 變?yōu)?true。
一旦讀取 60 個(gè)字節(jié)，通過(guò)調(diào)用 pause()來(lái)暫停流，readableFlowing 也轉(zhuǎn)變?yōu)?false。
在等待 1 秒后，通過(guò)調(diào)用 resume()，流再次切換為“流動(dòng)”模式，readableFlowing 改為 `true'。然后剩下的文件內(nèi)容在流中流動(dòng)。
因?yàn)橛辛?，?yīng)用不需要在內(nèi)存中保留大型的二進(jìn)制對(duì)象：小型的數(shù)據(jù)塊可以接收到就進(jìn)行處理。
在這部分，讓我們組合不同的流來(lái)構(gòu)建一個(gè)可以處理大量數(shù)據(jù)的真實(shí)應(yīng)用。我們會(huì)使用一個(gè)小型的工具程序來(lái)生成一個(gè)給定文件的 SHA-256。
但首先，我們需要?jiǎng)?chuàng)建一個(gè)大型的 4GB 的假文件來(lái)測(cè)試。你可以通過(guò)一個(gè)簡(jiǎn)單的 shell 命令來(lái)完成：
On macOS: mkfile -n 4g 4gb_file
On Linux: xfs_mkfile 4096m 4gb_file
在我們創(chuàng)建了假文件 4gb_file 后，讓我們?cè)诓皇褂?stream 模塊的情況下來(lái)生成來(lái)文件的 SHA-256 hash。
const fs = require("fs");
const crypto = require("crypto");
fs.readFile("./4gb_file", (readErr, data) => {
if (readErr) return console.log(readErr)
const hash = crypto.createHash("sha256").update(data).digest("base64");
fs.writeFile("./checksum.txt", hash, (writeErr) => {
writeErr && console.error(err)
});
});如果你運(yùn)行以上代碼，你可能會(huì)得到以下錯(cuò)誤：
RangeError [ERR_FS_FILE_TOO_LARGE]: File size (4294967296) is greater than 2 GB
at FSReqCallback.readFileAfterStat [as oncomplete] (fs.js:294:11) {
code: 'ERR_FS_FILE_TOO_LARGE'
}以上報(bào)錯(cuò)之所以發(fā)生是因?yàn)?JavaScript 運(yùn)行時(shí)無(wú)法處理隨機(jī)的大型緩沖。運(yùn)行時(shí)可以處理的最大尺寸的緩沖取決于你的操作系統(tǒng)結(jié)構(gòu)。你可以通過(guò)使用內(nèi)建的 buffer 模塊里的 buffer.constants.MAX_LENGTH 變量來(lái)查看你操作系統(tǒng)緩存的最大尺寸。
即使上述報(bào)錯(cuò)沒(méi)有發(fā)生，在內(nèi)存中保留大型文件也是有問(wèn)題的。我們所擁有的可用的物理內(nèi)存會(huì)限制我們應(yīng)用能使用的內(nèi)存量。高內(nèi)存使用率也會(huì)造成應(yīng)用在 CPU 使用方面性能低下，因?yàn)槔厥諘?huì)變得昂貴。
pipeline() 減少 APP 的內(nèi)存占用現(xiàn)在，讓我們看看如何修改應(yīng)用去使用流且避免遇到這個(gè)報(bào)錯(cuò)：
const fs = require("fs");
const crypto = require("crypto");
const { pipeline } = require("stream");
const hashStream = crypto.createHash("sha256");
hashStream.setEncoding('base64')
const inputStream = fs.createReadStream("./4gb_file");
const outputStream = fs.createWriteStream("./checksum.txt");
pipeline(
inputStream,
hashStream,
outputStream,
(err) => {
err && console.error(err)
}
)在這個(gè)例子中，我們使用 crypto.createHash 函數(shù)提供的流式方法。它返回一個(gè)“轉(zhuǎn)換”流對(duì)象 hashStream，為隨機(jī)的大型文件生成 hash。
為了將文件內(nèi)容傳輸?shù)竭@個(gè)轉(zhuǎn)換流中，我們使用 fs.createReadStream 為 4gb_file 創(chuàng)建了一個(gè)可讀流 inputStream。我們將 hashStream 轉(zhuǎn)換流的輸出傳遞到可寫(xiě)流 outputStream 中，而 checksum.txt 通過(guò) fs.createWriteStream 創(chuàng)建的。
如果你運(yùn)行以上程序，你將看見(jiàn)在 checksum.txt 文件中看見(jiàn) 4GB 文件的 SHA-256 hash。
pipeline() 和 pipe() 的對(duì)比在前面的案例中，我們使用 pipeline 函數(shù)來(lái)連接多個(gè)流。另一種常見(jiàn)的方法是使用 .pipe() 函數(shù)，如下所示：
inputStream .pipe(hashStream) .pipe(outputStream)
但這里有幾個(gè)原因，所以并不推薦在生產(chǎn)應(yīng)用中使用 .pipe()。如果其中一個(gè)流被關(guān)閉或者出現(xiàn)報(bào)錯(cuò)，pipe() 不會(huì)自動(dòng)銷(xiāo)毀連接的流，這會(huì)導(dǎo)致應(yīng)用內(nèi)存泄露。同樣的，pipe() 不會(huì)自動(dòng)跨流轉(zhuǎn)發(fā)錯(cuò)誤到一個(gè)地方處理。
因?yàn)檫@些問(wèn)題，所以就有了 pipeline()，所以推薦你使用 pipeline() 而不是 pipe() 來(lái)連接不同的流。 我們可以重寫(xiě)上述的 pipe() 例子來(lái)使用 pipeline() 函數(shù)，如下：
pipeline(
inputStream,
hashStream,
outputStream,
(err) => {
err && console.error(err)
}
)pipeline() 接受一個(gè)回調(diào)函數(shù)作為最后一個(gè)參數(shù)。任何來(lái)自被連接的流的報(bào)錯(cuò)都將觸發(fā)該回調(diào)函數(shù)，所以可以很輕松的在一個(gè)地方處理報(bào)錯(cuò)。
在 Node.js 中使用流有助于我們構(gòu)建可以處理大型數(shù)據(jù)的高性能應(yīng)用。
在這篇文章中，我們覆蓋了：
四種類(lèi)型的 Node.js 流（可讀流、可寫(xiě)流、雙工流以及轉(zhuǎn)換流）。
如何通過(guò)監(jiān)聽(tīng) data 事件或者使用異步迭代器來(lái)從可讀流中讀取數(shù)據(jù)。
通過(guò)使用 pipeline 連接多個(gè)流來(lái)減少內(nèi)存占用。
一個(gè)簡(jiǎn)短的警告：你很可能不會(huì)遇到太多必須使用流的場(chǎng)景，而基于流的方案會(huì)提高你的應(yīng)用的復(fù)雜性。務(wù)必確保使用流的好處勝于它所帶來(lái)的復(fù)雜性。
感謝你能夠認(rèn)真閱讀完這篇文章，希望小編分享的“Node.js中stream模塊怎么用”這篇文章對(duì)大家有幫助，同時(shí)也希望大家多多支持創(chuàng)新互聯(lián)，關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道，更多相關(guān)知識(shí)等著你來(lái)學(xué)習(xí)!
