[关闭]
@FunC 2017-11-29T17:08:09.000000Z 字数 7257 阅读 2441

Node.js Design Patterns | CH05

Node.js

来用 stream 编程吧！

先放一张思维导图（PDF版放大后还是勉强能看清的）：


书中的介绍已经很详细，这里有篇不错的 stream 入门文章：
stream-handbook：https://github.com/jabez128/stream-handbook
如果还想要更多的实操例子，可以试试 nodeschool 的对应专题：
npm install -g stream-adventure

发现 stream 的重要性

在事件驱动的 Node.js 中，最高效的处理 I/O 的模式，就是一有输入就对其进行处理并输出。
传统的是 buffer 模式，就是把整个文件都读入 buffer 后才进行处理：

而 stream 是将数据且分成一小个的 chunk，来一个处理一个；

Stream 的优点可以总结为空间效率高、时间效率高和组合性强。下面来简单介绍。

空间效率

主要问题在于并发处理大文件。
如果是 buffer 模式，并发处理大文件时，占用内存极大（因为要全部读取完才能处理）。同时，V8 的 buffer 上限约为 1GB，所以使用 stream 更合理。

时间效率

同样也是数据分成 chunk 带来的好处。观察下图的场景：

上图展示了一个：读取 -> 压缩 -> 发送至服务端 -> 服务端读取 -> 解压 -> 写入 的过程。
如果采用 stream，发挥了 Node.js 异步能力的同时，还让服务端能尽快接收，处理数据。

组合性

Stream 的高组合性得益于.pipe()方法，提供了一个 stream 组合的通用接口。
.pipe()除了让代码更易组合复用，还能让代码更清晰易懂且模块化。

使用 stream 来处理异步流程

回想一下，stream 顺序处理 chunk（遍历器），同时还能监听/触发事件（ EventEmitter 的实例），这不正是我们先前处理异步流程所需的东西吗。

1. 顺序执行

使用 stream 来做顺序执行任务，确保次序的同时还能保持优雅。
来看一段代码，重点见注释：
concatFiles.js

  1. const fromArray = require('from2-array');
  2. const through = require('through2');
  3. const fs = require('fs');
  5. function concatFiles(destination, files, callback) {
  6.   const destStream = fs.createWriteStream(destination);
  7.   fromArray.obj(files)             
  8.     .pipe(through.obj((file, enc, done) => {  
  9.       const src = fs.createReadStream(file);
  10.         // 每个 file 读取&写入完毕后不关闭 destStream，让下一个 file 继续
  11.       src.pipe(destStream, {end: false});
  12.       src.on('end', done);
  13.     }))
  14.     .on('finish', () => {
  15.         // through 生成的 transform stream 结束后，手动关闭 destStream       
  16.       destStream.end();
  17.       callback();
  18.     });
  19. }
  20. module.exports = concatFiles;

2. 无序并行执行

思路与使用 callback 类似，callback 中是立即返回，stream 中是立即调用 _transform 的 done。同时将 _flush中的 done 保留至全部任务完成。
来看一段代码，重点见注释：
parallelStream.js

  1. const stream = require('stream');
  3. class ParallelStream extends stream.Transform {
  4.   constructor(userTransform) {
  5.       // 这种并发场景主要用在 object stream 上
  6.     super({objectMode: true});
  7.       // 即 ._transform() 留给调用时实现
  8.     this.userTransform = userTransform;
  9.     this.running = 0;
  10.       // 用于稍后保存 ._flush(done) 中的 done
  11.     this.terminateCallback = null;
  12.   }
  14.   _transform(chunk, enc, done) {
  15.     this.running++;
  16.     this.userTransform(chunk, enc, this.push.bind(this), this._onComplete.bind(this));
  17.     done();
  18.   }
  20.   _flush(done) {
  21.       // 没运行完则不调用 done，保存到 this.terminateCallback中，以后用
  22.     if(this.running > 0) {
  23.       this.terminateCallback = done;
  24.     } else {
  25.       done();
  26.     }
  27.   }
  29.  // 该方法在每次完成一个 transform 中的 chunk 时都会触发
  30.   _onComplete(err) {
  31.     this.running--;
  32.     if(err) {
  33.       return this.emit('error', err);
  34.     }
  35.       // 如果这是最后一个任务了，先前有保存 done 的话就执行，否则在 _flush 中 done
  36.     if(this.running === 0) {
  37.       this.terminateCallback && this.terminateCallback();
  38.     }
  39.   }
  40. }

3. 无序并行执行（限制并发数）

并行处理任务时不限制并发数将会导致严重的后果。参考保存 _flush 中的 done 函数来延后 finish 的做法，transform stream 中达到并发上限时，也可以延后 done，来控制并发。
来看一段代码，重点见注释：
limitedParallelStream.js

  1. const stream = require('stream');
  3. class LimitedParallelStream extends stream.Transform {
  4.   constructor(concurrency, userTransform) {
  5.     super({objectMode: true});
  6.       // 并发数
  7.     this.concurrency = concurrency;
  8.     this.userTransform = userTransform;
  9.     this.running = 0;
  10.     this.terminateCallback = null;
  11.       // 与 terminateCallback 同理，不过保存的是 _transform 中的 done
  12.     this.continueCallback = null;
  13.   }
  15.   _transform(chunk, enc, done) {
  16.     this.running++;
  17.     this.userTransform(chunk, enc,  this.push.bind(this), this._onComplete.bind(this));
  18.     if(this.running < this.concurrency) {
  19.       done();
  20.     } else {
  21.       // 达到并发上限则将 done 延后到 _onComplete 中调用
  22.       this.continueCallback = done;
  23.     }
  24.   }
  26.   _flush(done) {
  27.     if(this.running > 0) {
  28.       this.terminateCallback = done;
  29.     } else {
  30.       done();
  31.     }
  32.   }
  34.   _onComplete(err) {
  35.     this.running--;
  36.     if(err) {
  37.       return this.emit('error', err);
  38.     }
  39.     const tmpCallback = this.continueCallback；
  40.     // reset
  41.     this.continueCallback = null;
  42.     // continue ._transform()
  43.     tmpCallback && tmpCallback();
  44.     if(this.running === 0) {
  45.       this.terminateCallback && this.terminateCallback();
  46.     }
  47.   }
  48. }
  50. module.exports = LimitedParallelStream;

4. 顺序并发执行

实际上，我们只要能把无序输出的chunk重新正确排序，就相当于做到顺序并发执行了。可以通过 npm 上的 through2-parallel包实现。

管道相关的模式（Piping patterns）

1. 组合（Combining streams）

优点：
1. 复用组合管道（且不用考虑内部组合细节）
2. 简化错误处理（只用处理合并管道的错误）

组合管道的实质：
1. 写入时，实际写入的是组合管道的第一个 stream
2. 读取时，实际读取的事组合管道的最后一个 stream

可直接使用 multipipe 包

2.分支（Forking streams）


fork 一个 stream，意味着将一个 Readable stream 接到多个 Writable streams 上。
在 Node.js 中，直接重复使用该 readable stream 即可：

  1. // 省略部分代码...
  2. const inputStream = fs.createReadStream(inputFile);
  3. inputStream
  4.   .pipe(sha1Stream)
  5.   .pipe(fs.createWriteStream(inputFile + '.sha1'))
  6. ;
  8. inputStream
  9.   .pipe(md5Stream)
  10.   .pipe(fs.createWriteStream(inputFile + '.md5'))
  11. ;

值得注意的是，fork 的 stream 接收到的是相同的 chunks，所以要注意有副作用的操作。因为会影响到其他 fork 的stream。
另外，数据流动的速度=最慢的 fork stream 的速度。

3.合并（Merging streams）

合并操作是分支操作的逆向版：

由于管道默认的 auto end 设置，其中一个 source stream 结束后 destination stream 就会马上关闭，从而导致错误。
解决方案是设置{end: false}，并在全部 source stream 完成后手动触发 destination 的end()。
示例代码如下：
mergeTar.js

  1. const tar = require('tar');
  2. const fstream = require('fstream');
  3. const path = require('path');
  5. const destination = path.resolve(process.argv[2]);
  6. const sourceA = path.resolve(process.argv[3]);
  7. const sourceB = path.resolve(process.argv[4]);
  9. // 还没有输入，可以先随意组合
  10. const pack = tar.Pack();
  11. pack.pipe(fstream.Writer(destination));
  13. // 计数器，在全部 source 都结束时调用 end()
  14. let endCount = 0;
  15. function onEnd() {
  16.   if(++endCount === 2) {
  17.     pack.end();
  18.   }
  19. }
  21. // 注意使用上面定义的 onEnd()回调
  22. const sourceStreamA = fstream.Reader({type: "Directory", path: sourceA})
  23.   .on('end', onEnd);
  25. const sourceStreamB = fstream.Reader({type: "Directory", path: sourceB})
  26.   .on('end', onEnd);
  27. //
  29. // 开始 pipe 时记得设置 end：false选项
  30. sourceStreamA.pipe(pack, {end: false});
  31. sourceStreamB.pipe(pack, {end: false});

注意，多个 source stream 的数据是随机顺序注入 destination stream 的，这在部分 object stream 中是可接受的。
如果想要按顺序（一个 source 接一个 source 地）消耗 chunk，可实现一个状态机来记录当前的 source，按顺序调用相应的 done。（npm packages：multistream）

多路复用与分路（multiplexing and demultiplexing）

如果我们想公用一个频道（channel），来传输多个独立的 streams（即传输完后可重新分离），则需要实现多路复用与分路，如下图所示：


可见，通过多路复用器（MUX）将多个 channel 通过一个共享的 channel 传输。然后通过分路器（DEMUX）将 channel 重新分离。这种技术经常用在通讯上。

实现多路复用的其中一种技术，就是分组交换（packet switching）。思路是将分路时所需的一些元信息（如 channel id，chunk length 等）和 chunk 打包在一起传输。

实现一个远程日志记录服务

我们先实现一个只有 channel id， data length 和 data 的 mini packet：

来看一下 client.js 的代码（重点见注释）：

  1. const child_process = require('child_process');
  2. const net = require('net');
  4. function multiplexChannels(sources, destination) {
  5.     let totalChannels = sources.length;
  6.     for (let i = 0; i < sources.length; i++) {
  7.         sources[i]
  8.             .on('readable', function() {
  9.                 let chunk;
  10.                 while((chunk = this.read()) !== null) {
  11.                     const outBuff = new Buffer(1 + 4 + chunk.length);
  12.                     // 写入1字节的 channel id
  13.                     outBuff.writeUInt8(i, 0);
  14.                     // 写入4字节的 chunk size
  15.                     outBuff.writeUInt32BE(chunk.length, 1);
  16.                     // 写入 chunk 的内容
  17.                     chunk.copy(outBuff, 5);
  18.                     console.log('Sending packet tp channel: ' + i);
  19.                     destination.write(outBuff);
  20.                 }
  21.             })
  22.             .on('end', () => {
  23.                 if (--totalChannels === 0) {
  24.                     destination.end();
  25.                 }
  26.             });
  27.     }
  28. }
  30. const socket = net.connect(3000, () => {
  31.       // silent: true 能让子进程不继承父进程的 stdout 和 stderr
  32.     const child = child_process.fork(process.argv[2],process.argv.slice(3),{silent:true});
  33.     multiplexChannels([child.stdout, child.stderr], socket);
  34. });

下面来看一下 server.js 的代码（重点见注释）：

  1. const net = require('net');
  2. const fs = require('fs');
  4. function demultiplexChannel(source, destinations) {
  5.       // 用于记录当前 chunk 的 channel id
  6.     let currentChannel = null;
  7.       // 用于记录当前 chunk 的长度
  8.     let currentLength = null;
  10.     source
  11.         .on('readable', () => {
  12.             let chunk;
  13.             if (currentChannel === null) {
  14.                   // 先读1个字节的 id 信息
  15.                 chunk = source.read(1);
  16.                 currentChannel = chunk && chunk.readUInt8(0);
  17.             }
  19.             if (currentLength === null) {
  20.                   // 读4个字节的长度信息
  21.                 chunk = source.read(4);
  22.                 currentLenth = chunk && chunk.readUInt32BE(0);
  23.                   // 有可能 buffer 中的数据不足，.read() return null
  24.                   // 等一下次 readable 再读取
  25.                 if (currentLength === null) {
  26.                     return;
  27.                 }
  28.             }
  29.               // 读取 chunk 的内容
  30.             chunk = source.read(currentLenth);
  31.             if (chunk === null) {
  32.                 return;
  33.             }
  34.             console.log('Received packet from: ' + currentChannel);
  35.               // 写入目标 channel
  36.             destinations[currentChannel].write(chunk);
  37.               // reset
  38.             currentChannel = null;
  39.             currentLength = null;
  40.         })
  41.         .on('end', () => {
  42.             destinations.forEach(destination => destination.end());
  43.             console.log('Source channel closed');
  44.         });
  45. }
  47. net.createServer(socket => {
  48.     const stdoutStream = fs.createWriteStream('stdout.log');
  49.     const stderrStream = fs.createWriteStream('stderr.log');
  50.     demultiplexChannel(socket, [stdoutStream, stderrStream]);
  51. }).listen(3000, () => console.log('Server started'));

object stream 的多路复用与分路

原理和 binary mode 基本一样，且 channelID 可以直接作为对象的属性 。

只进行分路操作

另一种模式是根据一定的规则对流入的 data 进行分路：

添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注