NodejsStream数据流使用手册_node.js

本文介绍了使用 node.js streams 开发程序的基本方法。

"We should have some ways of connecting programs like garden hose--screw in
another segment when it becomes necessary to massage data in
another way. This is the way of IO also."
Doug McIlroy. October 11, 1964

最早接触Stream是从早期的unix开始的数十年的实践证明Stream 思想可以很简单的开发出一些庞大的系统。在unix里，Stream是通过 |实现的；在node中，作为内置的stream模块，很多核心模块和三方模块都使用到。和unix一样，node Stream主要的操作也是.pipe()，使用者可以使用反压力机制来控制读和写的平衡。

Stream 可以为开发者提供可以重复使用统一的接口，通过抽象的Stream接口来控制Stream之间的读写平衡。

2、为什么使用Stream

node中的I/O是异步的，因此对磁盘和网络的读写需要通过回调函数来读取数据，下面是一个文件下载服务器的简单代码：

var http = require('http');
var fs = require('fs');
var server = http.createServer(function (req, res) {
fs.readFile(__dirname + '/data.txt', function (err, data) {
res.end(data);
});
});
server.listen(8000);

这些代码可以实现需要的功能，但是服务在发送文件数据之前需要缓存整个文件数据到内存，如果"data.txt"文件很大且并发量很大的话，会浪费很多内存。因为用户需要等到整个文件缓存到内存才能接受的文件数据，这样导致用户体验相当不好。不过还好(req, res)两个参数都是Stream，这样我们可以用fs.createReadStream()代替fs.readFile():

var http = require('http');
var fs = require('fs');
var server = http.createServer(function (req, res) {
var stream = fs.createReadStream(__dirname + '/data.txt');
stream.pipe(res);
});
server.listen(8000);

.pipe()方法监听fs.createReadStream()的'data' 和'end'事件，这样"data.txt"文件就不需要缓存整个文件，当客户端连接完成之后马上可以发送一个数据块到客户端。使用.pipe()另一个好处是可以解决当客户端延迟非常大时导致的读写不平衡问题。如果想压缩文件再发送，可以使用三方模块实现：

var http = require('http');
var fs = require('fs');
var oppressor = require('oppressor');
var server = http.createServer(function (req, res) {
var stream = fs.createReadStream(__dirname + '/data.txt');
stream.pipe(oppressor(req)).pipe(res);
});
server.listen(8000);

这样文件就会对支持gzip和deflate的浏览器进行压缩。oppressor 模块会处理所有的content-encoding。

Stream使开发程序变得简单。

3、基础概念

有五种基本的Stream: readable, writable, transform, duplex, and”classic”.

3-1、pipe

所有类型的Stream收是使用 .pipe() 来创建一个输入输出对，接收一个可读流src并将其数据输出到可写流dst，如下：

src.pipe(dst)

.pipe( dst )方法为返回dst流，这样就可以接连使用多个.pipe()，如下：

a.pipe( b ).pipe( c ).pipe( d )

功能与下面的代码相同：

a.pipe( b );
b.pipe( c );
c.pipe( d );

3-2、readable streams

通过调用Readable streams的 .pipe()方法可以把Readable streams的数据写入一个Writable , Transform, 或者Duplex stream。

readableStream.pipe( dst )

1>创建 readable stream

这里我们创建一个readable stream!

var Readable = require('stream').Readable;
var rs = new Readable;
rs.push('beep ');
rs.push('boop\n');
rs.push(null);
rs.pipe(process.stdout);
$ node read0.js
beep boop

rs.push( null ) 通知数据接收者数据已经发送完毕.

注意到我们在将所有数据内容压入可读流之前并没有调用rs.pipe(process.stdout);，但是我们压入的所有数据内容还是完全的输出了，这是因为可读流在接收者没有读取数据之前，会缓存所有压入的数据。但是在很多情况下, 更好的方法是只有数据接收着请求数据的时候，才压入数据到可读流而不是缓存整个数据。下面我们重写 一下._read()函数：

var Readable = require('stream').Readable;
var rs = Readable();
var c = 97;
rs._read = function () {
rs.push(String.fromCharCode(c++));
if (c > 'z'.charCodeAt(0)) rs.push(null);
};
rs.pipe(process.stdout);
$ node read1.js
abcdefghijklmnopqrstuvwxyz

上面的代码通过重写_read()方法实现了只有在数据接受者请求数据才向可读流中压入数据。_read()方法也可以接收一个size参数表示数据请求着请求的数据大小，但是可读流可以根据需要忽略这个参数。

注意我们也可以用util.inherits()继承可读流。为了说明只有在数据接受者请求数据时_read()方法才被调用，我们在向可读流压入数据时做一个延时，如下：

var Readable = require('stream').Readable;
var rs = Readable();
var c = 97 - 1;
rs._read = function () {
if (c >= 'z'.charCodeAt(0)) return rs.push(null);
setTimeout(function () {
rs.push(String.fromCharCode(++c));
}, 100);
};
rs.pipe(process.stdout);
process.on('exit', function () {
console.error('\n_read() called ' + (c - 97) + ' times');
});
process.stdout.on('error', process.exit);

用下面的命令运行程序我们发现_read()方法只调用了5次：

$ node read2.js | head -c5
abcde
_read() called 5 times

使用计时器的原因是系统需要时间来发送信号来通知程序关闭管道。使用process.stdout.on('error', fn) 是为了处理系统因为header命令关闭管道而发送SIGPIPE信号，因为这样会导致process.stdout触发EPIPE事件。如果想创建一个的可以压入任意形式数据的可读流，只要在创建流的时候设置参数objectMode为true即可，例如：Readable({ objectMode: true })。

2>读取readable stream数据

大部分情况下我们只要简单的使用pipe方法将可读流的数据重定向到另外形式的流，但是在某些情况下也许直接从可读流中读取数据更有用。如下：

process.stdin.on('readable', function () {
var buf = process.stdin.read();
console.dir(buf);
});
$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume0.js 



null

当可读流中有数据可读取时，流会触发'readable' 事件，这样就可以调用.read()方法来读取相关数据，当可读流中没有数据可读取时，.read() 会返回null，这样就可以结束.read() 的调用， 等待下一次'readable' 事件的触发。下面是一个使用.read(n)从标准输入每次读取3个字节的例子：

process.stdin.on('readable', function () {
var buf = process.stdin.read(3);
console.dir(buf);
});

如下运行程序发现，输出结果并不完全!

$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume1.js 

这是应为额外的数据数据留在流的内部缓冲区里了，而我们需要通知流我们要读取更多的数据.read(0)可以达到这个目的。

process.stdin.on('readable', function () {
var buf = process.stdin.read(3);
console.dir(buf);
process.stdin.read(0);
});

这次运行结果如下：

$ (echo abc; sleep 1; echo def; sleep 1; echo ghi) | node consume2.js 

我们可以使用 .unshift() 将数据重新押回流数据队列的头部，这样可以接续读取押回的数据。如下面的代码，会按行输出标准输入的内容：

var offset = 0;
process.stdin.on('readable', function () {
var buf = process.stdin.read();
if (!buf) return;
for (; offset 

src.pipe( writableStream );

var Writable = require('stream').Writable;
var ws = Writable();
ws._write = function (chunk, enc, next) {
console.dir(chunk);
next();
};
process.stdin.pipe(ws);
$ (echo beep; sleep 1; echo boop) | node write0.js 

Writable({ objectMode: true })

process.stdout.write('beep boop\n');

var fs = require('fs');
var ws = fs.createWriteStream('message.txt');
ws.write('beep ');
setTimeout(function () {
ws.end('boop\n');
}, 1000);
$ node writing1.js 
$ cat message.txt
beep boop

var Stream = require('stream');
var stream = new Stream;
stream.readable = true;
var c = 64;
var iv = setInterval(function () {
if (++c >= 75) {
clearInterval(iv);
stream.emit('end');
}
else stream.emit('data', String.fromCharCode(c));
}, 100);
stream.pipe(process.stdout);
$ node classic0.js
ABCDEFGHIJ

process.stdin.on('data', function (buf) {
console.log(buf);
});
process.stdin.on('end', function () {
console.log('__END__');
});
$ (echo beep; sleep 1; echo boop) | node classic1.js 


__END__

var through = require('through');
process.stdin.pipe(through(write, end));
function write (buf) {
console.log(buf);
}
function end () {
console.log('__END__');
}
$ (echo beep; sleep 1; echo boop) | node through.js 


__END__

var cOncat= require('concat-stream');
process.stdin.pipe(concat(function (body) {
console.log(JSON.parse(body));
}));
$ echo '{"beep":"boop"}' | node concat.js 
{ beep: 'boop' }

a.pipe(b).pipe(a)