一文学会 Node.js 中的流

时间 2020-06-09

标签一文学会 node.js node 栏目 Node.js 繁體版

原文原文链接

做者：Liz Parody
翻译：疯狂的技术宅javascript

原文：https://nodesource.com/blog/u...html

未经容许严禁转载前端

Node.js 中的流（Stream）是出了名的难用甚至是难以理解。java

用 Dominic Tarr 的话来讲：“流是 Node 中最好的，也是最容易被误解的想法。”即便是 Redux 的建立者和 React.js 的核心团队成员 Dan Abramov 也惧怕 Node 流。node

本文将帮助你了解流以及如何使用。不要惧怕，你彻底能够把它搞清楚！git

什么是流？

流是为 Node.js 应用提供动力的基本概念之一。它们是数据处理方法，用于将输入的数据顺序读取或把数据写入输出。程序员

流是一种以有效方式处理读写文件、网络通讯或任何类型的端到端信息交换的方式。github

流的处理方式很是独特，流不是像传统方式那样将文件一次所有读取到存储器中，而是逐段读取数据块并处理数据的内容，不将其所有保留在内存中。面试

这种方式使流在处理大量数据时很是强大，例如，文件的大小可能大于可用的内存空间，从而没法将整个文件读入内存进行处理。那是流的用武之地！算法

既能用流来处理较小的数据块，也能够读取较大的文件。

以 YouTube 或 Netflix 之类的“流媒体”服务为例：这些服务不会让你你当即下载视频和音频文件。取而代之的是，你的浏览器以连续的块流形式接收视频，从而使接收者几乎能够当即开始观看和收听。

可是，流不只涉及处理媒体和大数据。它们还在代码中赋予了咱们“可组合性”的力量。考虑可组合性的设计意味着可以以某种方式组合多个组件以产生相同类型的结果。在 Node.js 中，能够经过流在其余较小的代码段中传递数据，从而组成功能强大的代码段。

为何会用到流

与其余数据处理方法相比，流基本上具备两个主要优势：

内存效率：你无需事先把大量数据加载到内存中便可进行处理
时间效率：获得数据后当即开始处所需的时间大大减小，没必要等到整个有效数据所有发送完毕才开始处理

Node.js 中有 4 种类型的流：

可写：能够向其中写入数据的流。例如，fs.createWriteStream() 使咱们可使用流将数据写入文件。
可读：可从中读取数据的流。例如：fs.createReadStream() 让咱们读取文件的内容。
双工：可读和可写的流。例如，net.Socket
Transform：可在写入和读取时修改或转换数据。例如在文件压缩的状况下，你能够在文件中写入压缩数据，也能够从文件中读取解压缩的数据。

若是你已经使用过 Node.js，则可能遇到过流。例如在基于 Node.js 的 HTTP 服务器中，request 是可读流，而 response 是可写流。你可能用过 fs 模块，该模块可以让你用可读和可写文件流。每当使用 Express 时，你都在使用流与客户端进行交互，并且因为 TCP 套接字、TLS栈和其余链接都基于 Node.js，因此在每一个可使用的数据库链接驱动的程序中使用流。

一个实际的例子

如何建立可读流

首先须要可读性流，而后将其初始化。

const Stream = require('stream')
const readableStream = new Stream.Readable()

如今，流已初始化，能够向其发送数据了：

readableStream.push('ping!')
readableStream.push('pong!')

异步迭代器

强烈建议在使用流时配合异步迭代器(async iterator)。根据 Axel Rauschmayer 博士的说法，异步迭代是一种用于异步检索数据容器内容的协议（这意味着当前“任务”能够在检索项目以前被暂停）。另外必须说起的是，流异步迭代器实现使用内部的 readable 事件。

从可读流中读取时，可使用异步迭代器：

import * as fs from 'fs';

async function logChunks(readable) {
  for await (const chunk of readable) {
    console.log(chunk);
  }
}

const readable = fs.createReadStream(
  'tmp/test.txt', {encoding: 'utf8'});
logChunks(readable);

// Output:
// 'This is a test!\n'

也能够用字符串收集可读流的内容：

import {Readable} from 'stream';

async function readableToString2(readable) {
  let result = '';
  for await (const chunk of readable) {
    result += chunk;
  }
  return result;
}

const readable = Readable.from('Good morning!', {encoding: 'utf8'});
assert.equal(await readableToString2(readable), 'Good morning!');

注意，在这种状况下必须使用异步函数，由于咱们想返回 Promise。

请切记不要将异步功能与 EventEmitter 混合使用，由于当前在事件处理程序中发出拒绝时，没法捕获拒绝，从而致使难以跟踪错误和内存泄漏。目前的最佳实践是始终将异步函数的内容包装在 try/catch 块中并处理错误，但这很容易出错。这个 pull request 旨在解决一旦其落在 Node 核心上产生的问题。

要了解有关异步迭代的 Node.js 流的更多信息，请查看这篇很棒的文章。

Readable.from()：从可迭代对象建立可读流

stream.Readable.from(iterable, [options]) 这是一种实用方法，用于从迭代器中建立可读流，该迭代器保存可迭代对象中包含的数据。可迭代对象能够是同步可迭代对象或异步可迭代对象。参数选项是可选的，除其余做用外，还能够用于指定文本编码。

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

async function * generate() {
  yield 'hello';
  yield 'streams';
}

const readable = Readable.from(generate());

readable.on('data', (chunk) => {
  console.log(chunk);
});

两种读取模式

根据 Streams API，可读流有效地以两种模式之一运行：flowing和paused。可读流能够处于对象模式，不管处于 flowing 模式仍是 paused 模式。

在流模式下，将自动从底层系统读取数据，并经过 EventEmitter 接口使用事件将其尽快提供给程序。
在 paused 模式下，必须显式调用 stream.read() 方法以从流中读取数据块。

在 flowing 模式中，要从流中读取数据，能够监听数据事件并附加回调。当有大量数据可用时，可读流将发出一个数据事件，并执行你的回调。看下面的代码片断：

var fs = require("fs");
var data = '';

var readerStream = fs.createReadStream('file.txt'); //Create a readable stream

readerStream.setEncoding('UTF8'); // Set the encoding to be utf8. 

// Handle stream events --> data, end, and error
readerStream.on('data', function(chunk) {
   data += chunk;
});

readerStream.on('end',function() {
   console.log(data);
});

readerStream.on('error', function(err) {
   console.log(err.stack);
});

console.log("Program Ended");

函数调用 fs.createReadStream() 给你一个可读流。最初流处于静态状态。一旦你侦听数据事件并附加了回调，它就会开始流动。以后将读取大块数据并将其传递给你的回调。流实现者决定发送数据事件的频率。例如，每当有几 KB 的数据被读取时，HTTP 请求就可能发出一个数据事件。当从文件中读取数据时，你可能会决定读取一行后就发出数据事件。

当没有更多数据要读取（结束）时，流将发出结束事件。在以上代码段中，咱们监听此事件以在结束时获得通知。

另外，若是有错误，流将发出并通知错误。

在 paused 模式下，你只需在流实例上重复调用 read()，直到读完全部数据块为止，如如下示例所示：

var fs = require('fs');
var readableStream = fs.createReadStream('file.txt');
var data = '';
var chunk;

readableStream.on('readable', function() {
    while ((chunk=readableStream.read()) != null) {
        data += chunk;
    }
});

readableStream.on('end', function() {
    console.log(data)
});

read() 函数从内部缓冲区读取一些数据并将其返回。当没有内容可读取时返回 null。因此在 while 循环中，咱们检查是否为 null 并终止循环。请注意，当能够从流中读取大量数据时，将会发出可读事件。

全部 Readable 流均以 paused 模式开始，但能够经过如下方式之一切换为 flowing 模式：

添加一个 'data' 事件处理。
调用 stream.resume() 方法。
调用 stream.pipe() 方法将数据发送到可写对象。

Readable 可使如下方法之一切换回 paused 模式：

若是没有管道目标，则经过调用 stream.pause() 方法。
若是有管道目标，请删除全部管道目标。能够经过调用 stream.unpipe() 方法来删除多个管道目标。

一个须要记住的重要概念是，除非提供了一种用于消耗或忽略该数据的机制，不然 Readable 将不会生成数据。若是使用机制被禁用或取消，则 Readable 将会试图中止生成数据。添加 readable 事件处理会自动使流中止 flowing，并经过 read.read() 获得数据。若是删除了 readable 事件处理，那么若是存在 'data' 事件处理，则流将再次开始 flowing。

如何建立可写流

要将数据写入可写流，你须要在流实例上调用 write()。如如下示例所示：

var fs = require('fs');
var readableStream = fs.createReadStream('file1.txt');
var writableStream = fs.createWriteStream('file2.txt');

readableStream.setEncoding('utf8');

readableStream.on('data', function(chunk) {
    writableStream.write(chunk);
});

上面的代码很简单。它只是简单地从输入流中读取数据块，并使用 write() 写入目的地。该函数返回一个布尔值，指示操做是否成功。若是为 true，则写入成功，你能够继续写入更多数据。若是返回 false，则表示出了点问题，你目前没法写任何内容。可写流将经过发出 drain 事件来通知你何时能够开始写入更多数据。

调用 writable.end() 方法表示没有更多数据将被写入 Writable。若是提供，则可选的回调函数将做为 finish 事件的侦听器附加。

// Write 'hello, ' and then end with 'world!'.
const fs = require('fs');
const file = fs.createWriteStream('example.txt');
file.write('hello, ');
file.end('world!');
// Writing more now is not allowed!

你能够用可写流从可读流中读取数据：

const Stream = require('stream')

const readableStream = new Stream.Readable()
const writableStream = new Stream.Writable()

writableStream._write = (chunk, encoding, next) => {
    console.log(chunk.toString())
    next()
}

readableStream.pipe(writableStream)

readableStream.push('ping!')
readableStream.push('pong!')

writableStream.end()

还能够用异步迭代器来写入可写流，建议使用

import * as util from 'util';
import * as stream from 'stream';
import * as fs from 'fs';
import {once} from 'events';

const finished = util.promisify(stream.finished); // (A)

async function writeIterableToFile(iterable, filePath) {
  const writable = fs.createWriteStream(filePath, {encoding: 'utf8'});
  for await (const chunk of iterable) {
    if (!writable.write(chunk)) { // (B)
      // Handle backpressure
      await once(writable, 'drain');
    }
  }
  writable.end(); // (C)
  // Wait until done. Throws if there are errors.
  await finished(writable);
}

await writeIterableToFile(
  ['One', ' line of text.\n'], 'tmp/log.txt');
assert.equal(
  fs.readFileSync('tmp/log.txt', {encoding: 'utf8'}),
  'One line of text.\n');

stream.finished() 的默认版本是基于回调的，可是能够经过 util.promisify() 转换为基于 Promise 的版本（A行）。

在此例中，使用如下两种模式：

Writing to a writable stream while handling backpressure (line B):
在处理 backpressure 时写入可写流（B行）：

if (!writable.write(chunk)) {
  await once(writable, 'drain');
}

关闭可写流，并等待写入完成（C行）：

writable.end();
await finished(writable);

pipeline()

管道是一种机制，能够将一个流的输出做为另外一流的输入。它一般用于从一个流中获取数据并将该流的输出传递到另外一个流。管道操做没有限制。换句话说，管道可用于分多个步骤处理流数据。

在 Node 10.x 中引入了 stream.pipeline()。这是一种模块方法，用于在流转发错误和正确清理之间进行管道传输，并在管道完成后提供回调。

这是使用管道的例子：

const { pipeline } = require('stream');
const fs = require('fs');
const zlib = require('zlib');

// 使用 pipeline API 能够轻松将一系列流
// 经过管道传输在一块儿，并在管道彻底完成后获得通知。
// 一个有效地用 gzip压缩巨大视频文件的管道：

pipeline(
  fs.createReadStream('The.Matrix.1080p.mkv'),
  zlib.createGzip(),
  fs.createWriteStream('The.Matrix.1080p.mkv.gz'),
  (err) => {
    if (err) {
      console.error('Pipeline failed', err);
    } else {
      console.log('Pipeline succeeded');
    }
  }
);

因为pipe 不安全，应使用 pipeline 代替 pipe。