尝试catch将无法捕获流中发生的错误，因为在调用代码已经退出之后会抛出这些错误。您可以参考文档：

https://nodejs.org/dist/latest-v10.x/docs/api/errors.html

.on("error", handler)仅处理Stream错误，但是如果您使用自定义Transform流，.on("error", handler)请不要捕获_transform函数内部发生的错误。所以可以做这样的事情来控制应用程序流程：

this_transform函数中的关键字指向Stream自身，即一个EventEmitter。因此，您可以使用try catch如下所示的错误来捕获错误，然后将其传递给自定义事件处理程序。

// CustomTransform.js
CustomTransformStream.prototype._transform = function (data, enc, done) {
  var stream = this
  try {
    // Do your transform code
  } catch (e) {
    // Now based on the error type, with an if or switch statement
    stream.emit("CTError1", e)
    stream.emit("CTError2", e)
  }
  done()
}

// StreamImplementation.js
someReadStream
  .pipe(CustomTransformStream)
  .on("CTError1", function (e) { console.log(e) })
  .on("CTError2", function (e) { /*Lets do something else*/ })
  .pipe(someWriteStream)

这样，您可以将逻辑处理程序和错误处理程序分开。此外，您可以选择仅处理某些错误而忽略其他错误。

更新
替代：RXJS可观察

可以使用一个简单的函数将来自整个链的错误传播到最右边的流：

function safePipe (readable, transforms) {
    while (transforms.length > 0) {
        var new_readable = transforms.shift();
        readable.on("error", function(e) { new_readable.emit("error", e); });
        readable.pipe(new_readable);
        readable = new_readable;
    }
    return readable;
}

可以像这样使用：

safePipe(readable, [ transform1, transform2, ... ]);