使用 RxJava 执行多个 Observable答案

【问题标题】：Multiple Observable execution with RxJava使用 RxJava 执行多个 Observable
【发布时间】：2018-04-10 00:31:12
【问题描述】：

我定义了两个 Observable，如下所示：

Observable<Boolean> statusObser1 = reactiveApiConsumer.syncGradesSet1(subListDtos.get(0));
Observable<Boolean> statusObser2 = reactiveApiConsumer.syncGradesSet2(subListDtos.get(1));
statusObser1.toBlocking().first();
statusObser2.toBlocking().first();

但问题是statusObser2 仅在statusObser1 完成后执行。相反，我希望两个观察者并行执行，即 statusObser2 不应该等待 statusObser1 完成。

【问题讨论】：

标签： java rx-java

【解决方案1】：

它们按顺序执行，因为您正在阻塞 (toBlocking()) 并等待它们的响应。

相反，订阅它们。单独：

Observable<Boolean> statusObser1 = ...
Observable<Boolean> statusObser2 = ...
statusObser1.subscribe(System.out::println); //Example
statusObser2.subscribe(System.out::println); //Example

或使用 Zip 运算符：

public static int doSomethingWithBothValues(Boolean a, Boolean b) {
    ...
}

...

Observable<Boolean> statusObser1 = ...
Observable<Boolean> statusObser2 = ...
Observable.zip(statusObser1,statusObser2, this::doSomethingWithBothValues);

查看有关 Zip 运算符here 的更多详细信息。

在这两种情况下，如果 observables 是异步的，它们将被并行执行。

您可以使用其他运算符来组合两个运算符的结果，而不会阻塞其中任何一个。

【讨论】：

【解决方案2】：

尝试多线程。

Observable<Boolean> statusObser1 = reactiveApiConsumer.syncGradesSet1(subListDtos.get(0));
Observable<Boolean> statusObser2 = reactiveApiConsumer.syncGradesSet2(subListDtos.get(1));
startThread(statusObser1);
startThread(statusObser2);

public void startThread(Observable<Boolean> statusObser) {
    new Thread() {
        @Override
        public void run() {
            statusObser.toBlocking().first();
        }
    }.start();
}

startThread 方法将在一个新的线程中执行，并且两个执行都将在单独的线程中执行。

【讨论】：

这是一个非常糟糕的解决方案。它完全忽略了使用 Reacive 编程技术的意义。