如何将Writable的_write函数中抛出的错误传递给调用方？

问题描述 #

错误提示显示该错误未被捕获，但错误需通过回调传递，请问应在何处捕获？我已尝试在getWritable函数内多处添加try/catch块，但均无效。

代码示例 #

const { pipeline } = require('node:stream/promises');
const { Readable, Writable } = require('node:stream');

function getWritable() {
  const WritableStream = new Writable({
    objectMode: true,
    write(chunk, encoding, callback) {
      console.log(chunk);
      callback(('Throw error here'));
    },
  });
  return WritableStream;
}

try {
  const writestream = getWritable();
  const readstream = Readable.from('1,xcx\n2,xy\n3,xz');
  pipeline(
    readstream,
    writestream,
  );
} catch (err) {
  console.log(`Can error be caught here? ${err}`);
}

解决方案 #

核心问题是：你用的stream/promises模块中的pipeline返回的是Promise对象，当前仅调用了它却未处理Promise的拒绝状态，同步的try/catch无法捕获异步Promise的错误。

有两种常用的错误捕获方式：

1. 使用async/await配合try/catch #

把执行代码包裹在异步函数中，用await等待pipeline完成，这样try/catch就能捕获到错误：

const { pipeline } = require('node:stream/promises');
const { Readable, Writable } = require('node:stream');

function getWritable() {
  const WritableStream = new Writable({
    objectMode: true,
    write(chunk, encoding, callback) {
      console.log(chunk);
      callback(new Error('Throw error here')); // 建议传递Error对象而非字符串
    },
  });
  return WritableStream;
}

// 用立即执行的异步函数包裹
(async () => {
  try {
    const writestream = getWritable();
    const readstream = Readable.from('1,xcx\n2,xy\n3,xz');
    await pipeline(readstream, writestream); // 加上await等待异步操作完成
  } catch (err) {
    console.log(`Error caught here: ${err.message}`);
  }
})();

2. 使用Promise的.catch()方法 #

如果不想用async/await，可以直接在pipeline调用后链式调用.catch()来捕获错误：

const { pipeline } = require('node:stream/promises');
const { Readable, Writable } = require('node:stream');

function getWritable() {
  const WritableStream = new Writable({
    objectMode: true,
    write(chunk, encoding, callback) {
      console.log(chunk);
      callback(new Error('Throw error here'));
    },
  });
  return WritableStream;
}

const writestream = getWritable();
const readstream = Readable.from('1,xcx\n2,xy\n3,xz');
pipeline(readstream, writestream)
  .catch(err => {
    console.log(`Error caught here: ${err.message}`);
  });

关键说明 #

• 你之前在getWritable里加try/catch无效，是因为write回调是异步触发的（流有数据时才会执行），同步try/catch无法捕获异步回调里的错误，而且stream的错误是通过Promise或事件传递的，不是同步抛出的。
• 注意write回调的callback参数要传递Error对象，而不是字符串，这样错误信息更规范，也能被Promise正确识别为拒绝原因。

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