Node.js Stream 原理与高性能数据处理

Stream 是 Node.js 中最强大也最容易被误解的特性之一。掌握它，你就能用极少的内存处理海量数据。

从一个真实场景说起

假设你需要处理一个 2GB 的日志文件，从中提取所有包含 “ERROR” 的行并写入新文件。直觉写法大概是这样的：

const fs = require('fs');

fs.readFile('huge-log.txt', 'utf-8', (err, data) => {
  if (err) throw err;
  const lines = data
    .split('\n')
    .filter(line => line.includes('ERROR'))
    .join('\n');
  fs.writeFile('errors-only.txt', lines, () => {
    console.log('Done!');
  });
});

这段代码看起来无害，但执行时会发生什么？V8 的内存直接飙到 2GB+，如果文件再大一点，进程直接 OOM 崩溃。因为 readFile 会把整个文件一次性加载进内存。

而用 Stream 重写，同样的任务只需要几十 MB 内存——无论文件有多大。

这就是 Stream 的核心价值：以恒定内存处理无限数据。

Stream 到底是什么？

Stream 不是”一次性交付所有数据”，而是把数据拆成小块，像流水线一样逐块传递。它的核心抽象可以浓缩为四个字：

数据流 + 背压（Backpressure）

想象一根水管：一端进水（可读流），一端出水（可写流），中间你可以接各种”接头”（Transform 流）来加工水流。

Node.js 提供了四种基础 Stream 类型：

核心机制：背压（Backpressure）

Stream 最精妙的设计是 背压机制。当消费端处理速度跟不上生产端时，Stream 会自动”刹车”，防止内存中积压过多数据。

这背后的实现依赖两个关键方法：

const { Readable } = require('stream');

// 自定义可读流：每 500ms 生产一批数据
class NumberProducer extends Readable {
  constructor(options) {
    super({ objectMode: true, highWaterMark: 3 });
    this.counter = 0;
  }

  _read() {
    // push(null) 表示流结束
    if (this.counter >= 10) {
      return this.push(null);
    }
    // push 返回 false 表示消费者跟不上，暂停生产
    const canContinue = this.push({ num: ++this.counter });
    console.log(`生产: ${this.counter}  (缓冲区还能接受更多? ${canContinue})`);
  }
}

highWaterMark 是缓冲区的”高水位线”（默认 16KB for Buffer，或 objectMode 下的 16 个对象）。当内部缓冲区超过这个阈值时，push() 返回 false，生产者应暂停。

const { pipeline } = require('stream');
const { Writable } = require('stream');

const slowConsumer = new Writable({
  objectMode: true,
  highWaterMark: 2,
  write(chunk, encoding, callback) {
    // 模拟慢速消费：1 秒处理一条
    setTimeout(() => {
      console.log(`消费: ${chunk.num}`);
      callback(); // 调用 callback 通知流"我好了，可以继续"
    }, 1000);
  }
});

// pipeline 自动管理背压和错误
pipeline(
  new NumberProducer(),
  slowConsumer,
  (err) => {
    if (err) console.error('Pipeline failed:', err);
    else console.log('全部完成!');
  }
);

输出会清晰地展示背压如何工作：

生产: 1  (缓冲区还能接受更多? true)
生产: 2  (缓冲区还能接受更多? true)
生产: 3  (缓冲区还能接受更多? false)   ← 暂停！
消费: 1
生产: 4  (缓冲区还能接受更多? false)
消费: 2
生产: 5  (缓冲区还能接受更多? false)
...

实战：构建一个 CSV 大文件处理管道

下面是一个真实场景的完整示例——处理一个可能有几百 MB 的 CSV 文件，过滤、转换后写入 JSON 输出，同时用 Gzip 压缩。全程内存友好。

const fs = require('fs');
const { pipeline, Transform } = require('stream');
const { createGzip } = require('zlib');
const csv = require('csv-parser');

// Step 1: 创建可读流，读取原始 CSV
const source = fs.createReadStream('large-data.csv');

// Step 2: CSV 解析（csv-parser 本身就是 Transform 流）
const parser = csv();

// Step 3: 自定义 Transform：过滤 + 数据转换
const filterAndTransform = new Transform({
  objectMode: true,
  transform(row, encoding, callback) {
    // 只保留 active 状态的用户
    if (row.status !== 'active') {
      return callback(); // 跳过这条记录
    }

    // 转换字段
    const result = {
      id: parseInt(row.id, 10),
      name: row.name,
      email: row.email,
      score: parseFloat(row.score),
      processedAt: new Date().toISOString()
    };

    this.push(JSON.stringify(result) + '\n');
    callback();
  }
});

// Step 4: Gzip 压缩
const gzip = createGzip();

// Step 5: 写入文件
const destination = fs.createWriteStream('output.jsonl.gz');

// pipeline 串联所有步骤，自动处理背压
pipeline(
  source,      // 读取
  parser,      // 解析 CSV
  filterAndTransform,  // 过滤 + 转换
  gzip,        // 压缩
  destination, // 写入
  (err) => {
    if (err) {
      console.error('Pipeline 失败:', err);
    } else {
      console.log('处理完成！output.jsonl.gz 已生成');
    }
  }
);

// 可选：添加进度监控
let bytesRead = 0;
source.on('data', (chunk) => {
  bytesRead += chunk.length;
  process.stdout.write(`\r已读取: ${(bytesRead / 1024 / 1024).toFixed(1)} MB`);
});

这个管道的优雅之处在于：
– 每个组件只关心自己那一环
– 背压自动在整条链路上传导
– 任何环节出错，pipeline 都会自动清理资源

pipeline vs pipe：为什么你应该用前者

你可能见过老代码里到处是 .pipe()：

src.pipe(transform).pipe(gzip).pipe(dest);

这看起来很简洁，但有一个致命缺陷：错误不会自动传播。中间某个流崩了，其他流不会被销毁，导致资源泄漏。

pipeline 是 Node.js 10+ 推出的官方替代方案：

const { pipeline } = require('stream');

pipeline(src, transform, gzip, dest, (err) => {
  if (err) {
    // 所有流都会被正确销毁
    console.error('出错:', err);
  }
});

规则很简单：永远用 pipeline，别直接用 .pipe()。

另外，Node.js 15+ 还提供了 Promise 版本：

const { pipeline } = require('stream/promises');

async function run() {
  try {
    await pipeline(src, transform, gzip, dest);
    console.log('完成!');
  } catch (err) {
    console.error('失败:', err);
  }
}

五个常见的 Stream 陷阱

1. 忘记调用 callback()

在 Writable/Transform 的 _write 或 transform 方法中，必须调用 callback。否则流会永远卡住：

// ❌ 错误：永远不调用 callback
write(chunk, encoding, callback) {
  fs.appendFile('log.txt', chunk, () => {});
  // 忘了 callback() — 流会挂起！
}

// ✅ 正确
write(chunk, encoding, callback) {
  fs.appendFile('log.txt', chunk, callback);
}

2. 在 Transform 中忘记传数据

// ❌ 错误：调用 callback 时传了数据，但没 push
transform(chunk, encoding, callback) {
  callback(null, processedChunk); // 不会发出任何东西
}

// ✅ 正确方式一：push + callback
transform(chunk, encoding, callback) {
  this.push(processedChunk);
  callback();
}

// ✅ 正确方式二：callback(err, data) — 语法糖
transform(chunk, encoding, callback) {
  callback(null, processedChunk);
}

3. 忽略 objectMode

Buffer 流和对象流的 highWaterMark 含义完全不同：

// Buffer 模式：highWaterMark = 16384 字节
const bufStream = new Readable();

// Object 模式：highWaterMark = 16 个对象
const objStream = new Readable({ objectMode: true });

没有 objectMode 时不能 push 非 Buffer/string 的数据，否则会报错。

4. 在异步迭代中忽略错误

// Node.js 12+ 支持 for await...of 消费 Readable
async function process() {
  for await (const chunk of readableStream) {
    // 如果这里抛错，流可能不会被销毁
  }
}

最好包一层 try/catch 并手动销毁流。

5. 重复监听 ‘data’ 事件

// ❌ 每个 'data' 监听器都会把流切换到 flowing 模式
stream.on('data', handler1);
stream.on('data', handler2); // 可能导致数据被重复消费或竞争

性能数据：Stream vs 传统方式

用上面的 CSV 处理场景做基准测试（500MB CSV 文件，约 200 万行）：

Stream 方案以接近全量加载的速度，消耗不到 5% 的内存。

什么时候该用 Stream？

用 Stream 是一个好主意的场景：

• 处理大文件（日志、CSV、视频）
• HTTP 请求/响应体（尤其是文件上传下载）
• 数据压缩/解压、加密/解密
• 数据库大量数据的导入导出
• 实时数据流（WebSocket、MQTT）

不该用 Stream 的场景：

• 数据量很小（几 KB），直接读反而更简单
• 需要随机访问的数据（Stream 是顺序的）
• 业务逻辑高度同步、不需要流式处理

总结

Stream 不是 Node.js 的”高级特性”，而是构建高性能数据处理应用的基础设施。它的核心思想值得内化到你的编程思维中：

1. 分块处理：不要一次加载所有数据
2. 背压管理：尊重消费者的节奏
3. 组合优于继承：用 pipeline 串联小而专注的流

下次遇到”文件太大”或”内存飙升”的问题，第一个念头不应该是加内存——试着用 Stream 重构它。

本文适合有 Node.js 基础、想深入理解 Stream 机制的开发者。如果你对 Node.js 的 Event Loop、Worker Threads 等主题也感兴趣，欢迎在评论区留言。