前面兩篇（基礎(chǔ)篇和進(jìn)階篇）主要介紹流的基本用法和原理，本篇從應(yīng)用的角度，介紹如何使用管道進(jìn)行程序設(shè)計，主要內(nèi)容包括：

管道的概念

Browserify的管道設(shè)計

Gulp的管道設(shè)計

兩種管道設(shè)計模式比較

實例

所謂“管道”，指的是通過a.pipe(b)的形式連接起來的多個Stream對象的組合。

假如現(xiàn)在有兩個Transform：bold和red，分別可將文本流中某些關(guān)鍵字加粗和飄紅。 可以按下面的方式對文本同時加粗和飄紅：

// source: 輸入流 // dest: 輸出目的地 source.pipe(bold).pipe(red).pipe(dest)

bold.pipe(red)便可以看作一個管道，輸入流先后經(jīng)過bold和red的變換再輸出。

但如果這種加粗且飄紅的功能的應(yīng)用場景很廣，我們期望的使用方式是：

// source: 輸入流 // dest: 輸出目的地 // pipeline: 加粗且飄紅 source.pipe(pipeline).pipe(dest)

此時，pipeline封裝了bold.pipe(red)，從邏輯上來講，也稱其為管道。 其實現(xiàn)可簡化為：

var pipeline = new Duplex() var streams = pipeline._streams = [bold, red]// 底層寫邏輯：將數(shù)據(jù)寫入管道的第一個Stream，即bold pipeline._write = function (buf, enc, next) {streams[0].write(buf, enc, next) }// 底層讀邏輯：從管道的最后一個Stream（即red）中讀取數(shù)據(jù) pipeline._read = function () {var bufvar reads = 0var r = streams[streams.length - 1]// 將緩存讀空while ((buf = r.read()) !== null) {pipeline.push(buf)reads++}if (reads === 0) {// 緩存本來為空，則等待新數(shù)據(jù)的到來r.once('readable', function () {pipeline._read()})} }// 將各個Stream組合起來（此處等同于`bold.pipe(red)`） streams.reduce(function (r, next) {r.pipe(next)return next })

往pipeline寫數(shù)據(jù)時，數(shù)據(jù)直接寫入bold，再流向red，最后從pipeline讀數(shù)據(jù)時再從red中讀出。

如果需要在中間新加一個underline的Stream，可以：

pipeline._streams.splice(1, 0, underline) bold.unpipe(red) bold.pipe(underline).pipe(red)

如果要將red替換成green，可以：

// 刪除red pipeline._streams.pop() bold.unpipe(red)// 添加green pipeline._streams.push(green) bold.pipe(green)

可見，這種管道的各個環(huán)節(jié)是可以修改的。

stream-splicer對上述邏輯進(jìn)行了進(jìn)一步封裝，提供splice、push、pop等方法，使得pipeline可以像數(shù)組那樣被修改：

var splicer = require('stream-splicer') var pipeline = splicer([bold, red]) // 在中間添加underline pipeline.splice(1, 0, underline)// 刪除red pipeline.pop()// 添加green pipeline.push(green)

labeled-stream-splicer在此基礎(chǔ)上又添加了使用名字替代下標(biāo)進(jìn)行操作的功能：

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)

由于pipeline本身與其各個環(huán)節(jié)一樣，也是一個Stream對象，因此可以嵌套：

var splicer = require('labeled-stream-splicer') var pipeline = splicer(['style', [ bold, red ],'insert', [ comma ], ])pipeline.get('style') // 取得管道：[bold, red].splice(1, 0, underline) // 添加underline

Browserify的功能介紹可見substack/browserify-handbook，其核心邏輯的實現(xiàn)在于管道的設(shè)計：

var splicer = require('labeled-stream-splicer') var pipeline = splicer.obj([// 記錄輸入管道的數(shù)據(jù)，重建管道時直接將記錄的數(shù)據(jù)寫入。// 用于像watch時需要多次打包的情況'record', [ this._recorder() ],// 依賴解析，預(yù)處理'deps', [ this._mdeps ],// 處理JSON文件'json', [ this._json() ],// 刪除文件前面的BOM'unbom', [ this._unbom() ],// 刪除文件前面的`#!`行'unshebang', [ this._unshebang() ],// 語法檢查'syntax', [ this._syntax() ],// 排序，以確保打包結(jié)果的穩(wěn)定性'sort', [ depsSort(dopts) ],// 對擁有同樣內(nèi)容的模塊去重'dedupe', [ this._dedupe() ],// 將id從文件路徑轉(zhuǎn)換成數(shù)字，避免暴露系統(tǒng)路徑信息'label', [ this._label(opts) ],// 為每個模塊觸發(fā)一次dep事件'emit-deps', [ this._emitDeps() ],'debug', [ this._debug(opts) ],// 將模塊打包'pack', [ this._bpack ],// 更多自定義的處理'wrap', [], ])

每個模塊用row表示，定義如下：

{// 模塊的唯一標(biāo)識id: id,// 模塊對應(yīng)的文件路徑file: '/path/to/file',// 模塊內(nèi)容source: '',// 模塊的依賴deps: {// `require(expr)`expr: id,} }

在wrap階段前，所有的階段都處理這樣的對象流，且除pack外，都輸出這樣的流。 有的補充row中的一些信息，有的則對這些信息做一些變換，有的只是讀取和輸出。 一般row中的source、deps內(nèi)容都是在deps階段解析出來的。

下面提供一個修改Browserify管道的函數(shù)。

var Transform = require('stream').Transform // 創(chuàng)建Transform對象 function through(write, end) {return Transform({transform: write,flush: end,}) }// `b`為Browserify實例 // 該插件可打印出打包時間 function log(b) {// watch時需要重新打包，整個pipeline會被重建，所以也要重新修改b.on('reset', reset)// 修改當(dāng)前pipelinereset()function reset () {var time = nullvar bytes = 0b.pipeline.get('record').on('end', function () {// 以record階段結(jié)束為起始時刻time = Date.now()})// `wrap`是最后一個階段，在其后添加記錄結(jié)束時刻的Transformb.pipeline.get('wrap').push(through(write, end))function write (buf, enc, next) {// 累計大小bytes += buf.lengththis.push(buf)next()}function end () {// 打包時間var delta = Date.now() - timeb.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) // 應(yīng)用插件 b.plugin(log) b.bundle().pipe(fs.createWriteStream('bundle.js'))

事實上，這里的b.plugin(log)就是直接執(zhí)行了log(b)。

在插件中，可以修改b.pipeline中的任何一個環(huán)節(jié)。 因此，Browserify本身只保留了必要的功能，其它都由插件去實現(xiàn)，如watchify、factor-bundle等。

除了了上述的插件機制外，Browserify還有一套Transform機制，即通過b.transform(transform)可以新增一些文件內(nèi)容預(yù)處理的Transform。 預(yù)處理是發(fā)生在deps階段的，當(dāng)模塊文件內(nèi)容被讀出來時，會經(jīng)過這些Transform處理，然后才做依賴解析，如babelify、envify。

Gulp的核心邏輯分成兩塊：任務(wù)調(diào)度與文件處理。 任務(wù)調(diào)度是基于orchestrator，而文件處理則是基于vinyl-fs。

類似于Browserify提供的模塊定義（用row表示），vinyl-fs也提供了文件定義（vinyl對象）。

Browserify的管道處理的是row流，Gulp管道處理vinyl流：

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

任務(wù)中創(chuàng)建的管道起始于gulp.src，終止于gulp.dest，中間有若干其它的Transform（插件）。

如果與Browserify的管道對比，可以發(fā)現(xiàn)Browserify是確定了一條具有完整功能的管道，而Gulp本身只提供了創(chuàng)建vinyl流和將vinyl流寫入磁盤的工具，管道中間經(jīng)歷什么全由用戶決定。 這是因為任務(wù)中做什么，是沒有任何限制的，文件處理也只是常見的情況，并非一定要用gulp.src與gulp.dest。

Browserify與Gulp都借助管道的概念來實現(xiàn)插件機制。

Browserify定義了模塊的數(shù)據(jù)結(jié)構(gòu)，提供了默認(rèn)的管道以處理這樣的數(shù)據(jù)流，而插件可用來修改管道結(jié)構(gòu)，以定制處理行為。

Gulp雖也定義了文件的數(shù)據(jù)結(jié)構(gòu)，但只提供產(chǎn)生、消耗這種數(shù)據(jù)流的接口，完全由用戶通過插件去構(gòu)造處理管道。

當(dāng)明確具體的處理需求時，可以像Browserify那樣，構(gòu)造一個基本的處理管道，以提供插件機制。 如果需要的是實現(xiàn)任意功能的管道，可以如Gulp那樣，只提供數(shù)據(jù)流的抽象。

本節(jié)中實現(xiàn)一個針對Git倉庫自動生成changelog的工具，完整代碼見ezchangelog。

ezchangelog的輸入為git log生成的文本流，輸出默認(rèn)為markdown格式的文本流，但可以修改為任意的自定義格式。

輸入示意：

commit 9c5829ce45567bedccda9beb7f5de17574ea9437 Author: zoubin <zoubin04@gmail.com> Date: Sat Nov 7 18:42:35 2015 +0800CHANGELOGcommit 3bf9055b732cc23a9c14f295ff91f48aed5ef31a Author: zoubin <zoubin04@gmail.com> Date: Sat Nov 7 18:41:37 2015 +08004.0.3commit 87abe8e12374079f73fc85c432604642059806ae Author: zoubin <zoubin04@gmail.com> Date: Sat Nov 7 18:41:32 2015 +0800fix readmeadd more tests

輸出示意：

* [[`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 readmeadd more tests

其實需要的是這樣一個pipeline：

source.pipe(pipeline).pipe(dest)

可以分為兩個階段： * parse：從輸入文本流中解析出commit信息 * format: 將commit流變換為文本流

默認(rèn)的情況下，要想得到示例中的markdown，需要解析出每個commit的sha1、日期、消息、是否為tag。 定義commit的格式如下：

{commit: {// commit sha1long: '3bf9055b732cc23a9c14f295ff91f48aed5ef31a',short: '3bf9055',},committer: {// commit datedate: new Date('Sat Nov 7 18:41:37 2015 +0800'),},// raw message linesmessages: ['', ' 4.0.3', ''],// raw headers before the messagesheaders: [['Author', 'zoubin <zoubin04@gmail.com>'],['Date', 'Sat Nov 7 18:41:37 2015 +0800'],],// the first non-empty message linesubject: '4.0.3',// other message linesbody: '',// git tagtag: 'v4.0.3',// link to the commit. opts.baseUrl should be specified.url: 'https://github.com/zoubin/ezchangelog/commit/3bf9055', }

于是有：

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(),], ])

至此，基本功能已經(jīng)實現(xiàn)。 現(xiàn)在將其封裝并提供插件機制。

function Changelog(opts) {opts = opts || {}this._options = opts// 創(chuàng)建pipelinethis.pipeline = splicer.obj(['parse', ['split', split(),'commit', commit(),'tag', tag(),'url', url({ baseUrl: opts.baseUrl }),],'format', ['markdownify', markdownify(),],])// 應(yīng)用插件;[].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]}// 執(zhí)行插件函數(shù)，修改pipelinep(this, opts)return this }

上面的實現(xiàn)提供了兩種方式來應(yīng)用插件。 一種是通過配置傳入，另一種是創(chuàng)建實例后再調(diào)用plugin方法，本質(zhì)一樣。

為了使用方便，還可以簡單封裝一下。

function changelog(opts) {return new Changelog(opts).pipeline }

這樣，就可以如下方式使用：

source.pipe(changelog()).pipe(dest)

這個已經(jīng)非常接近我們的預(yù)期了。

現(xiàn)在來開發(fā)一個插件，修改默認(rèn)的渲染方式。

var through = require('through2')function customFormatter(c) {// c是`Changelog`實例// 添加解析author的transformc.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.shortsha1 = '[`' + 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)

同樣的輸入，輸出將會是：

* [`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

可以看出，通過創(chuàng)建可修改的管道，ezchangelog保持了本身邏輯的單一性，同時又提供了強大的自定義空間。

• GitHub，substack/browserify-handbook
• GitHub，zoubin/streamify-your-node-program

總結(jié)

以上是生活随笔為你收集整理的Node.js Stream - 实战篇的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。