流是基于事件的 API，用于管理和处理数据。

• 流是能够读写的
• 是基于事件实现的一个实例

理解流的最好方式就是想象一下没有流的时候怎么处理数据：

• fs.readFileSync同步读取文件，程序会阻塞，所有数据被读到内存
• fs.readFile阻止程序阻塞，但仍会将文件所有数据读取到内存中
• 希望少内存读取大文件，读取一个数据块到内存处理完再去索取更多的数据

流的类型

• 内置：许多核心模块都实现了流接口，如fs.createReadStream
• HTTP：处理网络技术的流
• 解释器：第三方模块 XML、JSON 解释器
• 浏览器：Node 流可以被拓展使用在浏览器
• Audio：流接口的声音模块
• RPC（远程调用）：通过网络发送流是进程间通信的有效方式
• 测试：使用流的测试库

使用内建流 API

静态 web 服务器

想要通过网络高效且支持大文件的发送一个文件到一个客户端。

不使用流

const http = require('http'); const fs = require('fs');

http.createServer((req, res) => {
  fs.readFile(`${__dirname}/index.html`, (err, data) => { if (err) {
      res.statusCode = 500;
      res.end(String(err)); return;
    }

    res.end(data);
  });
}).listen(8000); 复制代码

使用流

const http = require('http'); const fs = require('fs');

http.createServer((req, res) => {
  fs.createReadStream(`${__dirname}/index.html`).pipe(res);
}).listen(8000); 复制代码

• 更少代码，更加高效
• 提供一个缓冲区发送到客户端

使用流 + gzip

const http = require('http'); const fs = require('fs'); const zlib = require('zlib');

http.createServer((req, res) => {
  res.writeHead(200, { 'content-encoding': 'gzip',
  });
  fs.createReadStream(`${__dirname}/index.html`)
    .pipe(zlib.createGzip())
    .pipe(res);
}).listen(8000); 复制代码

流的错误处理

const fs = require('fs'); const stream = fs.createReadStream('not-found');

stream.on('error', (err) => { console.trace(); console.error('Stack:', err.stack); console.error('The error raised was:', err);
}); 复制代码

使用流基类

可读流 - JSON 行解析器

可读流被用来为 I/O 源提供灵活的 API，也可以被用作解析器：

• 继承自 steam.Readable 类
• 并实现一个_read(size)方法

json-lines.txt

{ "position": 0, "letter": "a" }
{ "position": 1, "letter": "b" }
{ "position": 2, "letter": "c" }
{ "position": 3, "letter": "d" }
{ "position": 4, "letter": "e" }
{ "position": 5, "letter": "f" }
{ "position": 6, "letter": "g" }
{ "position": 7, "letter": "h" }
{ "position": 8, "letter": "i" }
{ "position": 9, "letter": "j" } 复制代码

JSONLineReader.js

const stream = require('stream'); const fs = require('fs'); const util = require('util'); class JSONLineReader extends stream.Readable { constructor(source) { super(); this._source = source; this._foundLineEnd = false; this._buffer = '';

    source.on('readable', () => { this.read();
    });
  } // 所有定制 stream.Readable 类都需要实现 _read 方法 _read(size) { let chunk; let line; let result; if (this._buffer.length === 0) {
      chunk = this._source.read(); this._buffer += chunk;
    } const lineIndex = this._buffer.indexOf('\n'); if (lineIndex !== -1) {
      line = this._buffer.slice(0, lineIndex); // 从 buffer 的开始截取第一行来获取一些文本进行解析 if (line) {
        result = JSON.parse(line); this._buffer = this._buffer.slice(lineIndex + 1); this.emit('object', result); // 当一个 JSON 记录解析出来的时候，触发一个 object 事件 this.push(util.inspect(result)); // 将解析好的 SJON 发回内部队列 } else { this._buffer = this._buffer.slice(1);
      }
    }
  }
} const input = fs.createReadStream(`${__dirname}/json-lines.txt`, { encoding: 'utf8',
}); const jsonLineReader = new JSONLineReader(input); // 创建一个 JSONLineReader 实例，传递一个文件流给它处理 jsonLineReader.on('object', (obj) => { console.log('pos:', obj.position, '- letter:', obj.letter);
}); 复制代码

可写流 - 文字变色

可写的流可用于输出数据到底层 I/O:

• 继承自stream.Writable
• 实现一个_write方法向底层源数据发送数据

cat json-lines.txt | node stram_writable.js 复制代码

stram_writable.js

const stream = require('stream'); class GreenStream extends stream.Writable { constructor(options) { super(options);
  }

  _write(chunk, encoding, cb) {
    process.stdout.write(`\u001b[32m${chunk}\u001b[39m`);
    cb();
  }
}

process.stdin.pipe(new GreenStream()); 复制代码

双工流 - 接受和转换数据

双工流允许发送和接受数据：

• 继承自stream.Duplex
• 实现_read和_write方法

转换流 - 解析数据

使用流改变数据为另一种格式，并且高效地管理内存：

• 继承自stream.Transform
• 实现_transform方法

测试流

使用 Node 内置的断言模块测试

const assert = require('assert'); const fs = require('fs'); const CSVParser = require('./csvparser'); const parser = new CSVParser(); const actual = [];

fs.createReadStream(`${__dirname}/sample.csv`)
  .pipe(parser);

process.on('exit', function () {
  actual.push(parser.read());
  actual.push(parser.read());
  actual.push(parser.read()); const expected = [
    { name: 'Alex', location: 'UK', role: 'admin' },
    { name: 'Sam', location: 'France', role: 'user' },
    { name: 'John', location: 'Canada', role: 'user' },
  ];

  assert.deepEqual(expected, actual);
});