RxJava2可观察性抛出UndeliverableException

Question

据我了解，RxJava2 values.take(1)创建了另一个Observable，它仅包含原始Observable中的一个元素。这 绝不能
丢，因为它是由效果过滤掉一个例外take(1)，因为它是发生第二次。

如 以下 代码段所示

    Observable<Integer> values = Observable.create(o -> {
        o.onNext(1);
        o.onError(new Exception("Oops"));
    });

    values.take(1)
            .subscribe(
                    System.out::println,
                    e -> System.out.println("Error: " + e.getMessage()),
                    () -> System.out.println("Completed")
            );

输出量

1
Completed
io.reactivex.exceptions.UndeliverableException: java.lang.Exception: Oops
    at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:366)
    at io.reactivex.internal.operators.observable.ObservableCreate$CreateEmitter.onError(ObservableCreate.java:83)
    at ch02.lambda$main$0(ch02.java:28)
    at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
    at io.reactivex.Observable.subscribe(Observable.java:10841)
    at io.reactivex.internal.operators.observable.ObservableTake.subscribeActual(ObservableTake.java:30)
    at io.reactivex.Observable.subscribe(Observable.java:10841)
    at io.reactivex.Observable.subscribe(Observable.java:10827)
    at io.reactivex.Observable.subscribe(Observable.java:10787)
    at ch02.main(ch02.java:32)
Caused by: java.lang.Exception: Oops
    ... 8 more
Exception in thread "main" io.reactivex.exceptions.UndeliverableException: java.lang.Exception: Oops
    at io.reactivex.plugins.RxJavaPlugins.onError(RxJavaPlugins.java:366)
    at io.reactivex.internal.operators.observable.ObservableCreate$CreateEmitter.onError(ObservableCreate.java:83)
    at ch02.lambda$main$0(ch02.java:28)
    at io.reactivex.internal.operators.observable.ObservableCreate.subscribeActual(ObservableCreate.java:40)
    at io.reactivex.Observable.subscribe(Observable.java:10841)
    at io.reactivex.internal.operators.observable.ObservableTake.subscribeActual(ObservableTake.java:30)
    at io.reactivex.Observable.subscribe(Observable.java:10841)
    at io.reactivex.Observable.subscribe(Observable.java:10827)
    at io.reactivex.Observable.subscribe(Observable.java:10787)
    at ch02.main(ch02.java:32)
Caused by: java.lang.Exception: Oops
    ... 8 more

我的问题：

1. 我理解它正确吗？
2. 到底是什么导致了异常。
3. 如何从消费者那里解决呢？

Answer 1

1. 是的，但是因为可观察到的“结束”并不意味着内部运行的代码create(...)已停止。为了在这种情况下完全安全，您需要使用o.isDisposed()观察对象是否在下游结束。
2. 那里是一个例外，因为RxJava 2具有NEVER（从不）允许onError丢失呼叫的策略。UndeliverableException如果可观察对象已经终止，则将其传递到下游或作为全局对象抛出。由Observable的创建者来“适当地”处理Observable已结束且发生异常的情况。
3. 问题在于生产者（Observable）和消费者（Subscriber）在流何时结束时意见不一致。由于在这种情况下生产者胜过消费者，因此问题只能在生产者中解决。