java - 如何处理 RxJava 中观察者的 onNext 抛出的异常？

Question

考虑以下示例:

Observable.range(1, 10).subscribe(i -> {
    System.out.println(i);

    if (i == 5) {
        throw new RuntimeException("oops!");
    }
}, Throwable::printStackTrace);

这会输出从 1 到 5 的数字，然后打印异常。

我想要实现的是让观察者保持订阅状态并在抛出异常后继续运行，即打印从 1 到 10 的所有数字。

我尝试过使用 retry() 和 other various error handling operators , 但是，正如文档中所说，它们的目的是处理 observable 本身发出的错误。

最直接的解决方案就是将 onNext 的整个主体包装到一个 try-catch block 中，但这对我来说听起来不是一个好的解决方案。在 similar Rx.NET question ，建议的解决方案是创建一个扩展方法，通过创建一个可观察的代理来进行包装。我试图重新制作它:

Observable<Integer> origin = Observable.range(1, 10);
Observable<Integer> proxy = Observable.create((Observable.OnSubscribe<Integer>) s ->
        origin.subscribe(i -> {try { s.onNext(i); } catch (Exception ignored) {}}, s::onError, s::onCompleted));

proxy.subscribe(i -> {
    System.out.println(i);

    if (i == 5) {
        throw new RuntimeException("oops!");
    }
}, Throwable::printStackTrace);

这并没有改变任何东西，因为 RxJava 本身将订阅者包装到 SafeSubscriber 中。 .使用 unsafeSubscribe 来绕过它似乎也不是一个好的解决方案。

我能做些什么来解决这个问题？

Answer 1

这是学习 Rx 时常见的问题。

TL;DR

您建议将异常处理逻辑放在订阅者中比创建通用的可观察包装器更可取。

说明

请记住，Rx 是关于向订阅者推送事件。

从 observable 接口(interface)可以看出，除了订阅者处理事件所花费的时间或任何抛出的异常中包含的信息之外，observable 无法知道任何关于其订阅者的信息。

处理订阅者异常并继续向该订阅者发送事件的通用包装器是个坏主意。

为什么？好吧，observable 应该只知道订阅者现在处于未知的故障状态。在这种情况下继续发送事件是不明智的 - 例如，订阅者可能处于这样一种情况，即从现在开始的每个事件都会引发异常并需要一段时间才能完成。

一旦订阅者抛出异常，对于 observable 只有两种可行的行动方案:

• 重新抛出异常
• 实现通用处理以记录故障并停止向其发送(任何类型的)事件并清理该订阅者的所有资源并继续处理任何剩余的订阅。

订户异常的特定处理将是一个糟糕的设计选择；它会在订阅者和可观察者之间产生不适当的行为耦合。因此，如果您想对不良订阅者保持弹性，上述两个选择实际上是可观察对象本身的合理责任限制。

如果您希望您的订阅者具有弹性并继续使用，那么您绝对应该将其包装在异常处理逻辑中，该逻辑旨在处理您知道如何恢复的特定异常 from(也许是为了处理 transient 异常、日志记录、重试逻辑、断路等)。

只有订阅者自己有上下文来了解它是否适​​合在面对失败时接收更多事件。

如果您的情况需要开发可重用的错误处理逻辑，请将自己置于包装观察者的事件处理程序而不是 observable 的心态中 - 请注意不要盲目地继续传输事件的失败。 Release It!虽然不是关于 Rx 的文章，但它是一部有趣的软件工程经典，在最后一点上有很多话要说。如果您还没有阅读，我强烈建议您阅读。