RXJava - 一个接一个地运行多个observable(比如concat,但每个observable都有onCompleted)

Question

有没有办法实现以下目标:

我有3个类型的观察者Observable<MyData>.我想要的是:

• 运行第一个可观察的
• 观察onCompleted第一个观察者
• 运行第二个observable
• 观察onCompleted第二个可观测量
• 运行第三个observable
• 观察onCompleted第三个可观测量

这可以用,concat但我只能观察到最后一次onCompleted.

难看的解决方案

我知道,我可以实现这一目标,如果我onCompleted从前一个事件中开始下一个可以理解的话.

题

即使使用任意数量的可观察量,还有其他方法可以实现这一点吗？我想避免从onCompleted事件中将这一切联系在一起,因为这看起来非常丑陋,链接越深,它就越不清楚......

编辑 - UseCase

• 首先,我想发出代表我的应用程序上次启动时加载的数据的数据(我将数据序列化为光盘)=>原因是,我想要一个非常快速的应用程序启动
• 然后我想要第二次数据加载,以在空状态下加载最新数据
• 然后我想要第三次加载数据来加载最新数据的缺失深层数据

我想不断更新UI,我想知道,每个级别的数据加载完成

Answer 1

我希望下面的代码有帮助.

Observable<MyData> observable1 = ...;
Observable<MyData> observable2 = ...;
Observable<MyData> observable3 = ...;

Observable
        .concat(observable1.doOnCompleted(this::onCompleteObservable1),
                observable2.doOnCompleted(this::onCompleteObservable2),
                observable3.doOnCompleted(this::onCompleteObservable3))
        .subscribeOn(Schedulers.newThread())
        .observeOn(AndroidSchedulers.mainThread())
        .subscribe( ... );

这是示例方法.

void onCompleteObservable1() {
    //do some work
}

void onCompleteObservable2() {
    //do some work
}

void onCompleteObservable3() {
    //do some work
}

Answer 2

我认为concatMap可能是答案。使用 concatMap，您可以连接可观察值并订阅一次，因此您的代码可能类似于：

Observable.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)
    .concatMap(integer -> Observable.just(integer)
            .observeOn(Schedulers.computation())
            .concatMap(i -> {
                try {
                    Thread.sleep(new Random().nextInt(1000));
                    return Observable.just(2 * i);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                    return Observable.error(e);
                }
            }))
    .subscribe(System.out::println,
            Throwable::printStackTrace,
            () -> System.out.println("onCompleted"));