Node.js Stream 深度解析：从背压控制到管道模式的生产实践

后端开发

一次 OOM 排查引出的认知升级

两年前接手一个日志处理服务，功能很简单——读取 Nginx 访问日志，解析后写入 ClickHouse。老代码长这样：

const fs = require('fs');

async function processLogs(filePath) {
  const content = await fs.promises.readFile(filePath, 'utf-8');
  const lines = content.split('\n');

  for (const line of lines) {
    const parsed = parseLogLine(line);
    await insertToClickHouse(parsed);
  }
}

日常 200MB 的日志文件跑得没问题。直到某天运维把日志轮转周期从 1 天改成了 1 周，单个文件膨胀到 1.8GB。服务刚启动 3 秒，readFile 就把整个文件读进内存——直接 OOM Kill。

这是 Node.js Stream 最经典的使用场景。Stream 不是「流式处理」的花哨名词，而是当数据量超过可用内存时，你唯一的选择。

理解 Stream 的四种类型

Node.js 的 Stream 模块基于 EventEmitter，四种核心类型各有用途：

类型 职责 类比 核心方法
Readable 生产数据 水龙头 read(), pipe()
Writable 消费数据 下水道 write(), end()
Transform 边读边写边转换 净水器 _transform()
Duplex 可读可写（双向独立） 对讲机 _read() + _write()

Readable Stream

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

// 从数组创建可读流
const readable = Readable.from([
  '第一行数据\n',
  '第二行数据\n',
  '第三行数据\n',
]);

readable.on('data', (chunk) => {
  console.log('收到:', chunk.toString());
});

readable.on('end', () => {
  console.log('读取完毕');
});

最常用的是 fs.createReadStream：

const fs = require('fs');

// 文件再大也不会爆内存——每次只读 64KB
const stream = fs.createReadStream('1.8gb-logfile.log', {
  encoding: 'utf-8',
  highWaterMark: 64 * 1024,  // 内部缓冲区 64KB
});

Writable Stream

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

const writable = fs.createWriteStream('output.log', {
  flags: 'a',          // 追加模式
  highWaterMark: 16 * 1024,
});

// write() 返回 false 表示内部缓冲区满了，需要等待 drain
const canContinue = writable.write('这是一行日志\n');
if (!canContinue) {
  writable.once('drain', () => {
    console.log('缓冲区已排空，可以继续写入');
  });
}

write() 的返回值是理解背压的关键——后面会详细讲。

Transform Stream

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

// 将日志行转为 JSON 的转换流
const logParser = new Transform({
  readableObjectMode: true,    // 输出 JS 对象而非 Buffer
  writableObjectMode: false,   // 输入是 Buffer/字符串

  transform(chunk, encoding, callback) {
    const lines = chunk.toString().split('\n').filter(Boolean);

    for (const line of lines) {
      try {
        const parsed = JSON.parse(line);
        this.push(parsed);
      } catch (err) {
        // 解析失败的行可以通过 'error' 事件或单独的 error stream 传递
        this.emit('parse-error', { line, error: err.message });
      }
    }

    callback();
  },
});

Duplex Stream

顾名思义，可读可写。TCP Socket、加密解密流都是典型的 Duplex：

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

// 一个简单的回显流
const echo = new Duplex({
  read(size) {
    // 在这里实现读取逻辑
  },
  write(chunk, encoding, callback) {
    console.log('写入:', chunk.toString());
    this.push(chunk);   // 写入的数据立即变成可读
    callback();
  },
});

背压（Backpressure）——Stream 最核心的概念

背压是指消费速度跟不上生产速度时的流量控制机制。不用背压，数据会在内存中堆积直到 OOM。

const fs = require('fs');
const zlib = require('zlib');

// ❌ 没有背压保护
const readStream = fs.createReadStream('huge-file.log');
const gzipStream = zlib.createGzip();
const writeStream = fs.createWriteStream('huge-file.log.gz');

readStream.on('data', (chunk) => {
  // 不管写入端能不能消化，只管灌
  writeStream.write(chunk);
});
// 问题：如果 writeStream 内部缓冲区满了（highWaterMark），
// 它返回 false，但这里被忽略了。数据持续读取，内存堆积。

正确的背压感知写法：

// ✅ 正确：pipe() 自动处理背压
readStream.pipe(gzipStream).pipe(writeStream);

// 或者手动实现
readStream.on('data', (chunk) => {
  const canWrite = writeStream.write(chunk);
  if (!canWrite) {
    // 写入端缓冲区满了，暂停读取
    readStream.pause();
  }
});

writeStream.on('drain', () => {
  // 写入端消化完了，恢复读取
  readStream.resume();
});

highWaterMark——背压的阈值

场景 默认值 建议值
文件读取流 64 KB (65536) 保持默认或调大到 256KB
网络 Socket 16 KB (16384) 取决于带宽
对象模式 16 个对象 根据单对象大小调整
压缩流（Gzip） 16 KB 调大到 64KB 减少上下文切换

调大 highWaterMark 会提高吞吐但增加内存占用。一个经验法则：

// 根据文件大小动态调整
function createOptimalReadStream(filePath) {
  const stat = fs.statSync(filePath);
  const size = stat.size / (1024 * 1024);  // MB

  let highWaterMark;
  if (size < 10) {
    highWaterMark = 64 * 1024;       // 小文件 64KB
  } else if (size < 500) {
    highWaterMark = 256 * 1024;      // 中文件 256KB
  } else {
    highWaterMark = 1024 * 1024;     // 大文件 1MB
  }

  return fs.createReadStream(filePath, { highWaterMark });
}

pipe vs pipeline

pipe() 从 Node.js 诞生之初就存在，但有一个致命缺陷——它不会转发错误：

// ❌ pipe() 的错误陷阱
fs.createReadStream('nonexistent.log')
  .pipe(transformStream)
  .pipe(writeStream);

// 如果中间某个流报错，错误不会沿着 pipe 链传播
// 错误会导致未处理异常，进程退出

Node.js 10 引入的 pipeline() 解决了这个问题：

const { pipeline } = require('stream/promises');
// 或者 util.promisify(stream.pipeline)

async function processFile(inputPath, outputPath) {
  try {
    await pipeline(
      fs.createReadStream(inputPath),
      zlib.createGzip(),
      fs.createWriteStream(outputPath),
    );
    console.log('压缩完成');
  } catch (err) {
    console.error('压缩失败:', err.message);
    // pipeline 会自动销毁所有中间的 stream
  }
}

pipeline 的三个好处

  1. 自动错误传播——任一流报错，整个管道报错
  2. 自动销毁——报错后清理所有流，不会泄漏资源
  3. 回调/Promise 双支持——旧代码用回调，新代码用 async/await

自定义流实战

场景一：CSV 按行转换流

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

class CsvTransformer extends Transform {
  constructor(delimiter = ',') {
    super({ readableObjectMode: true });
    this.delimiter = delimiter;
    this.buffer = '';
  }

  _transform(chunk, encoding, callback) {
    this.buffer += chunk.toString();

    // 按行分割
    const lines = this.buffer.split('\n');
    // 最后一行可能不完整，留在 buffer 中
    this.buffer = lines.pop();

    for (const line of lines) {
      if (line.trim()) {
        this.push(line.split(this.delimiter));
      }
    }

    callback();
  }

  _flush(callback) {
    // 处理最后的残片
    if (this.buffer.trim()) {
      this.push(this.buffer.split(this.delimiter));
    }
    callback();
  }
}

// 使用
const csvStream = new CsvTransformer();
csvStream.on('data', (row) => {
  console.log('行数据:', row);
});

csvStream.write('name,age,city\n');
csvStream.write('Alice,28,NYC\n');
csvStream.write('Bob,35,SF\n');
csvStream.end();

场景二：限速写入流

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

class RateLimitedWritable extends Writable {
  constructor(options, maxOpsPerSecond) {
    super(options);
    this.interval = 1000 / maxOpsPerSecond;
    this.lastWrite = 0;
  }

  _write(chunk, encoding, callback) {
    const now = Date.now();
    const elapsed = now - this.lastWrite;
    const delay = Math.max(0, this.interval - elapsed);

    setTimeout(() => {
      this.lastWrite = Date.now();
      // 实际写入逻辑
      console.log('写入:', chunk.toString().trim());
      callback();
    }, delay);
  }
}

// 限制每秒 5 次写入
const rateLimited = new RateLimitedWritable({}, 5);

for (let i = 0; i < 20; i++) {
  rateLimited.write(`消息 ${i}\n`);
}
rateLimited.end();

场景三：分块聚合流（Batch Transform）

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

class BatchTransform extends Transform {
  constructor(batchSize = 100) {
    super({ readableObjectMode: true, writableObjectMode: true });
    this.batchSize = batchSize;
    this.buffer = [];
  }

  _transform(item, encoding, callback) {
    this.buffer.push(item);

    if (this.buffer.length >= this.batchSize) {
      this.push([...this.buffer]);
      this.buffer = [];
    }

    callback();
  }

  _flush(callback) {
    if (this.buffer.length > 0) {
      this.push([...this.buffer]);
    }
    callback();
  }
}

// 每 100 条数据组成一个批次批量写入数据库
await pipeline(
  incomingDataStream,
  new BatchTransform(100),
  async function* (source) {
    for await (const batch of source) {
      await db.bulkInsert(batch);
      console.log(`已入库 ${batch.length} 条`);
    }
  },
);

性能对比：Stream vs 全量加载

我写了一个简单的 benchmark 对比三种方式处理 500MB 文件：

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

// 方式 1：全量加载
async function fullLoad(path) {
  const data = await fs.promises.readFile(path, 'utf-8');
  const lines = data.split('\n');
  let count = 0;
  for (const line of lines) {
    if (line.includes('ERROR')) count++;
  }
  return count;
}

// 方式 2：逐行读取（readline）
const readline = require('readline');
async function readlineMethod(path) {
  const rl = readline.createInterface({
    input: fs.createReadStream(path),
    crlfDelay: Infinity,
  });
  let count = 0;
  for await (const line of rl) {
    if (line.includes('ERROR')) count++;
  }
  return count;
}

// 方式 3：Transform Stream
async function streamMethod(path) {
  let count = 0;
  const counter = new (require('stream').Transform)({
    transform(chunk, encoding, callback) {
      const lines = chunk.toString().split('\n');
      for (const line of lines) {
        if (line.includes('ERROR')) count++;
      }
      callback();
    },
  });

  await pipeline(fs.createReadStream(path), counter);
  return count;
}

实测结果（500MB 日志文件，MacBook Pro M1，Node.js 22）：

方式 耗时 峰值内存 说明
全量加载 0.8s 580 MB 文件 + 字符串开销
readline 1.5s 45 MB 逐行解析有开销
Transform Stream 1.2s 42 MB 块级处理，性能与内存平衡

Stream 版本比全量加载略慢 0.4 秒，但内存只用了 7%。当文件再大时，全量加载直接挂掉，Stream 照常运行。

生产级场景

场景一：大文件上传（前后端协作）

// Node.js 端接收分块上传
const http = require('http');
const { pipeline } = require('stream/promises');
const fs = require('fs');
const crypto = require('crypto');

http.createServer(async (req, res) => {
  if (req.method !== 'PUT') {
    res.writeHead(405);
    return res.end();
  }

  const fileId = crypto.randomUUID();
  const filePath = `./uploads/${fileId}`;

  try {
    await pipeline(
      req,                                    // 直接从 HTTP 请求读取
      fs.createWriteStream(filePath),         // 边收边写磁盘
    );
    res.writeHead(200, { 'Content-Type': 'application/json' });
    res.end(JSON.stringify({ fileId }));
  } catch (err) {
    res.writeHead(500);
    res.end('Upload failed');
  }
}).listen(3000);

场景二：实时日志清洗管道

const { pipeline } = require('stream/promises');
const { createReadStream } = require('fs');
const { Transform } = require('stream');
const { createGzip } = require('zlib');

// 创建一条完整的 ETL 管道
async function etlPipeline() {
  // 1. 过滤：只保留 ERROR 和 WARN 行
  const filterStream = new Transform({
    transform(chunk, encoding, callback) {
      const lines = chunk.toString().split('\n')
        .filter(line => line.includes('ERROR') || line.includes('WARN'))
        .join('\n');

      if (lines) {
        this.push(lines + '\n');
      }
      callback();
    },
  });

  // 2. 脱敏：替换 IP 地址
  const maskStream = new Transform({
    transform(chunk, encoding, callback) {
      const masked = chunk.toString()
        .replace(/\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}/g, '[MASKED_IP]');
      this.push(masked);
      callback();
    },
  });

  // 3. 组装管道
  await pipeline(
    createReadStream('/var/log/nginx/access.log'),
    filterStream,      // 过滤
    maskStream,        // 脱敏
    createGzip(),      // 压缩
    require('fs').createWriteStream('/tmp/filtered-log.gz'),
  );

  console.log('ETL 管道完成');
}

场景三：分页 API 的数据导出

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

// 将所有用户的订单数据导出为流
function createOrderExportStream(userIds) {
  return new Readable({
    objectMode: true,

    async read() {
      if (userIds.length === 0) {
        this.push(null);  // 结束流
        return;
      }

      const userId = userIds.shift();

      try {
        // 分页获取订单
        let page = 1;
        let hasMore = true;

        while (hasMore) {
          const response = await fetch(
            `/api/users/${userId}/orders?page=${page}&size=100`
          );
          const { data, total } = await response.json();

          for (const order of data) {
            // push 返回 false 就是背压——暂停 push，等 drain
            if (!this.push(order)) {
              // 把当前用户剩余数据放回去，等下一轮
              userIds.unshift(userId);
              return;
            }
          }

          hasMore = (page * 100) < total;
          page++;
        }
      } catch (err) {
        this.destroy(err);
      }
    },
  });
}

常见陷阱

陷阱 1：忘记处理 error 事件

// ❌ 错误：未处理的 error 会 crash 进程
const stream = fs.createReadStream('maybe-not-exist.txt');
stream.pipe(process.stdout);

// ✅ 正确
stream.on('error', (err) => {
  console.error('读取失败:', err.message);
  // 优雅降级
});

陷阱 2：在 Transform 的 flush 中忘记 callback

// ❌ 流永远不会结束
_flush(callback) {
  this.push('剩余数据');
  // 忘记调用 callback()！
}

// ✅ 正确
_flush(callback) {
  this.push('剩余数据');
  callback();
}

陷阱 3：对象模式和 Buffer 模式混用

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

// ❌ 输入是对象，但没声明 objectMode
const t = new Transform({
  transform(chunk, encoding, callback) {
    // chunk 会是 Buffer，不是对象
    callback();
  },
});

// ✅ 正确
const t = new Transform({
  readableObjectMode: true,
  writableObjectMode: true,
  transform(chunk, encoding, callback) {
    // chunk 是 JS 对象
    callback();
  },
});

相关工具

本站提供浏览器本地工具，免注册即可试用 →

#Node.js#Stream#后端#性能优化#教程