如何处理RxJava中observer的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（）和其他各种错误处理操作符，但正如文档中所述，它们的目的是处理可观察对象本身发出的错误。

最直接的解决方案是将onNext的整个主体包装成一个try-catch块，但对我来说这听起来不是一个好的解决方案。在类似的Rx中。NET问题，提出的解决方案是制作一个扩展方法，该方法将通过创建一个可观察的代理来进行包装。我试图重新制作它：

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本身将订阅者包装成一个安全订阅者。使用unsafeSubscribe来绕过它似乎也不是一个好的解决方案。

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

Answer 1

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

您建议将异常处理逻辑放在订阅服务器中，而不是创建一个通用的可观察包装器。

记住，Rx是关于将事件推送到订阅者。

从可观察的接口中，很明显，除了处理事件花费了多长时间，或者任何抛出的异常中包含的信息之外，可观察的用户实际上无法了解它的订阅者。

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

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

一旦订阅者抛出异常，对于可观察到的情况，只有两个可行的操作过程：

• 重新引发异常
• 实现通用处理，以记录故障并停止向其发送事件（任何类型），清理由于该订阅者而产生的任何资源，并继续进行任何剩余订阅

具体处理订阅者异常将是一个糟糕的设计选择；这将在订阅者和可观察者之间产生不适当的行为耦合。因此，如果你想对不好的订阅者保持弹性，上述两种选择实际上是可观察对象自身责任的合理限制。

如果您希望您的订阅者具有弹性并继续使用，那么您绝对应该将其包装在异常处理逻辑中，该逻辑旨在处理您知道如何从中恢复的特定异常（也许还可以处理瞬态异常、日志记录、重试逻辑、断路等）。

只有订阅者本身才有上下文来理解它是否适合在出现故障时接收进一步的事件。

如果您的情况需要开发可重用的错误处理逻辑，请将自己放在包装观察者的事件处理程序而不是可观察事件处理程序的心态中，并且注意不要在出现故障时盲目地传输事件。放开它！虽然不是关于Rx的，但这是一部有趣的软件工程经典，在最后一点上有很多话要说。如果你还没有读过，我强烈建议你。