Node.js Stream（流）是处理大文件与高并发 IO 的核心抽象：它把数据切成小块，边读边处理，而不是一次性把整个文件塞进内存。本文用可运行示例讲清流的四种类型、pipe 管道、Transform 转换与背压机制，帮你写出内存稳定、吞吐更高的 Node 服务。
一、为什么要用流：从「整读内存」到「边读边处理」
在早期写法里，读取一个 2GB 的日志文件常写成 fs.readFile('big.log')，Node 会先把整份文件读进 Buffer，再一次性交给回调。这意味着内存峰值等于文件大小，并发稍高就被 OOM killer 干掉。流的思路相反：数据源像水龙头，消费者像水杯，管道按节奏一段段流过，任意时刻内存只占「一小块」。这也是为什么几乎所有 Node 内置模块（HTTP、fs、zlib、crypto）都继承自 stream。
一个直观对比：用流拷贝大文件，内存常驻只有几十 KB；用 readFile 则等于文件体积。当请求体、响应体、文件、压缩都走流，服务的并发上限会显著抬升。
二、流的四种类型
| 类型 | 说明 | 典型场景 |
|---|---|---|
| Readable | 可读流，数据从源头流出 | 读取文件、HTTP 请求体 |
| Writable | 可写流，数据流入目标 | 写文件、HTTP 响应 |
| Duplex | 双向流，既读又写 | 网络 socket、TCP 连接 |
| Transform | 转换流，写入后变换再读出 | gzip 压缩、加密、转码 |
绝大多数业务只需要和 Readable / Writable / Transform 打交道。Transform 尤其好用：它天生是 Duplex，但读进来的数据会自动被变换后流出，比如压缩、加解密、行过滤。
三、用 pipe 读写大文件（基础实战）
最省心的写法是用 pipe() 把可读流接到可写流。Node 会自动处理数据节奏，不必手写缓冲。下面这段代码无论文件多大，内存都只占用一个 chunk 的大小。
const fs = require('fs');
// 大文件拷贝：边读边写，内存只占一小块
fs.createReadStream('big.iso')
.pipe(fs.createWriteStream('big-copy.iso'))
.on('finish', () => console.log('拷贝完成'))
.on('error', (err) => console.error('出错', err));注意：错误处理必须挂在管道两端。pipe 默认会把可读流的 error 转发到可写流，但稳妥起见两端都监听 error，否则流报错会让整个 Node 进程崩溃。
四、Transform 流：边传边处理
想要在传输途中做点什么（压缩、转码、脱敏），Transform 流是最自然的承载。下面先用 zlib 做实时 gzip，再用一个自定义 Transform 把输入全部转大写，两段都是「写进→变换→流出」的范式。
const fs = require('fs');
const zlib = require('zlib');
fs.createReadStream('access.log')
.pipe(zlib.createGzip()) // Transform：实时压缩
.pipe(fs.createWriteStream('access.log.gz'))
.on('finish', () => console.log('压缩完成'));const { Transform } = require('stream');
const upper = new Transform({
transform(chunk, enc, cb) {
this.push(chunk.toString().toUpperCase());
cb();
}
});
process.stdin.pipe(upper).pipe(process.stdout);自定义 Transform 只需实现 transform(chunk, encoding, callback)：处理完一块后调用 this.push(结果) 再 cb()。需要结束时做收尾（如压缩的 trailer），可实现 _flush(cb)。
五、理解背压（Backpressure）：流不会撑爆内存
背压是流的灵魂。当可读流生产太快、可写流消费太慢时，可写流的缓冲区会被填满并返回 false，pipe 会自动暂停可读流，等可写流排空（drain）后再恢复。这就是为什么 pipe 默认就是内存安全的。如果你手动用 data 事件，就必须自己处理背压，否则高速源会瞬间灌爆内存。
const fs = require('fs');
const rs = fs.createReadStream('big.bin');
const ws = fs.createWriteStream('out.bin');
rs.on('data', (chunk) => {
if (!ws.write(chunk)) { // 返回 false = 缓冲区已满
rs.pause(); // 暂停读取，等待排空
ws.once('drain', () => rs.resume());
}
});
rs.on('end', () => ws.end());六、对象流与高并发 IO 场景
除了字节流，Node 还支持「对象模式」流（objectMode），每个 chunk 是一个 JS 对象，非常适合 ETL、爬虫、批量写库等场景。Node 12+ 还支持把流当成异步可迭代对象，用 for await...of 逐块消费，语法比事件回调干净很多。
const fs = require('fs');
// 逐块消费，语法更干净
(async () => {
for await (const chunk of fs.createReadStream('big.txt', { encoding: 'utf8' })) {
process.stdout.write(chunk); // 每块约 highWaterMark 大小
}
})();// 默认 64KB；高吞吐场景可调大，降低系统调用次数
fs.createReadStream('big.bin', { highWaterMark: 1024 * 1024 }); // 1MB 一块把大文件落库、日志清洗、对象存储上传这类任务改写成对象流后，常能把内存从「文件级」降到「块级」，同时吞吐更高。配合容器化与 CI 自动化，这类脚本可以稳定跑在资源受限的节点上——关于把 Node 服务打包进容器，可参考本站Docker 实战；把流处理脚本纳入GitHub Actions做定时跑批也是常见做法。
七、常见坑与排错清单
| 坑 | 现象 | 对策 |
|---|---|---|
| 忘记监听 error | 流出错，进程直接崩溃 | 每个流都挂 .on(‘error’) |
| 用 data 事件却不处理背压 | 高速源撑爆内存 | 优先用 pipe，或手动 drain |
| 编码未指定 | 中文乱码 | 指定 encoding，或统一用 Buffer |
| finish 与 end 混淆 | 资源被提前关闭 | Writable 用 finish，Readable 用 end |
| data 监听导致暂停模式失效 | 流不自动流动 | 用 resume()/pipe 恢复流动模式 |
另外，流默认处于「暂停模式」，只有调用 pipe、resume 或监听 data 才会流动。调试时如果数据迟迟不来，先确认流处于流动模式。需要逐行处理文本，用 readline.createInterface({ input: rs }) 比手动按换行切分更稳。
八、与 REST / 缓冲方案的取舍
流不是银弹。对于返回小体积 JSON 的普通 REST 接口，直接序列化响应更简单；但当响应体很大（文件下载、报表导出）、或输入来自大文件 / 大请求体时，流是必选项。判断标准就一条：数据能否「边产生边消费」。能，就用流；不能，就别为了用而用。
在前端协作上，后端用流吐出大文件或分块响应，前端配合Vue3的渐进加载或 fetch + ReadableStream 就能做出丝滑的进度体验；本地开发调试流脚本，VS Code的断点与条件监视很好用。
九、用 pipeline() 替代裸 pipe：自动清理错误
裸 pipe 虽然方便，但错误清理不够彻底：一旦中间某个流报错，两端的流不会被自动销毁，可能留下文件句柄或 socket 泄漏，长期运行就会耗尽资源。Node 在 stream/promises 中提供的 pipeline() 会把整条链串起来，任何一个环节出错都会自动关闭所有流，并以 Promise 形式把错误抛给你，配合 try/catch 就能优雅收尾，不必为每个流单独挂监听。
const fs = require('fs');
const zlib = require('zlib');
const { pipeline } = require('stream/promises');
async function compress(src, dest) {
try {
await pipeline(
fs.createReadStream(src),
zlib.createGzip(),
fs.createWriteStream(dest)
);
console.log('完成，已自动清理所有流');
} catch (err) {
console.error('管道失败', err);
}
}几乎所有生产代码都建议用 pipeline 而非手动 pipe：它把「出错即释放资源」变成了默认行为，省去大量样板错误处理，也让异步流程更符合 async/await 的写法习惯。
十、实战：HTTP 响应边压缩边返回
流的另一个高频场景是 Web 服务：用户请求一个大报表或日志包，后端不必在内存里拼好再一次性发送，而是用流一边读文件、一边 gzip、一边写进响应，首字节时间大幅提前，内存也几乎不增长。下面用 Express 示范：只要把可读流 pipe 到 res，并加上正确的 Content-Encoding 头即可。
const fs = require('fs');
const zlib = require('zlib');
const express = require('express');
const app = express();
app.get('/download/:file', (req, res) => {
res.setHeader('Content-Type', 'application/octet-stream');
res.setHeader('Content-Encoding', 'gzip');
fs.createReadStream(req.params.file)
.pipe(zlib.createGzip())
.pipe(res); // 直接流向 HTTP 响应
});
app.listen(3000);这里的 res 本身就是一个 Writable 流，所以可以直接作为 pipe 的终点。响应体走流之后，无论文件多大，Node 进程内存都保持平稳，这也是文件下载、视频切片、大批量导出接口的标准写法。把它和前面提到的对象流、背压、pipeline 组合，就能搭出一套既省内存又易维护的数据处理管线。
十一、小结
Node.js Stream 用「小块流动 + 背压」解决了大文件和高并发 IO 的内存与吞吐难题。记住四件事：优先 pipe 或 pipeline 而非手写 data 循环；任何流都要挂 error（pipeline 可自动清理）；需要边传边改就用 Transform；对象流 + for await 让批处理代码既安全又清爽。把这些用进你的文件服务、日志管道和 ETL 任务，服务会更稳、更省内存。
