在处理大规模数据时，一次性将全部内容加载到内存的做法很快会遇到瓶颈。Node.js Stream 提供了一种优雅的解决方案：数据以连续块（chunk）的形式流动，消费者边读边处理，内存占用始终维持在固定水平。一个数十 GB 的日志文件用传统 fs.readFile 会直接撑爆 Node.js 的堆内存上限，而 fs.createReadStream 仅需几十 KB 即可逐块读取。Stream 的设计不仅体现在 API 表层，更深入到事件的发布订阅模型、背压的速率控制、管道的资源生命周期管理。Buffer 作为 Node.js 中处理二进制数据的基石，本身又是 Uint8Array 的子类，既继承了 TypedArray 的跨平台一致性，又补充了字符串编码转换、字符串切片拼接等 Node.js 特有的实用方法。
1. 四种流类型
Node.js 的 stream 模块定义了四种核心流类型，分别对应数据处理中不同的角色。理解它们的区别，是选择合适工具处理流数据的前提。
| 类型 | 角色 | 核心接口 | 典型场景 |
|---|---|---|---|
| Readable | 数据源 | read(), pipe() | 文件读取、HTTP 请求体 |
| Writable | 数据目的地 | write(), end() | 文件写入、HTTP 响应体 |
| Duplex | 双向数据通道 | 同时继承 Readable 和 Writable | TCP Socket、TLS 连接 |
| Transform | 读入数据经过处理后原样写回 | _transform() | 压缩、加密、数据转换 |
1.1 Readable 流
Readable 流是数据的源头。在 Node.js 中，所有可读的数据源都实现了 Readable 接口：fs.createReadStream 从磁盘读取，process.stdin 读取标准输入，HTTP 请求的 req 对象也继承自 Readable。
const fs = require('fs');
const readable = fs.createReadStream('access.log', {
encoding: 'utf8',
highWaterMark: 16 * 1024 // 每次从内核缓冲区拉取 16KB
});
readable.on('data', (chunk) => {
console.log(`收到 ${chunk.length} 字符, 内存占用 ${process.memoryUsage().heapUsed / 1024 / 1024} MB`);
});
readable.on('end', () => console.log('文件读取完毕'));
readable.on('error', (err) => console.error('读取失败:', err));
// Node.js v10 以后可以使用异步迭代器，代码更简洁
(async () => {
for await (const chunk of readable) {
process.stdout.write(chunk);
}
})();
Readable 流有两种消费模式。默认的流动模式（flowing mode）下，data 事件会不断触发，数据自动推送到消费者。暂停模式（paused mode）下，消费者主动调用 readable.read(size) 拉取数据。通过 pause() 和 resume() 可以在两种模式之间切换。highWaterMark 的定义在不同数据源中含义不同：对于文件流，它表示内核每次从磁盘读取的字节数；对于基于网络的数据源，它表示内部缓冲区的建议上限。
1.2 Writable 流
Writable 流是数据的消费终点。所有数据最终都需要写入某个目标：https 响应发送到客户端、日志写入磁盘、数据压缩后输出到新文件。write(chunk) 方法是最常用的写入接口。
const fs = require('fs');
const writable = fs.createWriteStream('output.txt', { flags: 'a' });
writable.write('第一行\n', 'utf8', (err) => {
if (err) console.error('写入失败', err);
});
writable.write('第二行\n');
writable.end('结束标记\n', () => {
console.log('所有数据已刷入磁盘');
});
writable.on('finish', () => console.log('finish 事件触发'));
write() 的回调在数据真正写入底层资源后触发，但这不意味着数据已经持久化到磁盘——操作系统可能仍在执行延迟写入。调用 end() 后，Writable 流不再接受新的 write() 调用，finish 事件在所有已排队的数据被写入后触发。特别值得关注的是 write() 的返回值：当内部缓冲区已满时，它返回 false，这正是下一节背压机制发挥作用的关键信号。
1.3 Duplex 流
Duplex 流同时具备 Readable 和 Writable 的能力，而且这两个端的数据流完全独立。与 Transform 不同，写入 Duplex 流的数据不会自动变成可读端的数据， Duplex 流通常用于表示双向的底层通信通道。
const { Duplex } = require('stream');
const net = require('net');
// TCP Socket 是最典型的 Duplex 流
const server = net.createServer((socket) => {
console.log('客户端连接');
socket.on('data', (chunk) => {
console.log('收到:', chunk.toString());
socket.write(`Echo: ${chunk}`);
});
socket.on('end', () => console.log('客户端断开'));
});
server.listen(8080);
const { Duplex } = require('stream');
// 自定义 Duplex：模拟一个可写可读的计数器
const counter = new Duplex({
read(size) {
this.push(String(this.count++) + '\n');
if (this.count > 10) this.push(null);
},
write(chunk, encoding, callback) {
console.log('写入端接收:', chunk.toString().trim());
callback();
}
});
counter.count = 0;
counter.on('data', (chunk) => process.stdout.write('读取端输出: ' + chunk));
counter.write('hello from write side');
1.4 Transform 转换流
Transform 流是 Duplex 的特化，写入的数据经过 _transform 处理后推送到读取端。这让它非常适合数据转换场景：压缩、加密、编码转换、日志过滤。Node.js 内置的 zlib.createGzip()、crypto.createCipher()、crypto.createHash() 都是 Transform 流的实现。
const { Transform } = require('stream');
const fs = require('fs');
// 自定义 Transform：将输入转为大写并统计行数
const upperCaseTransform = new Transform({
transform(chunk, encoding, callback) {
this.push(chunk.toString().toUpperCase());
callback();
}
});
fs.createReadStream('input.txt')
.pipe(upperCaseTransform)
.pipe(fs.createWriteStream('output.txt'));
transform(chunk, encoding, callback) 接收三个参数：当前数据块、编码格式、完成回调。处理完后通过 this.push(result) 将结果推送到下游，最后调用 callback(err) 标记当前块处理完成。如果需要在输入流结束前输出聚合结果，还可以实现 _flush(callback) 方法。
2. 背压与流量控制
背压（Backpressure）是流式系统中最容易被忽视却最容易导致内存泄漏的问题。当 Readable 产生的速度远超 Writable 消费的速度时，未被处理的数据会在内存中无休止地堆积，最终进程因为堆内存耗尽而崩溃。
2.1 手动处理背压
const fs = require('fs');
const readable = fs.createReadStream('big-file.bin');
const writable = fs.createWriteStream('output.bin');
readable.on('data', (chunk) => {
const canContinue = writable.write(chunk);
if (!canContinue) {
console.log('背压触发: 暂停读取, 等待 drain');
readable.pause();
writable.once('drain', () => {
console.log('drain 触发: 恢复读取');
readable.resume();
});
}
});
readable.on('end', () => {
writable.end();
});
这段代码揭示了背压的核心逻辑：writable.write() 返回 false 时，Writable 的内部缓冲区已达到或超过 highWaterMark，此时必须暂停 Readable 的生产。Writable 排空缓冲区后会触发 drain 事件，通知上游可以恢复读取。这个机制在 fs.createReadStream 和 fs.createWriteStream 的管道中被自动处理，但在自定义 Writable 或网络写入时，开发者必须手动关注 write() 的返回值。
2.2 pipe 自动处理背压
const fs = require('fs');
const zlib = require('zlib');
// pipe 内部自动完成了暂停、drain 监听、resume 恢复的全部逻辑
fs.createReadStream('input.txt')
.pipe(zlib.createGzip())
.pipe(fs.createWriteStream('input.txt.gz'));
pipe() 是最简洁的数据流管道方式，但它有一个严重缺陷：不会处理错误。如果链条中的任何一个流报错，其他流不会被自动清理，可能导致资源泄漏。这是引入 stream.pipeline 的主要原因。
2.3 pipeline 与错误处理
const { pipeline } = require('stream');
const fs = require('fs');
const zlib = require('zlib');
pipeline(
fs.createReadStream('archive.tar'),
zlib.createGzip(),
fs.createWriteStream('archive.tar.gz'),
(err) => {
if (err) {
console.error('Pipeline 失败:', err.message);
} else {
console.log('Pipeline 完成');
}
}
);
// Node.js v15+ 的 Promise 版本更适用于 async/await 上下文
const { pipeline } = require('stream/promises');
(async () => {
try {
await pipeline(
fs.createReadStream('input.txt'),
zlib.createGzip(),
fs.createWriteStream('output.txt.gz')
);
console.log('流式压缩成功');
} catch (err) {
console.error('压缩失败:', err);
}
})();
pipeline 相比 pipe 有三个关键改进。一是错误传播：任意流报错时回调或 Promise 的 reject 被触发。二是资源清理：所有参与的流都会被调用 destroy()，关闭底层文件描述符或网络连接。三是 Promise 支持：更适合现代 Node.js 的异步写法。在生产环境中，pipeline 是首选方案。
3. Object Mode 与非 Buffer 数据
默认情况下，Stream 内部流转的数据必须是 String 或 Buffer。但在解析 CSV、处理结构化日志、ETL 数据管道等场景中，直接传递 JavaScript 对象会更方便。Object Mode 开启后，流可以传递任意 JavaScript 对象，但 highWaterMark 的单位从字节数变成了对象个数。
const { Readable, Transform } = require('stream');
const objectSource = new Readable({
objectMode: true,
read() {
this.push({ id: 1, name: 'Alice', score: 95 });
this.push({ id: 2, name: 'Bob', score: 87 });
this.push({ id: 3, name: 'Carol', score: 92 });
this.push(null); // 结束信号
}
});
const filterHighScore = new Transform({
objectMode: true,
transform(chunk, encoding, callback) {
if (chunk.score >= 90) {
this.push(chunk);
}
callback();
}
});
const jsonSerialize = new Transform({
objectMode: true,
transform(chunk, encoding, callback) {
this.push(JSON.stringify(chunk) + '\n');
callback();
}
});
objectSource
.pipe(filterHighScore)
.pipe(jsonSerialize)
.pipe(process.stdout);
Object Mode 的可读流 highWaterMark 默认是 16 个对象，可写流也是 16 个对象。这意味着内存中最多缓存 16 个未消费的对象，即使每个对象占用 1MB，整体内存消耗也是可控的。许多高级 Node.js 库（如 csv-parser、JSONStream）内部都使用了 Object Mode 来实现逐行或逐条记录的流式处理。
4. 自定义 Transform 实现
自定义 Transform 流是处理数据转换的核心手段，需要准确理解 _transform 和 _flush 两个生命周期方法。
const { Transform } = require('stream');
class CSVParser extends Transform {
constructor(options) {
super({ ...options, objectMode: true });
this.buffer = '';
this.headers = null;
}
_transform(chunk, encoding, callback) {
this.buffer += chunk.toString();
let lines;
// 缓存可能包含不完整的最后一行，所以用 lastIndexOf 处理
const lastNewline = this.buffer.lastIndexOf('\n');
if (lastNewline === -1) {
return callback(); // 等待更多数据
}
lines = this.buffer.slice(0, lastNewline).split('\n');
this.buffer = this.buffer.slice(lastNewline + 1);
if (!this.headers) {
this.headers = lines.shift().split(',');
}
for (const line of lines) {
if (!line.trim()) continue;
const values = line.split(',');
const obj = {};
this.headers.forEach((h, i) => {
obj[h.trim()] = values[i]?.trim();
});
this.push(obj);
}
callback();
}
_flush(callback) {
// 缓冲区中可能残留最后一行没有换行符的数据
if (this.buffer.trim()) {
const values = this.buffer.split(',');
const obj = {};
this.headers.forEach((h, i) => {
obj[h.trim()] = values[i]?.trim();
});
this.push(obj);
}
callback();
}
}
// 使用
const parser = new CSVParser();
process.stdin.pipe(parser).on('data', (obj) => {
console.log('解析结果:', obj);
});
这个 CSV 解析器展示了 Transform 流处理的核心技巧。由于 chunk 的边界是任意的，data 事件中的数据可能从一个字段中间截断。因此必须在内部维护一个 buffer，按照逻辑边界（换行符）分割后再处理。_flush 负责处理输入流结束后缓冲区中残留的未完整数据，确保尾部数据不丢失。
5. Buffer 与 TypedArray
Buffer 是 Node.js 中处理二进制数据的核心类。它继承自 JavaScript 的 Uint8Array，因此在所有 TypedArray 操作的语义基础上，额外提供了字符串编码转换、切片拼接、内存池分配优化等 Node.js 特有的能力。
// Buffer 创建方式
const buf1 = Buffer.alloc(10); // 10 字节，初始化为 0
const buf2 = Buffer.allocUnsafe(10); // 10 字节，未初始化（更快但有残留风险）
const buf3 = Buffer.from('Hello 世界', 'utf8'); // 从字符串创建
const buf4 = Buffer.from([0x48, 0x65, 0x6c]); // 从数组创建
// 编码转换
console.log(buf3.toString('utf8')); // Hello 世界
console.log(buf3.toString('hex')); // 48656c6c6f20e4b896...（连续字节）
console.log(buf3.toString('base64')); // 短字符串的 Base64 表示
// 读写操作
const buf5 = Buffer.alloc(4);
buf5.writeUInt32BE(0x12345678, 0); // 大端序写入
console.log(buf5.toString('hex')); // 12345678
buf5.writeUInt32LE(0x12345678, 0); // 小端序写入
console.log(buf5.toString('hex')); // 78563412
// 切片（共享底层内存，O(1) 操作）
const slice = buf3.slice(0, 5);
console.log(slice.toString()); // Hello
slice[0] = 0x41; // 修改 slice 会影响 buf3 的第一个字节
// 拼接（创建新的 Buffer）
const combined = Buffer.concat([buf1, buf2, buf3]);
console.log(combined.length); // 3 个 buffer 的总长度
// 编码字节数计算
console.log(Buffer.byteLength('你好', 'utf8')); // 6
console.log(Buffer.byteLength('你好', 'ascii')); // 2（截断为多字节字符）
Buffer.allocUnsafe 直接从 Node.js 预分配的内存池中获取空间，不执行零填充填充，速度比 Buffer.alloc 快数倍。但这种速度优势是有代价的：如果分配到的内存之前存储过密钥等敏感数据，新 Buffer 中可能残留这些信息。因此，allocUnsafe 只适合在确认会立即写入完整数据的场景使用，如协议组装或文件读取缓冲。
5.1 TypedArray 与数据视图
Buffer 继承自 Uint8Array，因此可以无缝参与 TypedArray 体系：
const buf = Buffer.from([0x01, 0x02, 0x03, 0x04, 0x05, 0x06, 0x07, 0x08]);
// 使用 DataView 进行跨平台的字节操作
const view = new DataView(buf.buffer, buf.byteOffset, buf.byteLength);
console.log(view.getUint16(0, true)); // 小端序: 0x0201 = 513
console.log(view.getUint16(0, false)); // 大端序: 0x0102 = 258
console.log(view.getFloat64(0, true)); // 小端序: 解析为 64 位浮点数
// 通过不同 TypedArray 视图解读同一内存
const uint16 = new Uint16Array(buf.buffer, buf.byteOffset, 4);
console.log(uint16); // 根据平台字节序不同而变化
const float32 = new Float32Array(buf.buffer, buf.byteOffset, 2);
在网络协议解析、文件格式处理、硬件数据交互等场景中，精确控制字节序是不可或缺的。Node.js 的 Buffer 提供了 read/writeUInt16BE/LE、read/writeInt32BE/LE 等方法。对于更复杂的结构，DataView 提供了跨字节序的统一 API。但需要注意，直接创建 Uint16Array 或 Float64Array 视图时，字节序取决于运行平台的 CPU 架构；使用 DataView 的 true/false 参数可以明确指定字节序，避免平台差异带来的 bug。
6. 大文件流式处理实战
流式处理最常见的场景是大文件的读取、转换和持久化。以下示例展示了如何在有限内存中处理 GB 级数据：压缩、加密、分片写入。
const fs = require('fs');
const { pipeline } = require('stream/promises');
const zlib = require('zlib');
const crypto = require('crypto');
(async () => {
const input = fs.createReadStream('large-dataset.tar', {
highWaterMark: 256 * 1024 // 每次读取 256KB，减少系统调用次数
});
const gzip = zlib.createGzip({
level: 6, // 压缩级别 6 是速度与压缩率的平衡点
chunkSize: 16 * 1024
});
const key = Buffer.from(process.env.CIPHER_KEY, 'hex');
const iv = crypto.randomBytes(16);
const cipher = crypto.createCipheriv('aes-256-cbc', key, iv);
const output = fs.createWriteStream('large-dataset.tar.gz.enc');
output.write(iv); // 将 IV 写入文件头部，解密时需要
await pipeline(input, gzip, cipher, output);
console.log('流式压缩加密完成，内存峰值仅取决于 highWaterMark');
})();
// 对应解密流程
const fs = require('fs');
const { pipeline } = require('stream/promises');
const zlib = require('zlib');
const crypto = require('crypto');
(async () => {
const input = fs.createReadStream('large-dataset.tar.gz.enc');
// 先读取 IV
const iv = await new Promise((resolve) => {
input.once('data', (chunk) => {
resolve(chunk.slice(0, 16));
input.unshift(chunk.slice(16)); // 将剩余数据重新压回流
});
});
const key = Buffer.from(process.env.CIPHER_KEY, 'hex');
const decipher = crypto.createDecipheriv('aes-256-cbc', key, iv);
const gunzip = zlib.createGunzip();
const output = fs.createWriteStream('restored.tar');
await pipeline(input, decipher, gunzip, output);
console.log('解密解压完成');
})();
这段代码中有两个值得关注的技巧。一是 highWaterMark 的调优：对于磁盘 I/O，较大的 chunk 可以减少 read() 系统调用的次数。对于网络 I/O，过大的 chunk 会导致单次缓冲区填充时间过长，影响响应延迟。256KB 是文件流场景的常见选择。二是 input.unshift() 在解密示例中非常有用：可读流已经开始取出数据，但我们只需要前 16 字节作为 IV，剩余部分应保留给后续的解密流，unshift 可以将多余数据重新压回流的前端，确保数据不丢失。
延伸阅读
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。