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. |
