Node.js异步Pipeline流复制异常：Promise永久挂起无响应

核心问题 #

你在循环中重复使用了同一个response.body可读流，而Node.js的可读流只能被消费一次。第一次管道处理时已经将流的数据读取完毕，后续的管道没有数据可读，也无法触发end或error事件，导致Promise永远处于pending状态。

你提到同步代码能工作，大概率是测试时allKeys只有一个元素，或者同步模式下的错误被掩盖了——本质上重复复用同一个可读流是错误的用法。

解决方案 #

根据你的需求（为每个allKeys元素生成独立的可读流用于后续S3上传），提供两种可行方案：

方案1：缓存响应体为Buffer（适合小文件）

先把response.body的内容缓存成Buffer，再为每个任务创建新的可读流：

import { pipeline, PassThrough, Readable } from 'stream/promises';
import { buffer } from 'stream/consumers';

// 先将响应体缓存为Buffer
const bodyBuffer = await buffer(response.body);

// 为每个key创建独立的流管道
const streams = await Promise.all(allKeys.map(async () => {
  // 基于缓存的Buffer创建新的可读流
  const sourceStream = Readable.from(bodyBuffer);
  const destStream = new PassThrough();
  
  await pipeline(sourceStream, destStream);
  console.log('Pipeline succeeded.');
  return destStream;
}));

方案2：使用Tee流复制可读流（适合大文件）

如果图片文件较大，不想占用过多内存缓存Buffer，可以用tee流复制多个独立的可读流（需要安装第三方库stream-tee）：

npm install stream-tee

import { pipeline, PassThrough } from 'stream/promises';
import tee from 'stream-tee';

// 复制出与allKeys数量一致的可读流
const sourceStreams = tee(response.body, allKeys.length);

// 并行处理每个复制后的流
const streams = await Promise.all(sourceStreams.map(async (source) => {
  const destStream = new PassThrough();
  
  await pipeline(source, destStream);
  console.log('Pipeline succeeded.');
  return destStream;
}));

排查思路 #

如果后续遇到类似流挂起的问题，可以通过以下方式调试：

• 给源流添加事件监听，查看流的状态：

response.body.on('end', () => console.log('源流已结束'));
response.body.on('error', (err) => console.error('源流出错:', err));
response.body.on('data', (chunk) => console.log('读取到数据块，长度:', chunk.length));

• 确保每个管道使用的是独立的可读流实例，绝对不要重复复用同一个流。

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