Multiple Observable execution with RxJava

Question

I have two Observables define and I am calling them as below:

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

But the problem is that statusObser2 executes only after statusObser1 has completed. Instead I want both observers to execute in parallel, i.e. statusObser2 shouldn't wait for statusObser1 to complete.

Answer 1

They execute sequentially since you're blocking (toBlocking()) and waiting for their response.

Instead, subscribe to them. Either individually:

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

Or using the Zip operator:

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

...

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

see more details about the Zip operator here.

In both cases, if the observables are asynchronous, they will be executed in parallel.

There are other operators you can use to combine the results of both your operators without blocking either of them.

Answer 2

Try MultiThreading.

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 method will execute the execution in a new Thread, and both executions wil executed in seperate Thread.