爱你不容易 —— Stream详解

2023-08-19 23:04:28

作为前端，我们常常会和 Stream 有着频繁的接触。比如使用 gulp 对项目进行构建的时候，我们会使用 gulp.src 接口将匹配到的文件转为 stream（流）的形式，再通过 .pipe() 接口对其进行链式加工处理；

或者比如我们通过 http 模块创建一个 HTTP 服务：

const http = require('http');

http.createServer( (req, res) => {

  //...

}).listen(3000);

此处的 req 和 res 也属于 Stream 的消费接口（前者为 Readable Stream，后者为 Writable Stream）。

事实上像上述的 req/res，或者 process.stdout 等接口都属于 Stream 的实例，因此较少存在情况，是需要我们手动引入 Stream 模块的，例如：

//demo1.js

'use strict';

const Readable = require('stream').Readable;

const rs = Readable();

const s = 'VaJoy';

const l = s.length;

let i = 0;

rs._read = ()=>{

    if(i == l){

        rs.push(' is my name');

        return rs.push(null)

    }

    rs.push(s[i++])

};

rs.pipe(process.stdout);

如果不太能读懂上述代码，或者对 Stream 的概念感到模糊，那么可以放轻松，因为本文会进一步地对 Stream 进行剖析，并且谈谈直接使用它可能会存在的一些问题（这也是为何 gulp 要使用 through2 的原因）。

另外本文的示例均可在我的 github 仓库（https://github.com/VaJoy/stream/）获取到，读者可以自行下载和调试。

一. Stream的作用

在介绍 Stream（流）之前，我们先来看一个例子 —— 模拟服务器把本地某个文件内容吐给客户端：

//demo2

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(3000);

这段代码虽然可以正常执行，但存在一个显著的问题 —— 对于每一个客户端的请求，fs.readFile 接口都会把整个文件都缓存到内存中去，然后才开始把数据吐给用户。那么当文件体积很大、请求也较多（且特别当请求来自慢速用户）的时候，服务器需要消耗很大的内存，导致性能低下。

然而这个问题，则正是 stream 发挥所长的地方。如前文提及的，res 是流对象，那我们正好可以将其利用起来：

var server2 = http.createServer(function (req, res) {

    var stream = fs.createReadStream(__dirname + '/data.txt');

    stream.pipe(res);

});

server2.listen(4000);

在上方代码段里，fs.createReadStream 创建了 data.txt 的可读流（Readable Stream）。这里需要事先了解的是，流可以简单地分为“可读的（readable）”、“可写的（writable）”，或者“读写均可”三种类型，且所有的流都属于 EventEmitter 的实例。

回到代码，对于创建的可读流，我们通过 .pipe() 接口来监听其 data 和 end 事件，并把 data.txt （的可读流）拆分成一小块一小块的数据（chunks），像流水一样源源不断地吐给客户端，而不再需要等待整个文件都加载到内存后才发送数据。

其中 .pipe 可以视为流的“管道/通道”方法，任何类型的流都会有这个 .pipe 方法去成对处理流的输入与输出。

为了方便理解，我们把上述两种方式（不使用流/使用流）处理为如下的情景（卧槽我好好一个前端为啥要P这么萌的图）：

⑴ 不使用流：

⑵ 使用流：

由此可以得知，使用流（stream）的形式，可以大大提升响应时间，又能有效减轻服务器内存的压力。

二. Stream的分类

在上文我们曾提及到，stream 可以按读写权限来简单地分做三类，不过这里我们再细化下，可以把 stream 归为如下五个类别：

⑴ Readable Streams
⑵ Writable Streams
⑶ Transform Streams
⑷ Duplex Streams
⑸ Classic Streams

其中 Transform Streams 和 Duplex Streams 都属于即可读又可写的流，而最后一个 Classic Streams 是对 Node 古早版本上的 Stream 的一个统称。我们将照例对其进行逐一介绍。

2.1 Readable Streams

即可读流，通过 .pipe 接口可以将其数据传递给一个 writable、transform 或者 duplex流：

readableStream.pipe(dst)

常见的 Readable Streams 包括：

客户端上的 HTTP responses
服务端上的 HTTP requests
fs read streams
zlib streams
crypto streams
TCP sockets
子进程的 stdout 和 stderr
process.stdin

例如在前面 demo2 的代码段中，我们就使用了 fs.createReadStream 接口来创建了一个 fs read stream：

var server2 = http.createServer(function (req, res) {

    var stream = fs.createReadStream(__dirname + '/data.txt');

    stream.pipe(res);

});

server2.listen(4000);

这里有个有趣的地方 —— 虽然 Readable Streams 称为可读流，但在将其传入一个消耗对象之前，它都是可写的：

var Readable = require('stream').Readable;

var rs = new Readable;

rs.push('servers ');

rs.push('are listening on\n');

rs.push('3000 and 4000\n');

rs.push(null);

rs.pipe(process.stdout);

执行结果：

在这段代码中，我们通过 readStream.push(data) 的形式往可读流里注入数据，并以 readStream.push(null) 来结束可读流。

不过这种写法有个弊端 —— 从使用 .push() 将数据注入 readable 流中开始，直到另一个东西（process.stdout）来消耗数据之前，这些数据都会存在缓存中。

这里有个内置接口 ._read() 可以用来处理这个问题，它是从系统底层开始读取数据流时才会不断调用自身，从而减少缓存冗余。

我们可以回过头来看 demo1 的例子：

'use strict';

const Readable = require('stream').Readable;

const rs = Readable();

const s = 'VaJoy';

const l = s.length;

let i = 0;

rs._read = ()=>{

    if(i == l){

        rs.push(' is my name');

        return rs.push(null)

    }

    rs.push(s[i++])

};

rs.pipe(process.stdout);

我们是在 ._read 方法中才使用 readStream.push(data) 往可读流里注入数据供下游消耗（也会流经缓存），从而提升流处理的性能。

这里也有个小问题 —— 上一句话所提到的“供下游消耗”，这个下游通常又会以怎样的形式来消耗可读流的呢？

首先，可以使用我们熟悉的 .pipe() 方法将可读流推送给一个消耗对象（writable、transform 或者 duplex流）：

//ext1

const fs = require('fs');

const zlib = require('zlib');

const r = fs.createReadStream('data.txt');

const z = zlib.createGzip();

const w = fs.createWriteStream('data.txt.gz');

r.pipe(z).pipe(w);

其次，也可以通过监听可读流的“data”事件（别忘了文章前面提到的“所有的流都属于 EventEmitter 的实例”）来实现消耗处理 —— 在首次监听其 data 事件后，readStream 便会持续不断地调用 _read()，通过触发 data 事件将数据输出。当数据全部被消耗时，则触发 end 事件。

示例：

//demo3

const Readable = require('stream').Readable;

class ToReadable extends Readable {

    constructor(iterator) {

        super();

        this.iterator = iterator

    }

    _read() {

        const res = this.iterator.next();

        if (res.done) {

            // 迭代结束，顺便结束可读流

            this.push(null)

        }

        setTimeout(() => {

            // 将数据添加到流中

            this.push(res.value + '\n')

        }, 0)

    }

}

const gen = function *(a){

    let count = 5,

        res = a;

    while(count--){

        res = res*res;

        yield res

    }

};

const readable = new ToReadable(gen(2));

// 监听`data`事件，一次获取一个数据

readable.on('data', data => process.stdout.write(data));

// 可读流消耗完毕

readable.on('end', () => process.stdout.write('readable stream ends~'));

执行结果为：

这里需要留意的是，在使用 .push() 往可读流里注入数据的代码段，我们使用了 setTimeout 将其包裹起来，这是为了让系统能有足够时间优先处理接收流结束信号的事务。当然你也可以改写为：

        if (res.done) {

            // 直接 return

            return this.push(null)

        }

        this.push(res.value + '\n')

2.2 Writable Streams

Writable（可写）流接口是对写入数据的目标的抽象：

src.pipe(writableStream)

常见的 Writable Streams 包括：

客户端的 HTTP requests
服务端的 HTTP responses
fs write streams
zlib streams
crypto streams
TCP sockets
子进程的 stdin
process.stdout 和 process.stderr

可写流有两个重要的方法：

writableStream.write(chunk[, encoding, callback]) —— 往可写流里写入数据；
writableStream.end([chunk, encoding, callback]) —— 停止写入数据，结束可写流。在调用 .end() 后，再调用 .write() 方法会产生错误。

上方两方法的 encoding 参数表示编码字符串（chunk为String时才可以用）。

write 方法的 callback 回调参数会在 chunk 被消费后（从缓存中移除后）被触发；end 方法的 callback 回调参数则在 Stream 结束时触发。

另外，如同通过 readable._read() 方法可以处理可读流，我们可以通过 writable._write(chunk, enc, next) 方法在系统底层处理流写入的逻辑中，对数据进行处理。

其中参数 chunk 代表写进来的数据；enc 代表编码的字符串；next(err) 则是一个回调函数，调用它可以告知消费者进行下一轮的数据流写入。

示例：

//demo4

const Writable = require('stream').Writable;

const writable = Writable();

writable._write = (chunck, enc, next) => {

    // 输出打印

    process.stdout.write(chunck.toString().toUpperCase());

    // 写入完成时，调用`next()`方法通知流传入下一个数据

    process.nextTick(next)

};

// 所有数据均已写入底层

writable.on('finish', () => process.stdout.write('DONE'));

// 将一个数据写入流中

writable.write('a' + '\n');

writable.write('b' + '\n');

writable.write('c' + '\n');

// 再无数据写入流时，需要调用`end`方法

writable.end();

执行如下：

2.3 Duplex Streams

Duplex 是双工的意思，因此很容易猜到 Duplex 流就是既能读又能写的一类流，它继承了 Readable 和 Writable 的接口。

常见的 Duplex Streams 有：

TCP sockets
zlib streams
crypto streams

示例：

//demo5

const Duplex = require('stream').Duplex;

const duplex = Duplex();

duplex._read = function () {

    var date = new Date();

    this.push( date.getFullYear().toString() );

    this.push(null)

};

duplex._write = function (buf, enc, next) {

    console.log( buf.toString() + '\n' );

    next()

};

duplex.on('data', data => console.log( data.toString() ));

duplex.write('the year is');

duplex.end();

执行结果：

2.4 Transform Streams

Transform Stream 是在继承了 Duplex Streams 的基础上再进行了扩展，它可以把写入的数据和输出的数据，通过 ._transform 接口关联起来。

常见的 Transform Streams 有：

zlib streams
crypto streams

示例：

//demo6

const Transform = require('stream').Transform;

class SetName extends Transform {

    constructor(name, option) {

        super(option || {});

        this.name = name || ''

    }

    // .write接口写入的数据，处理后直接从 data 事件的回调中可取得

    _transform(buf, enc, next) {

        var res = buf.toString().toUpperCase();

        this.push(res + this.name + '\n');

        next()

    }

}

var transform = new SetName('VaJoy');

transform.on('data', data => process.stdout.write(data));

transform.write('my name is ');

transform.write('here is ');

transform.end();

执行结果：

其中的 _transform 是 Transform Streams 的内置方法，所有 Transform Streams 都需要使用该接口来接收输入和处理输出，且该方法只能由子类来调用。

_transform 接口格式如下：

transform._transform(chunk, encoding, callback)

第一个参数表示被转换（transformed）的数据块（chunk），除非构造方法 option 参数（可选）传入了 “decodeString : false”，否则其类型均为 Buffer；

第二个参数用于设置编码，但只有当 chunck 为 String 格式（即构造方法传入 “decodeString : false”参数）的时候才可配置，否则默认为“buffer”；

第三个参数 callback 用于在 chunk 被处理后调用，通知系统进入下一轮 _transform 调用。该回调方法接收两个可选参数 —— callback([error, data])，其中的 data 参数可以将 chunck 写入缓存中（供更后面的消费者去消费）：

transform.prototype._transform = function(data, encoding, callback){

    this.push(data);

    callback()

};

///////等价于

transform.prototype._transform = function(data, encoding, callback){

    callback(null, data)

};

另外 Transform Streams 还有一个 _flush(callback) 内置方法，它会在没有更多可消耗的数据时、在“end”事件之前被触发，而且会清空缓存数据并结束 Stream。

该内置方法同样只允许由子类来调用，而且执行后，不能再调用 .push 方法。

关于 Transform Streams 的更多细节还可以参考这篇文章，推荐阅读。

2.5 Classic Streams

在较早版本的 NodeJS 里，Stream 的实现相较简陋，例如上文提及的“Stream.Readable”接口均是从 Node 0.9.4 开始才有，因此我们往往需要对其进行多次封装扩展才能更好地用来开发。

而 Classic Streams 便是对这种古旧模式的 Stream 接口的统称。

需要留意的是，只要往任意一个 stream 注册一个“data”事件监听器，它就会自动切换到“classic”模式，并按照旧的 API 去执行。

classic 流可以当作一个带有 .pipe 接口的事件发射器（event emitter），当它要为消耗者提供数据时会发射“data”事件，当要结束生产数据时，则发射“end”事件。

另外只有当设置 Stream.readable 为 true 时，.pipe 接口才会将当前流视作可读流：

//demo7

var Stream = require('stream');

var stream = new Stream();

stream.readable = true; //告诉 .pipe 这是个可读流

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);

另外，Classic readable streams 还有 .pause() 和 .resume() 两个接口可用于暂停/恢复流的读取：

createServer(function(q,s) {

  // ADVISORY only!

  q.pause()

  session(q, function(ses) {

    q.on('data', handler)

    q.resume()

  })

})

3. Object Mode

对于可读流来说，push(data) 时，data 的类型只能是 String 或Buffer，且消耗时 data 事件输出的数据类型都为 Buffer；

对于可写流来说，write(data) 时，data 的类型也只能是 String 或 Buffer，_write(data) 调用时所传进来的 data 类型都为 Buffer。

示例：

//demo8

writable._write = (chunck, enc, next) => {

    // 输出打印

    console.log(chunck);   //Buffer

    //console.log(chunck.toString());  //转为String

    process.nextTick(next)

};

writable.write('Happy Chinese Year');

writable.end();

执行结果：

不过，为了增强数据类型的灵活性，无论是可读流或是可写流，只需要往其构造函数里传入配置参数“{ objectMode: true }”，便可往流里传入/获取任意类型（null除外）的数据：

const objectModeWritable = Writable({ objectMode: true });

objectModeWritable._write = (chunck, enc, next) => {

    // 输出打印

    console.log(typeof chunck);

    console.log(chunck);

    process.nextTick(next)

};

objectModeWritable.write('Happy Chinese Year');

objectModeWritable.write( { year : 2017 } );

objectModeWritable.end( 2017 );

执行结果：

4. Stream的兼容问题

在前文我们介绍了 classic streams，它属于陈旧版本的 Node 上的 Stream 接口，可以把它称为 Streams1。而从 Node 0.10 开始，Stream 新增了系列实用的新接口，可以做更多除了 .pipe() 之外的事情，我们把其归类为 Streams2（事实上，在 Node 0.11+开始，Stream有些许新的变动，从该版本开始的 Stream 也可称为 Streams3）。

那么这里存在一个问题 —— 那些使用了 Stream1 的项目（特别是 npm 包），想升级使用环境的 Node 版本到 0.10+，会否导致兼容问题呢？

还好 Streams2 虽然改头换面，但本质上是设计为向后兼容的。

打个比方，如果你同时推送了一条 Streams2 流和一条旧格式的、基于事件发射器的流，Stream2 将降级为旧模式（shim mode）来向后兼容。

但是，如果我们的开发环境使用的是 Node 0.8（且因为某些原因不能升级），但又想使用 Streams2 的API怎么办呢？或者比如 npm 上的某些开源的工具包，想要拥抱 Streams2 的便利，又想保持对使用 Node 0.8 的用户进行兼容处理，这样又得怎么处理？

针对上述问题，早在 Node 0.10 释放之前，Issacs 就把 Node-core 中操作 Stream 的核心接口独立拷贝了一份出来，开源到了 npm 上并持续更新，它就是 readable-stream。

通过使用 readable-stream，我们就可以在那些核心里没有 Streams2/3 的低版本 Node 中，直接使用 Streams2/3：

var Readable = require('stream').Readable || require('readable-stream').Readable

readable-stream 现在有 v1.0.x 和 v1.1.x 两个主要版本，前者跟进 Streams2 的迭代，后者跟进 Streams3 的迭代，用户可以根据需求使用对应版本的包。

5. through2

readable-stream 虽然提供了一个 Streams 的兼容方案，但我们也希望能对 Stream 复杂的API进行精简。

而 through2 便基于 readable-stream 对 Stream 接口进行了封装，并提供了更简单和灵活的方法。

through2 会为你生成 Transform Streams（貌似旧版本是 Duplex Streams）来处理任意你想使用的流 —— 如前文介绍，相比其它流，Transform 流处理起数据会更加灵活方便。

来看下 through2 的示例：

//demo9

const fs = require('fs');

const through2 = require('through2');

fs.createReadStream('data.txt')

    .pipe(through2(function (chunk, enc, callback) {

        for (var i = 0; i < chunk.length; i++)

            if (chunk[i] == 97)

                chunk[i] = 122; // 把 'a' 替换为 'z'

        this.push(chunk);

        callback()

    }))

    .pipe(fs.createWriteStream('out.txt'))

    .on('finish', ()=> {

        console.log('DONE')

    });

使用 through2.obj 接口操作 Object Mode 下的流：

//demo10

const fs = require('fs');

const through2 = require('through2');

const csv2 = require('csv2');

let all = [];

fs.createReadStream('list.csv')

    .pipe(csv2())

    // through2.obj(fn) 是 through2({ objectMode: true }, fn) 的简易封装

    .pipe(through2.obj(function (chunk, enc, callback) {

        var data = {

            name: chunk[0],

            sex: chunk[1],

            addr: chunk[2]

        };

        this.push(data);

        callback()

    }))

    .on('data', function (data) {

        all.push(data)

    })

    .on('end', function () {

        console.log(all)

    });

对比原生的 Stream API，through2 简洁了不少，加上有 readable-stream 依赖加持，也很好理解为何像 gulp 及其插件都会使用 through2 来操作和处理 stream 了。

顺便贴下对 through2 的源码注释：

var Transform = require('readable-stream/transform'),

    inherits = require('util').inherits,

    xtend = require('xtend');

//构造方法，继承了Transform

function DestroyableTransform(opts) {

    Transform.call(this, opts);

    this._destroyed = false

}

inherits(DestroyableTransform, Transform);

//原型接口 destroy，用于关闭当前流

DestroyableTransform.prototype.destroy = function (err) {

    if (this._destroyed) return;

    this._destroyed = true;

    var self = this;

    process.nextTick(function () {

        if (err)

            self.emit('error', err);

        self.emit('close')

    })

};

// a noop _transform function

function noop(chunk, enc, callback) {

    callback(null, chunk)

}

// 闭包，用于返回对外接口方法

function through2(construct) {

    //最终返回此匿名函数

    return function (options, transform, flush) {

        if (typeof options == 'function') {

            flush = transform

            transform = options

            options = {}

        }

        if (typeof transform != 'function')

            transform = noop

        if (typeof flush != 'function')

            flush = null

        return construct(options, transform, flush)

    }

}

// 出口，执行 throuh2 闭包函数，返回一个 DestroyableTransform 的实例（t2）

module.exports = through2(function (options, transform, flush) {

    //t2 为 Transform Stream 对象

    var t2 = new DestroyableTransform(options);

    //Transform Streams 的内置接口 _transform(chunk, encoding, next) 方法

    t2._transform = transform;

    if (flush)

        t2._flush = flush;

    return t2

});

// 对外暴露一个可以直接 new （或者不加 new）来创建实例的的构造函数

module.exports.ctor = through2(function (options, transform, flush) {

    function Through2(override) {

        if (!(this instanceof Through2))

            return new Through2(override)

        this.options = xtend(options, override)

        DestroyableTransform.call(this, this.options)

    }

    inherits(Through2, DestroyableTransform)

    Through2.prototype._transform = transform

    if (flush)

        Through2.prototype._flush = flush

    return Through2

})

//Object Mode接口的简易封装

module.exports.obj = through2(function (options, transform, flush) {

    var t2 = new DestroyableTransform(xtend({objectMode: true, highWaterMark: 16}, options))

    t2._transform = transform

    if (flush)

        t2._flush = flush

    return t2

})

以上是本文对 Stream 的一个介绍，但事实上 Stream 还有许多未露面的 API，感兴趣的同学可以直接阅读官方 API文档做进一步了解。

本篇文章是对后续 gulp 源码解析系列的一个基础铺垫，想了解更多 gulp 相关内容的话可以留意我的博客。最后恭祝大家鸡年大吉！共勉~

Reference

⑴ Stream API Doc - https://nodejs.org/api/stream.html

⑵ stream-handbook - https://github.com/substack/stream-handbook

⑶ Node.js Stream - 基础篇 - http://www.cnblogs.com/zapple/p/5759670.html

⑷ Why I don't use Node's core 'stream' module - https://r.va.gg/2014/06/why-i-dont-use-nodes-core-stream-module.html

码农公寓

相关文章