onNext 中未捕获的异常被传播到 RxJava 全局异常处理程序

Question

以下是我试图解决的问题的简化示例。在我的代码库中，问题出在 PublishSubject 上，但 Observable 可用于重现此问题。

问题：

我在订阅者的 onNext() 中有一个未捕获的异常，该异常将异常传播到 RxJava 的全局异常处理程序，导致崩溃。然而，这不是期望的行为。 objective 是优雅地处理未捕获的异常。

问题：

为什么没有为未捕获的异常调用 onError()？我可以使用什么方法来捕获此类未捕获的异常？

快速解决方案：

我可以使用 RxJavaPlugins.setErrorHandler 来应对这种情况，但如果我们没有任何其他解决方案来处理手头的问题，我宁愿将此解决方案作为最后的手段。

代码：

Observable.just(true)
                .subscribeWith(new DisposableObserver<Boolean>() {
                    @Override
                    public void onNext(Boolean aBoolean) {
                        String x = null;
                        x.getBytes();
                    }

                    @Override
                    public void onError(Throwable e) { <--- NOT CALLED UPON EXCEPTION
                        System.out.println("inside test observable onError");
                        e.printStackTrace();
                    }

                    @Override
                    public void onComplete() {

                    }
                });

堆栈跟踪：

12-03 11:09:01.454 4750-4750/? E/MDMApplication: UNCAUGHT EXCEPTION IN APP: java.lang.RuntimeException: Unable to start activity ComponentInfo{com.test.mdm.app/mdm.connector.app.MainActivity}: java.lang.NullPointerException
        at android.app.ActivityThread.performLaunchActivity(ActivityThread.java:2195)
        at android.app.ActivityThread.handleLaunchActivity(ActivityThread.java:2245)
        at android.app.ActivityThread.access0(ActivityThread.java:135)
        at android.app.ActivityThread$H.handleMessage(ActivityThread.java:1196)
        at android.os.Handler.dispatchMessage(Handler.java:102)
        at android.os.Looper.loop(Looper.java:136)
        at android.app.ActivityThread.main(ActivityThread.java:5017)
        at java.lang.reflect.Method.invokeNative(Native Method)
        at java.lang.reflect.Method.invoke(Method.java:515)
        at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:779)
        at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:595)
        at dalvik.system.NativeStart.main(Native Method)
     Caused by: java.lang.NullPointerException
        at mdm.connector.app.MainActivity.onNext(MainActivity.java:63)
        at mdm.connector.app.MainActivity.onNext(MainActivity.java:59)
        at io.reactivex.internal.operators.observable.ObservableDoOnEach$DoOnEachObserver.onNext(ObservableDoOnEach.java:103)
        at io.reactivex.internal.operators.observable.ObservableMap$MapObserver.onNext(ObservableMap.java:64)
        at io.reactivex.internal.operators.observable.ObservableScalarXMap$ScalarDisposable.run(ObservableScalarXMap.java:248)
        at io.reactivex.internal.operators.observable.ObservableJust.subscribeActual(ObservableJust.java:35)
        at io.reactivex.Observable.subscribe(Observable.java:12036)
        at io.reactivex.internal.operators.observable.ObservableMap.subscribeActual(ObservableMap.java:33)
        at io.reactivex.Observable.subscribe(Observable.java:12036)
        at io.reactivex.internal.operators.observable.ObservableDoOnEach.subscribeActual(ObservableDoOnEach.java:42)
        at io.reactivex.Observable.subscribe(Observable.java:12036)
        at io.reactivex.Observable.subscribeWith(Observable.java:12088)
        at mdm.connector.app.MainActivity.onCreate(MainActivity.java:59)
        at android.app.Activity.performCreate(Activity.java:5231)
        at android.app.Instrumentation.callActivityOnCreate(Instrumentation.java:1087)
        at android.app.ActivityThread.performLaunchActivity(ActivityThread.java:2159)
        at android.app.ActivityThread.handleLaunchActivity(ActivityThread.java:2245) 
        at android.app.ActivityThread.access0(ActivityThread.java:135) 
        at android.app.ActivityThread$H.handleMessage(ActivityThread.java:1196) 
        at android.os.Handler.dispatchMessage(Handler.java:102) 
        at android.os.Looper.loop(Looper.java:136) 
        at android.app.ActivityThread.main(ActivityThread.java:5017) 
        at java.lang.reflect.Method.invokeNative(Native Method) 
        at java.lang.reflect.Method.invoke(Method.java:515) 
        at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:779) 
        at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:595) 
        at dalvik.system.NativeStart.main(Native Method)

Answer 1

解决上述问题的方法是，我们应该在订阅方法中使用 Consumer 和 Consumer，而不是使用 DisposableObserver。这确保如果在 Consumer accept 方法中有任何未捕获的异常，它们将在 Consumer accept 方法中被捕获，而不是传播到 RxJava 的全局异常处理程序并导致崩溃。

示例：

Observable.just(true)
.subscribe(new Consumer<Boolean> {
  public void accept(Boolean aboolean) {
      String x = null;
      x.getBytes();
  }
}, new Consumer<Throwable> {

  public void accept(Throwable e) {
      e.printStackTrace()
  }
});

如果有其他更好的方案来应对未处理的异常，欢迎在评论中提出

onNext 中未捕获的异常被传播到 RxJava 全局异常处理程序

Uncaught exception inside onNext being propagated to RxJava global exception handler

java

android

observable

rx-java

rx-java2