RxJava and parallel execution of observer code

Question

I am having the following code using RxJava Observable api :

Observable<Info> observable = fileProcessor.processFileObservable(processedFile.getAbsolutePath());
    observable
      .buffer(10000)
      .observeOn(Schedulers.computation())
      .subscribe(recordInfo -> {
        _logger.info("Running stage2 on thread with id : " + Thread.currentThread().getId());
          for(Info info : recordInfo) {
            // some I/O operation logic
         }
      }, 
      exception -> {
      }, 
      () -> {
      });

My expectation was that the observation code i.e. code inside the subscribe() method will be executed in parallel after I have specified the computation scheduler. Instead the code is still being executed sequentially on single thread. How can make the code run in parallel using RxJava api.

Answer 1

RxJava is often misunderstood when it comes to the asynchronous/multithreaded aspects of it. The coding of multithreaded operations is simple, but understanding the abstraction is another thing.

A common question about RxJava is how to achieve parallelization, or emitting multiple items concurrently from an Observable. Of course, this definition breaks the Observable Contract which states that onNext() must be called sequentially and never concurrently by more than one thread at a time.

To achieve parallelism you need multiple Observables.

This runs in a single thread:

Observable<Integer> vals = Observable.range(1,10);

vals.subscribeOn(Schedulers.computation())
          .map(i -> intenseCalculation(i))
          .subscribe(val -> System.out.println("Subscriber received "
                  + val + " on "
                  + Thread.currentThread().getName()));

This runs in multiple threads:

Observable<Integer> vals = Observable.range(1,10);

vals.flatMap(val -> Observable.just(val)
            .subscribeOn(Schedulers.computation())
            .map(i -> intenseCalculation(i))
).subscribe(val -> System.out.println(val));

Code and text comes from this blog post.

Answer 2

RxJava 2.0.5 introduced parallel flows and ParallelFlowable, which makes parallel execution simpler and more declarative.

You no longer have to create Observable/Flowable within flatMap, you can simply call parallel() on Flowable and it returns ParallelFlowable.

It's not as feature rich as a regular Flowable, because concurrency raises many issues with Rx contracts, but you have basic map(), filter() and many more, which should be enough in most cases.

So instead of this flow from @LordRaydenMK answer:

Observable<Integer> vals = Observable.range(1,10);

vals.flatMap(val -> Observable.just(val)
        .subscribeOn(Schedulers.computation())
        .map(i -> intenseCalculation(i))
    ).subscribe(val -> System.out.println(val));

Now you can do:

Flowable<Integer> vals = Flowable.range(1, 10);

vals.parallel()
        .runOn(Schedulers.computation())
        .map(i -> intenseCalculation(i))
        .sequential()
        .subscribe(val -> System.out.println(val));

Answer 3

You have to specify subscribeOn(Schedulers.computation()) instead of observeOn(Schedulers.computation()) for that purpose. In subscribeOn you declare in which thread you are going to emit your values. In observeOn you declare in which thread you are going to handle and observe them.

Answer 4

Using flatMap and specify to subscribe on Schedulers.computation() will achieve concurrency.

Here is a more practical example using Callable, from the output, we can see it will take about 2000 milliseconds to finish all the tasks.

static class MyCallable implements Callable<Integer> {

    private static final Object CALLABLE_COUNT_LOCK = new Object();
    private static int callableCount;

    @Override
    public Integer call() throws Exception {
        Thread.sleep(2000);
        synchronized (CALLABLE_COUNT_LOCK) {
            return callableCount++;
        }
    }

    public static int getCallableCount() {
        synchronized (CALLABLE_COUNT_LOCK) {
            return callableCount;
        }
    }
}

private static void runMyCallableConcurrentlyWithRxJava() {
    long startTimeMillis = System.currentTimeMillis();

    final Semaphore semaphore = new Semaphore(1);
    try {
        semaphore.acquire();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
    Observable.just(new MyCallable(), new MyCallable(), new MyCallable(), new MyCallable())
            .flatMap(new Function<MyCallable, ObservableSource<?>>() {
                @Override
                public ObservableSource<?> apply(@NonNull MyCallable myCallable) throws Exception {
                    return Observable.fromCallable(myCallable).subscribeOn(Schedulers.computation());
                }
            })
            .subscribeOn(Schedulers.computation())
            .subscribe(new Observer<Object>() {
                @Override
                public void onSubscribe(@NonNull Disposable d) {

                }

                @Override
                public void onNext(@NonNull Object o) {
                    System.out.println("onNext " + o);
                }

                @Override
                public void onError(@NonNull Throwable e) {

                }

                @Override
                public void onComplete() {
                    if (MyCallable.getCallableCount() >= 4) {
                        semaphore.release();
                    }
                }
            });


    try {
        semaphore.acquire();
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        semaphore.release();
    }
    System.out.println("durationMillis " + (System.currentTimeMillis()-startTimeMillis));
}

Answer 5

This still comes in the same sequence. Even on new threads

    Observable<Integer> ob3 = Observable.range(1, 5);

    ob3.flatMap(new Func1<Integer, Observable<Integer>>() {

        @Override
        public Observable<Integer> call(Integer pArg0) {

            return Observable.just(pArg0);
        }

    }).subscribeOn(Schedulers.newThread()).map(new Func1<Integer, Integer>() {

        @Override
        public Integer call(Integer pArg0) {

            try {
                Thread.sleep(1000 - (pArg0 * 100));
                System.out.println(pArg0 + "  ccc   " + Thread.currentThread().getName());
            } catch (Exception e) {
                e.printStackTrace();
            }

            return pArg0;
        }

    }).subscribe();

Output

1 ccc RxNewThreadScheduler-1

2 ccc RxNewThreadScheduler-1

3 ccc RxNewThreadScheduler-1

4 ccc RxNewThreadScheduler-1

5 ccc RxNewThreadScheduler-1