Node.js中的流（Stream）介紹

 這篇文章主要介紹了Node.js中的流（Stream）介紹,本文講解了什麼是流、pipe方法、流的分類、Readable流狀態的切換等內容,需要的朋友可以參考下

   

什麼是流？

說到流，就涉及到一個*nix的概念：管道——在*nix中，流在Shell中被實現為可以通過 |(管道符) 進行橋接的數據，一個進程的輸出（stdout）可被直接作為下一個進程的輸入（stdin）。

在Node中，流（Stream）的概念與之類似，代表一種數據流可供橋接的能力。

pipe

流化的精髓在於 .pipe()方法。可供橋接的能力，在於數據流的兩端（上游/下游 或稱為 讀/寫流）以一個 .pipe(）方法進行橋接。

偽代碼的表現形式為：

代碼如下:
//上游.pipe（下游)
Readable.pipe(Writable);

 

流的分類

這裡並不打算討論所謂的Node v0.4 之前的“經典”流。那麼，流分為這麼幾類（皆為抽象接口：

1.stream.Readable 可讀流（需要實現_read方法，關注點在於對數據流讀取的細節
2.stream.Writable 可寫流（需要實現_write方法，關注點在於對數據流寫入的細節
3.stream.Duplex 可讀/寫流（需要實現以上兩接口，關注點為以上兩接口的細節
4.stream.Transform 繼承自Duplex（需要實現_transform方法，關注點在於對數據塊的處理

簡單來說：

1）.pipe() 的擁有者一定具備 Readable 流（並不局限於）能力，它擁有 'readable'/'data'/'end'/'close'/'error' 一系列事件可供訂閱，也提供 .read()/.pause()/.resume()等一系列方法供調用；
2）.pipe() 的參數一定具備Writable 流（並不局限於 ）能力，它擁有 'drain'/'pipe'/'unpipe'/'error'/'finish' 事件可供訪問，也提供 .write()/.end() 等一系列方法供調用

什麼鬼

有沒有一絲絲焦慮？別急，做為一個說人話的低級碼工，我會把Stream掰開了和您扯一扯的。

Stream類，在 Node.js的源碼 裡，是這麼定義的：

代碼如下:
var EE = require('events').EventEmitter;
var util = require('util');
util.inherits(Stream, EE);

function Stream() {
EE.call(this);
}

 

可以看出，本質上，Stream是一個EventEmitter，那意味著它具備事件驅動的功能（.emit/.on...）。眾所周知，“Node.js 就是基於V8的事件驅動平台”，實現了事件驅動的流式編程，具備了和Node一樣的異步回調的特征。

比如在 Readable 流中，有一個 readable 事件，在一個暫停的只讀流中，只要有數據塊准備好可讀時，它就會被發送給訂閱者（Readable 流有哪些呢？express中的 req，ftp或者mutli-form上傳組件的req.part，系統中的標准輸入 process.stdin等）。有了readable 事件，我們可以做個處理shell 命令輸出的分析器之類的工具：

代碼如下:
process.stdin.on('readable', function(){
var buf = process.stdin.read();
if(buf){
var data = buf.toString();
// parsing data ...
}
});

 

這樣調用：

 

代碼如下:
head -10 some.txt | node parser.js

 

對於 Readable 流，我們還可以訂閱它的 data 和 end 事件，以獲取數據塊並在流枯竭時獲得通知，如 經典socket示例 中那樣：

 代碼如下:
req.on('connect', function(res, socket, head) {
socket.on('data', function(chunk) {
console.log(chunk.toString());
});
socket.on('end', function() {
proxy.close();
});
});

 

Readable流狀態的切換
需要注意的是，Readable 流有兩種狀態：flowing mode（激流） 和 pause mode（暫停）。前者根本停不下來，誰被pipe上了就馬上不停的給；後者會暫停，直到下游顯式的調用 Stream.read() 請求才讀取數據塊。Readable 流初始化時是 pause mode的。

這兩種狀態可以互為切換的，其中，

有以下任一行為，pause 轉 flowing：

1.對 Readable 流添加一個data事件訂閱
2.對 Readable 調用 .resume() 顯式開啟flowing
3.調用 Readable 流的 .pipe(writable) ，橋接到一個 Writable 流上

有以下任一行為，flowing 轉回 pause：

1.Readable 流還沒有 pipe 到任何流上，可調 .pause() 暫停
2.Readable 流已經 pipe 到了流上，需 remove 掉所有 data 事件訂閱，並且調用 .unpipe()方法逐一解除與下游流的關系

妙用

結合流的異步特性，我可以寫出這樣的應用：直接將 用戶A 的輸出橋接到 用戶B 的頁面上輸出：

代碼如下:
router.post('/post', function(req, res) {
var destination = req.headers['destination']; //發給誰
cache[destionation] = req;
//是的，並不返回，所以最好是個ajax請求
});

 

用戶B請求的時候：

 

代碼如下:
router.get('/inbox', function(req, res){
var user = req.headers['user'];
cache.find(user, function(err, previousReq){ //找到之前存的req
var form = new multiparty.Form();
form.parse(previousReq); // 有文件給我
form.on('part', function (part) {
part.pipe(res); //流式大法好:)

part.on('error', function (err) {
console.log(err);
messaging.setRequestDone(uniqueID);
return res.end(err);
});
});
});
});

 

參考

how to write node programs with streams: stream-handbook