Node.js 流式处理的三个反模式：别再让内存成为瓶颈

前言

不夸张地说，90% 的 Node.js 后端开发者在处理文件上传、日志聚合、数据导出时都有过一个共同的疑问："为什么我的服务处理 500MB 的 CSV 文件会 OOM？我明明用了 Stream 啊。"

答案很残酷：你只是"写"了 Stream，但没有真正"流"起来。

这篇文章拆解三个最常见的流式处理反模式，每一个都来自真实的生产事故。

反模式一：读完再写——假装在流

最常见的"假流"代码：

// ❌ 反模式：看似用了 createReadStream，实际上全读进内存
const fs = require('fs');

async function transformFile(input, output) {
  const chunks = [];
  const stream = fs.createReadStream(input);
  
  for await (const chunk of stream) {
    chunks.push(chunk);  // 全堆在内存里
  }
  
  const allData = Buffer.concat(chunks);
  const transformed = allData.toString()
    .split('\n')
    .map(line => line.toUpperCase())
    .join('\n');
  
  fs.writeFileSync(output, transformed);
}

这段代码的问题：for await 虽然逐块读取，但 chunks.push 把每一块都留在了内存。500MB 文件 = 500MB 内存占用，完全不"流"。

正确做法：

// ✅ 真正流式：pipe + Transform
const { Transform } = require('stream');
const { pipeline } = require('stream/promises');

async function transformFile(input, output) {
  const upperCaseTransform = new Transform({
    transform(chunk, encoding, callback) {
      callback(null, chunk.toString().toUpperCase());
    }
  });

  await pipeline(
    fs.createReadStream(input),
    upperCaseTransform,
    fs.createWriteStream(output)
  );
}

关键差异：

方式	内存占用	处理 1GB 文件
for await + push	~1GB	OOM
pipeline + Transform	~64KB（默认 highWaterMark）	稳定

反模式二：忽视背压——上游洪水，下游旱地

背压（Backpressure）是 Stream 的核心机制。当消费者速度跟不上生产者时，Stream 会自动"捏闸"。但手工 data 事件会绕开这个机制：

// ❌ 反模式：data 事件无视背压
const readStream = fs.createReadStream('large.log');
const writeStream = fs.createWriteStream('output.log');

readStream.on('data', (chunk) => {
  // writeStream.write 返回 false 表示缓冲区满了
  // 但 data 事件照样轰炸，完全不减速
  writeStream.write(processChunk(chunk));
});

readStream.on('end', () => {
  console.log('done, maybe with OOM');
});

这个问题在生产中的表现是：处理小文件正常，文件一大就 OOM。因为 data 事件不管 write() 返回 false，继续推送数据。

正确做法：pipe 或 pipeline 自动处理背压：

// ✅ pipeline 内置背压管理
const { pipeline } = require('stream/promises');
const { Transform } = require('stream');

const processTransform = new Transform({
  transform(chunk, encoding, callback) {
    // 模拟耗时处理
    setTimeout(() => {
      callback(null, processChunk(chunk));
    }, 10);
  }
});

await pipeline(
  fs.createReadStream('large.log'),
  processTransform,
  fs.createWriteStream('output.log')
);
// pipeline 会在 writeStream 缓冲区满时自动暂停 readStream

如果你必须手动处理，用 readable + read() 配合 drain 事件：

// ✅ 手动背压管理
readStream.on('readable', () => {
  let chunk;
  while ((chunk = readStream.read()) !== null) {
    const canContinue = writeStream.write(processChunk(chunk));
    if (!canContinue) {
      // 缓冲区满了，暂停读取
      readStream.pause();
      writeStream.once('drain', () => readStream.resume());
      break;
    }
  }
});

反模式三：高水位开航母——默认配置踩坑

Stream 的 highWaterMark 默认值是 16KB（对于 fs.createReadStream 是 64KB）。这个值对大多数场景够用，但两个极端会让你翻车：

极端一：水位太低，IO 碎片化

// ❌ 读 100GB 的数据库 dump，64KB 一次
// 结果：150 万次系统调用，IO 开销巨大
fs.createReadStream('db_dump.sql')
  .pipe(transform)
  .pipe(fs.createWriteStream('output.sql'));

极端二：水位太高，内存吃紧

// ❌ 100MB 高水位，并发 10 个请求 = 1GB 内存
const stream = new Readable({
  highWaterMark: 100 * 1024 * 1024, // 100MB
  read() { /* ... */ }
});

工程化建议：

// ✅ 根据场景调整 highWaterMark
const { pipeline } = require('stream/promises');

// 大文件顺序处理 → 提高水位减少系统调用
const reader = fs.createReadStream('dump.sql', {
  highWaterMark: 1024 * 1024  // 1MB，系统调用减少 16 倍
});

// 高并发场景 → 降低水位控制内存
const concurrentReader = fs.createReadStream(file, {
  highWaterMark: 16 * 1024  // 16KB，配合背压稳定内存
});

await pipeline(reader, transform, writer);

一个经验法则：

• 单任务大文件 → 1-4 MB highWaterMark
• 并发任务 → 16-64 KB（默认值通常就对了）
• 网络流 → 16KB（TCP 默认窗口的友好值）

别忘了错误处理

pipeline 相比 pipe 的最大优势是错误传播：

// ❌ pipe：错误不会自动传播
readStream.pipe(transform).pipe(writeStream);
readStream.on('error', handleErr);
transform.on('error', handleErr);  // 容易遗漏
writeStream.on('error', handleErr);

// ✅ pipeline：任一环节出错都会清理并 reject
const { pipeline } = require('stream/promises');
try {
  await pipeline(readStream, transform, writeStream);
} catch (err) {
  // 所有 stream 已自动销毁
  console.error('Pipeline failed:', err);
}

总结

反模式	症状	修复
读完再写	大文件 OOM	pipeline + Transform
忽视背压	内存持续增长	pipeline 或 readable + drain
高水位不当	IO 慢或内存高	按场景调 highWaterMark

Node.js Stream 的 API 设计很优雅，但"能用"和"用好"之间隔着一层对背压机制的理解。下次有人问你为什么服务 OOM 了，先看看他的 Stream 是不是真在"流"。