Node.js Stream
四种流、pipe、背压、以及用 pipeline 正确处理错误。
Key Points流把大数据切成小块处理，`pipeline` 会自动管好错误和关闭。
四种流
| Type | Example | Direction |
|---|---|---|
| Readable | fs.createReadStream | 读入 |
| Writable | fs.createWriteStream | 写出 |
| Duplex | net.Socket | 双向 |
| Transform | zlib.createGzip | 读入→转换→写出 |
pipe 一条链
JavaScript
const fs = require('node:fs');
const zlib = require('node:zlib');
// 压缩：读 → gzip → 写
fs.createReadStream('access.log')
.pipe(zlib.createGzip())
.pipe(fs.createWriteStream('access.log.gz'))
.on('finish', () => console.log('压缩完成'));Output
压缩完成pipe 的坑：错误不传递
JavaScript
// 危险：中间环节报错，目标流不会被关闭，文件句柄泄漏
src.pipe(gzip).pipe(dest);
// 安全：pipeline 自动传播错误并销毁所有流
const { pipeline } = require('node:stream/promises');
await pipeline(
fs.createReadStream('access.log'),
zlib.createGzip(),
fs.createWriteStream('access.log.gz')
);pipe only forwards errors from the source; a failing Transform does not close the destination. Production code should always use stream.pipeline.背压
When writes cannot keep up, write() returns false. Stop reading and resume on 'drain'. pipe and pipeline handle this; hand-written loops must do it themselves:
JavaScript
const src = fs.createReadStream('big.bin', { highWaterMark: 64 * 1024 });
const dest = fs.createWriteStream('copy.bin');
src.on('data', (chunk) => {
const ok = dest.write(chunk);
if (!ok) {
src.pause();
dest.once('drain', () => src.resume());
}
});
src.on('end', () => dest.end());自己写一个 Transform
JavaScript
const { Transform } = require('node:stream');
const upper = new Transform({
transform(chunk, encoding, callback) {
callback(null, chunk.toString().toUpperCase());
},
});
process.stdin.pipe(upper).pipe(process.stdout);Output
$ echo hello | node upper.js
HELLO用 async 迭代器读流
JavaScript
for await (const chunk of fs.createReadStream('big.log')) {
if (chunk.includes('ERROR')) console.log('发现错误行');
}A Readable is an async iterable, so for await reads better than listening to 'data' and applies backpressure naturally — one chunk at a time.
四种流
| 类型 | 例子 | 方向 |
|---|---|---|
| 可读 | fs.createReadStream | 读入 |
| 可写 | fs.createWriteStream | 写出 |
| 双工 | net.Socket | 双向 |
| 转换 | zlib.createGzip | 读入→转换→写出 |
pipe 一条链
JavaScript
const fs = require('node:fs');
const zlib = require('node:zlib');
// 压缩：读 → gzip → 写
fs.createReadStream('access.log')
.pipe(zlib.createGzip())
.pipe(fs.createWriteStream('access.log.gz'))
.on('finish', () => console.log('压缩完成'));Output
压缩完成pipe 的坑：错误不传递
JavaScript
// 危险：中间环节报错，目标流不会被关闭，文件句柄泄漏
src.pipe(gzip).pipe(dest);
// 安全：pipeline 自动传播错误并销毁所有流
const { pipeline } = require('node:stream/promises');
await pipeline(
fs.createReadStream('access.log'),
zlib.createGzip(),
fs.createWriteStream('access.log.gz')
);pipe 只在源流出错时把错误传下去，中间 Transform 出错不会关闭目标流。生产代码一律用 stream.pipeline。背压
写入速度跟不上读取速度时，write() 返回 false。此时应停止读，等 'drain' 再继续。pipe 和 pipeline 已经帮你做了这件事；手写循环时要自己处理：
JavaScript
const src = fs.createReadStream('big.bin', { highWaterMark: 64 * 1024 });
const dest = fs.createWriteStream('copy.bin');
src.on('data', (chunk) => {
const ok = dest.write(chunk);
if (!ok) {
src.pause();
dest.once('drain', () => src.resume());
}
});
src.on('end', () => dest.end());自己写一个 Transform
JavaScript
const { Transform } = require('node:stream');
const upper = new Transform({
transform(chunk, encoding, callback) {
callback(null, chunk.toString().toUpperCase());
},
});
process.stdin.pipe(upper).pipe(process.stdout);Output
$ echo hello | node upper.js
HELLO用 async 迭代器读流
JavaScript
for await (const chunk of fs.createReadStream('big.log')) {
if (chunk.includes('ERROR')) console.log('发现错误行');
}可读流本身就是异步可迭代对象，配合 for await 比监听 'data' 更好读，也能自然形成背压（每次只取一块）。