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.

Malt · Accepted Answer · 2017-10-28 15:44:24Z

2

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.

edited Oct 28, 2017 at 15:44

answered Oct 28, 2017 at 15:38

Malt

30.5k10 gold badges73 silver badges115 bronze badges

Sign up to request clarification or add additional context in comments.

Comments

Saqib Ahmed · Accepted Answer · 2017-10-28 16:02:10Z

0

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.

answered Oct 28, 2017 at 16:02

Saqib Ahmed

1,2692 gold badges15 silver badges25 bronze badges

1 Comment

Malt Over a year ago

That's a very bad solution. It completely misses the point of using Reacive programming techniques.

Collectives™ on Stack Overflow

Multiple Observable execution with RxJava

2 Answers 2

Comments

1 Comment

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

2 Answers 2

Comments

1 Comment

Your Answer

Sign up or log in

Post as a guest

Linked

Related