NodeJS中如何限制流间流量？解决读写流速度不匹配内存溢出

当网络读取流的速度远快于数据库写入流时，未处理的数据会堆积在内存中，最终导致JavaScript heap out of memory错误。以下是几种解决方法：

1. 确保自定义可写流正确实现背压逻辑 #

如果你的writeStream是自定义的数据库写入流，必须在_write方法完成写入操作后调用回调函数——这是Node.js判断可写流是否准备好接收新数据的关键：

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

class DBWriteStream extends Writable {
  _write(chunk, encoding, callback) {
    // 执行数据库写入操作
    yourDBInstance.insert(chunk, (err) => {
      // 写入完成后必须调用callback，告知流可以继续接收数据
      callback(err);
    });
  }
}

如果遗漏了回调调用，流会一直认为写入未完成，导致读取的数据持续堆积在内存缓冲区。

2. 手动监听事件控制读取节奏 #

如果默认的pipe方法没生效，可以手动绑定事件，通过pause()和resume()控制读取流的速度：

readStream.on('data', (chunk) => {
  // 尝试写入数据，返回false表示缓冲区已满
  const canAcceptMore = writeStream.write(chunk);
  if (!canAcceptMore) {
    // 暂停读取流，避免继续加载数据
    readStream.pause();
  }
});

// 当可写流缓冲区排空后，恢复读取流
writeStream.on('drain', () => {
  readStream.resume();
});

// 处理结束和错误事件，避免资源泄漏
readStream.on('end', () => writeStream.end());
readStream.on('error', (err) => {
  console.error('读取错误:', err);
  writeStream.destroy(err);
});
writeStream.on('error', (err) => {
  console.error('写入错误:', err);
  readStream.destroy(err);
});

3. 使用stream.pipeline替代pipe（推荐） #

Node.js内置的stream.pipeline方法比原生pipe更健壮，它会自动处理背压、错误传播和资源清理，能有效避免内存堆积问题：

const { pipeline } = require('stream');
const { promisify } = require('util');
const pipelineAsync = promisify(pipeline);

async function transferData() {
  try {
    await pipelineAsync(readStream, writeStream);
    console.log('数据传输完成');
  } catch (err) {
    console.error('传输失败:', err);
  }
}

transferData();

pipeline会自动根据可写流的处理能力调节读取速度，同时在发生错误时销毁所有相关流，防止内存泄漏。

内容的提问来源于stack exchange，提问作者ablaszkiewicz1