本文深入解析 Node.js 中 Stream（流）的核心概念与工作机制。通过对比传统文件读取方式，阐述流在内存效率和时间效率上的优势。详细讲解 Readable、Writable、Duplex 和 Transform 四种流类型，以及 pipe 管道操作和关键事件处理，并提供创建自定义流的代码示例，帮助开发者高效处理大数据量场景。
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!
*/
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持本站。