rxjava 抛出 OnErrorNotImplementedException

Question

Observable.create(new ObservableOnSubscribe())
    .subscribeOn(Schedulers.io())
    .compose(new ParserTransformer())
    .map()
    .subscribe()

当我的 ObservableOnSubscribe 出现问题时，它会抛出异常（java.net.UnknownHostException ...）。

可观察订阅：

public void subscribe(ObservableEmitter oe) throws Exception {
    try{
        ...
    } catch {
        if (!oe.isDisposed()) {
            if (ex instanceof IOException) {
                throw new NetException(...); //sometimes, it make the app crash!
            }
        } else{
            ...
        }
    }
    oe.onComplete();
}

这是日志：

D OkHttp: error: java.net.UnknownHostException
D OkHttp: oe.isDisposed: false
W System.err: io.reactivex.exceptions.OnErrorNotImplementedException: Unable to resolve host ******: No address associated with hostname
W System.err:   at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:704)
W System.err:   at io.reactivex.internal.functions.Functions$OnErrorMissingConsumer.accept(Functions.java:701)
W System.err:   at io.reactivex.internal.observers.LambdaObserver.onError(LambdaObserver.java:77)
W System.err:   at io.reactivex.internal.observers.BasicFuseableObserver.onError(BasicFuseableObserver.java:100)
W System.err:   at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.checkTerminated(ObservableObserveOn.java:276)
W System.err:   at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.drainNormal(ObservableObserveOn.java:172)
W System.err:   at io.reactivex.internal.operators.observable.ObservableObserveOn$ObserveOnObserver.run(ObservableObserveOn.java:252)
W System.err:   at io.reactivex.android.schedulers.HandlerScheduler$ScheduledRunnable.run(HandlerScheduler.java:109)
W System.err:   at android.os.Handler.handleCallback(Handler.java:754)
W System.err:   at android.os.Handler.dispatchMessage(Handler.java:95)
W System.err:   at android.os.Looper.loop(Looper.java:163)
W System.err:   at android.app.ActivityThread.main(ActivityThread.java:6343)
W System.err:   at java.lang.reflect.Method.invoke(Native Method)
W System.err:   at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:880)
W System.err:   at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:770)

为什么会发生这种情况？为什么 rxjava 说 OnError 没有实现？

Answer 1

发生错误，并且您没有在订阅者中实现 onError() 方法。

Observable.create(new ObservableOnSubscribe())
    .subscribeOn(Schedulers.io())
    .compose(new ParserTransformer())
    .map()
    .subscribe(new Consumer<Object>() {
        @Override
        public void accept(Object object) {

        }
     }, new Consumer<Throwable>() {
        @Override
        public void accept(Throwable throwable) {
            Log.e(TAG, throwable.getMessage(), throwable);
        }
     });)

Answer 2

您应该使用Emitter.onError而不是抛出异常。

编辑：记住 javadocs: OnErrorNotImplementedException

public void subscribe(ObservableEmitter oe) throws Exception {
    try{
        ...
    } catch {
        if (!oe.isDisposed()) {
            if (ex instanceof IOException) {
                oe.onError(new NetException(...)); //pass error to emitter
            }
        } else{
            ...
        }
    }
    oe.onComplete();
}