订阅者可以在不关闭流的情况下处理飞镖流中抛出的异常吗？

Question

我无法理解的简短示例：

Stream<int> getNumbersWithException() async* {
  for (var i = 0; i < 10; i++) {
    yield i;
    if (i == 3) throw Exception();
  }
}

有用法：

getNumbersWithException()
    .handleError((x) => print('Exception caught for $x'))
    .listen((event) {
  print('Observed: $event');
});

这将在 3 处停止并输出：

Observed: 0
Observed: 1
Observed: 2
Observed: 3
Exception caught for Exception: foo

根据文档 (https://dart.dev/tutorials/language/streams) 和 (https://api.dart.dev/stable/2.9.1/dart-async/Stream/handleError.html)，这是符合预期的，因为抛出的异常会自动关闭流。

1. 这是否意味着在流中处理异常以使订阅可以长期存在于此类事件中的正确方法是在流本身内处理异常？从外部无法做到这一点？
2. 广播流也一样吗？
3. 如果我以错误的方式思考这个问题，有什么建议可以开始正确思考？

我目前认为流是异步数据事件的来源，有时可能是错误事件。从文档和示例来看，一切看起来都很整洁，但我认为想要处理错误并继续观察数据流是一个正常的用例。我很难编写代码来做到这一点。但是，我可能做错了。任何见解将不胜感激。

---

编辑：我可以补充一点，我已经尝试了各种方法，例如使用流转换器，结果相同：

var transformer = StreamTransformer<int, dynamic>.fromHandlers(
  handleData: (data, sink) => sink.add(data),
  handleError: (error, stackTrace, sink) =>
      print('Exception caught for $error'),
  handleDone: (sink) => sink.close(),
);
getNumbersWithException().transform(transformer).listen((data) {
  print('Observed: $data');
});

另外，listen() 有一个可选参数 cancelOnError，看起来很有希望，但它默认为 false，所以这里没有雪茄。

Answer 1

生成器方法

Stream<int> getNumbersWithException() async* {
  for (var i = 0; i < 10; i++) {
    yield i;
    if (i == 3) throw Exception();
  }
}

会在你抛出异常时终止。 throw 正常工作，它不会直接将异常添加到流中。因此，它通过循环和方法体传播出去，直到整个方法体以抛出的异常结束。 此时未处理的异常被添加到流中，然后流被关闭，因为主体已经结束。

所以，问题不在于处理，而在于流的生成。 您确实必须在本地处理错误以避免它结束流生成主体。

您不能在 async* 方法中使用 throw 向流中添加多个错误，并且错误将是流所做的最后一件事。

实际发出多个错误的可用技巧是产生异常：

  if (i == 3) yield* () async* { throw Exception(); }();
  // or:      yield* Stream.fromFuture(Future.error(Exception());

这会将异常直接发送到生成的流中，而不会在本地抛出并结束生成器方法体。