RxJava sequential result of parallel async task

Question

I have stream of objects that I have to process in parallel (it is long running task), however, at same time I have to keep sequence of results. I am trying to use RxJava for this, and could not find any good solution.

Basic idea, is translate every object to function of future and then call get on each future in sequence. In case of future completed with Exception I need to retry it, till it is actually get completed successfully.

public class RxTest {

    private static CompletableFuture<String> futureFunction(int i) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(300);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

            if (ThreadLocalRandom.current().nextInt(1, 100) < 10) {
                throw new RuntimeException("failed.. sorry");
            }

            return Thread.currentThread().getName() + " " + String.valueOf(i);
        });
    }

    public static void main(String[] args) throws InterruptedException {

        Observable.range(1, 100)
                .flatMap(i ->
                        Observable.just(i).map(RxTest::futureFunction).flatMap(Observable::from).retry()
                )

                .observeOn(Schedulers.from(Executors.newSingleThreadExecutor()))

                .subscribe(Subscribers.create((x) -> {
                    System.out.println(Thread.currentThread().getName() + " " + x);
                }));

    }
}

In this example I am receive sequential output

pool-1-thread-1 ForkJoinPool.commonPool-worker-9 1
pool-1-thread-1 ForkJoinPool.commonPool-worker-9 2
pool-1-thread-1 ForkJoinPool.commonPool-worker-9 3
pool-1-thread-1 ForkJoinPool.commonPool-worker-9 4
pool-1-thread-1 ForkJoinPool.commonPool-worker-9 5
pool-1-thread-1 ForkJoinPool.commonPool-worker-9 6

But each future is created and called only after previous complete, not in parallel as I expected.

I was playing around with observeOn() and subscribeOn(), nothing works well.

akarnokd · Accepted Answer · 2016-01-27 07:56:06Z

1

You can use concatMapEager or concatEager which will run the sources but keeps the item order when emitting:

Observable.range(1, 100)
.concatMapEager(v -> 
    Observable.fromCallable(() -> { ... })
    .subscribeOn(Schedulers.computation())
    .retry()
)
.subscribe(...)

answered Jan 27, 2016 at 7:56

akarnokd

70.1k14 gold badges161 silver badges199 bronze badges

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

3 Comments

Andrii Liubimov Over a year ago

Works like a charm :) Thank you!

Simon Baslé Over a year ago

@akarnokd how much of an experimental API is concatMapEager?

akarnokd Over a year ago

I don't know about any plan for major API restructuring for the 1.x branch for a year.

Collectives™ on Stack Overflow

RxJava sequential result of parallel async task

1 Answer 1

3 Comments

Your Answer

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

3 Comments

Your Answer

Sign up or log in

Post as a guest

Related