通过 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如果它已经被释放就不会回退的方法。