Node.js Streams 核心机制：从原理到实战应用

Node.js Streams 核心机制：从原理到实战应用

什么是流（streams）

在 Node.js 中，流（stream）是一种用于处理流式数据的抽象接口。

这一概念并非 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

利用 Stream 处理数据主要基于以下两个优点：

内存效率：在处理数据之前，无需占用大量内存空间。

时间效率：由于流是逐块处理数据，而非等待整个数据有效负载加载完毕才启动，因此处理时间更短。

首先看内存效率，与会缓冲整个文件的 fs.readFile 相比，流式传输充分利用了超过 8kb 的 Buffer 不受 V8 内存限制的特点，利用堆外内存完成高效传输。相关验证可参考这篇博文（地址）。在时间效率方面，相比 fs.FileSync 有一定优势，但与异步的 fs.readFile 相比，优势并不显著。

Node.js 中 Stream 的使用与示例

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)

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

src.pipe(dest1).pipe(dest2)

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

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

通常，使用 pipe 方法时不需要使用事件；但如果场景需要更灵活、自定义的方式使用流，则需考虑事件机制。

Stream 事件处理机制

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：即可读也可写；
• Transform：与 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!
*/

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