淺析Node.js 中 Stream API 的使用

2024-05-06 16:24:52

字體：大中小

供稿：網(wǎng)友

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

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

基本介紹

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

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

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

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

本文是我學(xué)習(xí) Node.js 中的 Stream API 中的一點(diǎn)總結(jié)，希望對(duì)大家有用。

特點(diǎn)

基于事件通訊

可以通過 pipe 來連接流

種類

Readable Stream 可讀數(shù)據(jù)流

Writeable Stream 可寫數(shù)據(jù)流

Duplex Stream 雙向數(shù)據(jù)流，可以同時(shí)讀和寫

Transform Stream 轉(zhuǎn)換數(shù)據(jù)流，可讀可寫，同時(shí)可以轉(zhuǎn)換(處理)數(shù)據(jù)

事件

可讀數(shù)據(jù)流的事件

readable 數(shù)據(jù)向外流時(shí)觸發(fā)

data 對(duì)于那些沒有顯式暫停的數(shù)據(jù)流，添加data事件監(jiān)聽函數(shù)，會(huì)將數(shù)據(jù)流切換到流動(dòng)態(tài)，盡快向外提供數(shù)據(jù)

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

close 數(shù)據(jù)源關(guān)閉時(shí)觸發(fā)

error 讀取數(shù)據(jù)發(fā)生錯(cuò)誤時(shí)觸發(fā)

可寫數(shù)據(jù)流的事件

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

finish 調(diào)用 .end 方法時(shí)，所有緩存的數(shù)據(jù)釋放后觸發(fā)，類似于可讀數(shù)據(jù)流中的 end 事件，表示寫入過程結(jié)束

pipe 作為 pipe 目標(biāo)時(shí)觸發(fā)

unpipe 作為 unpipe 目標(biāo)時(shí)觸發(fā)

error 寫入數(shù)據(jù)發(fā)生錯(cuò)誤時(shí)觸發(fā)

狀態(tài)

可讀數(shù)據(jù)流有兩種狀態(tài)：流動(dòng)態(tài) 和暫停態(tài) ，改變數(shù)據(jù)流狀態(tài)的方法如下：

暫停態(tài) -> 流動(dòng)態(tài)

添加 data 事件的監(jiān)聽函數(shù)

調(diào)用 resume 方法

調(diào)用 pipe 方法

注意：如果轉(zhuǎn)為流動(dòng)態(tài)時(shí)，沒有 data 事件的監(jiān)聽函數(shù)，也沒有 pipe 方法的目的地，那么數(shù)據(jù)將遺失。

流動(dòng)態(tài) -> 暫停態(tài)

不存在 pipe 方法的目的地時(shí)，調(diào)用 pause 方法

存在 pipe 方法的目的地時(shí)，移除所有 data 事件的監(jiān)聽函數(shù)，并且調(diào)用 unpipe 方法，移除所有 pipe 方法的目的地

注意：只移除 data 事件的監(jiān)聽函數(shù)，并不會(huì)自動(dòng)引發(fā)數(shù)據(jù)流進(jìn)入「暫停態(tài)」。另外，存在 pipe 方法的目的地時(shí)，調(diào)用 pause 方法，并不能保證數(shù)據(jù)流總是處于暫停態(tài)，一旦那些目的地發(fā)出數(shù)據(jù)請(qǐng)求，數(shù)據(jù)流有可能會(huì)繼續(xù)提供數(shù)據(jù)。

用法

讀寫文件

var fs = require('fs');

// 新建可讀數(shù)據(jù)流

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

// 新建可寫數(shù)據(jù)流

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

// 監(jiān)聽可讀數(shù)據(jù)流結(jié)束事件

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

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

});

// 監(jiān)聽可寫數(shù)據(jù)流結(jié)束事件

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

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

});

// 把可讀數(shù)據(jù)流轉(zhuǎn)換成流動(dòng)態(tài)，流進(jìn)可寫數(shù)據(jù)流中

rs.pipe(ws);

讀取 CSV 文件，并上傳數(shù)據(jù)（我在生產(chǎn)環(huán)境中寫過）

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)

// 處理前一個(gè) stream 傳遞過來的數(shù)據(jù)

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

upload(data, function(err) {

callback(err);

});

}))

// 相當(dāng)于監(jiān)聽前一個(gè) stream 的 end 事件

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

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

}));

更多用法

可以參考一下https://github.com/jeresig/node-stream-playground，進(jìn)去示例網(wǎng)站之后直接點(diǎn) add stream 就能看到結(jié)果了。

常見坑

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

已結(jié)束/關(guān)閉的流不能重復(fù)使用，必須重新創(chuàng)建數(shù)據(jù)流

pipe 方法返回的是目標(biāo)數(shù)據(jù)流，如 a.pipe(b) 返回的是 b，因此監(jiān)聽事件的時(shí)候請(qǐng)注意你監(jiān)聽的對(duì)象是否正確

如果你要監(jiān)聽多個(gè)數(shù)據(jù)流，同時(shí)你又使用了 pipe 方法來串聯(lián)數(shù)據(jù)流的話，你就要寫成：

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');

});