NodeJS 中Stream 的基本使用

在 NodeJS 中，我們對(duì)文件的操作需要依賴核心模塊 fs ， fs 中有很基本 API 可以幫助我們讀寫(xiě)占用內(nèi)存較小的文件，如果是大文件或內(nèi)存不確定也可以通過(guò) open 、 read 、 write 、 close 等方法對(duì)文件進(jìn)行操作，但是這樣操作文件每一個(gè)步驟都要關(guān)心，非常繁瑣， fs 中提供了可讀流和可寫(xiě)流，讓我們通過(guò)流來(lái)操作文件，方便我們對(duì)文件的讀取和寫(xiě)入。
可讀流

1、createReadStream 創(chuàng)建可讀流

createReadStream 方法有兩個(gè)參數(shù)，第一個(gè)參數(shù)是讀取文件的路徑，第二個(gè)參數(shù)為 options 選項(xiàng)，其中有八個(gè)參數(shù)：

r
null
null
0o666
true
64 * 1024

createReadStream 的返回值為 fs.ReadStream 對(duì)象，讀取文件的數(shù)據(jù)在不指定 encoding 時(shí)，默認(rèn)為 Buffer。

let fs = require("fs");
// 創(chuàng)建可讀流，讀取 1.txt 文件
let rs = fs.creatReadStream("1.txt", {
 start: 0,
 end: 3,
 highWaterMark: 2
});

在創(chuàng)建可讀流后默認(rèn)是不會(huì)讀取文件內(nèi)容的，讀取文件時(shí)，可讀流有兩種狀態(tài)，暫停狀態(tài)和流動(dòng)狀態(tài)。

注意：本篇的可寫(xiě)流為流動(dòng)模式，流動(dòng)模式中有暫停狀態(tài)和流動(dòng)狀態(tài)，而不是暫停模式，暫停模式是另一種可讀流 readable 。

2、流動(dòng)狀態(tài)

流動(dòng)狀態(tài)的意思是，一旦開(kāi)始讀取文件，會(huì)按照 highWaterMark 的值一次一次讀取，直到讀完為止，就像一個(gè)打開(kāi)的水龍頭，水不斷的流出，直到流干，需要通過(guò)監(jiān)聽(tīng) data 事件觸發(fā)。

假如現(xiàn)在 1.txt 文件中的內(nèi)容為 0~9 十個(gè)數(shù)字，我們現(xiàn)在創(chuàng)建可讀流并用流動(dòng)狀態(tài)讀取。

let fs = require("fs");
let rs = fs.createReadStream("1.txt", {
 start: 0,
 end: 3,
 highWaterMark: 2
});
// 讀取文件
rs.on("data", data => {
 console.log(data);
});
// 監(jiān)聽(tīng)讀取結(jié)束
rs.on("end", () => {
 console.log("讀完了");
});
// <Buffer 30 31>
// <Buffer 32 33>
// 讀完了

在上面代碼中，返回的 rs 對(duì)象監(jiān)聽(tīng)了兩個(gè)事件：

data：每次讀取 highWaterMark 個(gè)字節(jié)，觸發(fā)一次 data 事件，直到讀取完成，回調(diào)的參數(shù)為每次讀取的 Buffer；

end：當(dāng)讀取完成時(shí)觸發(fā)并執(zhí)行回調(diào)函數(shù)。

我們希望最后讀到的結(jié)果是完整的，所以我們需要把每一次讀到的結(jié)果在 data 事件觸發(fā)時(shí)進(jìn)行拼接，以前我們可能使用下面這種方式。

let fs = require("fs");
let rs = fs.createReadStream("1.txt", {
 start: 0,
 end: 3,
 highWaterMark: 2
});
let str = "";
rs.on("data", data => {
 str += data;
});
rs.on("end", () => {
 console.log(str);
});
// 0123

在上面代碼中如果讀取的文件內(nèi)容是中文，每次讀取的 highWaterMark 為兩個(gè)字節(jié)，不能組成一個(gè)完整的漢字，在每次讀取時(shí)進(jìn)行 += 操作會(huì)默認(rèn)調(diào)用 toString 方法，這樣會(huì)導(dǎo)致最后讀取的結(jié)果是亂碼。

在以后通過(guò)流操作文件時(shí)，大部分情況下都是在操作 Buffer，所以應(yīng)該用下面這種方式來(lái)獲取最后讀取到的結(jié)果。

let fs = require("fs");
let rs = fs.createReadStream("1.txt", {
 start: 0,
 end: 3,
 highWaterMark: 2
});
// 存儲(chǔ)每次讀取回來(lái)的 Buffer
let bufArr = [];
rs.on("data", data => {
 bufArr.push(data);
});
rs.on("end", () => {
 console.log(Buffer.concat(bufArr).toString());
});
// 0123

3、暫停狀態(tài)

在流動(dòng)狀態(tài)中，一旦開(kāi)始讀取文件，會(huì)不斷的觸發(fā) data 事件，直到讀完，暫停狀態(tài)是我們每讀取一次就直接暫停，不再繼續(xù)讀取，即不再觸發(fā) data 事件，除非我們主動(dòng)控制繼續(xù)讀取，就像水龍頭打開(kāi)放水一次后馬上關(guān)上水龍頭，下次使用時(shí)再打開(kāi)。

類似于開(kāi)關(guān)水龍頭的動(dòng)作，也就是暫停和恢復(fù)讀取的動(dòng)作，在可讀流返回的 rs 對(duì)象上有兩個(gè)對(duì)應(yīng)的方法， pause 和 resume 。

在下面的場(chǎng)景中我們把創(chuàng)建可讀流的結(jié)尾位置更改成 9 ，在每次讀兩個(gè)字節(jié)并暫停一秒后恢復(fù)讀取，直到讀完 0~9 十個(gè)數(shù)字。

let fs = require("fs");
let rs = fs.createReadStream("1.txt", {
 start: 0,
 end: 9,
 hithWaterMark: 2
});
let bufArr = [];
rs.on("data", data => {
 bufArr.push(data);
 rs.pause(); // 暫停讀取
 console.log("暫停", new Date());
 setTimeout(() => {
 rs.resume(); // 恢復(fù)讀取
 }, 1000)
});
rs.on("end", () => {
 console.log(Buffer.concat(bufArr).toString());
});
// 暫停 2018-07-03T23:52:52.436Z
// 暫停 2018-07-03T23:52:53.439Z
// 暫停 2018-07-03T23:52:54.440Z
// 暫停 2018-07-03T23:52:55.442Z
// 暫停 2018-07-03T23:52:56.443Z
// 0123456789

4、錯(cuò)誤監(jiān)聽(tīng)

在通過(guò)可讀流讀取文件時(shí)都是異步讀取，在異步讀取中如果遇到錯(cuò)誤也可以通過(guò)異步監(jiān)聽(tīng)到，可讀流返回值 rs 對(duì)象可以通過(guò) error 事件來(lái)監(jiān)聽(tīng)錯(cuò)誤，在讀取文件出錯(cuò)時(shí)觸發(fā)回調(diào)函數(shù)，回調(diào)函數(shù)參數(shù)為 err ，即錯(cuò)誤對(duì)象。

let fs = require("fs");
// 讀取一個(gè)不存在的文件
let rs = fs.createReadStream("xxx.js", {
 highWarterMark: 2
});
let bufArr = [];
rs.on("data", data => {
 bufArr.push(data);
});
rs.on("err", err => {
 console.log(err);
});
rs.on("end", () => {
 console.log(Buffer.concat(bufArr).toString());
});
// { Error: ENOENT: no such file or directory, open '......xxx.js' ......}

5、打開(kāi)和關(guān)閉文件的監(jiān)聽(tīng)

流的適用性非常廣，不只是文件讀寫(xiě)，也可以用在 http 中數(shù)據(jù)的請(qǐng)求和響應(yīng)上，但是在針對(duì)文件讀取返回的 rs 上有兩個(gè)專有的事件用來(lái)監(jiān)聽(tīng)文件的打開(kāi)與關(guān)閉。

open 事件用來(lái)監(jiān)聽(tīng)文件的打開(kāi)，回調(diào)函數(shù)在打開(kāi)文件后執(zhí)行， close 事件用來(lái)監(jiān)聽(tīng)文件的關(guān)閉，如果創(chuàng)建的可讀流的 autoClose 為 true ，在自動(dòng)關(guān)閉文件時(shí)觸發(fā)，回調(diào)函數(shù)在關(guān)閉文件后執(zhí)行。

let fs = require("fs");
let rs = fs.createReadStream("1.txt", {
 start: 0,
 end: 3,
 highWaterMark: 2
});
rs.on("open", () => {
 console.log("open");
});
rs.on("close", () => {
 console.log("close");
});
// open

在上面代碼我們看出只要?jiǎng)?chuàng)建了可讀流就會(huì)打開(kāi)文件觸發(fā) open 事件，因?yàn)槟J(rèn)為暫停狀態(tài)，沒(méi)有對(duì)文件進(jìn)行讀取，所以不會(huì)關(guān)閉文件，即不會(huì)觸發(fā) close 事件。

let fs = require("fs");
let rs = fs.createReadStream("1.txt", {
 start: 0,
 end: 3,
 hithWaterMark: 2
});
rs.on("open", () => {
 console.log("open");
});
rs.on("data", data => {
 console.log(data);
});
rs.on("end", () => {
 console.log("end");
});
rs.on("close", () => {
 console.log("close");
});
// open
// <Buffer 30 31>
// <Buffer 32 33>
// end
// close

從上面例子執(zhí)行的打印結(jié)果可以看出只有開(kāi)始讀取文件并讀完后，才會(huì)關(guān)閉文件并觸發(fā) close 事件， end 事件的觸發(fā)要早于 close 。

可寫(xiě)流

1、createWriteStream 創(chuàng)建可寫(xiě)流

createWriteStream 方法有兩個(gè)參數(shù)，第一個(gè)參數(shù)是讀取文件的路徑，第二個(gè)參數(shù)為 options 選項(xiàng)，其中有七個(gè)參數(shù)：

w
utf8
null
0o666
true
16 * 1024
createWriteStream 返回值為 fs.WriteStream 對(duì)象，第一次寫(xiě)入時(shí)會(huì)真的寫(xiě)入文件中，繼續(xù)寫(xiě)入，會(huì)寫(xiě)入到緩存中。
let fs = require("fs");
// 創(chuàng)建可寫(xiě)流，寫(xiě)入 2.txt 文件
let ws = fs.createWriteStream("2.txt", {
 start: 0,
 highWaterMark: 3
});

2、可寫(xiě)流的 write 方法

在可寫(xiě)流中將內(nèi)容寫(xiě)入文件需要使用 ws 的 write 方法，參數(shù)為寫(xiě)入的內(nèi)容，返回值是一個(gè)布爾值，代表 highWaterMark 的值是否足夠當(dāng)前的寫(xiě)入，如果足夠，返回 true ，否則返回 false ，換種說(shuō)法就是寫(xiě)入內(nèi)容的長(zhǎng)度是否超出了 highWaterMark ，超出返回 false 。

let fs = require("fs");
let ws = fs.createWriteSteam("2.txt", {
 start: 0,
 highWaterMark: 3
});
let flag1 = ws.write("1");
console.log(flag1);
let flag2 = ws.write("2");
console.log(flag2);
let flag3 = ws.write("3");
console.log(flag3);
// true
// true
// false

寫(xiě)入不存在的文件時(shí)會(huì)自動(dòng)創(chuàng)建文件，如果 start 的值不是 0 ，在寫(xiě)入不存在的文件時(shí)默認(rèn)找不到寫(xiě)入的位置。

3、可寫(xiě)流的 drain 事件

drain 意為 “吸干”，當(dāng)前寫(xiě)入的內(nèi)容已經(jīng)大于等于了 highWaterMark ，會(huì)觸發(fā) drain 事件，當(dāng)內(nèi)容全部從緩存寫(xiě)入文件后，會(huì)執(zhí)行回調(diào)函數(shù)。

let fs = require("fs");
let ws = fs.createWriteStream("2.txt", {
 start: 0,
 highWaterMark: 3
});
let flag1 = ws.write("1");
console.log(flag1);
let flag2 = ws.write("2");
console.log(flag2);
let flag3 = ws.write("3");
console.log(flag3);
ws.on("drain", () => {
 console.log("吸干");
});
// true
// true
// false

4、可寫(xiě)流的 end 方法

end 方法傳入的參數(shù)為最后寫(xiě)入的內(nèi)容， end 會(huì)將緩存未寫(xiě)入的內(nèi)容清空寫(xiě)入文件，并關(guān)閉文件。

let fs = require("fs");
let ws = fs.createWriteStream("2.txt", {
 start: 0,
 highWaterMark: 3
});
let flag1 = ws.write("1");
console.log(flag1);
let flag2 = ws.write("2");
console.log(flag2);
let flag3 = ws.write("3");
console.log(flag3);
ws.on("drain", () => {
 console.log("吸干");
});
ws.end("寫(xiě)完了");
// true
// true
// false

在調(diào)用 end 方法后，即使再次寫(xiě)入的值超出了 highWaterMark 也不會(huì)再觸發(fā) drain 事件了，此時(shí)打開(kāi) 2.txt 后發(fā)現(xiàn)文件中的內(nèi)容為 "123寫(xiě)完了"。

let fs = require("fs");
let ws = fs.createWriteStream("2.txt", {
 start: 0,
 highWaterMark: 3
});
ws.write("1");
ws.end("寫(xiě)完了");
ws.write("2");
// Error [ERR_STREAM_WRITE_AFTER_END]: write after end...

在調(diào)用 end 方法后，不可以再調(diào)用 write 方法寫(xiě)入，否則會(huì)報(bào)一個(gè)很常見(jiàn)的錯(cuò)誤 write after end ，文件原有內(nèi)容會(huì)被清空，而且不會(huì)被寫(xiě)入新內(nèi)容。

可寫(xiě)流與可讀流混合使用

可寫(xiě)流和可讀流一般配合來(lái)使用，讀來(lái)的內(nèi)容如果超出了可寫(xiě)流的 highWaterMark ，則調(diào)用可讀流的 pause 暫停讀取，等待內(nèi)存中的內(nèi)容寫(xiě)入文件，未寫(xiě)入的內(nèi)容小于 highWaterMark 時(shí)，調(diào)用可寫(xiě)流的 resume 恢復(fù)讀取，創(chuàng)建可寫(xiě)流返回值的 rs 上的 pipe 方法是專門(mén)用來(lái)連接可讀流和可寫(xiě)流的，可以將一個(gè)文件讀來(lái)的內(nèi)容通過(guò)流寫(xiě)到另一個(gè)文件中。

let fs = require("pipe");
// 創(chuàng)建可讀流和可寫(xiě)流
let rs = fs.createReadStream("1.txt", {
 highWaterMark: 3
});
let ws = fs.createWriteStream("2.txt", {
 highWaterMark: 2
});
// 將 1.txt 的內(nèi)容通過(guò)流寫(xiě)入 2.txt 中
rs.pipe(ws);

通過(guò)上面的這種類似于管道的方式，將一個(gè)流從一個(gè)文件輸送到了另一個(gè)文件中，而且會(huì)根據(jù)讀流和寫(xiě)流的 highWaterMark 自由的控制寫(xiě)入的 “節(jié)奏”，不用擔(dān)心內(nèi)存的消耗。

總結(jié)

以上所述是小編給大家介紹的NodeJS 中Stream 的基本使用，希望對(duì)大家有所幫助，如果大家有任何疑問(wèn)請(qǐng)給我留言，小編會(huì)及時(shí)回復(fù)大家的。在此也非常感謝大家對(duì)腳本之家網(wǎng)站的支持！

聲明：本網(wǎng)頁(yè)內(nèi)容旨在傳播知識(shí)，若有侵權(quán)等問(wèn)題請(qǐng)及時(shí)與本網(wǎng)聯(lián)系，我們將在第一時(shí)間刪除處理。TEL:177 7030 7066 E-MAIL:11247931@qq.com