等待两个 Observable 完成后再继续？答案

【问题标题】：Waiting for two Observables to complete before proceeding?等待两个 Observable 完成后再继续？
【发布时间】：2019-01-18 21:41:06
【问题描述】：

我有一个加载页面，我想在其中执行两个网络请求 (retrofit2-rxjava) 以获取不相关的信息。在这两个请求完成之前，我不想继续到下一页，即使其中一个或两个都失败了。

使用 zip 将请求捆绑在一起。有没有办法不强制使用 BiFunction，并且不必返回 null？

请求 A 和 B 有一个 .doOnNext 和 .doOnError。如果其中一个返回错误，zip observable 是否继续？ zip 订阅者是否也返回错误？

这是最好的方法吗？

private Disposable retrieveBothThings() {
return Observable.zip(getThingA(), getThingB(),
                    (A, B) -> {
                        onAllCallsComplete();
                        return null;
                    })
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .subscribe(o -> {}, Logger::e);
}
    
    

private Observable<...> getThingA() {
            return SessionManager.getInstance().getApi()
                    .getA()
                    .subscribeOn(Schedulers.io())
                    .observeOn(AndroidSchedulers.mainThread())
                    .doOnNext(this::onACompleted)
                    .doOnError(this::onAFailed);
}

private Observable<...> getThingB() {
        return SessionManager.getInstance().getApi()
                .getB()
                .subscribeOn(Schedulers.io())
              .observeOn(AndroidSchedulers.mainThread()).toObservable()
                .doOnNext(this::onBSuccess)
                .doOnError(this::onBFailure);
}
    
private void onBSuccess(...) {
    ...        
}
    
private void onBFailure(final Throwable throwable) {
    Logger.e(throwable);
}

private void onACompleted(...) {
    ...        
}
    
private void onAFailed(final Throwable throwable) {
    Logger.e(throwable);
}

【问题讨论】：

如果你返回 null 你会得到 RuntimeException 因为空值在 RxJava2 中是不被接受的

标签： java android observable retrofit2 rx-java2

【解决方案1】：

您可以使用merge() 运算符组合两个可观察对象。您可以使用onErrorResumeNext() 将错误转换为onComplete()。

Completable.merge(
  observable1
    .doOnNext(this::onACompleted)
    .doOnError(this::onAFailed)
    .onErrorResumeNext( Completable.complete() )
    .toCompletable(),
  observable2
    .doOnNext(this::onBCompleted)
    .doOnError(this::onBFailed)
    .onErrorResumeNext( Completable.complete() ),
    .toCompletable() )
.subscribe( ignore -> {}, 
            error -> {},
            () -> { processCompletion(); } );

【讨论】：