Node.js數據流Stream之Readable流和Writable流

一、前傳

Stream在很多語言都會有,當然Node.js也不例外。數據流是可讀、可寫、或即可讀又可寫的內存結構。Node.js中主要包括Readable、Writable、Duplex(雙工)和Transform(變換)流。但是在學這些之前先學會util模塊中的一個從其他對象繼承的功能.

util模塊提供了util.inherits()方法來允許你創建一個繼承另一個對象的prototype(原形)方法的對象。當創建一個新對象時，prototype方法自動被使用。

util.inherits(constructor,superconstructor)原形constructor被設定為原形superConstructor,並在一個新的對象被創建時執行。可以通過使用constructor.super_屬性從自定義對象的構造函數訪問supercontructor.

二、Readable流

有的前傳util模塊從其他對象繼承的功能的了解，Readable就很好理解了.主要它包含以下方法和事件。

1.事件:

readable:在數據塊可以從流中讀取的時候發出。

data:類似readable,不同之處在於,當數據的事件處理程序被連接時，流被轉變成流動的模式,並且數據處理程序被連續的調用，直到所有數據都被用盡

end:當數據不再被提供時由流發出

close:當底層資源,如文件，已關閉時發出。

error:在接收數據中出錯是發出。

2.方法:

read([size]):從流中讀數據.數據可以是String、Buffer、null(下面代碼會有),當指定size，那么只讀僅限於那個字節數

setEncoding(encoding)：設置read()請求讀取返回String時使用的編碼

pause():暫停從該對象發出的data事件

resume():恢復從該對象發出的data事件

pipe(destination,[options]):把這個流的輸出傳輸到一個由deatination(目的地)指定的Writable流對象。options是一個js對象.例如:{end:true}當Readable結束時就結束Writable目的地。

unpipe([destination])：從Writale目的地斷開這一對象。

3.demo：

var stream = require('stream');
var util = require('util');
util.inherits(Answers, stream.Readable);
function Answers(opt) {
  stream.Readable.call(this, opt);
  this.quotes = ["yes", "no", "maybe"];
  this._index = 0;
}
Answers.prototype._read = function() {
  if (this._index > this.quotes.length){
    this.push(null);
  } else {
    this.push(this.quotes[this._index]);
    this._index += 1;
  }
};
var r = new Answers();
console.log("Direct read: " + r.read().toString());
r.on('data', function(data){
  console.log("Callback read: " + data.toString());
});
r.on('end', function(data){
  console.log("No more answers.");
});
r.on('readable',function(data)
{
   console.log('readable');
});

"C:\Program Files (x86)\JetBrains\WebStorm 11.0.3\bin\runnerw.exe" F:\nodejs\node.exe stream_read.js
Direct read: yes
readable
Callback read: no
Callback read: maybe
readable
readable
readable
No more answers.

Process finished with exit code 0

上面定義了一個通過util.inherits()繼承Readable流的對象,從輸出結果可以看到輸出了3個字符串,但readable事件確執行了4次,其實前面也有寫,read()可以是null,最后是push(null)了。

三、Writable流

有讀就會有寫,畢竟是可逆的,它和readable一樣也有一些事件和方法

1.方法

write(chunk,[encoding],[callback])：將數據寫入流。chunk(數據塊)中包含要寫入的數據,encoding指定字符串的編碼,callback指定當數據已經完全刷新時執行的一個回調函數。如果成功寫入，write()返回true.

end([chunk],[encoding],[callback]):與write()相同，它把Writable對象設為不再接受數據的狀態，並發送finish事件。

2.事件

drain:在write()調用返回false后，當准備好開始寫更多數據時，發出此事件通知監視器。

finish:當end()在Writable對象上調用，所以數據被刷新，並不會有更多的數據被接受時觸發

pipe:當pipe()方法在Readable流上調用,已添加此writable為目的地時發出

unpipe:當unpipe()方法被調用，以刪除Writable為目的地時發出。

3.demo

var stream = require('stream');
var util = require('util');
util.inherits(Writer, stream.Writable);
function Writer(opt) {
  stream.Writable.call(this, opt);
  this.data = new Array();
}
Writer.prototype._write = function(data, encoding, callback) {
  this.data.push(data.toString('utf8'));
  console.log("Adding: " + data);
  callback();
};
var w = new Writer();
for (var i=1; i<=5; i++){  
  w.write("Item" + i, 'utf8');
}
w.end("ItemLast");
console.log(w.data);

"C:\Program Files (x86)\JetBrains\WebStorm 11.0.3\bin\runnerw.exe" F:\nodejs\node.exe stream_write.js
Adding: Item1
Adding: Item2
Adding: Item3
Adding: Item4
Adding: Item5
Adding: ItemLast
[ 'Item1', 'Item2', 'Item3', 'Item4', 'Item5', 'ItemLast' ]

Process finished with exit code 0

四、把Readable流用管道輸送到Writable流

上面也介紹了Readable流pipe()方法,這個主要是來測試

var stream = require('stream');
var util = require('util');
util.inherits(Reader, stream.Readable);
util.inherits(Writer, stream.Writable);
function Reader(opt) {
  stream.Readable.call(this, opt);
  this._index = 1;
}
Reader.prototype._read = function(size) {
  var i = this._index++;
  if (i > 10){
    this.push(null);
  } else {
    this.push("Item " + i.toString());
  }
};
function Writer(opt) {
  stream.Writable.call(this, opt);
  this._index = 1;
}
Writer.prototype._write = function(data, encoding, callback) {
  console.log(data.toString());
  callback();
};

var r = new Reader();
var w = new Writer();
w.on('pipe',function(){
  console.log('pipe');
});
r.pipe(w);

"C:\Program Files (x86)\JetBrains\WebStorm 11.0.3\bin\runnerw.exe" F:\nodejs\node.exe stream_piped.js
pipe
Item 1
Item 2
Item 3
Item 4
Item 5
Item 6
Item 7
Item 8
Item 9
Item 10

Process finished with exit code 0