<h2>背景</h2>    <p>前面两篇（基础篇和进阶篇）主要介绍流的基本用法和原理，本篇从应用的角度，介绍如何使用管道进行程序设计，主要内容包括：</p>    <h2>Pipeline</h2>    <p>所谓“管道”，指的是通过 a.pipe(b) 的形式连接起来的多个Stream对象的组合。</p>    <p>假如现在有两个 Transform ： bold 和 red ，分别可将文本流中某些关键字加粗和飘红。</p>    <p>可以按下面的方式对文本同时加粗和飘红：</p>    <pre>  <code class="language-javascript">// source: 输入流  // dest: 输出目的地  source.pipe(bold).pipe(red).pipe(dest)</code></pre>    <p>bold.pipe(red) 便可以看作一个管道，输入流先后经过 bold 和 red 的变换再输出。</p>    <p>但如果这种加粗且飘红的功能的应用场景很广，我们期望的使用方式是：</p>    <pre>  <code class="language-javascript">// source: 输入流  // dest: 输出目的地  // pipeline: 加粗且飘红  source.pipe(pipeline).pipe(dest)</code></pre>    <p>此时， pipeline 封装了 bold.pipe(red) ，从逻辑上来讲，也称其为管道。</p>    <p>其实现可简化为：</p>    <pre>  <code class="language-javascript">var pipeline = new Duplex()  var streams = pipeline._streams = [bold, red]    // 底层写逻辑：将数据写入管道的第一个Stream，即bold  pipeline._write = function (buf, enc, next) {    streams[0].write(buf, enc, next)  }    // 底层读逻辑：从管道的最后一个Stream（即red）中读取数据  pipeline._read = function () {    var buf    var reads = 0    var r = streams[streams.length - 1]    // 将缓存读空    while ((buf = r.read()) !== null) {      pipeline.push(buf)      reads++    }    if (reads === 0) {      // 缓存本来为空，则等待新数据的到来      r.once('readable', function () {        pipeline._read()      })    }  }    // 将各个Stream组合起来（此处等同于`bold.pipe(red)`）  streams.reduce(function (r, next) {    r.pipe(next)    return next  })</code></pre>    <p>往 pipeline 写数据时，数据直接写入 bold ，再流向 red ，最后从 pipeline 读数据时再从 red 中读出。</p>    <p>如果需要在中间新加一个 underline 的Stream，可以：</p>    <pre>  <code class="language-javascript">pipeline._streams.splice(1, 0, underline)  bold.unpipe(red)  bold.pipe(underline).pipe(red)</code></pre>    <p>如果要将 red 替换成 green ，可以：</p>    <pre>  <code class="language-javascript">// 删除red  pipeline._streams.pop()  bold.unpipe(red)    // 添加green  pipeline._streams.push(green)  bold.pipe(green)</code></pre>    <p>可见，这种管道的各个环节是可以修改的。</p>    <p><a href="/misc/goto?guid=4959675733599059174" rel="nofollow,noindex">stream-splicer</a> 对上述逻辑进行了进一步封装，提供 splice 、 push 、 pop 等方法，使得 pipeline 可以像数组那样被修改：</p>    <pre>  <code class="language-javascript">var splicer = require('stream-splicer')  var pipeline = splicer([bold, red])  // 在中间添加underline  pipeline.splice(1, 0, underline)    // 删除red  pipeline.pop()    // 添加green  pipeline.push(green)</code></pre>    <p><a href="/misc/goto?guid=4959675733684455920" rel="nofollow,noindex">labeled-stream-splicer</a> 在此基础上又添加了使用名字替代下标进行操作的功能：</p>    <pre>  <code class="language-javascript">var splicer = require('labeled-stream-splicer')  var pipeline = splicer([    'bold', bold,    'red', red,  ])    // 在`red`前添加underline  pipeline.splice('red', 0, underline)    // 删除`bold`  pipeline.splice('bold', 1)</code></pre>    <p>由于 pipeline 本身与其各个环节一样，也是一个Stream对象，因此可以嵌套：</p>    <pre>  <code class="language-javascript">var splicer = require('labeled-stream-splicer')  var pipeline = splicer([    'style', [ bold, red ],    'insert', [ comma ],  ])    pipeline.get('style')     // 取得管道：[bold, red]    .splice(1, 0, underline) // 添加underline</code></pre>    <h2>Browserify</h2>    <p><a href="/misc/goto?guid=4958836570912107601" rel="nofollow,noindex">Browserify</a> 的功能介绍可见 <a href="/misc/goto?guid=4959615073390951462" rel="nofollow,noindex">substack/browserify-handbook</a> ，其核心逻辑的实现在于管道的设计：</p>    <pre>  <code class="language-javascript">var splicer = require('labeled-stream-splicer')  var pipeline = splicer.obj([      // 记录输入管道的数据，重建管道时直接将记录的数据写入。      // 用于像watch时需要多次打包的情况      'record', [ this._recorder() ],      // 依赖解析，预处理      'deps', [ this._mdeps ],      // 处理JSON文件      'json', [ this._json() ],      // 删除文件前面的BOM      'unbom', [ this._unbom() ],      // 删除文件前面的`#!`行      'unshebang', [ this._unshebang() ],      // 语法检查      'syntax', [ this._syntax() ],      // 排序，以确保打包结果的稳定性      'sort', [ depsSort(dopts) ],      // 对拥有同样内容的模块去重      'dedupe', [ this._dedupe() ],      // 将id从文件路径转换成数字，避免暴露系统路径信息      'label', [ this._label(opts) ],      // 为每个模块触发一次dep事件      'emit-deps', [ this._emitDeps() ],      'debug', [ this._debug(opts) ],      // 将模块打包      'pack', [ this._bpack ],      // 更多自定义的处理      'wrap', [],  ])</code></pre>    <p>每个模块用 row 表示，定义如下：</p>    <pre>  <code class="language-javascript">{    // 模块的唯一标识    id: id,    // 模块对应的文件路径    file: '/path/to/file',    // 模块内容    source: '',    // 模块的依赖    deps: {      // `require(expr)`      expr: id,    }  }</code></pre>    <p>在 wrap 阶段前，所有的阶段都处理这样的对象流，且除 pack 外，都输出这样的流。</p>    <p>有的补充 row 中的一些信息，有的则对这些信息做一些变换，有的只是读取和输出。</p>    <p>一般 row 中的 source 、 deps 内容都是在 deps 阶段解析出来的。</p>    <p>下面提供一个修改 <a href="/misc/goto?guid=4958836570912107601" rel="nofollow,noindex">Browserify</a> 管道的函数。</p>    <pre>  <code class="language-javascript">var Transform = require('stream').Transform  // 创建Transform对象  function through(write, end) {    return Transform({      transform: write,      flush: end,    })  }    // `b`为Browserify实例  // 该插件可打印出打包时间  function log(b) {    // watch时需要重新打包，整个pipeline会被重建，所以也要重新修改    b.on('reset', reset)    // 修改当前pipeline    reset()      function reset () {      var time = null      var bytes = 0      b.pipeline.get('record').on('end', function () {        // 以record阶段结束为起始时刻        time = Date.now()      })        // `wrap`是最后一个阶段，在其后添加记录结束时刻的Transform      b.pipeline.get('wrap').push(through(write, end))      function write (buf, enc, next) {        // 累计大小        bytes += buf.length        this.push(buf)        next()      }      function end () {        // 打包时间        var delta = Date.now() - time        b.emit('time', delta)        b.emit('bytes', bytes)        b.emit('log', bytes + ' bytes written ('          + (delta / 1000).toFixed(2) + ' seconds)'        )        this.push(null)      }    }  }    var fs = require('fs')  var browserify = require('browserify')  var b = browserify(opts)  // 应用插件  b.plugin(log)  b.bundle().pipe(fs.createWriteStream('bundle.js'))</code></pre>    <p>事实上，这里的 b.plugin(log) 就是直接执行了 log(b) 。</p>    <p>在插件中，可以修改 b.pipeline 中的任何一个环节。</p>    <p>因此， <a href="/misc/goto?guid=4958836570912107601" rel="nofollow,noindex">Browserify</a> 本身只保留了必要的功能，其它都由插件去实现，如 <a href="/misc/goto?guid=4959675733844821438" rel="nofollow,noindex">watchify</a> 、 <a href="/misc/goto?guid=4959675733914604836" rel="nofollow,noindex">factor-bundle</a> 等。</p>    <p>除了了上述的插件机制外， <a href="/misc/goto?guid=4958836570912107601" rel="nofollow,noindex">Browserify</a> 还有一套Transform机制，即通过 b.transform(transform) 可以新增一些文件内容预处理的Transform。</p>    <p>预处理是发生在 deps 阶段的，当模块文件内容被读出来时，会经过这些Transform处理，然后才做依赖解析，如 <a href="/misc/goto?guid=4959658690685375103" rel="nofollow,noindex">babelify</a> 、 <a href="/misc/goto?guid=4959675734036127151" rel="nofollow,noindex">envify</a> 。</p>    <h2>Gulp</h2>    <p><a href="/misc/goto?guid=4958828538346190979" rel="nofollow,noindex">Gulp</a> 的核心逻辑分成两块：任务调度与文件处理。</p>    <p>任务调度是基于 <a href="/misc/goto?guid=4959642516177679433" rel="nofollow,noindex">orchestrator</a> ，而文件处理则是基于 <a href="/misc/goto?guid=4959675734165542225" rel="nofollow,noindex">vinyl-fs</a> 。</p>    <p>类似于 <a href="/misc/goto?guid=4958836570912107601" rel="nofollow,noindex">Browserify</a> 提供的模块定义（用 row 表示）， <a href="/misc/goto?guid=4959675734165542225" rel="nofollow,noindex">vinyl-fs</a> 也提供了文件定义（ <a href="/misc/goto?guid=4959659945105329496" rel="nofollow,noindex">vinyl</a> 对象）。</p>    <p><a href="/misc/goto?guid=4958836570912107601" rel="nofollow,noindex">Browserify</a> 的管道处理的是 row 流， <a href="/misc/goto?guid=4958828538346190979" rel="nofollow,noindex">Gulp</a> 管道处理 <a href="/misc/goto?guid=4959659945105329496" rel="nofollow,noindex">vinyl</a> 流：</p>    <pre>  <code class="language-javascript">gulp.task('scripts', ['clean'], function() {    // Minify and copy all JavaScript (except vendor scripts)     // with sourcemaps all the way down     return gulp.src(paths.scripts)      .pipe(sourcemaps.init())      .pipe(coffee())      .pipe(uglify())      .pipe(concat('all.min.js'))      .pipe(sourcemaps.write())      .pipe(gulp.dest('build/js'));  });</code></pre>    <p>任务中创建的管道起始于 gulp.src ，终止于 gulp.dest ，中间有若干其它的Transform（插件）。</p>    <p>如果与 <a href="/misc/goto?guid=4958836570912107601" rel="nofollow,noindex">Browserify</a> 的管道对比，可以发现 <a href="/misc/goto?guid=4958836570912107601" rel="nofollow,noindex">Browserify</a> 是确定了一条具有完整功能的管道，而 <a href="/misc/goto?guid=4958828538346190979" rel="nofollow,noindex">Gulp</a> 本身只提供了创建 <a href="/misc/goto?guid=4959659945105329496" rel="nofollow,noindex">vinyl</a> 流和将 <a href="/misc/goto?guid=4959659945105329496" rel="nofollow,noindex">vinyl</a> 流写入磁盘的工具，管道中间经历什么全由用户决定。</p>    <p>这是因为任务中做什么，是没有任何限制的，文件处理也只是常见的情况，并非一定要用 gulp.src 与 gulp.dest 。</p>    <h2>两种模式比较</h2>    <p><a href="/misc/goto?guid=4958836570912107601" rel="nofollow,noindex">Browserify</a> 与 <a href="/misc/goto?guid=4958828538346190979" rel="nofollow,noindex">Gulp</a> 都借助管道的概念来实现插件机制。</p>    <p><a href="/misc/goto?guid=4958836570912107601" rel="nofollow,noindex">Browserify</a> 定义了模块的数据结构，提供了默认的管道以处理这样的数据流，而插件可用来修改管道结构，以定制处理行为。</p>    <p><a href="/misc/goto?guid=4958828538346190979" rel="nofollow,noindex">Gulp</a> 虽也定义了文件的数据结构，但只提供产生、消耗这种数据流的接口，完全由用户通过插件去构造处理管道。</p>    <p>当明确具体的处理需求时，可以像 <a href="/misc/goto?guid=4958836570912107601" rel="nofollow,noindex">Browserify</a> 那样，构造一个基本的处理管道，以提供插件机制。</p>    <p>如果需要的是实现任意功能的管道，可以如 <a href="/misc/goto?guid=4958828538346190979" rel="nofollow,noindex">Gulp</a> 那样，只提供数据流的抽象。</p>    <h2>实例</h2>    <p>本节中实现一个针对 <a href="/misc/goto?guid=4958964328408675079" rel="nofollow,noindex">Git</a> 仓库自动生成changelog的工具，完整代码见 <a href="/misc/goto?guid=4959675734466981825" rel="nofollow,noindex">ezchangelog</a> 。</p>    <p><a href="/misc/goto?guid=4959675734466981825" rel="nofollow,noindex">ezchangelog</a> 的输入为 git log 生成的文本流，输出默认为markdown格式的文本流，但可以修改为任意的自定义格式。</p>    <p>输入示意：</p>    <pre>  <code class="language-javascript">commit 9c5829ce45567bedccda9beb7f5de17574ea9437  Author: zoubin <zoubin04@gmail.com>  Date:   Sat Nov 7 18:42:35 2015 +0800        CHANGELOG    commit 3bf9055b732cc23a9c14f295ff91f48aed5ef31a  Author: zoubin <zoubin04@gmail.com>  Date:   Sat Nov 7 18:41:37 2015 +0800        4.0.3    commit 87abe8e12374079f73fc85c432604642059806ae  Author: zoubin <zoubin04@gmail.com>  Date:   Sat Nov 7 18:41:32 2015 +0800        fix readme      add more tests</code></pre>    <p>输出示意：</p>    <pre>  <code class="language-javascript">* [[`9c5829c`](https://github.com/zoubin/ezchangelog/commit/9c5829c)] CHANGELOG    ## [v4.0.3](https://github.com/zoubin/ezchangelog/commit/3bf9055) (2015-11-07)    * [[`87abe8e`](https://github.com/zoubin/ezchangelog/commit/87abe8e)] fix readme        add more tests</code></pre>    <p>其实需要的是这样一个 pipeline ：</p>    <pre>  <code class="language-javascript">source.pipe(pipeline).pipe(dest)</code></pre>    <p>可以分为两个阶段：</p>    <ul>     <li>parse：从输入文本流中解析出commit信息</li>     <li>format: 将commit流变换为文本流</li>    </ul>    <p>默认的情况下，要想得到示例中的markdown，需要解析出每个commit的sha1、日期、消息、是否为tag。</p>    <p>定义commit的格式如下：</p>    <pre>  <code class="language-javascript">{    commit: {      // commit sha1      long: '3bf9055b732cc23a9c14f295ff91f48aed5ef31a',      short: '3bf9055',    },    committer: {      // commit date      date: new Date('Sat Nov 7 18:41:37 2015 +0800'),    },    // raw message lines    messages: ['', '    4.0.3', ''],    // raw headers before the messages    headers: [      ['Author', 'zoubin <zoubin04@gmail.com>'],      ['Date', 'Sat Nov 7 18:41:37 2015 +0800'],    ],    // the first non-empty message line    subject: '4.0.3',    // other message lines    body: '',    // git tag    tag: 'v4.0.3',    // link to the commit. opts.baseUrl should be specified.    url: 'https://github.com/zoubin/ezchangelog/commit/3bf9055',  }</code></pre>    <p>于是有：</p>    <pre>  <code class="language-javascript">var splicer = require('labeled-stream-splicer')  pipeline = splicer.obj([    'parse', [      // 按行分隔      'split', split(),      // 生成commit对象，解析出sha1和日期      'commit', commit(),      // 解析出tag      'tag', tag(),      // 解析出url      'url', url({ baseUrl: opts.baseUrl }),    ],    'format', [      // 将commit组合成markdown文本      'markdownify', markdownify(),    ],  ])</code></pre>    <p>至此，基本功能已经实现。</p>    <p>现在将其封装并提供插件机制。</p>    <pre>  <code class="language-javascript">function Changelog(opts) {    opts = opts || {}    this._options = opts    // 创建pipeline    this.pipeline = splicer.obj([      'parse', [        'split', split(),        'commit', commit(),        'tag', tag(),        'url', url({ baseUrl: opts.baseUrl }),      ],      'format', [        'markdownify', markdownify(),      ],    ])      // 应用插件    ;[].concat(opts.plugin).filter(Boolean).forEach(function (p) {      this.plugin(p)    }, this)  }    Changelog.prototype.plugin = function (p, opts) {    if (Array.isArray(p)) {      opts = p[1]      p = p[0]    }    // 执行插件函数，修改pipeline    p(this, opts)    return this  }</code></pre>    <p>上面的实现提供了两种方式来应用插件。</p>    <p>一种是通过配置传入，另一种是创建实例后再调用 plugin 方法，本质一样。</p>    <p>为了使用方便，还可以简单封装一下。</p>    <pre>  <code class="language-javascript">function changelog(opts) {    return new Changelog(opts).pipeline  }</code></pre>    <p>这样，就可以如下方式使用：</p>    <pre>  <code class="language-javascript">source.pipe(changelog()).pipe(dest)</code></pre>    <p>这个已经非常接近我们的预期了。</p>    <p>现在来开发一个插件，修改默认的渲染方式。</p>    <pre>  <code class="language-javascript">var through = require('through2')    function customFormatter(c) {    // c是`Changelog`实例      // 添加解析author的transform    c.pipeline.get('parse').push(through.obj(function (ci, enc, next) {      // parse the author name from: 'zoubin <zoubin04@gmail.com>'      ci.committer.author = ci.headers[0][1].split(/\s+/)[0]      next(null, ci)    }))      // 替换原有的渲染    c.pipeline.get('format').splice('markdownify', 1, through.obj(function (ci, enc, next) {      var sha1 = ci.commit.short      sha1 = '[`' + sha1 + '`](' + c._options.baseUrl + sha1 + ')'      var date = ci.committer.date.toISOString().slice(0, 10)      next(null, '* ' + sha1 + ' ' + date + ' @' + ci.committer.author + '\n')    }))  }    source    .pipe(changelog({      baseUrl: 'https://github.com/zoubin/ezchangelog/commit/',      plugin: [customFormatter],    }))    .pipe(dest)</code></pre>    <p>同样的输入，输出将会是：</p>    <pre>  <code class="language-javascript">* [`9c5829c`](https://github.com/zoubin/ezchangelog/commit/9c5829c) 2015-11-07 @zoubin  * [`3bf9055`](https://github.com/zoubin/ezchangelog/commit/3bf9055) 2015-11-07 @zoubin  * [`87abe8e`](https://github.com/zoubin/ezchangelog/commit/87abe8e) 2015-11-07 @zoubin</code></pre>    <p>可以看出，通过创建可修改的管道， <a href="/misc/goto?guid=4959675734466981825" rel="nofollow,noindex">ezchangelog</a> 保持了本身逻辑的单一性，同时又提供了强大的自定义空间。</p>    <h2>系列文章</h2>    <ul>     <li>第一部分：<a href="http://www.open-open.com/lib/view/open1469618649894.html">《Node.js Stream - 基础篇》</a>，介绍<a href="/misc/goto?guid=4958963900143935496">Stream</a>接口的基本使用。</li>     <li>第二部分：<a href="http://www.open-open.com/lib/view/open1469619171753.html">《Node.js Stream - 进阶篇》</a>，重点剖析Stream底层如何支持流式数据处理，及其<a href="/misc/goto?guid=4959675728905058726">back pressure</a>机制。</li>     <li>第三部分：<a href="http://www.open-open.com/lib/view/open1469619230397.html">《Node.js Stream - 实战篇》</a>，介绍如何使用Stream进行程序设计。从<a href="/misc/goto?guid=4958836570912107601">Browserify</a>和<a href="/misc/goto?guid=4958828538346190979">Gulp</a>总结出两种设计模式，并基于Stream构建一个为<a href="/misc/goto?guid=4958964328408675079">Git</a>仓库自动生成changelog的应用作为示例。</li>    </ul>    <h2>参考文献</h2>    <ul>     <li>GitHub， <a href="/misc/goto?guid=4959615073390951462" rel="nofollow,noindex">substack/browserify-handbook</a></li>     <li>GitHub， <a href="/misc/goto?guid=4959675729265092912" rel="nofollow,noindex">zoubin/streamify-your-node-program</a></li>    </ul>    <p> </p>    <p>来自：http://tech.meituan.com/stream-in-action.html</p>    <p> </p>