RxJava mergeDelayError OnErrorFailedException 错误：尝试将错误传播到 Observer.onError 时发生答案

【问题标题】：RxJava mergeDelayError OnErrorFailedException Error: occurred when trying to propagate error to Observer.onErrorRxJava mergeDelayError OnErrorFailedException 错误：尝试将错误传播到 Observer.onError 时发生
【发布时间】：2016-03-23 07:08:57
【问题描述】：

当我在 observables 列表上使用 mergeDelayError 在出错之前执行每个 observable 时，我得到：

OnErrorFailedException: Error occurred when trying to propagate error to Observer.onError

代码：

return Observable.mergeDelayError(Observable.from(observableList));

我使用 onError 的参数订阅 observable（也未执行）。

但是，当我尝试使用 onErrorResumeNext 进行调试时，我发现传播可观察到的错误不会导致任何问题。这里发生了什么？

return Observable.mergeDelayError(Observable.from(observableList)).onErrorResumeNext(new Func1<Throwable, Observable<? extends Pair<String, Map<Long, Set<ProgressGradedItemBL>>>>>() {
    @Override
    public Observable<? extends Pair<String, Map<Long, Set<ProgressGradedItemBL>>>> call(Throwable throwable) {
        return Observable.error(throwable);
    }
});

这是我在没有onErrorResumeNext 的情况下得到的堆栈跟踪：

rx.exceptions.OnErrorFailedException: Error occurred when trying to propagate error to Observer.onError
    at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:175)
    at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:97)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:466)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.completeInner(OperatorMerge.java:449)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:599)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onCompleted(OperatorMerge.java:542)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:472)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.completeInner(OperatorMerge.java:449)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:614)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onCompleted(OperatorMerge.java:542)
    at rx.internal.operators.OnSubscribeCombineLatest$MultiSourceProducer.tick(OnSubscribeCombineLatest.java:151)
    at rx.internal.operators.OnSubscribeCombineLatest$MultiSourceProducer.onCompleted(OnSubscribeCombineLatest.java:183)
    at rx.internal.operators.OnSubscribeCombineLatest$MultiSourceRequestableSubscriber.onCompleted(OnSubscribeCombineLatest.java:236)
    at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:472)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.onCompleted(OperatorMerge.java:435)
    at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:472)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.completeInner(OperatorMerge.java:449)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:614)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onCompleted(OperatorMerge.java:542)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:472)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.onCompleted(OperatorMerge.java:435)
    at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.drainAndComplete(OperatorMerge.java:472)
    at rx.internal.operators.OperatorMerge$MergeSubscriber.completeInner(OperatorMerge.java:449)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.emit(OperatorMerge.java:614)
    at rx.internal.operators.OperatorMerge$InnerSubscriber.onCompleted(OperatorMerge.java:542)
    at rx.internal.operators.OperatorDistinctUntilChanged$1.onCompleted(OperatorDistinctUntilChanged.java:64)
    at rx.internal.operators.OperatorDoOnEach$1.onCompleted(OperatorDoOnEach.java:53)
    at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
    at rx.internal.operators.OperatorFilter$1.onCompleted(OperatorFilter.java:42)
    at rx.internal.operators.OperatorSubscribeOn$1$1$1.onCompleted(OperatorSubscribeOn.java:66)
    at rx.observers.SerializedObserver.onCompleted(SerializedObserver.java:99)
    at rx.observers.SerializedSubscriber.onCompleted(SerializedSubscriber.java:46)
    at rx.internal.operators.OperatorConcat$ConcatSubscriber.subscribeNext(OperatorConcat.java:158)
    at rx.internal.operators.OperatorConcat$ConcatSubscriber.completeInner(OperatorConcat.java:150)
    at rx.internal.operators.OperatorConcat$ConcatInnerSubscriber.onCompleted(OperatorConcat.java:212)
    at rx.internal.operators.OperatorDoOnEach$1.onCompleted(OperatorDoOnEach.java:53)
    at rx.internal.operators.OperatorDoOnEach$1.onCompleted(OperatorDoOnEach.java:53)
    at rx.internal.operators.OperatorMap$1.onCompleted(OperatorMap.java:43)
    at rx.internal.operators.OperatorDoOnEach$1.onCompleted(OperatorDoOnEach.java:53)

【问题讨论】：

标签： java rx-java

【解决方案1】：

当您使用subscribe(Action1) 而不是subscribe(Action1, Action1) 时，会抛出OnErrorFailedException。此外，您似乎使用的是旧版本的 RxJava（名称 drainAndComplete 类似于 1.0.12 及之前的版本）。当前版本是 1.1.2。

【讨论】：

我正在使用 subscribe(final Action1<? super T> onNext, final Action1<Throwable> onError 订阅 observable，所以我认为这不是问题所在。

【解决方案2】：

如果您的流对 Observable 抛出未经检查的异常，则 observable 处理将中止并抛出 OnErrorFailedException。

【讨论】：