淺析Node.js 中 Stream API 的使用

2024-05-06 16:24:52

字體：大中小

來源：轉載

供稿：網友

這篇文章給大家淺析node.js中stream api的使用，本文介紹的非常詳細，涉及到node.js api，node.js stream相關知識，感興趣的朋友可以參考下

本文由淺入深給大家介紹node.js stream api，具體詳情請看下文吧。

基本介紹

在 Node.js 中，讀取文件的方式有兩種，一種是用 fs.readFile ，另外一種是利用 fs.createReadStream 來讀取。

fs.readFile 對于每個 Node.js 使用者來說最熟悉不過了，簡單易懂，很好上手。但它的缺點是會先將數據全部讀入內存，一旦遇到大文件的時候，這種方式讀取的效率就非常低下了。

而 fs.createReadStream 則是通過 Stream 來讀取數據，它會把文件(數據)分割成小塊，然后觸發一些特定的事件，我們可以監聽這些事件，編寫特定的處理函數。這種方式相對上面來說，并不好上手，但它效率非常高。

事實上， Stream 在 Node.js 中并非僅僅用在文件處理上，其他地方也可以看到它的身影，如 process.stdin/stdout , http , tcp sockets , zlib , crypto 等都有用到。

本文是我學習 Node.js 中的 Stream API 中的一點總結，希望對大家有用。

特點

基于事件通訊

可以通過 pipe 來連接流

種類

Readable Stream 可讀數據流

Writeable Stream 可寫數據流

Duplex Stream 雙向數據流，可以同時讀和寫

Transform Stream 轉換數據流，可讀可寫，同時可以轉換(處理)數據

事件

可讀數據流的事件

readable 數據向外流時觸發

data 對于那些沒有顯式暫停的數據流，添加data事件監聽函數，會將數據流切換到流動態，盡快向外提供數據

end 讀取完數據時觸發。注意不能和 writeableStream.end() 混淆，writeableStream 并沒有 end 事件，只有 .end() 方法

close 數據源關閉時觸發

error 讀取數據發生錯誤時觸發

可寫數據流的事件

drain writable.write(chunk) 返回 false 之后，緩存全部寫入完成，可以重新寫入時就會觸發

finish 調用 .end 方法時，所有緩存的數據釋放后觸發，類似于可讀數據流中的 end 事件，表示寫入過程結束

pipe 作為 pipe 目標時觸發

unpipe 作為 unpipe 目標時觸發

error 寫入數據發生錯誤時觸發

狀態

可讀數據流有兩種狀態：流動態和暫停態，改變數據流狀態的方法如下：

暫停態 -> 流動態

添加 data 事件的監聽函數

調用 resume 方法

調用 pipe 方法

注意：如果轉為流動態時，沒有 data 事件的監聽函數，也沒有 pipe 方法的目的地，那么數據將遺失。

流動態 -> 暫停態

不存在 pipe 方法的目的地時，調用 pause 方法

存在 pipe 方法的目的地時，移除所有 data 事件的監聽函數，并且調用 unpipe 方法，移除所有 pipe 方法的目的地

注意：只移除 data 事件的監聽函數，并不會自動引發數據流進入「暫停態」。另外，存在 pipe 方法的目的地時，調用 pause 方法，并不能保證數據流總是處于暫停態，一旦那些目的地發出數據請求，數據流有可能會繼續提供數據。

用法

讀寫文件

var fs = require('fs');

// 新建可讀數據流

var rs = fs.createReadStream('./test1.txt');

// 新建可寫數據流

var ws = fs.createWriteStream('./test2.txt');

// 監聽可讀數據流結束事件

rs.on('end', function() {

console.log('read text1.txt successfully!');

});

// 監聽可寫數據流結束事件

ws.on('finish', function() {

console.log('write text2.txt successfully!');

});

// 把可讀數據流轉換成流動態，流進可寫數據流中

rs.pipe(ws);

讀取 CSV 文件，并上傳數據（我在生產環境中寫過）

var fs = require('fs');

var es = require('event-stream');

var csv = require('csv');

var parser = csv.parse();

var transformer = csv.transform(function(record) {

return record.join(',');

});

var data = fs.createReadStream('./demo.csv');

data

.pipe(parser)

.pipe(transformer)

// 處理前一個 stream 傳遞過來的數據

.pipe(es.map(function(data, callback) {

upload(data, function(err) {

callback(err);

});

}))

// 相當于監聽前一個 stream 的 end 事件

.pipe(es.wait(function(err, body) {

process.stdout.write('done!');

}));

更多用法

可以參考一下https://github.com/jeresig/node-stream-playground，進去示例網站之后直接點 add stream 就能看到結果了。

常見坑

用 rs.pipe(ws) 的方式來寫文件并不是把 rs 的內容 append 到 ws 后面，而是直接用 rs 的內容覆蓋 ws 原有的內容

已結束/關閉的流不能重復使用，必須重新創建數據流

pipe 方法返回的是目標數據流，如 a.pipe(b) 返回的是 b，因此監聽事件的時候請注意你監聽的對象是否正確

如果你要監聽多個數據流，同時你又使用了 pipe 方法來串聯數據流的話，你就要寫成：

data

.on('end', function() {

console.log('data end');

})

.pipe(a)

.on('end', function() {

console.log('a end');

})

.pipe(b)

.on('end', function() {

console.log('b end');

});