Node.js 使用Stream模块传送数据

Stream模块是Node.js一个重要模块。数据流是可读、可写或即可读又可写得内存结构。接下来介绍使用Stream模块来创建流，以及从它们中读出数据和向它们写入数据。

1 Readable流

Readable流可以方便地读取从其他来源进入应用程序的数据。

Readable流公开了一些事件：

- readable：在数据块可以从流中读取时发出。
- data：类似于readable，但数据的处理程序被连接时，流被转换成流动的模式，并且数据处理程序被连续地调用，直到所有数据被用尽。
- end：当数据不再被提供时由流发出。
- close：当底层的资源，如文件，关闭时发出。
- error：出错时发出。

示例：

var stream = require('stream');
var util = require('util');
util.inherits(Output, stream.Readable);

function Output() {
stream.Readable.call(this);

this.content = ['开始数据', '中间数据', '结束数据'];
this._index = 0;
}

Output.prototype._read = function () {
if (this._index > this.content.length) {
this.push(null);
} else {
this.push(this.content[this._index]);
this._index ++;
}
}

var out = new Output();
console.log('开始读取：' + out.read().toString('utf8'));

out.on('data', function (data) {
console.log('回调函数读取：' + data.toString());
})

out.on('end', function () {
console.log('已全部读取！');
})

输出：

开始读取：开始数据
回调函数读取：中间数据
回调函数读取：结束数据
已全部读取！

2 Writable流

示例：

var util = require('util');
var stream = require('stream');

util.inherits(Write, stream.Writable);

function Write() {
stream.Writable.call(this);
this.data = new Array();
}

Write.prototype._write = function (data, encoding, callback) {
this.data.push(data.toString('utf8'));
console.log('写入数据：' + data);
callback();
}

var w = new Write();
for (var i = 0; i < 3; i++) {
w.write('数据' + i);
}

w.end('写入最后的数据', 'utf8', function () {
console.log('写入最后数据成功，不再接收数据！');
});

console.log('现在w中的数据是：' + w.data);

输出：

写入数据：数据0
写入数据：数据1
写入数据：数据2
写入数据：写入最后的数据
现在w中的数据是：数据0,数据1,数据2,写入最后的数据
写入最后数据成功，不再接收数据！

3 Duplex流

Duplex流是结合可读可写功能的流。

示例：

var stream = require('stream');
var util = require('util');
util.inherits(Duplexer, stream.Duplex);

function Duplexer(opt) {
stream.Duplex.call(this, opt);
this.data = [];
}
Duplexer.prototype._read = function readItem(size) {
var chunk = this.data.shift();
if (chunk == 'stop') {
this.push(null);
} else if (chunk) {
this.push(chunk);
} else {
setTimeout(readItem.bind(this), 500, size);
}
}

Duplexer.prototype._write = function (data, encoding, callback) {
this.data.push(data);
callback();
}

var d = new Duplexer();
d.on('data', function (chunk) {
console.log('读取：' + chunk.toString('utf8'));
})

d.write('我认为', 'utf8');
d.write('我很好看', 'utf8');
d.write('结束！', 'utf8');
d.write('stop');
console.log(d.data.toString('utf8'));

输出：

我认为,我很好看,结束！,stop
读取：我认为
读取：我很好看
读取：结束！

4 Transform流

Transform流扩展了Duplex流，但它修改Readable和Writable流之间的数据。

示例：

var stream = require('stream');
var util = require('util');

util.inherits(JSONObjTransform, stream.Transform);

function JSONObjTransform() {
stream.Transform.call(this);
}

JSONObjTransform.prototype._transform = function (data, encoding, callback) {
var object = data ? JSON.parse(data.toString('utf8')) : "";
this.emit('object', object);
object.handled = true;
this.push(JSON.stringify(object));
callback();
}

JSONObjTransform.prototype._flush = function (callback) {
callback();
}

var tc = new JSONObjTransform();
tc.on('object', function (object) {
console.log('姓名：%s', object.name);
console.log('性别：%s', object.sex);
})

tc.on('data', function (data) {
console.log('数据：%s', data.toString('utf8'));
})

tc.write('{"name": "李明", "sex": "男"}');
tc.write('{"name": "张三", "sex": "女"}');
tc.write('{"name": "二狗", "sex": "男"}');

输出：

姓名：李明
性别：男
数据：{"name":"李明","sex":"男","handled":true}
姓名：张三
性别：女
数据：{"name":"张三","sex":"女","handled":true}
姓名：二狗
性别：男
数据：{"name":"二狗","sex":"男","handled":true}

5 把Readable流用管道输送到Writable流

实现方法：pipe(writableStream, [options])

示例：

var stream = require('stream');
var util = require('util');

util.inherits(Reader, stream.Readable);
util.inherits(Writer, stream.Writable);

function Reader() {
stream.Readable.call(this);
this._index = 1;
}

Reader.prototype._read = function (size) {
var i = this._index++;
if (i > 10) {
this.push(null);
} else {
this.push('条目：' + i.toString());
}
}

function Writer() {
stream.Writable.call(this);
this._index = 1;
}

Writer.prototype._write = function (data, encoding, callback) {
console.log(data.toString('utf8'));
callback();
}
var r = new Reader();
var w = new Writer();
r.pipe(w);

输出：

条目：1
条目：2
条目：3
条目：4
条目：5
条目：6
条目：7
条目：8
条目：9
条目：10