Node.js数据流Stream之Readable流和Writable流用法

一、前传

Stream在许多编程语言中都存在，Node.js自然也不例外。数据流是一种内存结构，支持读取、写入或同时读写。在Node.js中，Stream主要包含Readable（可读）、Writable（可写）、Duplex（双工）和Transform（变换）这几种类型。但在深入学习这些流之前，需要先了解util模块中一个用于对象继承的功能。

util模块提供了util.inherits()方法，允许创建一个继承自另一个对象prototype（原型）方法的对象。当创建新对象时，其prototype方法会自动被使用。

util.inherits(constructor, superconstructor)将原形constructor设定为原形superConstructor，并在创建新对象时执行。可以通过使用constructor.super_属性从自定义对象的构造函数访问superconstructor。

二、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]):把这个流的输出传输到一个由destination（目的地）指定的Writable流对象。options是一个JS对象。例如：{end:true}，当Readable结束时，Writable目的地也会结束。

unpipe([destination])：从Writable目的地断开这一对象。

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

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。