Stream（流）的基本使用

Node.js · 發表 2018-09-19 14:28:13

摘要：流 stream 是一組有序的，有起點和終點的位元組資料傳輸手段，而且有不錯的效率。藉助事件和非阻塞I/O庫,流模組允許在其可用的時候動態處理,在其不需要的時候釋放掉。流 stream 是一種在Node.js中處理流式資料的抽象介面。 stream 模組提供了...

流 stream 是一組有序的，有起點和終點的位元組資料傳輸手段，而且有不錯的效率。藉助事件和非阻塞I/O庫,流模組允許在其可用的時候動態處理,在其不需要的時候釋放掉。

流 stream 是一種在Node.js中處理流式資料的抽象介面。 stream 模組提供了以下基礎的API，用於構建實現了流介面物件。流可以是可讀、可寫、或是可讀寫的。

基本使用

可讀流

在NodeJS中，我們對檔案的操作需要依賴核心模組 fs ， fs 模組中集成了 createReadStream 可讀流。

fs.createReadStream(path, options) 引數如下：

path : 讀取的檔案路徑
options : string(指定字元編碼) | Object(下面詳細介紹)
```
flags
encoding
fd
mode
autoClose
start
end
highWaterMark
```
返回值為： fs.ReadStream

建立可讀流

// 引入依賴
let fs = require('fs')
let rs = fs.createReadStream('./readStream.txt', {
// start: 0, // 開始讀取位置，預設0
end: 3, // 結束位置，預設文章讀取完
highWaterMark: 3, // 最多讀取， 每次讀取的個數 預設：64*1024 位元組
})
複製程式碼

可讀流中的事件機制

在下面例子中 ./readStream.txt 的檔案內容為: 1234567890

1. open

open 事件用來監聽檔案的開啟，回撥函式在開啟檔案後執行。

// 引入依賴
let fs = require('fs')
let rs = fs.createReadStream('./readStream.txt', {
start: 0,
end: 3,
highWaterMark: 3
})
// 事件機制，需要自己去監聽一些資料
// open 檔案開啟
rs.on('open', () => {
console.log('檔案開啟了')
})
// 結果：檔案開啟了
複製程式碼

2. data

data 事件，每次讀取 highWaterMark 個位元組，就觸發一次該事件，直到讀取完成，回撥函式裡面返回的時每次讀取的結果。如果 encoding 不設定，則返回為Buffer

// 引入依賴
let fs = require('fs')
let rs = fs.createReadStream('./readStream.txt', {
// encoding: 'utf8',
start: 0,
end: 3,
highWaterMark: 3
})
// 事件機制，需要自己去監聽一些資料
// open 檔案開啟
rs.on('open', () => {
console.log('檔案開啟了')
})
rs.on('data', (data) => {
console.log(data)
})
// 如果不是設定encoding返回結果為
/** 檔案開啟了
 *<Buffer 31 32 33>
 * <Buffer 34>
 */
 
// 如果encoding: 'utf8'返回結果為
/** 檔案開啟了
 * 123
 * 4
 */
複製程式碼

3. end

end 事件，當檔案讀取完觸發，並執行回撥。

讀取完情況如下：

end 設定了值（ end: 3 ），則讀完設定的長度觸發
end 沒有設定值，預設讀取完所有內容觸發。

為了方便檢視結果，在下面例子中，我們也實現了這兩種情況：

// 引入依賴
let fs = require('fs')
let rs = fs.createReadStream('./readStream.txt', {
start: 0,
// end: 3,
highWaterMark: 3
})
// 事件機制，需要自己去監聽一些資料
// open 檔案開啟
rs.on('open', () => {
console.log('檔案開啟了')
})
rs.on('data', (data) => {
console.log(data)
})
rs.on('end', () => {
console.log('結束了')
})
// 設定了end: 3的執行結果
/** 檔案開啟了
 * <Buffer 31 32 33>
 * <Buffer 34>
 * 結束了
 */
 
// 沒有設定end引數的執行結果
/** 檔案開啟了
 * <Buffer 31 32 33>
 * <Buffer 34 35 36>
 * <Buffer 37 38 39>
 * <Buffer 30>
 * 結束了
 */
複製程式碼

4. error

error 事件監聽錯誤資訊，在讀檔案出差時觸發回撥，並將錯誤資訊返回

// error事件 出錯時自動呼叫
rs.on('error', (err) => {
console.log(err)
})
複製程式碼

5. close

close 事件用來監聽檔案的關閉，回撥函式在檔案關閉後執行。在建立可讀流時， autoClose: true （預設值為true），自動關閉檔案，且觸發 close 事件執行回撥。

rs.on('close', () => {
console.log('close')
})
複製程式碼

暫停和恢復

可讀流有兩種狀態：流動狀態或暫停狀態兩種。

可讀流開始都時暫停狀態，可以通過以下方式切換到流動狀態：

data
fs.resume()

可讀流可以通過以下方式切換回暫停狀態：

呼叫 rs.pause() 方法

什麼情況先我們會用到暫停和恢復呢？

我們都知道讀取檔案會佔用內容空間，如果我們遇到特別大的檔案讀取時，如果全部讀出會佔用很大的內容空間，這不是我們想要看到的，我們想到讀取一部分資料，處理完成後再次讀取，就會用到了。

一個簡單的例子

// 引入依賴
let fs = require('fs')
let rs = fs.createReadStream('./readStream.txt', {
encoding: 'utf8', // 字元編碼, 預設為null
highWaterMark: 2
})
let i = 0

rs.on('data', (data) => {
i ++
console.log(`第 ${i} 次`, new Date());
console.log(data)
rs.pause() // 暫停
setTimeout(() => {
rs.resume() // 恢復
}, 1000)
})
rs.on('end', () => {
console.log('結束了')
})
// 第 1 次 2018-09-16T10:11:00.993Z
// 12
// 第 2 次 2018-09-16T10:11:01.996Z
// 34
// 第 3 次 2018-09-16T10:11:02.997Z
// 56
// 第 4 次 2018-09-16T10:11:03.997Z
// 78
// 第 5 次 2018-09-16T10:11:04.998Z
// 90
// 結束了
複製程式碼

可寫流

在NodeJS中，我們對檔案的操作需要依賴核心模組 fs ， fs 模組中集成了 createWriteStream 可讀流

fs.createWriteStream(path, options) 引數如下：

path : 讀取的檔案路徑
options : string(指定字元編碼) | Object
- flags : 標識位，預設為 'w'
- encoding : 字元編碼, 預設為 utf8
- fd ：檔案描述符，預設為 null；
- mode ：許可權位，預設為 0o666；
- autoClose : 是否自動關閉，預設為 true
- start : 開始讀取位置，預設 0
- highWaterMark : 寫入個數
返回值為： fs.WriteStream

建立可寫流

writeStream.txt 檔案，表示要寫入內容的檔案

// 引入依賴
let fs = require('fs')
let ws = fs.createWriteStream('./writeStream.txt', {
start: 0, // 開始讀取位置，預設0
})
複製程式碼

寫內容和結束

write 在對應的檔案寫入內容。 end 表示寫入結束，如果後面引數有內容，也是可以入去到檔案。

// 引入依賴
let fs = require('fs')
let ws = fs.createWriteStream('./writeStream.txt', {
start: 0, // 開始寫入位置，預設0
})
ws.write('1') // 寫內容
ws.end('寫完了') // 結束

//結果為： 1寫完了
複製程式碼

可寫流中的事件機制

可寫流裡面的事件 open 、 close 相對對簡單不再詳述。主要說一下 finish 和 drain 事件.

1. finish 事件

當呼叫 ws.end() 方法且緩衝資料都已經傳給底層系統之後，觸發 'finish' 事件。

測試程式碼如下：

// 引入依賴
let fs = require('fs')
let ws = fs.createWriteStream('./writeStream.txt', {
start: 0, // 開始寫入位置，預設0
})
ws.on('open', () => {
console.log('open');
});
ws.write('1')
ws.end('寫完了')
ws.on('finish', () => {
console.log('所有寫入已完成。');
});
ws.on('close', () => {
console.error('close');
});

// 輸出結果為：
// open
// 所有寫入已完成。
// close
複製程式碼

2. drain 事件

如果呼叫 ws.write(chunk) 方法返回 false ，也就是寫入的內容到達快取區的大小，觸發 drain 事件

let fs = require('fs')
let ws = fs.createdWriteStream('writeStream.txt', {
start: 0,
highWaterMark: 5
})

// 寫入15個數，每次寫五個，只希望佔用五個位元組的記憶體
let i = 0
function write() {
let flag = true
while(i < 15 && flag) {
flag = ws.write(i++'','utf8')
}
}
write() // 如果寫入的內容到達快取區的大小了，當他寫入完成後會觸發一個事件
ws.on('drain', () => {
console.log('佔滿')
write() //清空快取 繼續寫入
})

// 佔滿
// 佔滿
複製程式碼

可讀可寫混合使用

通過 pipe 方式讀取一個檔案內容，寫入到另外一個檔案【常用】

// 引入fs模組
const fs = require("fs");

// 建立可讀流和可寫流
let rs = fs.createReadStream("./ReadStream/readStream.txt", {
highWaterMark: 3
});
let ws = fs.createWriteStream("./writeStream.txt", {
highWaterMark: 2
});

// 將 readStream.txt 的內容通過流寫入 writeStream.txt 中
rs.pipe(ws);
複製程式碼

附原始碼

重要：為了方便大家瞭解、檢視、除錯程式碼，完整的原始碼參見 ofollow,noindex">gitHub

總結

本篇文章是 NodeJs 中流（stream）的基礎瞭解。希望對大家瞭解流起到一定的作用。

下篇： NodeJS 流的實現原理和簡單實現—— 《流之原，源之理》