如何处理RxJava中观察者的onNext引发的异常？ [英] How to handle exceptions thrown by observer's onNext in RxJava?

问题描述

请考虑以下示例：

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

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

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

This outputs numbers from 1 to 5 and then prints the exception.

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

What I want to achieve is make the observer stay subscribed and continue to run after throwing an exception, i.e. print all numbers from 1 to 10.

我尝试使用 retry（）和其他各种错误处理操作符，但是，如文档中所述，它们的目的是处理observable本身发出的错误。

I have tried using retry() and other various error handling operators, but, as said in the documentation, their purpose is handling errors emitted by the observable itself.

最多直截了当的解决方案就是将整个 onNext 包装成一个try-catch块，但这对我来说听起来不是一个好方法。在类似的Rx.NET问题中，建议的解决方案是制作一个扩展方法，通过创建代理来进行包装观察到的。我试图重拍它：

The most straightforward solution is just to wrap the whole body of onNext into a try-catch block, but that doesn't sound like a good solution to me. In the similar Rx.NET question, the proposed solution was to make an extension method which would do the wrapping by creating a proxy observable. I tried to remake it:

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 来解决它似乎也不是一个好的解决方案。

This does not change anything, because RxJava itself wraps the subscriber into a SafeSubscriber. Using unsafeSubscribe to get around it doesn't seem to be a good solution either.

我该怎么办？解决这个问题？

What can I do to solve this problem?

推荐答案

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

This is a common question that arises when learning Rx.

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

Your suggestion to put your exception handling logic in the subscriber is preferable over creating a generic observable wrapper.

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

Remember, that Rx is about pushing events to subscribers.

从可观察的界面可以清楚地看出，除了他们处理事件所花费的时间或任何抛出异常中包含的信息之外，没有任何观察者可以知道它的订阅者。

From the observable interface, it's clear there's not really anything an observable can know about it's subscribers other than how long they took to handle an event, or the information contained in any thrown exceptions.

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

A generic wrapper to handle subscriber exceptions and carry on sending events to that subscriber is a bad idea.

为什么？那么observable应该只知道订户现在处于未知的故障状态。在这种情况下继续发送事件是不明智的 - 也许，例如，订阅者处于这样一种情况，即从这一点开始的每个事件都会抛出异常并花费一些时间来完成它。

Why? Well the observable should only really know that the subscriber is now in an unknown failure state. To carry on sending events in this situation is unwise - perhaps, for example, the subscriber is in a condition where every event from this point forward is going to throw an exception and take a while to do it.

一旦订阅者抛出异常，观察者只有两种可行的行动方案：

Once a subscriber has thrown an exception, there are only two viable courses of action for the observable:

• 重新抛出异常


• 实施通用处理以记录失败并停止向其发送事件（任何类型）并清理由于该订户而导致的任何资源并继续进行任何剩余订阅。

订阅者例外的具体处理将是一个糟糕的设计选择;它会在订阅者和可观察者之间产生不适当的行为耦合。因此，如果您想要对不良订阅者具有弹性，那么上述两种选择实际上是观察者自身的合理责任限制。

Specific handling of subscriber exceptions would be a poor design choice; it would create inappropriate behavioural coupling between subscriber and observable. So if you want to be resilient to bad subscribers the two choices above are really the sensible limit of responsibility of the observable itself.

如果您想要订阅者要有弹性并继续，那么你应该把它包装在异常处理逻辑中，用于处理你知道如何从中恢复的特定异常（并且可能处理瞬态异常，记录，重试逻辑，电路中断等。）。

If you want your subscriber to be resilient and carry on, then you should absolutely wrap it in exception handling logic designed to handle the specific exceptions you know how to recover from (and perhaps to handle transient exceptions, logging, retry logic, circuit breaking etc.).

只有订户本身才会有上下文来了解它是否适​​合在失败时接收更多事件。

Only the subscriber itself will have the context to understand whether it is fit to receive further events in the face of failure.

如果您的情况需要开发可重复使用的错误处理逻辑，请将自己置于包装观察者事件处理程序而不是 observable 的思维模式中 - 并且不要小心在失败的情况下盲目地进行传播事件。 发布！，但未写入Rx，是一个有趣的软件工程经典，在最后一点上有很多话要说。如果你还没看过，我强烈建议。

If your situation warrants developing reusable error handling logic, put yourself in the mindset of wrapping the observer's event handlers rather than the observable - and do take care not to blindly carry on transmitting events in the face of failure. Release It! whilst not written about Rx, is an entertaining software engineering classic has plenty to say on this last point. If you've not read it, I highly advise it.

这篇关于如何处理RxJava中观察者的onNext引发的异常？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！