java - 如何返回保证不会抛出 OnErrorNotImplementedException 的 RxJava Observable？

Question

我想在我的应用程序中创建一个模式，其中所有 Observable<T>返回的对象有一些默认的错误处理，这意味着订阅者可以使用 .subscribe(onNext)重载而不必担心应用程序崩溃。 (通常您必须使用 .subscribe(onNext, onError) )。有什么办法可以达到这个目的吗？

我尝试使用 onErrorReturn 附加到 Observable , doOnError和onErrorResumeNext - 他们中没有人对我的案子有帮助。也许我做错了，但我仍然得到 rx.exceptions.OnErrorNotImplementedException如果 Observable 内发生错误。

编辑 1:这是发出错误的 Observable 的示例，我想在某个中间层处理该错误:

Observable.create(subscriber -> {
    subscriber.onError(new RuntimeException("Somebody set up us the bomb"));
});

编辑2:我已经尝试使用此代码代表消费者处理错误，但我仍然得到 OnErrorNotImplementedException :

// obs is set by the method illustrated in edit 1
obs = obs.onErrorResumeNext(throwable -> {
    System.out.println("This error is handled by onErrorResumeNext");
    return null;
});
obs = obs.doOnError(throwable -> System.out.println("A second attempt at handling it"));
// Consumer code:
obs.subscribe(
    s -> System.out.println("got: " + s)
);

Answer 1

这会起作用 - 关键是返回 Observable.empty();

private <T> Observable<T> attachErrorHandler(Observable<T> obs) {
    return obs.onErrorResumeNext(throwable -> {
        System.out.println("Handling error by printint to console: " + throwable);
        return Observable.empty();
    });
}

// Use like this:
Observable<String> unsafeObs = getErrorProducingObservable(); 
Observable<String> safeObservable = attachErrorHandler(unsafeObs);
// This call will now never cause OnErrorNotImplementedException
safeObservable.subscribe(s -> System.out.println("Result: " + s));