Home » Java » Enterprise Java » Scatter-Gather using Spring Reactor Core

About Biju Kunjummen

Biju Kunjummen

Scatter-Gather using Spring Reactor Core

I have a good working experience in using the Netflix Rx-Java libraries and have previously blogged about using Rx-Java and Java 8 CompletableFuture for a scatter-gather kind of problems. Here I want to explore applying the same pattern using the Spring Reactor Core library.

tldr – If you are familiar with Netflix Rx-Java, you already know Spring Reactor Core, the API’s map beautifully and I was thrilled to see that the Spring Reactor team has diligently used Marble diagrams in their Javadoc API’s

Another quick point is that rx.Observable maps to Flux or Mono based on whether many items are being emitted or whether one or none is being emitted.

With this let me directly jump into the sample – I have a simple task(simulated using a delay) that is spawned a few times, I need to execute these tasks concurrently and then collect back the results, represented the following way using a rx.Observable code:

@Test
public void testScatterGather() throws Exception {
    ExecutorService executors = Executors.newFixedThreadPool(5);

    List<Observable<String>> obs =
            IntStream.range(0, 10)
                .boxed()
                .map(i -> generateTask(i, executors)).collect(Collectors.toList());


    Observable<List<String>> merged = Observable.merge(obs).toList();
    List<String> result = merged.toBlocking().first();

    logger.info(result.toString());

}

private Observable<String> generateTask(int i, ExecutorService executorService) {
    return Observable
            .<String>create(s -> {
                Util.delay(2000);
                s.onNext( i + "-test");
                s.onCompleted();
            }).subscribeOn(Schedulers.from(executorService));
}

Note that I am blocking purely for the test. Now, a similar code using Spring Reactor Core translates to the following:

@Test
public void testScatterGather() {
    ExecutorService executors = Executors.newFixedThreadPool(5);

    List<Flux<String>> fluxList = IntStream.range(0, 10)
            .boxed()
            .map(i -> generateTask(executors, i)).collect(Collectors.toList());

    Mono<List<String>> merged = Flux.merge(fluxList).toList();

    List<String> list = merged.get();

    logger.info(list.toString());


}

public Flux<String> generateTask(ExecutorService executorService, int i) {
    return Flux.<String>create(s -> {
        Util.delay(2000);
        s.onNext(i + "-test");
        s.onComplete();
    }).subscribeOn(executorService);
}

It more or less maps one to one. A small difference is in the Mono type, I personally felt that this type was a nice introduction to the reactive library as it makes it very clear whether more than 1 item is being emitted vs only a single item which I have made use of in the sample. These are still early explorations for me and I look forward to getting far more familiar with this excellent library.

Reference: Scatter-Gather using Spring Reactor Core from our JCG partner Biju Kunjummen at the all and sundry blog.
Do you want to know how to develop your skillset to become a Java Rockstar?
Subscribe to our newsletter to start Rocking right now!
To get you started we give you our best selling eBooks for FREE!
1. JPA Mini Book
2. JVM Troubleshooting Guide
3. JUnit Tutorial for Unit Testing
4. Java Annotations Tutorial
5. Java Interview Questions
6. Spring Interview Questions
7. Android UI Design
and many more ....
Email address:

Leave a Reply

Be the First to Comment!

avatar
  Subscribe  
Notify of