什麼是流?
說到流,就涉及到一個*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