通过 RxJava 发送异常时崩溃

Question

我开始使用 RXAndroid，并且我已经发布了我的第一个测试版，但我遇到了一个我没有预料到的异常。我确信这是因为我对 onError 方法的工作原理缺乏了解。

我的代码是这样的：

disposables.add(Observable.create(new ObservableOnSubscribe<String>() {
                    @Override
                    public void subscribe(@NonNull ObservableEmitter<String> e) throws Exception {
                        if (e.isDisposed()) {
                            return;
                        }
                        //do stuff
                        if (stuff didn't fail) {
                            e.onNext(myString);
                        } else {
                            e.onError(new Exception("stuff failed"));
                        }
                        e.onComplete();
                    }


                })
                        .subscribeOn(Schedulers.io())
                        .observeOn(AndroidSchedulers.mainThread())
                        .subscribeWith(new DisposableObserver<String>() {
                            @Override
                            public void onNext(@NonNull String myString) {
                                //do stuff

                            }

                            @Override
                            public void onError(@NonNull Throwable e) {
                                //handle the error
                            }

                            @Override
                            public void onComplete() {
                            }
                        })
        );

我在这里遇到了崩溃：

e.onError(new Exception("stuff failed"));

但我个人认为这可以处理这种情况？

public void onError(@NonNull Throwable e) {
    //handle the error
}

我已经更改了我的代码以用不同的方式处理这个问题，但我仍然想知道onError 应该如何工作。

谢谢。

编辑：例外：

java.lang.Exception: stuff failed
       at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:349)
       at io.reactivex.internal.operators.observable.ObservableCreate$CreateEmitter.onError(ObservableCreate.java:83)
       at MyClass$4.subscribe(MyClass.java:295)
       at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
       at io.reactivex.Observable.subscribe(Observable.java:10842)
       at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)
       at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:452)
       at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61)
       at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52)
       at java.util.concurrent.FutureTask.run(FutureTask.java:237)
       at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272)
       at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133)
       at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607)
       at java.lang.Thread.run(Thread.java:762)
Caused by java.lang.Exception: stuff failed
       at MyClass$4.subscribe(MyClass.java:295)
       at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
       at io.reactivex.Observable.subscribe(Observable.java:10842)
       at io.reactivex.internal.operators.observable.ObservableSubscribeOn$SubscribeTask.run(ObservableSubscribeOn.java:96)
       at io.reactivex.Scheduler$DisposeTask.run(Scheduler.java:452)
       at io.reactivex.internal.schedulers.ScheduledRunnable.run(ScheduledRunnable.java:61)
       at io.reactivex.internal.schedulers.ScheduledRunnable.call(ScheduledRunnable.java:52)
       at java.util.concurrent.FutureTask.run(FutureTask.java:237)
       at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:272)
       at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1133)
       at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:607)
       at java.lang.Thread.run(Thread.java:762)

Answer 1

在检查e.isDisposed() 和您对e.onError() 的调用之间，发射器最终以某种方式被处置。这会导致对onError 的调用退回到全局错误处理程序。

您需要再次检查e.isDisposed()，或者如果您愿意使用实验性功能，RxJava 2.1.1 添加了e.tryOnError 方法，如果它已经被释放，则不会回退。