深圳市网站建设科技,微网站建设报价方案模板,大型建站网站,阜阳市住房和城乡建设部网站背景 在构建较复杂的系统时#xff0c;通常将其拆解为功能独立的若干部分。这些部分的接口遵循一定的规范#xff0c;通过某种方式相连#xff0c;以共同完成较复杂的任务。譬如#xff0c;shell通过管道|连接各部分#xff0c;其输入输出的规范是文本流。 在Node.js中通常将其拆解为功能独立的若干部分。这些部分的接口遵循一定的规范通过某种方式相连以共同完成较复杂的任务。譬如shell通过管道|连接各部分其输入输出的规范是文本流。 在Node.js中内置的Stream模块也实现了类似功能各部分通过.pipe()连接。 鉴于目前国内系统性介绍Stream的文章较少而越来越多的开源工具都使用了Stream本系列文章将从以下几方面来介绍相关内容 流的基本类型以及Stream模块的基本使用方法流式处理与back pressure的工作原理如何开发流式程序包括对Gulp与Browserify的剖析以及一个实战示例。 本文为系列文章的第一篇。 流的四种类型 Stream提供了以下四种类型的流 var Stream require(stream)var Readable Stream.Readable
var Writable Stream.Writable
var Duplex Stream.Duplex
var Transform Stream.Transform使用Stream可实现数据的流式处理如 var fs require(fs)
// fs.createReadStream创建一个Readable对象以读取bigFile的内容并输出到标准输出
// 如果使用fs.readFile则可能由于文件过大而失败
fs.createReadStream(bigFile).pipe(process.stdout)Readable 创建可读流。 实例流式消耗迭代器中的数据。 use strict
const Readable require(stream).Readableclass ToReadable extends Readable {constructor(iterable) {super()this.iterator new function *() {yield * iterable}}// 子类需要实现该方法// 这是生产数据的逻辑_read() {const res this.iterator.next()if (res.done) {// 数据源已枯竭调用push(null)通知流this.push(null)} else {// 通过push方法将数据添加到流中this.push(res.value \n)}}
}module.exports ToReadable实际使用时new ToReadable(iterable)会返回一个可读流下游可以流式的消耗迭代器中的数据。 const iterable function *(limit) {while (limit--) {yield Math.random()}
}(1e10)const readable new ToReadable(iterable)// 监听data事件一次获取一个数据
readable.on(data, data process.stdout.write(data))// 所有数据均已读完
readable.on(end, () process.stdout.write(DONE))执行上述代码将会有100亿个随机数源源不断地写进标准输出流。 创建可读流时需要继承Readable并实现_read方法。 _read方法是从底层系统读取具体数据的逻辑即生产数据的逻辑。在_read方法中通过调用push(data)将数据放入可读流中供下游消耗。在_read方法中可以同步调用push(data)也可以异步调用。当全部数据都生产出来后必须调用push(null)来结束可读流。流一旦结束便不能再调用push(data)添加数据。 可以通过监听data事件的方式消耗可读流。 在首次监听其data事件后readable便会持续不断地调用_read()通过触发data事件将数据输出。第一次data事件会在下一个tick中触发所以可以安全地将数据输出前的逻辑放在事件监听后同一个tick中。当数据全部被消耗时会触发end事件。 上面的例子中process.stdout代表标准输出流实际是一个可写流。下小节中介绍可写流的用法。 Writable 创建可写流。 前面通过继承的方式去创建一类可读流这种方法也适用于创建一类可写流只是需要实现的是_write(data, enc, next)方法而不是_read()方法。 有些简单的情况下不需要创建一类流而只是一个流对象可以用如下方式去做 const Writable require(stream).Writableconst writable Writable()
// 实现_write方法
// 这是将数据写入底层的逻辑
writable._write function (data, enc, next) {// 将流中的数据写入底层process.stdout.write(data.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()上游通过调用writable.write(data)将数据写入可写流中。write()方法会调用_write()将data写入底层。在_write中当数据成功写入底层后必须调用next(err)告诉流开始处理下一个数据。 next的调用既可以是同步的也可以是异步的。上游必须调用writable.end(data)来结束可写流data是可选的。此后不能再调用write新增数据。在end方法调用后当所有底层的写操作均完成时会触发finish事件。 Duplex 创建可读可写流。 Duplex实际上就是继承了Readable和Writable的一类流。 所以一个Duplex对象既可当成可读流来使用需要实现_read方法也可当成可写流来使用需要实现_write方法。 var Duplex require(stream).Duplexvar duplex Duplex()// 可读端底层读取逻辑
duplex._read function () {this._readNum this._readNum || 0if (this._readNum 1) {this.push(null)} else {this.push( (this._readNum))}
}// 可写端底层写逻辑
duplex._write function (buf, enc, next) {// a, bprocess.stdout.write(_write buf.toString() \n)next()
}// 0, 1
duplex.on(data, data console.log(ondata, data.toString()))duplex.write(a)
duplex.write(b)duplex.end()上面的代码中实现了_read方法所以可以监听data事件来消耗Duplex产生的数据。 同时又实现了_write方法可作为下游去消耗数据。 因为它既可读又可写所以称它有两端可写端和可读端。 可写端的接口与Writable一致作为下游来使用可读端的接口与Readable一致作为上游来使用。 Transform 在上面的例子中可读流中的数据0, 1与可写流中的数据a, b是隔离开的但在Transform中可写端写入的数据经变换后会自动添加到可读端。Tranform继承自Duplex并已经实现了_read和_write方法同时要求用户实现一个_transform方法。 use strictconst Transform require(stream).Transformclass Rotate extends Transform {constructor(n) {super()// 将字母旋转n个位置this.offset (n || 13) % 26}// 将可写端写入的数据变换后添加到可读端_transform(buf, enc, next) {var res buf.toString().split().map(c {var code c.charCodeAt(0)if (c a c z) {code this.offsetif (code z.charCodeAt(0)) {code - 26}} else if (c A c Z) {code this.offsetif (code Z.charCodeAt(0)) {code - 26}}return String.fromCharCode(code)}).join()// 调用push方法将变换后的数据添加到可读端this.push(res)// 调用next方法准备处理下一个next()}}var transform new Rotate(3)
transform.on(data, data process.stdout.write(data))
transform.write(hello, )
transform.write(world!)
transform.end()// khoor, zruog!objectMode 前面几节的例子中经常看到调用data.toString()。这个toString()的调用是必需的吗 本节介绍完如何控制流中的数据类型后自然就有了答案。 在shell中用管道|连接上下游。上游输出的是文本流标准输出流下游输入的也是文本流标准输入流。在本文介绍的流中默认也是如此。 对于可读流来说push(data)时data只能是String或Buffer类型而消耗时data事件输出的数据都是Buffer类型。对于可写流来说write(data)时data只能是String或Buffer类型_write(data)调用时传进来的data都是Buffer类型。 也就是说流中的数据默认情况下都是Buffer类型。产生的数据一放入流中便转成Buffer被消耗写入的数据在传给底层写逻辑时也被转成Buffer类型。 但每个构造函数都接收一个配置对象有一个objectMode的选项一旦设置为true就能出现“种瓜得瓜种豆得豆”的效果。 Readable未设置objectMode时 const Readable require(stream).Readableconst readable Readable()readable.push(a)
readable.push(b)
readable.push(null)readable.on(data, data console.log(data))输出 Buffer 61
Buffer 62Readable设置objectMode后 const Readable require(stream).Readableconst readable Readable({ objectMode: true })readable.push(a)
readable.push(b)
readable.push({})
readable.push(null)readable.on(data, data console.log(data))输出 a
b
{}可见设置objectMode后push(data)的数据被原样地输出了。此时可以生产任意类型的数据。 系列文章 第一部分《Node.js Stream - 基础篇》介绍Stream接口的基本使用。第二部分《Node.js Stream - 进阶篇》重点剖析Stream底层如何支持流式数据处理及其back pressure机制。第三部分《Node.js Stream - 实战篇》介绍如何使用Stream进行程序设计。从Browserify和Gulp总结出两种设计模式并基于Stream构建一个为Git仓库自动生成changelog的应用作为示例。 参考文献 GitHubsubstack/browserify-handbook GitHubzoubin/streamify-your-node-program 来自http://tech.meituan.com/stream-basics.html
