Node 流转换分块管道

示例说明

把几个 GB 的日志一次性读进内存必然崩溃。流式管道让数据按块流动：可读流按 highWaterMark 分块吐出，Transform 逐块转换，gzip 压缩后写入目标文件，任意时刻内存里只有一个块。pipeline 的价值在于任一环节出错会沿管道双向传播并销毁全部流，不会悬挂句柄。

技术要点

• Transform 子类实现 _transform 与 _flush 两个钩子
• pipeline 自动处理背压与错误传播
• highWaterMark 控制分块大小

代码解读

LineCount 继承 Transform，_transform 里统计换行符数量并把文本转大写后下推；_flush 在流结束时追加总计行数注释，保证尾部统计随数据一起到达下游。pipeline 依次接收四个流，返回的 Promise 在全部数据落盘后 resolve。

使用建议

对象模式适合结构化数据但注意引用共享问题。压缩级别与分块大小是吞吐与 CPU 的权衡项，按文件类型调优。pipeline 失败后要检查目标文件是否为半成品并清理。

代码示例

import { createReadStream, createWriteStream } from 'node:fs';
import { Transform } from 'node:stream';
import { pipeline } from 'node:stream/promises';
import { createGzip } from 'node:zlib';

class LineCount extends Transform {
  constructor() {
    super({ objectMode: false });
    this.lines = 0;
  }

  _transform(chunk, encoding, callback) {
    const text = chunk.toString('utf8');
    this.lines += (text.match(/\n/g) ?? []).length;
    this.push(text.toUpperCase());
    callback();
  }

  _flush(callback) {
    this.push('\n# total lines: ' + this.lines + '\n');
    callback();
  }
}

await pipeline(
  createReadStream('input.log', { encoding: 'utf8', highWaterMark: 64 * 1024 }),
  new LineCount(),
  createGzip({ level: 6 }),
  createWriteStream('output.log.gz'),
);

console.log('pipeline finished');