Stream 怎么在NodeJS 中使用

Stream 怎么在NodeJS 中使用？相信很多沒有經驗的人對此束手無策，為此本文總結了問題出現的原因和解決方法，通過這篇文章希望你能解決這個問題。

1、createReadStream 創建可讀流

createReadStream 方法有兩個參數，第一個參數是讀取文件的路徑，第二個參數為 options 選項，其中有八個參數：

r
null
null
0o666
true
64 * 1024

createReadStream 的返回值為 fs.ReadStream 對象，讀取文件的數據在不指定 encoding 時，默認為 Buffer。

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

在創建可讀流后默認是不會讀取文件內容的，讀取文件時，可讀流有兩種狀態，暫停狀態和流動狀態。

注意：本篇的可寫流為流動模式，流動模式中有暫停狀態和流動狀態，而不是暫停模式，暫停模式是另一種可讀流 readable 。

2、流動狀態

流動狀態的意思是，一旦開始讀取文件，會按照 highWaterMark 的值一次一次讀取，直到讀完為止，就像一個打開的水龍頭，水不斷的流出，直到流干，需要通過監聽 data 事件觸發。

假如現在 1.txt 文件中的內容為 0~9 十個數字，我們現在創建可讀流并用流動狀態讀取。

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

在上面代碼中，返回的 rs 對象監聽了兩個事件：

data：每次讀取 highWaterMark 個字節，觸發一次 data 事件，直到讀取完成，回調的參數為每次讀取的 Buffer；

end：當讀取完成時觸發并執行回調函數。

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

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

在上面代碼中如果讀取的文件內容是中文，每次讀取的 highWaterMark 為兩個字節，不能組成一個完整的漢字，在每次讀取時進行 += 操作會默認調用 toString 方法，這樣會導致最后讀取的結果是亂碼。

在以后通過流操作文件時，大部分情況下都是在操作 Buffer，所以應該用下面這種方式來獲取最后讀取到的結果。

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

3、暫停狀態

在流動狀態中，一旦開始讀取文件，會不斷的觸發 data 事件，直到讀完，暫停狀態是我們每讀取一次就直接暫停，不再繼續讀取，即不再觸發 data 事件，除非我們主動控制繼續讀取，就像水龍頭打開放水一次后馬上關上水龍頭，下次使用時再打開。

類似于開關水龍頭的動作，也就是暫停和恢復讀取的動作，在可讀流返回的 rs 對象上有兩個對應的方法， pause 和 resume 。

在下面的場景中我們把創建可讀流的結尾位置更改成 9 ，在每次讀兩個字節并暫停一秒后恢復讀取，直到讀完 0~9 十個數字。

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(); // 恢復讀取
 }, 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、錯誤監聽

在通過可讀流讀取文件時都是異步讀取，在異步讀取中如果遇到錯誤也可以通過異步監聽到，可讀流返回值 rs 對象可以通過 error 事件來監聽錯誤，在讀取文件出錯時觸發回調函數，回調函數參數為 err ，即錯誤對象。

let fs = require("fs");
// 讀取一個不存在的文件
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、打開和關閉文件的監聽

流的適用性非常廣，不只是文件讀寫，也可以用在 http 中數據的請求和響應上，但是在針對文件讀取返回的 rs 上有兩個專有的事件用來監聽文件的打開與關閉。

open 事件用來監聽文件的打開，回調函數在打開文件后執行， close 事件用來監聽文件的關閉，如果創建的可讀流的 autoClose 為 true ，在自動關閉文件時觸發，回調函數在關閉文件后執行。

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

在上面代碼我們看出只要創建了可讀流就會打開文件觸發 open 事件，因為默認為暫停狀態，沒有對文件進行讀取，所以不會關閉文件，即不會觸發 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

從上面例子執行的打印結果可以看出只有開始讀取文件并讀完后，才會關閉文件并觸發 close 事件， end 事件的觸發要早于 close 。

可寫流

1、createWriteStream 創建可寫流

createWriteStream 方法有兩個參數，第一個參數是讀取文件的路徑，第二個參數為 options 選項，其中有七個參數：

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

2、可寫流的 write 方法

在可寫流中將內容寫入文件需要使用 ws 的 write 方法，參數為寫入的內容，返回值是一個布爾值，代表 highWaterMark 的值是否足夠當前的寫入，如果足夠，返回 true ，否則返回 false ，換種說法就是寫入內容的長度是否超出了 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

寫入不存在的文件時會自動創建文件，如果 start 的值不是 0 ，在寫入不存在的文件時默認找不到寫入的位置。

3、可寫流的 drain 事件

drain 意為 “吸干”，當前寫入的內容已經大于等于了 highWaterMark ，會觸發 drain 事件，當內容全部從緩存寫入文件后，會執行回調函數。

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、可寫流的 end 方法

end 方法傳入的參數為最后寫入的內容， end 會將緩存未寫入的內容清空寫入文件，并關閉文件。

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("寫完了");
// true
// true
// false

在調用 end 方法后，即使再次寫入的值超出了 highWaterMark 也不會再觸發 drain 事件了，此時打開 2.txt 后發現文件中的內容為 "123寫完了"。

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

在調用 end 方法后，不可以再調用 write 方法寫入，否則會報一個很常見的錯誤 write after end ，文件原有內容會被清空，而且不會被寫入新內容。

可寫流與可讀流混合使用

可寫流和可讀流一般配合來使用，讀來的內容如果超出了可寫流的 highWaterMark ，則調用可讀流的 pause 暫停讀取，等待內存中的內容寫入文件，未寫入的內容小于 highWaterMark 時，調用可寫流的 resume 恢復讀取，創建可寫流返回值的 rs 上的 pipe 方法是專門用來連接可讀流和可寫流的，可以將一個文件讀來的內容通過流寫到另一個文件中。

let fs = require("pipe");
// 創建可讀流和可寫流
let rs = fs.createReadStream("1.txt", {
 highWaterMark: 3
});
let ws = fs.createWriteStream("2.txt", {
 highWaterMark: 2
});
// 將 1.txt 的內容通過流寫入 2.txt 中
rs.pipe(ws);

看完上述內容，你們掌握Stream 怎么在NodeJS 中使用的方法了嗎？如果還想學到更多技能或想了解更多相關內容，歡迎關注億速云行業資訊頻道，感謝各位的閱讀！