Over a million developers have joined DZone.
{{announcement.body}}
{{announcement.title}}

Scatter-Gather Using Spring Reactor Core

DZone's Guide to

Scatter-Gather Using Spring Reactor Core

If you are familiar with Netflix Rx-Java, you already know Spring Reactor Core. The API's map beautifully.

· Java Zone ·
Free Resource

Verify, standardize, and correct the Big 4 + more– name, email, phone and global addresses – try our Data Quality APIs now at Melissa Developer Portal!

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

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 this task multiple times concurrently and then collect back the results, represented the following way using an rx.Observable:

@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, similar code using Spring Reactor Core:

@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, as in the sample.

These are still early explorations for me and I look forward to getting far more familiar with this excellent library.

Developers! Quickly and easily gain access to the tools and information you need! Explore, test and combine our data quality APIs at Melissa Developer Portal – home to tools that save time and boost revenue. 

Topics:
spring ,netflix

Published at DZone with permission of

Opinions expressed by DZone contributors are their own.

{{ parent.title || parent.header.title}}

{{ parent.tldr }}

{{ parent.urlSource.name }}