深入浅出了解Node.js Streams

什么是流（steams）

流（stream）是 node.js 中处理流式数据的抽象接口。

streams 不是 node.js 独有的概念。它们是几十年前在 unix 操作系统中引入的。

它们能够以一种有效的方式来处理文件的读、写，网络通信或任何类型的端到端信息交换。
例如，当你编写了一段程序用来读取文件时，传统的方法是将文件从头到尾读入内存，然后再进行处理。而使用流的话，你就可以逐块读取它，处理其内容而不将其全部保存在内存中。
以如下代码为例

const fs = require('fs');
const rs = fs.createreadstream('test.md');
let data = '';
rs.on("data", function (chunk) {
data += chunk;
});
rs.on("end", function() {
console.log(data);
});

利用 createreadstream 创建一个读取数据的流，来读取 test.md 文件的内容，此时监听 data 事件，它是在当流将数据块传送给消费者后触发。并在对应的 eventhandler 中，拼接 chunk。在 end 事件中，打印到终端上。
之前说流，可以逐块读取文件内容，那么这个块，也就是 chunk 是什么？
一般情况下是 buffer，修改 data 事件的 eventhandler 来验证下

rs.on("data", function (chunk) {
console.log("chunk", buffer.isbuffer(chunk)) // log true
data += chunk;
});

流的工作方式可以具体的表述为，在内存中准备一段 buffer，然后在 fs.read() 读取时逐步从磁盘中将字节复制到 buffer 中。

为什么要使用 stream

利用 stream 来处理数据，主要是因为它的两个优点：

内存效率：在够处理数据之前，不需要占用大量内存；

时间效率：处理数据花费的时间更少，因为流是逐块来处理数据，而不是等到整个数据有效负载才启动。

首先内存效率，与 fs.readfile 这种会缓冲整个文件相比，流式传输充分地利用 buffer （超过 8kb）不受 v8 内存控制的特点，利用堆外内存完成高效地传输。相关验证可以参考这篇博文，地址。
时间效率，与 fs.filesync 相比，有些优势，但是与异步的 fs.readfile 相比，优势不大。

node.js 中 stream 的使用

首先用一张图来了解下 node.js 中有哪些内置的 stream 接口

图中提供了一些 node.js 原生的流的示例，有些是可读、写的流。 也有一些是可读写的流，如 tcp sockets、zlib 以及 crypto。

特别注意： 流的读、写与环境是密切相关的。例如 http 响应在客户端上的可读流，但它是服务器上的可写流。同时还需要注意，stdio streams（stdin，stdout，stderr） 在子进程上是相反的流。

使用一个例子来展示流的使用

首先利用如下脚本创建一个比较大的文件（大概 430 mb）

const fs = require('fs');
const file = fs.createwritestream('test.md');
for(let i=0; i<= 1e6; i++) {
file.write('hello world.\n');
}
file.end();

在当前目录下，启动 http 服务

const http = require('http')
const fs = require('fs')
const server = http.createserver(function (req, res) {
fs.readfile(__dirname + '/test.md', (err, data) => {
res.end(data)
})
})
server.listen(3000)

得到的结果，如图

const http = require('http')
const fs = require('fs')
const server = http.createserver((req, res) => {
const stream = fs.createreadstream(__dirname + '/test.md')
stream.pipe(res)
})
server.listen(3000)

时间减少了 2s 多。这可以解释为，在读取文件内容，并且不需要改变内容的场景下，流能够完成只读取 buffer，然后直接传输，不做额外的转换，避免损耗，提高性能。
上述代码中，应用了 stream.pipe(...) 。它主要是对流进行链式地管道操作，例如

src.pipe(dest1).pipe(dest2)

这样数据流会被自动管理。

如果可读流发生错误，目标可写流不会自动关闭，需要手动关闭所有流以避免内存泄漏。

通常，当你使用 pipe 方法时，就不需要使用事件，但如果场景需要以更灵活、自定义的方式使用流，那么就要考虑事件。

stream events

在上述例子中，我们使用了可读流的 data 、end 事件来控制文件的读取，它本质上与 pipe 方法相同，例如

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

只不过，使用 event 会更加灵活，可控。

图中简单罗列了可读流、可写流的相关事件、方法，其中最重要的是

可读流：

• data 事件：每当流将一大块数据传递时，就会触发；
• end 事件：当没有更多数据要从流发出时，就会触发。

可写流：

• drain 事件：当可以继续写入数据到流时会触发事件；
• finish 事件：处理完全部数据块之后触发。

流的不同类型

除了上面涉及到的可读、写流之后，还有 duplex、transform 两类：

• readable ：可以接收数据，但不能向其发送数据。当你将数据推送到可读流中时，它会被缓冲，直到消费者开始读取数据；
• writable ：可以发送数据，但不能从中接收；
• duplex ：即可读也可写；
• tranform ：与 duplex 一样是可写又可读的，但它的输出与输入是相关联的。

如何创建一个可读流

这里只做简单介绍，具体见 stream module。

const stream = require('stream')
const readablestream = new stream.readable()
readablestream._read = (size) => {
console.log('read', size)
}

利用 stream 模块初始化一个可读流，然后向其中发送数据

readablestream.push('hi!')
readablestream.push('ho!')

如何创建一个可写流

为了创建可写流，需要扩展了基本的 writable 对象，并实现了它的 _write 方法。

const stream = require('stream')
const writablestream = new stream.writable()

实现 _write 方法：

writablestream._write = (chunk, encoding, next) => {
console.log(chunk.tostring())
next()
}

结合上述例子实现

利用 readablestream 读入数据，并输出到 writablestream

const stream = require('stream')
const readablestream = new stream.readable()
readablestream._read = (size) => {
console.log('read', size)
}
const writablestream = new stream.writable()
writablestream._write = (chunk, encoding, next) => {
console.log('write', chunk.tostring())
next()
}
readablestream.pipe(writablestream)
readablestream.push('hi!')
readablestream.push('ho!')
/* 
log:
read 16384
write hi!
write ho!
*/

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持。