Node Stream 流（二）流的四种基本类型

Z时代
2024-01-10
分类：综合

在Node Stream 流（一）中对流的机制以及使用流的优点做了一个简单的介绍，在这篇文章中我们继续介绍流，对流的四种基本类型——Readable、Writable、Duplex和Transform——进行一个简单的介绍。

在对这几种类型进行介绍之前，我们先来介绍一个非常重要的函数——.pipe()。

pipe

不同类型流之间都可以使用pipe函数来对输入和输出进行匹配。

什么意思呢？pipe我们知道是管道的意思。其实理解起来也非常容易，无非就是在Readable和Writable之间使用.pipe()函数建立一个管道，将Readable流中的数据传输给Writable流。

没错，.pipe()就是一个函数，其调用方式如下：


readable.pipe(writeable);

.pipe()前面必须是Readable或者Transform，括号内是Writeable。也就是说这个管道是单向的。但是，.pipe()函数的返回值也可以看做是一个资源流，也就是说.pipe()函数可以链式调用。


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

它就相当于


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

但是这里需要注意的是，中间的b和c必须是Transform流。


var fs = require('fs');
var stream = fs.createReadStream('./data.txt');
stream.pipe(process.stdout);

上面的一段程序就是创建data.txt文件的读取流，然后将其传输给process.stdout（标准输出）

Readable Stream

Readable Stream其实可以理解为生产者，将生产的数据通过.pipe()函数传输给Writeable 、Transform或者Duplex流，从而进行相应的处理。


Readable.pipe(destination);

首先看一个简单的例子


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

rs.push()是向Readable stream中放入数据。其中rs.push(null)是告诉消费者数据已经生产完了。

但是看上面的例子，在消费者读取数据之前，还是需要将生产的数据缓存到内存中。那是不是有一种更好的方式是当消费者请求数据的时候再去生产数据呢。这个还真可以有，可以通过._read函数来实现


var stream = require('stream');
var c = 97;
function readableStream(num){
    var rs = stream.Readable();
    var count = 0;
    rs._read = function(){
        if(count > num || c > 'z'.charCodeAt(0)){
            return rs.push(null);
        }
        setTimeout(function(){
            rs.push(String.fromCharCode(c++));
        },100)
        count++;
    }
    return rs;
}
rws = readableStream();
rws.pipe(process.stdout);

这个例子就是当有消费者准备好读取数据的时候我们才将a-z放入Readable stream中。默认情况下只能向流中放入string或者buffer类型的数据，如果想让如任意合法类型的数据，可以在Readable()函数中加入以下选项


var rs = stream.Readable({ objectMode: true });

Writable Stream

Writable stream相对于Readable stream就很好理解了，前者只能作为数据的目的地，不能作为数据的来源。相反后者只能作为数据的来源不能作为数据的目的地。


src.pipe(Writable);

在上面案例的基础上我们来扩展Writable stream


var stream = require('stream');
...
function writeableStream(){
    var ws = stream.Writable();
    ws._write = function(chunk,enc,next){
        console.log(chunk.toString());
        next();
    }
    return ws;
}
rws = readableStream();
wws = writeableStream();
rws.pipe(wws);

这里我们是将放入Readable stream中的a-z传输给Writable stream，由其来进行输出。

其中._write 函数我们看到有三个参数，chunk就是由Readable stream写入的数据——默认情况下是经过编码的buffer类型的数据；enc是编码的类型，它是一个字符串，默认情况下是“buffer”；最后next是一个回调函数——next()，该函数的作用就是告诉消费者可以写入更多的数据。

默认情况下Writable stream写入的数据都是经过编码的buffer数据。当然我们可以通过以下代码来改变这个默认设置。


var ws = stream.Writable({ decodeStrings: false });

Duplex Stream

Duplex stream既可以作为Readable stream也可以作为Writable stream。也就是说该类型的数据流是可以双向流动的。


a.pipe(b).pipe(a)

其实理解起来很简单，就相当于我们现实中的电话，我们既可以发出声音也可以接收声音。

Transform Stream

Transform stream可以认为是Duplex stream的一种特定的类型。首先它可以作为Writable stream接收Readable stream中的数据进行特定的处理然后再作为Readable stream将经过特定处理的数据通过.pipe函数传输给Writable stream。

我们继续扩展上面的应用


var stream = require('stream');
...
function transformStream_ToUper(){
    var ts = stream.Transform();
    ts._transform = function(chunk,enc,next){
        chunk = chunk.toString().toUpperCase();
        ts.push(chunk);
        next();
    }
    return ts;
}
rws = readableStream();
wws = writeableStream();
tws = transformStream_ToUper();
rws.pipe(tws).pipe(wws);