流（stream）是Nodejs中出处理流式数据的抽象接口。stream模块提供了一些API，用于构建实现了流接口的对象。
Node.js提供了多种流对象。例如，HTTP服务器的请求和process.stdout都是流的实例。
流可以是可读的，可写的，或者是可读可写的。所有的流都是EventEmitter的实例
stream模块可以通过以下方式使用

const stream = require('stream')

尽管理解流的工作方式很重要，但是stream模块本身主要用于开发者创建新类型的流实例，对于以消费对象为主的开发者，极少需要直接使用stream模块

流的类型

• Writable 可写流
• Readable 可读流
• Duplex 双公流，可读又可写流
• Transform 在读写过程中可以修改或者转换数据的Duplex

对象模式

node创建的流都是运作在字符串和Buffer（或者uint8Array）上，当然，流的实现也可以使用其他的类型的JavaScript值（除了null）。这些流会以“对象模式”进行操作。当创建流时，可以使用objectMode 选项把流实例切换到对象模式。将已存在的流切换到对象模式是不安全的

用于消费流的API

几乎所有的Nodejs应用都在某种程度上使用了流。
比如：

const http = require('http');

const server = http.createServer((req, res) => {
  // req 是一个 http.IncomingMessage 实例，它是可读流。
  // res 是一个 http.ServerResponse 实例，它是可写流。

  let body = '';
  // 接收数据为 utf8 字符串，
  // 如果没有设置字符编码，则会接收到 Buffer 对象。
  req.setEncoding('utf8');

  // 如果添加了监听器，则可读流会触发 'data' 事件。
  req.on('data', (chunk) => {
    body += chunk;
  });

  // 'end' 事件表明整个请求体已被接收。 
  req.on('end', () => {
    try {
      const data = JSON.parse(body);
      // 响应信息给用户。
      res.write(typeof data);
      res.end();
    } catch (er) {
      // json 解析失败。
      res.statusCode = 400;
      return res.end(`错误: ${er.message}`);
    }
  });
});

server.listen(1337);

// $ curl localhost:1337 -d "{}"
// object
// $ curl localhost:1337 -d "\"foo\""
// string
// $ curl localhost:1337 -d "not json"
// 错误: Unexpected token o in JSON at position 1

可写流（比如例子中的 res）会暴露了一些方法，比如 write() 和 end() 用于写入数据到流。
当数据可以从流读取时，可读流会使用 EventEmitter API 来通知应用程序。 从流读取数据的方式有很多种。
可写流和可读流都通过多种方式使用 EventEmitter API 来通讯流的当前状态。
Duplex 流和 Transform 流都是可写又可读的。
对于只需写入数据到流或从流消费数据的应用程序，并不需要直接实现流的接口，通常也不需要调用 require('stream')。

可写流

可写流是对数据要被写入的目的地的一种抽象。

可写流的例子包括：

• 客户端的 HTTP 请求
• 服务器的 HTTP 响应
• fs 的写入流
• zlib 流
• crypto 流
• TCP socket
• 子进程 stdin
• process.stdout、process.stderr

上面的一些例子事实上是实现了可写流接口的 Duplex 流。

所有可写流都实现了 stream.Writable 类定义的接口。

可读流

可读流的例子包括：

• 客户端的 HTTP 响应
• 服务器的 HTTP 请求
• fs 的读取流
• zlib 流
• crypto 流
• TCP socket
• 子进程 stdout 与 stderr
• process.stdin

所有可读流都实现了 stream.Readable 类定义的接口。

可写流的例子

下面的例子是一个简单的可写流的实现

const {Writable} = require('stream')
class MyWritable extends Writable {
    constructor (options) {
      super(options);
      // todo
    }
  _write(chunk,encoding,callback){
    if(chunk.toString().index('a') >= 0){
         callback(new Error('无效的数据块'))
    }else {
      callback();
    }
  }
}

在可写流中解码buffer

const { Writable } = require('stream');
const { StringDecoder } = require('string_decoder');

class StringWritable extends Writable {
  constructor(options) {
    super(options);
    this._decoder = new StringDecoder(options && options.defaultEncoding);
    this.data = '';
  }
  _write(chunk, encoding, callback) {
    if (encoding === 'buffer') {
      chunk = this._decoder.write(chunk);
    }
    this.data += chunk;
    callback();
  }
  _final(callback) {
    this.data += this._decoder.end();
    callback();
  }
}

const euro = [[0xE2, 0x82], [0xAC]].map(Buffer.from);
const w = new StringWritable();

w.write('货币: ');
w.write(euro[0]);
w.end(euro[1]);

console.log(w.data); // 货币: €

可读流的例子

class Counter extends stream.Readable {
  constructor(opt){
    super(opt)
    this._max = 1000000
    this._index = 1
  }
  _read(){
    const i = this._index++
    if(i > this._max){
      this.push(null)
    }else{
      const str = String(i)
      const buf = Buffer.from(str,'ascii')
      this.push(buf)
    }
  }
}

双工流的例子

下面是一个双工流的例子，封装了一个可读可写的底层资源对象。

const { Duplex } = require('stream');
const kSource = Symbol('source');

class MyDuplex extends Duplex {
  constructor(source, options) {
    super(options);
    this[kSource] = source;
  }

  _write(chunk, encoding, callback) {
    // 底层资源只处理字符串。
    if (Buffer.isBuffer(chunk))
      chunk = chunk.toString();
    this[kSource].writeSomeData(chunk);
    callback();
  }

  _read(size) {
    this[kSource].fetchSomeData(size, (data, encoding) => {
      this.push(Buffer.from(data, encoding));
    });
  }
}

在下面的例子中，创建了一个变换流（双工流的一种），对象模式的可写端接收 JavaScript 数值，并在可读端转换为十六进制字符串。

const { Transform } = require('stream');

// 转换流也是双工流。
const myTransform = new Transform({
  writableObjectMode: true,

  transform(chunk, encoding, callback) {
    // 强制把 chunk 转换成数值。
    chunk |= 0;

    // 将 chunk 转换成十六进制。
    const data = chunk.toString(16);

    // 推送数据到可读队列。
    callback(null, '0'.repeat(data.length % 2) + data);
  }
});

myTransform.setEncoding('ascii');
myTransform.on('data', (chunk) => console.log(chunk));

myTransform.write(1);
// 打印: 01
myTransform.write(10);
// 打印: 0a
myTransform.write(100);
// 打印: 64