AI 助理
备案控制台
开发者社区 开发与运维 文章 正文

Node.js Stream - 实战篇

2016-08-26 1902
版权
版权声明：
本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《 阿里云开发者社区用户服务协议》和 《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写 侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。
本文涉及的产品
公共DNS（含HTTPDNS解析），每月1000万次HTTP解析
全局流量管理 GTM，标准版 1个月
云解析 DNS，旗舰版 1个月
简介： 背景 前面两篇（基础篇和进阶篇）主要介绍流的基本用法和原理，本篇从应用的角度，介绍如何使用管道进行程序设计，主要内容包括： Pipeline 所谓“管道”，指的是通过 a.pipe(b) 的形式连接起来的多个Stream对象的组合。 假如现在有两个 Transform ： bold 和 red ，分别可将文本流中某些关键字加粗和飘红。 可以按下面的方式对文

背景

前面两篇（基础篇和进阶篇）主要介绍流的基本用法和原理，本篇从应用的角度，介绍如何使用管道进行程序设计，主要内容包括：

Pipeline

所谓“管道”，指的是通过 a.pipe(b) 的形式连接起来的多个Stream对象的组合。

假如现在有两个 Transform ： bold 和 red ，分别可将文本流中某些关键字加粗和飘红。

可以按下面的方式对文本同时加粗和飘红：

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

bold.pipe(red) 便可以看作一个管道，输入流先后经过 bold 和 red 的变换再输出。

但如果这种加粗且飘红的功能的应用场景很广，我们期望的使用方式是：

// source: 输入流
// dest: 输出目的地
// pipeline: 加粗且飘红
source.pipe(pipeline).pipe(dest)

此时， pipeline 封装了 bold.pipe(red) ，从逻辑上来讲，也称其为管道。

其实现可简化为：

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

往 pipeline 写数据时，数据直接写入 bold ，再流向 red ，最后从 pipeline 读数据时再从 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)

可见，这种管道的各个环节是可以修改的。

stream-splicer 对上述逻辑进行了进一步封装，提供 splice 、 push 、 pop 等方法，使得 pipeline 可以像数组那样被修改：

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 在此基础上又添加了使用名字替代下标进行操作的功能：

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 本身与其各个环节一样，也是一个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

Browserify 的功能介绍可见 substack/browserify-handbook ，其核心逻辑的实现在于管道的设计：

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', [],
])

每个模块用 row 表示，定义如下：

{
  // 模块的唯一标识
  id: id,
  // 模块对应的文件路径
  file: '/path/to/file',
  // 模块内容
  source: '',
  // 模块的依赖
  deps: {
    // `require(expr)`
    expr: id,
  }
}

在 wrap 阶段前，所有的阶段都处理这样的对象流，且除 pack 外，都输出这样的流。

有的补充 row 中的一些信息，有的则对这些信息做一些变换，有的只是读取和输出。

一般 row 中的 source 、 deps 内容都是在 deps 阶段解析出来的。

下面提供一个修改 Browserify 管道的函数。

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

事实上，这里的 b.plugin(log) 就是直接执行了 log(b) 。

在插件中，可以修改 b.pipeline 中的任何一个环节。

因此， Browserify 本身只保留了必要的功能，其它都由插件去实现，如 watchify 、 factor-bundle 等。

除了了上述的插件机制外， Browserify 还有一套Transform机制，即通过 b.transform(transform) 可以新增一些文件内容预处理的Transform。

预处理是发生在 deps 阶段的，当模块文件内容被读出来时，会经过这些Transform处理，然后才做依赖解析，如 babelify 、 envify 。

Gulp

Gulp 的核心逻辑分成两块：任务调度与文件处理。

任务调度是基于 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'));
});

任务中创建的管道起始于 gulp.src ，终止于 gulp.dest ，中间有若干其它的Transform（插件）。

如果与 Browserify 的管道对比，可以发现 Browserify 是确定了一条具有完整功能的管道，而Gulp 本身只提供了创建 vinyl 流和将 vinyl 流写入磁盘的工具，管道中间经历什么全由用户决定。

这是因为任务中做什么，是没有任何限制的，文件处理也只是常见的情况，并非一定要用 gulp.src 与 gulp.dest 。

两种模式比较

Browserify 与 Gulp 都借助管道的概念来实现插件机制。

Browserify 定义了模块的数据结构，提供了默认的管道以处理这样的数据流，而插件可用来修改管道结构，以定制处理行为。

Gulp 虽也定义了文件的数据结构，但只提供产生、消耗这种数据流的接口，完全由用户通过插件去构造处理管道。

当明确具体的处理需求时，可以像 Browserify 那样，构造一个基本的处理管道，以提供插件机制。

如果需要的是实现任意功能的管道，可以如 Gulp 那样，只提供数据流的抽象。

实例

本节中实现一个针对 Git 仓库自动生成changelog的工具，完整代码见 ezchangelog 。

ezchangelog 的输入为 git log 生成的文本流，输出默认为markdown格式的文本流，但可以修改为任意的自定义格式。

输入示意：

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

输出示意：

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

其实需要的是这样一个 pipeline ：

source.pipe(pipeline).pipe(dest)

可以分为两个阶段：

  • parse：从输入文本流中解析出commit信息
  • format: 将commit流变换为文本流

默认的情况下，要想得到示例中的markdown，需要解析出每个commit的sha1、日期、消息、是否为tag。

定义commit的格式如下：

{
  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',
}

于是有：

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

至此，基本功能已经实现。

现在将其封装并提供插件机制。

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
}

上面的实现提供了两种方式来应用插件。

一种是通过配置传入，另一种是创建实例后再调用 plugin 方法，本质一样。

为了使用方便，还可以简单封装一下。

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

这样，就可以如下方式使用：

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

这个已经非常接近我们的预期了。

现在来开发一个插件，修改默认的渲染方式。

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)

同样的输入，输出将会是：

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

可以看出，通过创建可修改的管道， ezchangelog 保持了本身逻辑的单一性，同时又提供了强大的自定义空间。

系列文章

  • 第一部分：《Node.js Stream - 基础篇》，介绍Stream接口的基本使用。
  • 第二部分：《Node.js Stream - 进阶篇》，重点剖析Stream底层如何支持流式数据处理，及其back pressure机制。
  • 第三部分：《Node.js Stream - 实战篇》，介绍如何使用Stream进行程序设计。从Browserify和Gulp总结出两种设计模式，并基于Stream构建一个为Git仓库自动生成changelog的应用作为示例。

参考文献

 

来自：http://tech.meituan.com/stream-in-action.html

文章标签：
云解析DNS
JavaScript
开发工具
git
前端开发
数据处理
数据格式
调度
缓存
JSON
设计模式
关键词：
JavaScript实战
node.js Stream
node.js实战
冰荷
目录
相关文章
一缕微风绕指柔
|
2月前
|
自然语言处理 JavaScript 前端开发
深入理解JavaScript中的闭包：原理与实战
【10月更文挑战第12天】深入理解JavaScript中的闭包：原理与实战
一缕微风绕指柔
54 1
白雾茫茫丶
|
22天前
Next.js 实战 (二)：搭建 Layouts 基础排版布局
本文介绍了作者在Next.js v15.x版本发布后，对一个旧项目的重构过程。文章详细说明了项目开发规范配置、UI组件库选择（最终选择了Ant-Design）、以及使用Ant Design的Layout组件实现中后台布局的方法。文末展示了布局的初步效果，并提供了GitHub仓库链接供读者参考学习。
白雾茫茫丶
27 1
Next.js 实战 (二)：搭建 Layouts 基础排版布局
白雾茫茫丶
|
16天前
|
存储 网络架构
Next.js 实战 (四)：i18n 国际化的最优方案实践
这篇文章介绍了Next.js国际化方案，作者对比了网上常见的方案并提出了自己的需求：不破坏应用程序的目录结构和路由。文章推荐使用next-intl库来实现国际化，并提供了详细的安装步骤和代码示例。作者实现了国际化切换时不改变路由，并把当前语言的key存储到浏览器cookie中，使得刷新浏览器后语言不会失效。最后，文章总结了这种国际化方案的优势，并提供Github仓库链接供读者参考。
白雾茫茫丶
42 5
白雾茫茫丶
|
17天前
Next.js 实战 (三)：优雅的实现暗黑主题模式
这篇文章介绍了在Next.js中实现暗黑模式的具体步骤。首先，需要安装next-themes库。然后，在/components/ThemeProvider/index.tsx文件中新增ThemeProvider组件，并在/app/layout.tsx文件中注入该组件。如果想要加入过渡动画，可以修改代码实现主题切换时的动画效果。最后，需要在需要的位置引入ThemeModeButton组件，实现暗黑模式的切换。
白雾茫茫丶
32 1
土木林森
|
1月前
|
设计模式 前端开发 JavaScript
JavaScript设计模式及其在实战中的应用，涵盖单例、工厂、观察者、装饰器和策略模式
本文深入探讨了JavaScript设计模式及其在实战中的应用，涵盖单例、工厂、观察者、装饰器和策略模式，结合电商网站案例，展示了设计模式如何提升代码的可维护性、扩展性和可读性，强调了其在前端开发中的重要性。
土木林森
30 2
cqtianxin1
|
2月前
|
JavaScript
Node.js Stream(流)
10月更文挑战第4天
cqtianxin1
33 1
Node.js Stream(流)
小周sir
|
2月前
|
JavaScript 前端开发 开发者
探索JavaScript原型链：深入理解与实战应用
【10月更文挑战第21天】探索JavaScript原型链：深入理解与实战应用
小周sir
39 1
白雾茫茫丶
|
2月前
|
SQL 前端开发 JavaScript
Nest.js 实战 (十五)：前后端分离项目部署的最佳实践
这篇文章介绍了如何使用现代前端框架Vue3和后端Node.js框架Nest.js实现的前后端分离架构的应用，并将其部署到生产环境。文章涵盖了准备阶段，包括云服务器的设置、1Panel面板的安装、数据库的安装、域名的实名认证和备案、SSL证书的申请。在部署Node服务环节，包括了Node.js环境的创建、数据库的配置、用户名和密码的设置、网站信息的填写、静态网站的部署、反向代理的配置以及可能遇到的常见问题。最后，作者总结了部署经验，并希望对读者有所帮助。
白雾茫茫丶
212 11
代码bug生产队
|
2月前
|
存储 JavaScript 前端开发
前端开发：Vue.js入门与实战
【10月更文挑战第9天】前端开发：Vue.js入门与实战
代码bug生产队
41 1
bruce_xiaowei
|
2月前
|
数据采集 JSON 前端开发
JavaScript逆向爬虫实战分析
JavaScript逆向爬虫实战分析
bruce_xiaowei
44 4

热门文章

最新文章

  • 1
    Node.js 的事件驱动机制
  • 2
    基于 JavaScript/VuePress 搭建的远程工作平台：YuanCheng.works
  • 3
    如何使用内存监控工具来优化 Node.js 应用的性能
  • 4
    html5 + Three.js 3D风雪封印在棱镜中的梅花鹿动效源码
  • 5
    js+jquery实现贪吃蛇经典小游戏
  • 6
    在 golang 中执行 javascript 代码的方案详解
  • 7
    Node.js新作《循序渐进Node.js企业级开发实践》简介
  • 8
    Node.js 的性能
  • 9
    基于SVG的js圆形菜单插件
  • 10
    如何使用内存快照分析工具来分析Node.js应用的内存问题？
  • 1
    JavaScript闭包允许内部函数访问并保留外部函数的变量，即使外部函数执行结束
    56
  • 2
    js设计模式-观察者模式与发布/订阅模式
    74
  • 3
    js设计模式-策略模式与代理模式的应用
    75
  • 4
    深入理解JavaScript中的闭包机制
    62
  • 5
    使用Vue.js构建交互式前端的技术探索
    92
  • 6
    使用Node.js构建实时聊天应用的技术指南
    136
  • 7
    深入理解JavaScript中的事件循环机制
    91
  • 8
    【MongoDB 专栏】MongoDB 的 JavaScript 引擎与脚本执行
    124
  • 9
    JavaScript 数组方法概览
    39
  • 10
    在JavaScript中，函数原型（Function Prototype）是一个特殊的对象
    71

    • 相关课程

    更多
  • Node.js 入门与实战
  • JavaScript入门与实战
  • Node.js 入门教程文档
  • JavaScript 自学手册文档教程

    • 相关电子书

    更多
  • 探究 Node.js 的服务端之路
  • 个推微服务实践 基于OpenResty 和Node.js
  • 沪江基于Node.js大规模应用实践

    • 相关实验场景

    更多
  • Html5和Webpack2：Webpack5打包JS和样式表
  • 前端开发基础6：Node.js和LESS预编译工具
  • 搭建Node.js编程环境
    • 下一篇
    手把手教你白嫖阿里云服务器(免费领服务器)