Node.js Stream - 进阶篇

上篇（基础篇）主要介绍了Stream的基本概念和用法，本篇将深入剖析背后工作原理，重点是如何实现流式数据处理和 back pressure 机制。

目录

本篇介绍 stream 是如何实现流式数据处理的。

数据生产和消耗的媒介

为什么使用流取数据

下面是一个读取文件内容的例子：

const fs = require('fs') fs.readFile(file, function (err, body) {   console.log(body)   console.log(body.toString()) })

但如果文件内容较大，譬如在440M时，执行上述代码的输出为：

<Buffer 64 74 09 75 61 09 63 6f 75 6e 74 0a 0a 64 74 09 75 61 09 63 6f 75 6e 74 0a 32 30 31 35 31 32 30 38 09 4d 6f 7a 69 6c 6c 61 2f 35 2e 30 20 28 63 6f 6d ... > buffer.js:382     throw new Error('toString failed');     ^  Error: toString failed     at Buffer.toString (buffer.js:382:11)

可以考虑使用流来读取文件内容。

const fs = require('fs') fs.createReadStream(file).pipe(process.stdout)

fs.createReadStream 创建一个可读流，连接了源头（上游，文件）和消耗方（下游，标准输出）。

如何通过流取到数据

当 readable 连接了数据源后，下游便可以调用 readable.read(n) 向流请求数据，同时监听 readable 的 data 事件来接收取到的数据。

这个流程可简述为：

read

read 方法中的逻辑可用下图表示，后面几节将对该图中各环节加以说明。

push方法

消耗方调用 read(n) 促使流输出数据，而流通过 _read() 使底层调用 push 方法将数据传给流。

如果流在流动模式下（ state.flowing 为 true ）输出数据，数据会自发地通过 data 事件输出，不需要消耗方反复调用 read(n) 。

所以，如果 _read 异步调用 push 时发现缓存为空，则意味着当前数据是下一个需要的数据，且不会被 read 方法输出，应当在 push 方法中立即以 data 事件输出。

因此，上图中“立即输出”的条件是：

state.flowing && state.length === 0 && !state.sync

end事件

readable事件

doRead

var doRead = state.needReadable  if (state.length === 0 || state.length - n < state.highWaterMark) {   doRead = true }  if (state.ended || state.reading) {   doRead = false }  if (doRead) {   state.reading = true   state.sync = true   if (state.length === 0) {     state.needReadable = true   }   this._read(state.highWaterMark)   state.sync = false }

howMuchToRead

• state.length 为0， state.ended 为 true 。
  数据源已枯竭，且缓存为空，无数据可取， m 为0.
• state.objectMode 为 true 。
  n 为0，则 m 为0；
  否则 m 为1，将缓存的第一个元素输出。
• n 是数字。
  若 n <= 0 ，则 m 为0；
  若 n > state.length ，表示缓存中数据量不够。
  此时如果还有数据可读（ state.ended 为 false ），则 m 为0，同时设置 state.needReadable ，下次执行 read() 时 doRead 会为 true ，将从底层再取数据。
  如果已无数据可读（ state.ended 为 true ），则 m 为 state.length ，将剩下的数据全部输出。
  若 0 < n <= state.length ，则缓存中数据够用， m 为 n 。
• 其它情况。
  state.flowing 为 true （流动模式），则 m 为缓存中第一个元素（ Buffer ）的长度，实则还是将第一个元素输出；
  否则 m 为 state.length ，将缓存读空。

上面的规则中：

• n 通常是 undefined 或 0 ，即不指定读取的字节数。
• read(0) 不会有数据输出，但从前面对 doRead 的分析可以看出，是有可能从底层读取数据的。
• 执行 read() 时，由于流动模式下数据会不断输出，所以每次只输出缓存中第一个元素输出，而非流动模式则会将缓存读空。
• objectMode 为 true 时， m 为 0 或 1 。此时，一次 push() 对应一次 data 事件。

数据的流式消耗

所谓“流式数据”，是指按时间先后到达的数据序列。

数据消耗模式

可以在两种模式下消耗可读流中的数据：暂停模式（paused mode）和流动模式（flowing mode）。

暂停模式下，需要显示地调用 read() ，触发 data 事件。

• true 。流动模式。
• false 。暂停模式。
• null 。初始状态。

暂停模式

const Readable = require('stream').Readable  // 底层数据 const dataSource = ['a', 'b', 'c']  const readable = Readable() readable._read = function () {   if (dataSource.length) {     this.push(dataSource.shift())   } else {     this.push(null)   } }  // 进入暂停模式 readable.pause() readable.on('data', data => process.stdout.write('/ndata: ' + data))  var data = readable.read() while (data !== null) {   process.stdout.write('/nread: ' + data)   data = readable.read() }

执行上面的脚本，输出如下：

data: a read: a data: b read: b data: c read: c

const Readable = require('stream').Readable  // 底层数据 const dataSource = ['a', 'b', 'c']  const readable = Readable() readable._read = function () {   process.nextTick(() => {     if (dataSource.length) {       this.push(dataSource.shift())     } else {       this.push(null)     }   }) }  readable.pause() readable.on('data', data => process.stdout.write('/ndata: ' + data))  while (null !== readable.read()) ;

执行上述脚本，可以发现没有任何数据输出。

此时，需要使用 readable 事件：

const Readable = require('stream').Readable  // 底层数据 const dataSource = ['a', 'b', 'c']  const readable = Readable() readable._read = function () {   process.nextTick(() => {     if (dataSource.length) {       this.push(dataSource.shift())     } else {       this.push(null)     }   }) }  readable.pause() readable.on('data', data => process.stdout.write('/ndata: ' + data))  readable.on('readable', function () {   while (null !== readable.read()) ;; })

输出：

data: a data: b data: c

所以，一旦 read 被调用，上面的例子中就会形成一个循环： readable 事件导致 read 方法调用， read 方法又触发 readable 事件。

首次监听 readable 事件时，还会触发一次 read(0) 的调用，从而引起 _read 和 push 方法的调用，从而启动循环。

总之，在暂停模式下需要使用 readable 事件和 read 方法来消耗流。

流动模式

流动模式使用起来更简单一些。

本节解释 data 事件如何能触发流动模式。

先看一下 Readable 是如何处理 data 事件的监听的：

Readable.prototype.on = function (ev, fn) {   var res = Stream.prototype.on.call(this, ev, fn)   if (ev === 'data' && false !== this._readableState.flowing) {     this.resume()   }    // 处理readable事件的监听   // 省略    return res }

if (state.flowing) do {   var chunk = stream.read() } while (null !== chunk && state.flowing)

下面再详细看一下 push() 的两个分支：

if (state.flowing && state.length === 0 && !state.sync) {   stream.emit('data', chunk)   stream.read(0) } else {   state.length += state.objectMode ? 1 : chunk.length   state.buffer.push(chunk)    if (state.needReadable)     emitReadable(stream) }

称第一个分支为立即输出。

在立即输出的情况下，输出数据后，执行 read(0) ，进一步引起 _read() 和 push() 的调用，从而使数据源源不断地输出。

在非立即输出的情况下，数据先被添加到缓存中。此时有两种情况：

• state.length 为0。
  这时，在调用 _read() 前， state.needReadable 就会被设为 true 。
  因此，一定会调用 emitReadable() 。
  这个方法会在下一个tick中触发 readable 事件，同时再调用 flow() ，从而形成流动。
• state.length 不为0。
  由于流动模式下，每次都是从缓存中取第一个元素，所以这时 read() 返回值一定不为 null 。
  故 flow() 中的循环还在继续。

事实上， pause 方法就是这样使流从流动模式转换到暂停模式的。

背压反馈机制

考虑下面的例子：

const fs = require('fs') fs.createReadStream(file).on('data', doSomething)

pipe

用 pipe 方法连接上下游：

const fs = require('fs') fs.createReadStream(file).pipe(writable)

于是上游可以根据 write() 的返回值在流动模式和暂停模式间切换：

readable.on('data', function (data) {   if (false === writable.write(data)) {     readable.pause()   } })  writable.on('drain', function () {   readable.resume() })

上面便是 pipe 方法的核心逻辑。

看一个例子：

const stream = require('stream')  var c = 0 const readable = stream.Readable({   highWaterMark: 2,   read: function () {     process.nextTick(() => {       var data = c < 6 ? String.fromCharCode(c + 65) : null       console.log('push', ++c, data)       this.push(data)     })   } })  const writable = stream.Writable({   highWaterMark: 2,   write: function (chunk, enc, next) {     console.log('write', chunk)   } })  readable.pipe(writable)

输出：

push 1 A write <Buffer 41> push 2 B push 3 C push 4 D

下面使用 tick-node 将 Readable 的debug信息按tick分组：

⌘ NODE_DEBUG=stream tick-node pipe.js STREAM 18930: pipe count=1 opts=undefined STREAM 18930: resume ---------- TICK 1 ---------- STREAM 18930: resume read 0 STREAM 18930: read 0 STREAM 18930: need readable false STREAM 18930: length less than watermark true STREAM 18930: do read STREAM 18930: flow true STREAM 18930: read undefined STREAM 18930: need readable true STREAM 18930: length less than watermark true STREAM 18930: reading or ended false ---------- TICK 2 ---------- push 1 A STREAM 18930: ondata write <Buffer 41> STREAM 18930: read 0 STREAM 18930: need readable true STREAM 18930: length less than watermark true STREAM 18930: do read ---------- TICK 3 ---------- push 2 B STREAM 18930: ondata STREAM 18930: call pause flowing=true STREAM 18930: pause STREAM 18930: read 0 STREAM 18930: need readable true STREAM 18930: length less than watermark true STREAM 18930: do read ---------- TICK 4 ---------- push 3 C STREAM 18930: emitReadable false STREAM 18930: emit readable STREAM 18930: flow false ---------- TICK 5 ---------- STREAM 18930: maybeReadMore read 0 STREAM 18930: read 0 STREAM 18930: need readable false STREAM 18930: length less than watermark true STREAM 18930: do read ---------- TICK 6 ---------- push 4 D ---------- TICK 7 ----------

• TICK 0: readable.resume()
• TICK 1: readable 在流动模式下开始从底层读取数据
• TICK 2: A 被输出，同时执行 readable.read(0) 。
• TICK 3: B 被输出，同时执行 readable.read(0) 。
  writable.write('B') 返回 false 。
  执行 readable.pause() 切换至暂停模式。
• TICK 4: TICK 3中 read(0) 引起 push('C') 的调用， C 被加到 readable 缓存中。
  此时， writable 中有 A 和 B ， readable 中有 C 。
  这时已在暂停模式，但在 readable.push('C') 结束前，发现缓存中只有1个数据，小于设定的 highWaterMark （2），故准备在下一个tick再读一次数据。
• TICK 5: 调用 read(0) 从底层取数据。
• TICK 6: push('D') ， D 被加到 readable 缓存中。
  此时， writable 中有 A 和 B ， readable 中有 C 和 D 。
  readable 缓存中有2个数据，等于设定的 highWaterMark （2），不再从底层读取数据。

消耗驱动的数据生产

用喝饮料来说明拉式流和普通流的区别的话，普通流就像是将杯子里的饮料往嘴里倾倒，动力来源于上游，数据是被推往下游的；拉式流则是用吸管去喝饮料，动力实际来源于下游，数据是被拉去下游的。

当使用 Transform 作为下游时，尤其需要注意消耗。

const stream = require('stream')  var c = 0 const readable = stream.Readable({   highWaterMark: 2,   read: function () {     process.nextTick(() => {       var data = c < 26 ? String.fromCharCode(c++ + 97) : null       console.log('push', data)       this.push(data)     })   } })  const transform = stream.Transform({   highWaterMark: 2,   transform: function (buf, enc, next) {     console.log('transform', buf)     next(null, buf)   } })  readable.pipe(transform)

以上代码执行结果为：

push a transform <Buffer 61> push b transform <Buffer 62> push c push d push e push f

可见，并没有将26个字母全生产出来。

Transform 中有两个缓存：可写端的缓存和可读端的缓存。

此时，由前面一节的分析可知， readable 将缓存 ef ，之后便不再生产数据。

这三个缓存加起来的长度恰好为6，所以一共就生产了6个数据。

readable.pipe(transform).pipe(process.stdout)

第二种是，不要将数据存入可读端中，这样可读端的缓存便会一直处于数据不足状态，上游便会源源不断地生产数据：

const transform = stream.Transform({   highWaterMark: 2,   transform: function (buf, enc, next) {     next()   } })

参考文献

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