Stream in NodeJS

数据块： 每次读入或写入的一段数据，可以是任何类型。

读取流: 代表数据源的模型，数据的来源。
潜在资源(underlying source)：读取流所封装的底层I/O源
推送源(push sources): 无论如何都在将数据推出的源。也许提供了暂停和恢复的功能。
拉取源(pull sources): 需要有消费数据的对象从他们中请求数据的源。
队列(queue): 潜在资源将数据块推入流内部队列，这些数据块可以通过流的公共接口一次一个读出。
消费者(Consumer): 使用读取流的公共接口从其读取数据的程序。
取消(cancel): 消费者能够取消读取流。当其不再对流中数据有兴趣，立即关闭流，放弃任何队列中的数据块，执行任何潜在资源的取消机制。
分流(tee): 消费者能够复制读取流。锁定流使其不再直接可用，创建两个新的能独立消费的分支(branches)。

写入流： 代表数据的目的地，数据的流向处。

潜在汇入(underlying sink)： 写入流所封装的底层I/O汇入点。写入流通过将之后的写入缓存到队列中，一次传递一个给潜在汇入，来抽象底层汇入的复杂实现。数据块被一次一个地通过写入流的公共接口传递给潜在汇入。

供给者(producer): 通过使用写入流的公共接口将数据写入的代码为供给者。

终止(abort): 供给者有能力终止写入流。如果供给者认为有错误，任何之后的写入都应该被禁止，可以将写入流转换到错误状态而无需潜在汇入的信号。

转换流包含一对流：一个写入流，一个读取流。数据写入写入流，处理转换后，将可以从读取流读取新数据。

管道: 流一个接一个好像管道一样连接起来，比如一个读取流连接上写入流。这些流可以一个接一个连接起来。

管道链： 一系列像管道一样连接的流成为管道链。

初始源(original source): 是链中第一个读取流所代表的潜在资源。

终点汇入(ultimate sink): 是链中最后一个写入流代表的潜在资源。

反压: 一旦管道链建立起来，就可以传播流速信号。如果任何链中环节还不能接受数据块，就将该信号反向传回初始源，告知其不要这么快生成数据。这种根据链中数据处理速度将信号反馈给初始源的过程叫做反压。

如果分裂一个读取流，两个分支的反压信号将聚合，任何一个分支有反压信号将被传送给读取流的潜在资源。

读取流和写入流都维护一个内部队列(internal queues)。他们有相似的目的。在读取流中，内部队列缓存了潜在来源推入队列的数据。在写入流中，内部队列缓存了将要传递给潜在汇入还未来得及被其处理的数据。

队列策略(queuing strategy)是一种决定流如何根据它的内部状态给出反压信号的策略。队列策略赋给每个数据块一个尺寸，比较队列中所有数据块总体尺寸和一个特殊的叫做水平标记(high water mark)的值做比较。结果之差，即水平标记减去数据块总尺寸，用来决定填入流队列中意向尺寸(desired size to fill the stream's queue。

流意向读取的尺寸 = 水平标记 - 队列中已有数据块总大小

对读取流来说，一个潜在来源可以使用该意向作为反压信号，减慢数据块生成速度来保持意向读取尺寸大于等于零。

对写入流来说，供给源可以类似避免写入速度过快。

简单例子，如果每个数据块尺寸为1，水平标志为3，在流考虑应用反压之前流中队列最多有三个数据块。

读取流读者(readable stream reader)是允许直接从读取流中读取数据块的对象。没有这个对象，消费者只能对读取流进行高级操作：等待流关闭或者出错，取消流，用管道将读取流连接上写入流。许多这些高级操作实际自身使用流读者。

给定的流一次只能有一个读者。叫做流被锁定到读者(locked to the reader)，读者被成为激活的(active)。

读者也有能力释放它的锁(release its read lock)，让其不再激活。其他读者可以自由获取这个锁。如果流因为潜在来源或这取消而关闭或出错，读者将自动释放锁。

• 延迟处理缓冲的数据
• 时间驱动非阻塞
• 内存消耗低
• 自动 back-pressure 处理
• 突破 V8 堆内存限制
• NodeJS 核心结构

五个类

• Readable: 读出
• Writable: 写入
• Duplex: 双向流
• Transform: 流变形
• Passthrough: 流间谍

Streams 自动处理缓冲和 backpressure

Stream是EventEmitter。要么监听=error=事件，要么就让它们冒泡

最早的Stream是从http.js和net.js模块抽象出来，说明其源头正是因处理网络而生。参见Add Stream Base class with stream.pipe, ry committed on 11 Oct 2010

这时候的Stream是朴素的基础类，仅仅从EventEmitter继承而来，有一个=pipe=方法，实现从src读取数据写入dest的功能。

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  
16  

function Stream () {  
 events.EventEmitter.call(this);  
}  
inherits(Stream, events.EventEmitter);  
exports.Stream = Stream;  
  
Stream.prototype.pipe = function (dest, options) {  
 var source = this;  
  
source.on("data", function (chunk) {  
 if (false === dest.write(chunk)) source.pause();  
});  
  
dest.on("drain", function () {  
 if (source.readable) source.resume();  
});  

pipe实现了一种简单的流量管理。有数据时，就往dest写入。而dest.write如果返回false就表示dest无法及时处理数据，将source暂停(pause)。一旦dest的缓冲区为空，drain事件触发，如果src是可读的，就恢复(resume)src。看上去似乎很好。

接下来，几经修修改改，增增减减，没有太大变化。

这时的Stream有四个特点：

• =pause=是建议性质的，并不一定真能暂停
• =data=事件不管你准备没准备好都会发生
• 没法处理特定数目的数据然后将剩下的数据交给其他部分处理
• 实现流难到极点，需要顾及缓冲、暂停、恢复、数据事件、状态等等。缺少共享的类致使反复解决同样的问题，产生同样的错误和相似的bug。

两年后，Node 0.10 发布，Stream 2 应运而生。参见streams2: The new stream base classes , isaacs committed on 3 Oct 2012

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  

util.inherits(Stream, events.EventEmitter);  
module.exports = Stream;  
Stream.Readable = require('_stream_readable');  
Stream.Writable = require('_stream_writable');  
Stream.Duplex = require('_stream_duplex');  
Stream.Transform = require('_stream_transform');  
Stream.PassThrough = require('_stream_passthrough');  
  
// Backwards-compat with node 0.4.x  
// Backwards-compat with node 0.4.x  
Stream.Stream = Stream;  
Stream.Stream = Stream;  

提供了五种基本流类，还提供原始的Stream以兼容旧版。这几种基本类提供了强大的抽象，但也给nodejs的核心模块带来了巨大的复杂度。Stream一跃成为NodeJS核心模块中最复杂的部分之一。以至于社区内有很多人认为，应该将这些模块分离维护以维持NodeJS较小的稳定核心。

但不管怎么说，托Stream 2的福，用户实现流简单很多。Stream 2的流API也着重解决Stream 1 的一些问题。将Push stream改成Pull stream。

之后，Stream2全面更新NodeJS其他核心模块。在实践中增增减减。稳定成如下形式：

当流的缓冲队列内有数据时会触发=readable=事件，流通过=read=读取潜在资源推入缓冲队列中的数据块。这样，就可以指定想要读取的数据大小。

为了兼容，可以通过=data=事件继续使用Stream 1的api，当监听=data=事件时会取消流的暂停状态。

stream2还有很多细节。NodeJS团队为了更优雅的流量控制和更高的效率，不断锤炼Stream 2的设计和实现。接口和设计频繁变化，lowWaterMark特性被移除，push和unshift被加入，\_read()不再接受回调，read(0)频繁变化，实现自动扩展的highWaterMark、添加ObjectMode等等，同时其他使用stream的核心模块也渐渐使用新的stream类。

比较重要的API倒不多，比如readable stream除了=stream.read()=。有当流想要拉取更多数据时调用的=stream._read=(实际上就是read调用的)。有操作缓冲队列的=stream.push=和=stream.unshift=。

这时的stream，push和pull stream两种模式不能同时使用，=read()=时并不会触发相应的=data=事件。

修修改改修修改改。

2014年12月，贡献者因对NodeJS开发公司Joyent在NodeJS管理上的长期不满，社区分裂。15年1月，iojs发布。

Summary of changes from Node.js v0.10.35 to io.js v1.0.0里写道：这次流的变化没有从streams1倒streams2这么大：他们都是从现存的想法中提炼出来的，这些设计应该让人类更少意外，让计算机能更快处理。最终所有这些改变, 成为了stream3，但大部分改变将不会被流消费者和实现者察觉。

也就是说，Streams 3是两者的结合，两者将更加一致与和谐。比如读取流不再只能从非flowing状态转移到flowing状态而不可逆，=read()=也将触发=data=事件。

Changelog里一副图完整描绘了Stream 3

修修补补修修补补。

stream的API仍然非常复杂，社区创造了through等来简化流的创建。终于，官方实现了简化版的流构建(construction)

1  
2  

if (options && typeof options.read === 'function')  
 this._read = options.read;  

就是上面这种黑魔法。。。

顺便说到黑魔法，isaac实现的扩展highWaterMarkstream: Raise readable high water mark in powers of 2 。

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  

function roundUpToNextPowerOf2(n) {  
 if (n >= MAX_HWM) {  
 n = MAX_HWM;  
 } else {  
 // Get the next highest power of 2  
 n--;  
 for (var p = 1; p < 32; p <<= 1) n |= n >> p;  
 n++;  
 }  
 return n;  
}  

2015年8月21日，更新了这么一个微优化。。。第一眼看去我擦咧这是啥？！直觉似乎hackers delight里会有这种东西，果然。。。这也是神奇的算法。

1  
2  
3  
4  
5  
6  
7  
8  
9  
10  
11  
12  
13  
14  
15  

function computeNewHighWaterMark(n) {  
 if (n >= MAX_HWM) {  
 n = MAX_HWM;  
 } else {  
 // Get the next highest power of 2  
 n--;  
 n |= n >>> 1;  
 n |= n >>> 2;  
 n |= n >>> 4;  
 n |= n >>> 8;  
 n |= n >>> 16;  
 n++;  
 }  
 return n;  
}  

eventEmitter实现里关于函数参数也有些优化黑魔法，如果哪天写那个再说吧。

之后stream的接口和实现趋于稳定，修修改改提升效率，更新文档，专注应用。