如何在NodeJS中克隆readstream以获取两份可独立消费的流副本

问题原因 #

你直接分别pipe两次失败的核心原因是：Node.js 的Readable流默认只能被消费一次，如果你先后分两次执行pipe操作，第一次pipe触发后流就会进入流动模式开始传输数据，等你执行第二次pipe时，部分或全部数据已经被第一个消费者消费完毕，第二个流自然拿不到完整数据。

解决方案1：使用 PassThrough 流实时分流（推荐适合绝大多数场景） #

用内置的stream.PassThrough作为中转流，一次性将原始流同时pipe到多个PassThrough实例，即可得到多个可独立消费的副本。

代码实现 #

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

/**
 * 克隆可读流为多个独立可消费的副本
 * @param {import('stream').Readable} originalStream 原始可读流
 * @param {number} cloneCount 需要克隆的份数，默认2份
 * @returns {import('stream').PassThrough[]} 克隆后的流数组
 */
function cloneReadStream(originalStream, cloneCount = 2) {
  const clones = Array.from({ length: cloneCount }, () => new PassThrough());

  // 一次性将原始流pipe到所有克隆流，避免数据丢失
  clones.forEach(clone => originalStream.pipe(clone));

  // 错误透传：原始流出错时销毁所有克隆流，避免内存泄漏
  originalStream.on('error', (err) => {
    clones.forEach(clone => clone.destroy(err));
  });

  return clones;
}

使用示例 #

// 示例：克隆文件读取流，分别写入两个不同文件
const { createReadStream, createWriteStream } = require('fs');

const originalStream = createReadStream('./source.txt');
const [stream1, stream2] = cloneReadStream(originalStream);

// 两个流独立消费，互不干扰
stream1.pipe(createWriteStream('./copy1.txt'));
stream2.pipe(createWriteStream('./copy2.txt'));

注意事项 #

• 必须一次性完成所有克隆流的pipe操作，不要先消费其中一个流后再创建第二个克隆流，会丢失已经传输的数据
• 如果多个克隆流的消费速度差异较大，速度较慢的流会触发背压，拖慢整体传输速度，同时未消费的数据会暂存在内存中，差异过大时会导致内存占用过高。如果存在这种场景，建议使用方案2。

解决方案2：缓存全量数据后生成独立流（适合小数据量/消费速度差异大的场景） #

如果流数据量不大，或者不同克隆流的消费速度差异非常大，可以先把原始流的全量数据读取到内存缓存，再基于缓存生成多个完全独立的可读流。

代码实现 #

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

async function cloneReadStreamByCache(originalStream) {
  // 读取全量流数据到Buffer
  const chunks = [];
  for await (const chunk of originalStream) {
    chunks.push(chunk);
  }
  const fullData = Buffer.concat(chunks);

  // 返回多个完全独立的可读流
  return [
    Readable.from(fullData),
    Readable.from(fullData)
  ];
}

注意事项 #

• 该方案会将全量流数据加载到内存，不适合GB级别的大文件/大流量流使用
• 生成的克隆流完全独立，消费速度互不影响，不存在背压互相干扰的问题

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