node中的stream（前篇）

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了。这么做的好处是无需等待所有数据传输完毕就可以开始处理和传输，能够尽快给客户响应。