Stream在node中是一个无处不在的概念,但凡和IO沾边的程序都离不开stream,所以不弄懂stream是无法真正使用node的。在这份简短的笔记中,让我们来一起看看这个融入node血肉中的stream到底是什么,以及该如何正确使用stream吧。
Stream这个概念一点都不新鲜,早在unix早期就被用于命令行中了。在当时,计算机内部带宽非常有限,输入输出都只能按字节码顺序进行传输,这也就是最早的stream的概念。后来stream这个概念被发扬广大,几乎所有IO数据都按stream的形式,通过pipe(管道)进行传输,甚至连进程间通信都用到了stream和pipe的概念。
在node中,stream作为一个各种可读写的对象的抽象层,方便我们用统一的方法对各种对象进行操作。常见的http request response,标准输出流stdout和输入流stdin都可以用stream来进行处理。
node中的stream分为Readable(可读的)、Writable(可写)或是Duplex两者兼有的三大类,让我们依次了解吧。
stream.Readable
Readable stream是对所有可读数据源的抽象。node中的Readable stream有两个模式:flowing和paused。flowing指的是数据流不停歇地流入,一有数据就立刻传输,以便尽快将数据传输给出去;而paused模式则每次只传输一部分数据,读取了一部分数据后再传输这部分数据,按块传输数据直到传输结束。
前面说过,stream是通过pipe(管道)传输的。readable的pipe方法非常直观:
readable.pipe(destination[, options])
destination就是数据传输的目标。options接收一个end参数,该参数为true时,只要接收方明确停止数据传输,数据流就立即停止。
在node学习笔记中我们见过的Readable stream包括:
- 客户端的http response
- 服务器端的http request
- fs模块中的read streams
- tcp中的socket
- process的stdin和stdout方法
下面让我们来看一个fs的例子。fs有一个createReadStream方法,可以按stream的方式读取文件。我们就用这个方法来读取命令行传入的指定路径的文件,然后用pipe(管道)传输给stdout:
var fs = require('fs');
fs.createReadStream(process.argv[2]).pipe(process.stdout);
另外一个简单的例子,stdin产生的流传输给stdout:
process.stdin.pipe(process.stdout);
如果你用node执行这个程序,就会看到每输入一行回车后,屏幕(stdout)就会重复这一行的内容。
Transform Stream
如果只是简单的将数据从一端传到另一端,那么流的也未免太简单了。事实上,stream是非常灵活的一种数据传输手段。现在我们就来看看transform stream是如何实现这种灵活性的。
许多时候,我们希望对数据先做一些处理,再将处理后的结果传输给接收方。这就是的transform stream的作用了,顾名思义,它的输出流是基于输入流变动得来的。它本身属于Deplux类型,可读可写。
我们来看一个简单例子,将前面例子的stdin输入流转换为大写后再传输给输出流stdout。
也许你会想到process.stdin.toUpperCase().pipe.(process.stdout),但这样行不通的。前面讲过,process.stdin的流是一个Readable stream,是不能被改变的。所以我们需要将readable stream转换为Deplux stream,再将内容改成大写,最后输出给stdout。
这个转换流的功能有些复杂,我们这里借助through2模块来做演示。
var through = require('through2');
使用through我们就可以创建一个Deplux流了:
var stream = through(transformFunction, flushFunction)
这里的transformFunction和flushFunction都是callback函数。transformFunction用来对出入流做处理,该函数的签名为:
transform._transform(chunk, encoding, callback)
chunk就是输入流的数据块,encoding是输入流的编码,callback在数据处理完成后调用。每个chunk处理完成后用push方法来写入,并转换为deplux stream进行输出。
flushFunction将在所有输入流传输结束后调用,用来结束transform stream。
有了这个模块的帮助,程序就可以写成:
var through = require('through2');
var stream = through(write, end);
function write(buffer, encoding, next) {
this.push(buffer.toString().toUpperCase());
next();
}
function end(done) {
done();
}
process.stdin.pipe(stream).pipe(process.stdout);
用write函数将输入流逐块转换为大写,并用push转换为deplux stream。end则在输入流结束后发出done来明确结束transform stream。
下面再介绍一个很有用的模块split,可以将输入流按指定的分隔符分开。比如前面的输入的例子,我们可以按换行符来分割输入流,然后再按行输出:
process.stdin.pipe(split()).pipe(process.stdot);
另外如果我们不想按块或者按行处理,而是想在一起处理所有的输入流怎么办呢?我们需要一个方法将输入的数据块块连接起来,有个很方便的模块concat-stream可以完成这个任务:
var concat = require('concat-stream');
process.stdin.pipe(concat(function (buf) {
console.log(buf);
}));
注意这个concat的callback的buf参数不再是一个stream了,不能再使用pipe传输了,所以这里我们console就可以看到完整的输入了。
Stream与http服务
在node学习笔记里,我们介绍过http服务。我们说过,利用node核心的http模块,可以很方便的创建http服务。http.createServer()的callback带有两个参数,一个代表用户请求的request,还有一个代表响应的response。其实request和response都可以处理stream。比如可以将某个流输出到response中:pipe(response)。也可以将用户请求中附带的数据按流输入,比如request.pipe()。
这里我们来看一个例子,将request数据按stream输入,转换为大写后再通过response返回给客户:
var http = require('http');
var through = require('through2');
var server = http.createServer(function (req, res) {
req.pipe(tr).pipe(res);
});
server.listen(process.argv[2]);
var tr = through(function (buf, _, next) {
var buffer = buf.toString().toUpperCase();
this.push(buffer);
next();
});
内容前面都讲过,应该不陌生了。处理stream也很直观,只要按req.pipe(tr).pipe(res)就可以按前面提过的方法处理stream了。这么做的好处是无需等待所有数据传输完毕就可以开始处理和传输,能够尽快给客户响应。