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

基本介紹

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

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

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

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

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

特點

基于事件通訊

可以通過 pipe 來連接流

種類

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

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

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

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

事件

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

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

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

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

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

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

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

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

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

pipe 作為 pipe 目標時觸發(fā)

unpipe 作為 unpipe 目標時觸發(fā)

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

狀態(tài)

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

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

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

調(diào)用 resume 方法

調(diào)用 pipe 方法

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

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

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

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

注意：只移除 data 事件的監(jiān)聽函數(shù)，并不會自動引發(fā)數(shù)據(jù)流進入「暫停態(tài)」。另外，存在 pipe 方法的目的地時，調(diào)用 pause 方法，并不能保證數(shù)據(jù)流總是處于暫停態(tài)，一旦那些目的地發(fā)出數(shù)據(jù)請求，數(shù)據(jù)流有可能會繼續(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)換成流動態(tài)，流進可寫數(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)
 // 處理前一個 stream 傳遞過來的數(shù)據(jù)
 .pipe(es.map(function(data, callback) {
  upload(data, function(err) {
   callback(err);
  });
 }))
 // 相當于監(jiān)聽前一個 stream 的 end 事件
 .pipe(es.wait(function(err, body) {
  process.stdout.write('done!');
 }));

更多用法

可以參考一下 https://github.com/jeresig/node-stream-playground ，進去示例網(wǎng)站之后直接點 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 方法返回的是目標數(shù)據(jù)流，如 a.pipe(b) 返回的是 b，因此監(jiān)聽事件的時候請注意你監(jiān)聽的對象是否正確

如果你要監(jiān)聽多個數(shù)據(jù)流，同時你又使用了 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');
});

常用類庫

event-stream 用起來有函數(shù)式編程的感覺，個人比較喜歡

awesome-nodejs#streams 由于其他 stream 庫我都沒用過，所以有需求的就直接看這里吧

以上內(nèi)容是小編給大家介紹的Node.js 中 Stream API 的使用，希望大家喜歡。

最新評論