[译]关于Node.js streams你需要知道的一切

frontoldman

发布于 2019-09-02 16:30:41

1.1K0

发布于 2019-09-02 16:30:41

文章被收录于专栏：樯橹代码樯橹代码

Node.js的stream模块是有名的应用困难，更别说理解了。那现在可以告诉你，这些都不是问题了。

多年来，开发人员在那里创建了大量的软件包，其唯一目的就是使用stream使用起来更简单，但是在这篇文章里，我们专注于介绍原生的Node.js Steam Api。

"Stream 是Node.js中最好的却最容易被误解的部分" ----- Dominic Tarr

Streams到底是什么

Streams是数据的集合，就跟数组和字符串一样。不同点就在于Streams可能不是立刻就全部可用，并且不会全部载入内存。这使得他非常适合处理大量数据，或者处理每隔一段时间有一个数据片段传入的情况。

但是，Stream并不仅仅适用于处理大数据(大块的数据。。。)。使用它，同样也有利于组织我们大代码。就像我们使用管道去和合并强大的Linux命令。在Node.js中，我们也可以做同样的事情。

const grep = ... // A stream for the grep output
const wc = ... // A stream for the wc input
grep.pipe(wc)

Node.js的很多内置模块都实现了Stream接口

上面例子里面的Node.js对象列表包括了可读流和可写流，有一些对象既是可读流也是可写流，像TCP sockets, zlib 和 crypto streams。

注意这些对象是有很密切的关联的。当一个客户端的HTTP 响应对象是一个可读流，那么在服务器端这就是一个可写流。因为在HTTP例子中，我们通常是从一个对象(http.IncomingMessage)读取再写入到另外一个对象(http.ServerResponse)中去。

还要注意，当涉及到子进程时，stdio流（stdin，stdout，stderr）具有逆流类型。这就允许我们非常方便的使用管道从主进程连接子进程的Streams。

一些实例的Streams例子

理论都是很好的，但事实到底是怎么样子的呢？让我们看一些例子示范代码Streams在内存使用方面的比较。

我们先创建一个大文件

const fs = require('fs');
const file = fs.createWriteStream('./big.file');

for(let i=0; i<= 1e6; i++) {
  file.write('Lorem ipsum dolor sit amet, consectetur adipisicing elit, sed do eiusmod tempor incididunt ut labore et dolore magna aliqua. Ut enim ad minim veniam, quis nostrud exercitation ullamco laboris nisi ut aliquip ex ea commodo consequat. Duis aute irure dolor in reprehenderit in voluptate velit esse cillum dolore eu fugiat nulla pariatur. Excepteur sint occaecat cupidatat non proident, sunt in culpa qui officia deserunt mollit anim id est laborum.\n');
}

file.end();

看看我使用什么创建文件的？一个可写流嘛

fs模块可以通过Stream接口来读取和写入文件。在上面的例子中，我们在循环中通过可写流向big.file写入了1百万行数据。

运行上面的代码会生成一个大概400M的文件

这是一个简单的Node web服务器，专门为big.file提供服务：

const fs = require('fs');
const server = require('http').createServer();

server.on('request', (req, res) => {
  fs.readFile('./big.file', (err, data) => {
    if (err) throw err;
  
    res.end(data);
  });
});

server.listen(8000);

当server收到请求，它会使用异步方法fs.readFile处理这个big file。但是这并不代表我们会打断事件循环机制。一切都是正确的吗？？

那现在当我们启动server，看看内存监视器都发生了什么。

clipboard.png

现在访问这个服务器，看看内存的使用情况。

内存占用立刻飙升到434.8 MB。

在我们把文件内容输出到客户端之前，我们就把整个文件读入了内存。这是很低效的。

HTTP response对象(上文中的res对象)也是一个可写流，这就意味着如果我们有一个代表着big file的可读流，我们可以通过管道把他们俩连接起来实现同样的功能，而不需要使用400M内存。

Node的fs模块给我们提供了一个可以操作任何文件的可读流,通过createReadStream方法创建。我们可以把它和response对象连接起来。

const fs = require('fs');
const server = require('http').createServer();

server.on('request', (req, res) => {
  const src = fs.createReadStream('./big.file');
  src.pipe(res);
});

server.listen(8000);

现在再去访问server的时候，令人惊讶的事情发生了(看内存监视器)

clipboard.png

发生了什么？

当我们访问服务器的时候，我们通过流每次使用一段数据，这意味着我们不是把全部的数据都加载到内存中。内存使用量只上升了不到25M。

可以把上面的例子用到极致，生成5百万行数据而不是1百万行。这样子的话，这个文件的大小会超过2GB，这实际上大于Node中的默认缓冲区限制。

如果你想在server上使用fs.readFile,这在默认情况下是行不通的，除非你改了Node.js的默认缓冲区限制。但是使用fs.createReadStream，把2 GB的数据返回给客户端根本不存在问题，甚至内存使用量都没有任何变化。

准备好学习Steam了吗？

Streams 101

在Node.js中有4中基本的流类型：Readable, Writable, Duplex, and Transform streams。

Readable 可读流是可以从中消耗数据的源的抽象，一个例子就是fs.createReadStream方法
Writable 可写流是可以写入数据的目标的抽象，一个例子就是fs.createWriteStream方法
duplex Steam是一个同时具有读写功能的流，一个例子就是TCP socket
Transform 是一个双工流，它可以在交换数据的时候做转换。一个例子就是zlib.createGzip使用gzip压缩数据。你可以把Transform streams当成是一个传入可读流，返回一个可写流的函数。它还有一个别名through streams

所有的Stream都是EventEmitter的实例对象。当流读和写的时候都会触发相应的事件。但是还有一个更简单的使用方法，那就是使用pipe。

The pipe method

要记住下面这个魔幻方法

readableSrc.pipe(writableDest)

在这一行里面，我们通过管道把可读流(源)输出到一个可写流里面去(目标)，源必须是一个可写流，目标必须是可写流。当然，他们也都可以是duplex/Transform。事实上，当我们使用管道连接流的时候，我们可以像在linux中一样使用链式连接。

readableSrc
  .pipe(transformStream1)
  .pipe(transformStream2)
  .pipe(finalWrtitableDest)

pipe方法返回目标流，这保证了我们可以使用链式调用。对于streams a(可读流)，b,c(可读可写流)，d可写流，我们可以使用：

a.pipe(b).pipe(c).pipe(d)
# Which is equivalent to:
a.pipe(b)
b.pipe(c)
c.pipe(d)
# Which, in Linux, is equivalent to:
$ a | b | c | d

pipe方法是使用流最简便的方法。通常通过管道和事件的方法使用流，但是要尽量避免两者混用。通常当你使用pipe方法就不需要使用事件了。但是当你需要更多定制的操作的话，使用事件的方式会更好。

Stream events

除了从可读流读取数据传输到可写流，pipe方法还自动处理一些其他事情。比如处理错误，处理文件结束操作，流之间速度快慢问题。

同时，流也可以直接使用事件操作。以下是和管道相等的通过事件操作流的方法。

# readable.pipe(writable)

readable.on('data', (chunk) => {
  writable.write(chunk);
});

readable.on('end', () => {
  writable.end();
});

下面是一些重要流的事件和方法。

clipboard.png

这些事件和方法在某种程度上是相关的，因为它们通常被一起使用。

可读流上的最重要的事件是

data事件，当可读流传输了一段数据的时候会触发
end事件，当没有数据被传输时触发

可写流上的最重要的事件是

drain事件，当可写流可以接收事件的时候被触发
finish事件，当所有数据被接收时被触发

事件和方法可以结合起来，以便定制和优化流的使用。读取可读流，我们可以使用pipe/unpipe方法，或者read/unshift/resume方法。使用可写流，我们可以可写流作为pipe/unpipe方法的参数，或者使用write方法写入，使用end方法关闭。

可读流的暂停和流动

可读流有两个很重要的模式影响了我们使用的方式。

暂停模式
流动模式

这些模式有时候被称为拉和推模式

所有的可读流开始的时候都是默认暂停模式，但是它们可以轻易的被切换成流动模式，当我们需要的时候又可以切换成暂停模式。有时候这个切换是自动的。

当一个可读流是暂停模式的时候，我们可以使用read方法从流中读取。但是当一个流是流动模式的时候，数据是持续的流动，我们需要使用事件去监听数据的变化。

在流动模式中，如果可读流没有监听者，可读流的数据会丢失。这就是为什么当可读流逝流动模式的时候，我们必须使用data事件去监听数据的变化。事实上，只需添加一个数据事件处理程序即可将暂停的流转换为流模式，删除数据事件处理程序将流切换回暂停模式。其中一些是为了与旧的Node Stream接口进行向后兼容。

可以使用resume()和pause()方法在这两种模式之间切换。

clipboard.png

当我们使用pipe方法操作可读流的时候是不需要担心上面的这些操作的，因为pipe方法会自动帮我们处理这些问题。

流的创建

当我们讨论Node.js中的流时，有两项重要的任务：

流的创建
流的使用

我们到现在为止讨论的都是如何使用流，那下面来看看如何创建吧！

Streams的创建通常使用stream模块。

创建一个可写流

为了创建一个可写流，我们需要使用stream模块里面的Writable类。

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

我们可以有很多种方式实现一个可写流。例如，我们可以继承Writable类。

class myWritableStream extends Writable {

}

但是我更喜欢使用构造函数的方式创建。通过给Writable传递一些参数来创建一个对象。唯一必须要传的选项时write方法，它需要暴漏需要写入的数据块。

const { Writable } = require('stream');
const outStream = new Writable({
  write(chunk, encoding, callback) {
    console.log(chunk.toString());
    callback();
  }
});

process.stdin.pipe(outStream);

write方法接收3个参数

chunk通常是一个buffer对象，我们可以通过配置修改
encoding在这种情况下就需要了，不过通常情况是可以忽略的
callback是当我们处理完这个数据块的时候需要调用的函数。这是一个写入是否成功的信号。如果失败了，给这个回调传递一个Error对象

在outStream中，我们简单的把chunk打印出来，因为并没有发生错误，我们直接调用了callback方法。这是这是简单并不实用的打印流。它会打印接收到的所有值。

为了使用这个流，我们可以简单的process.stdin这个可读流。通过pipe方法连接起来。

当我们运行上面的例子，任何我们在控制台输入的内容都会被console.log打印出来。

这不是一个非常实用的流的实现，但是它已经被Node.js内置实现了。outStream功能和process.stdout基本类似。我们也可以通过pipe方法把stdin和stdout连接起来并实现同样的功能。

process.stdin.pipe(process.stdout);

创建一个可读流

创建可读流，我们需要Readable类

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

创建一个可读流非常简单。可以使用push方法推入数据给其他流使用

const { Readable } = require('stream'); 
const inStream = new Readable();
inStream.push('ABCDEFGHIJKLM');
inStream.push('NOPQRSTUVWXYZ');
inStream.push(null); // No more data
inStream.pipe(process.stdout);

当我们push一个null对象进去的时候，这就标志着我们要终止传输了。

我们可以简单的把这个流通过pipe方法连接到一个可写流process.stdout

运行上面的代码，会获取所有的inStream的数据并打印出来。非常简单但有效。

我们在通过pipe连接之前，就会把所有的数据推送到流里面。更好的方法是在消费者要求时按需推送数据。可以通过修改可读流配置里面的read()方法实现。

const inStream = new Readable({
  read(size) {
    // there is a demand on the data... Someone wants to read it.
  }
});

当读取方法在可读流上被调用时，该实现可以将部分数据推送到队列。例如，我们可以一次推一个字母，从字符代码65（表示A）开始，并在每次推送时递增：

const inStream = new Readable({
  read(size) {
    this.push(String.fromCharCode(this.currentCharCode++));
    if (this.currentCharCode > 90) {
      this.push(null);
    }
  }
});
inStream.currentCharCode = 65;
inStream.pipe(process.stdout);

当有流在读取可读流的数据的时候，read方法会持续执行，这样就会一直推出更多的字符。我们需要在某个时刻终止它，这就是为什么我们设置了一个终止条件推入了null。

我们应该始终按需推送数据。

Duplex/Transform 流的实现

使用Duplex流，我们通过同一个对象实现可读流和可写流。这类似同时实现了两个接口。

下面这个例子就结合了上面两个可读流和可写流的综合例子。

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

const inoutStream = new Duplex({
  write(chunk, encoding, callback) {
    console.log(chunk.toString());
    callback();
  },

  read(size) {
    this.push(String.fromCharCode(this.currentCharCode++));
    if (this.currentCharCode > 90) {
      this.push(null);
    }
  }
});

inoutStream.currentCharCode = 65;
process.stdin.pipe(inoutStream).pipe(process.stdout);

通过合并这些方法，我们可以使用这个duplex流读取从A-Z的字母也同样可以使用它的打印功能。我们把stdin流连接到这个duplex上去使用它的打印功能，再把这个duplex流本身连接到stdout上去就在控制台看到了A-Z。

双工流的可读写的两侧完全独立运行。就像一个对象上两种独立的功能。

transform流是一种更有趣的duplex流。因为它的输出来源于她的输入。

对于一个transform流，我们不需要实现read和write方法，我们仅仅需要实现transform方法，这个方法合并了它们两个。它具有写入方法的功能，也可以用它推送数据。

这是一个简单的transform例子，把任何输入转换成大写。

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

const upperCaseTr = new Transform({
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().toUpperCase());
    callback();
  }
});

process.stdin.pipe(upperCaseTr).pipe(process.stdout);

在这个transformstream里面，像上个例子中双工流一样。但是我们只实现了transform()方法。我们把chunk转换成大写，再把大写字母作为可读流的输入。

Streams Object Mode

默认，流会接收 Buffer/String 类型的数据。还有个字段 objectMode 设置，可以让stream 接收任意类型的对象。

下面是一个这种类型的例子。以下变换流的组合使得将逗号分隔值的字符串映射为JavaScript对象的功能。所以“a，b，c，d”成为{a：b，c：d}。

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

const commaSplitter = new Transform({
  readableObjectMode: true,
  transform(chunk, encoding, callback) {
    this.push(chunk.toString().trim().split(','));
    callback();
  }
});

const arrayToObject = new Transform({
  readableObjectMode: true,
  writableObjectMode: true,
  transform(chunk, encoding, callback) {
    const obj = {};
    for(let i=0; i < chunk.length; i+=2) {
      obj[chunk[i]] = chunk[i+1];
    }
    this.push(obj);
    callback();
  }
});

const objectToString = new Transform({
  writableObjectMode: true,
  transform(chunk, encoding, callback) {
    this.push(JSON.stringify(chunk) + '\n');
    callback();
  }
});

process.stdin
  .pipe(commaSplitter)
  .pipe(arrayToObject)
  .pipe(objectToString)
  .pipe(process.stdout)

我们通过commasplitter传递输入字符串（例如，“a，b，c，d”），它将数组作为其可读数据（[“a”，“b”，“c”，“d”]））。在该流上添加可读的ObjectMode标志是必要的，因为我们正在将对象推送到其上，而不是字符串。

然后我们把数组导入到arrayToObject数据流中，我们需要把writableObjectMode设置为 true，以表示arrayToObject会接收一个对象。另外它还会推送一个对象出去，所以还要把他的readableObjectMode为true。最后一个objectToString接收一个对象但是输出字符串，所以就只需要设置一个writableObjectMode。

clipboard.png

Node.js内置transform streams对象

Node有一些非常有用的内置transform streams对象。这包括zlib和crypto。

下面这个例子使用了zlib.createGzip()结合了额fs readable/writable streams实现了文件压缩。

const fs = require('fs');
const zlib = require('zlib');
const file = process.argv[2];

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .pipe(fs.createWriteStream(file + '.gz'));

你可以使用上面的脚本压缩任何你传入的参数文件。我们把文件的可读流传入了zlib的内置转换流。再写入到新的.gz文件中。

使用管道还有一个很酷的事情，就是可以和事件结合起来。比如我想用户看到进度，并在结束的时候发个消息。因为pipe方法会返回目标流，我们也可以通过链式注册事件。

const fs = require('fs');
const zlib = require('zlib');
const file = process.argv[2];

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .on('data', () => process.stdout.write('.'))
  .pipe(fs.createWriteStream(file + '.zz'))
  .on('finish', () => console.log('Done'));

所以使用管道方法，我们可以轻松地操作流，但是我们还可以使用需要的事件进一步定制与这些流的交互。

管道方法的好处是，我们可以用它来以一种可读的方式逐一构成我们的程序。例如，我们可以简单地创建一个变换流来报告进度，而不用监听上面的数据事件，并用另一个.pipe()调用替换 .on() 调用：

const fs = require('fs');
const zlib = require('zlib');
const file = process.argv[2];

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

const reportProgress = new Transform({
  transform(chunk, encoding, callback) {
    process.stdout.write('.');
    callback(null, chunk);
  }
});

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .pipe(reportProgress)
  .pipe(fs.createWriteStream(file + '.zz'))
  .on('finish', () => console.log('Done'));

reportProgress流是一个简单的pass-through流，但是也跟标准事件一样报告进度。注意callback()函数的第二个参数，这相当于把数据推送出去。

结合流的应用是无止境的。例如，如果我们需要在我们gzip之前或之后加密文件，我们需要做的就是按照我们需要的确切顺序来管理另一个转换流。使用Node的crypto模块处理这个事情。

const crypto = require('crypto');
// ...

fs.createReadStream(file)
  .pipe(zlib.createGzip())
  .pipe(crypto.createCipher('aes192', 'a_secret'))
  .pipe(reportProgress)
  .pipe(fs.createWriteStream(file + '.zz'))
  .on('finish', () => console.log('Done'));

上面的脚本压缩然后加密传递的文件，只有具有密码的人才可以使用文件。我们无法使用正常的解压缩实用程序解压缩此文件，因为它已被加密。

为了能够解压缩文件，我们需要使用完全相反的操作，这也很简单。

fs.createReadStream(file)
  .pipe(crypto.createDecipher('aes192', 'a_secret'))
  .pipe(zlib.createGunzip())
  .pipe(reportProgress)
  .pipe(fs.createWriteStream(file.slice(0, -3)))
  .on('finish', () => console.log('Done'));

假设传递的文件是压缩版本，上面的代码将创建一个读取流，将其传输到crypto createDecipher（）流中（使用相同的秘密），将其输出管道输入到zlib createGunzip（）流中，然后将文件写回到没有扩展名的文件中。

以上就是全部了，谢谢阅读！！

翻译自Node.js Streams: Everything you need to know

本文参与腾讯云自媒体分享计划，分享自作者个人站点/博客。

如有侵权请联系 cloudcommunity@tencent.com 删除

css