流在Node.js中是处理流数据的抽象接口。stream模块提供了基础的API。使用这些API可以很容易地来构建实现流接口的对象。 Node.js里提供了很多流对象。例如http.IncomingMessage类、fs.createReadStream类等等。都继承流的私有属性和公有方法。 所以学习流，有助于学习Node的其他模块。

文章结构

• 简介stream
• pipe源码

stream

const EE = require('events');
const util = require('util');
function Stream() {EE.call(this);
}
util.inherits(Stream, EE);
复制代码

• Stream继承EventEmitter。流可以是可读的、可写的，或是可读写的。
• Stream分为Readable（可读流）、Writable（可写流）、Duplex（可读写流）、Transform（读写过程中可以修改和变换数据的 Duplex 流）。

pipe

Stream.prototype.pipe = function(dest, options){var source = this;source.on('data', ondata);dest.on('drain', ondrain);if (!dest._isStdio && (!options || options.end !== false)) {source.on('end', onend);source.on('close', onclose);}source.on('error', onerror);dest.on('error', onerror);source.on('end', cleanup);source.on('close', cleanup);  dest.on('close', cleanup);dest.emit('pipe', source);return dest;
};
复制代码

• Stream公有方法pipe
  • source是可读流：Readable。dest是可写流：Writable。
  • Readable.pipe(Writable) 。
  • Readable订阅事件：data、error、end、close。Readable接收到事件执行相应的方法。
  • Writable订阅事件：drain、error、close，并发布pipe事件。Writable接收到事件执行相应的方法。
  • 返回Writable。

ondata

function ondata(chunk) {if (dest.Writable) {if (false === dest.write(chunk) && source.pause) {source.pause();}}}
复制代码

• Readable订阅data事件。
• Readable触发data事件，表示读入数据。
• dest.Writable当写完时会赋值为false。
• 如果读的太快，没有写完dest.write(chunk)返回false。
• source.pause暂停写入。
• 总结：订阅data事件，触发ondata方法，如果Readable读入数据太快，来不及写入，要暂停读入数据。

ondrain

function ondrain() {if (source.Readable && source.resume) {source.resume();}}
复制代码

• Writable订阅drain事件。
• Writable触发drain事件，表示这时才可以继续向流中写入数据。
• source.Readable在读到末尾时会赋值为false。
• source.resume()表示会重新触发Writable的data事件。
• 总结：订阅drain事件，表示这时才可以继续向流中写入数据，调用source.resume()，触发Writable的data事件。

cleanup

 function cleanup() {source.removeListener('data', ondata);dest.removeListener('drain', ondrain);source.removeListener('end', onend);source.removeListener('close', onclose);source.removeListener('error', onerror);dest.removeListener('error', onerror);source.removeListener('end', cleanup);source.removeListener('close', cleanup);dest.removeListener('close', cleanup);}复制代码

• 移除Writable和Readable订阅的事件。

error

function onerror(er) {cleanup();if (EE.listenerCount(this, 'error') === 0) {throw er; // Unhandled stream error in pipe.}}
复制代码

• Writable和Readable有错误是执行的方法。

options

if (!dest._isStdio && (!options || options.end !== false)) {source.on('end', onend);source.on('close', onclose);
}
复制代码

• dest._isStdio 暂时没理解。
• 如果不传options或者options.end不是false，给Readable订阅end和close事件。

onend

var didOnEnd = false;function onend() {if (didOnEnd) return;didOnEnd = true;dest.end();}
复制代码

• 当Readable触发end事件时，执行dest.end()，停止写入。

onclose

function onclose() {if (didOnEnd) return;didOnEnd = true;if (typeof dest.destroy === 'function') dest.destroy();}复制代码

• 当Readable触发close事件后，该流将不会再触发任何事件。
• dest.destroy()摧毁这个流，并发出传过来的错误。当这个函数被调用后，这个写入流就结束了。

小记

• pipe方法大白话：读东西，写东西，读快了，来不及写，暂停读，来得及写了，再读东西，再写。。。
• 流在Node.js中应用广泛，http、文件、打包工具等等。可见流的重要性。
• 学习源码，有助于理解底层的实现。

参考

• Stream
• pipe