Node.js中如何捕获处理多pipe串联流传输时触发的错误

现有代码无法捕获错误、响应挂起的核心原因：

• pipe 方法不会返回Promise，流运行时的错误通过error事件派发，默认不会进入外层async/await的try/catch捕获范围
• 仅给上游流绑定了销毁下游的错误逻辑，没有覆盖所有转换流的错误场景，也没有在错误发生时主动透传错误、结束所有流，导致流状态卡在传输中直接挂起。

直接使用Node.js内置的stream/promises.pipeline工具方法，它原生支持多流顺序拼接，会自动处理错误传播、资源销毁，适配async/await语法，能从根源上避免流挂起问题：

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

const seriesOfPipes = async () => {
  const transformStreams = [transformStream1, transformStream2, transformStream3];
  const res = await fetch("foo");

  try {
    // 按顺序拼接可读流、所有转换流
    await pipeline(res.body, ...transformStreams);
    return transformStreams.at(-1);
  } catch (e) {
    // 此处可拿到完整错误对象，做针对性处理
    console.log("管道传输错误：", e.message);
    // 兜底销毁所有流，避免资源泄漏
    transformStreams.forEach(s => !s.destroyed && s.destroy());
    !res.body.destroyed && res.body.destroy();
    throw e;
  }
};

如果运行环境不支持pipeline，可以手动将流逻辑包装为Promise，给所有参与管道的流统一绑定错误监听，确保错误能正常抛出、所有流能被正常销毁：

const seriesOfPipes = async () => {
  const transformStreams = [transformStream1, transformStream2, transformStream3];
  const res = await fetch("foo");
  const allStreams = [res.body, ...transformStreams];
  let outStream = res.body;

  try {
    return await new Promise((resolve, reject) => {
      // 任意流出错都统一处理
      allStreams.forEach(stream => {
        stream.on('error', (err) => {
          // 销毁所有流避免挂起
          allStreams.forEach(s => !s.destroyed && s.destroy());
          reject(err);
        });
      });

      // 依次拼接管道
      for (const streamItem of transformStreams) {
        outStream = outStream.pipe(streamItem);
      }

      // 传输完成返回最终流
      outStream.on('finish', () => resolve(outStream));
    });
  } catch (e) {
    console.log("捕获到管道错误：", e.message);
    // 自定义错误处理逻辑
    throw e;
  }
};

• 管道中的所有流（包括初始可读流、每一个转换流）都可能抛出错误，必须全部覆盖错误监听，不能只给上游流绑定事件
• 错误触发后必须销毁所有关联流，否则会残留资源句柄导致流挂起、内存泄漏
• 不建议使用outStream.close(<message>)的方式处理错误，该方式不会传递标准Error对象，无法支撑精细化的错误处理逻辑

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