java - RxJava2 concat() 只运行第一个 Observable

Question

我有 3 个返回 Observable(String) 的方法。我需要一个一个地执行这些方法。每种方法大约需要 2 或 3 分钟。我正在使用下一个代码:

Observable.concat(one(),two(),three())
          .subscribeWith(new DisposableObserver<String>() {
                    @Override
                    public void onNext(String s) {
                        LOG.debug(s);
                    }

                    @Override
                    public void onError(Throwable e) {
                        LOG.error(e.toString());
                    }

                    @Override
                    public void onComplete() {

                    }
                });

只执行第一种方法。忽略第二种和第三种方法。我不明白这是什么问题？我正在使用 RxJava2 版本 2.1.6

Answer 1

显然，one() 是一个 Observable，它不 发出 onComplete() 事件，这意味着，它还没有完成，而 concat() 运算符将等到 observable 发出一个终端事件，然后才开始下一个 observable。

如果您不具备第一个 observable 的创建逻辑，您可以使用 Single.fromObservable() 将 observable 转换为 Single API。

Observable.concat(Single.fromObservable(one()), two(), three())

如果您确实有权访问第一个可观察对象的创建逻辑，则发出一个终端事件(即 emitter.onComplete())。

---

正如 David Karnok 在评论部分概述的那样:

Single.fromObservable() also requires a completing source so it can error you if that source happens to have no items.