Over a million developers have joined DZone.

Using Java 8 CompletableFuture and Rx-Java Observable

DZone's Guide to

Using Java 8 CompletableFuture and Rx-Java Observable

A simple scatter-gather scenario using Java 8 CompletableFuture and using Rx-Java Observable.

· Java Zone
Free Resource

Just released, a free O’Reilly book on Reactive Microsystems: The Evolution of Microservices at Scale. Brought to you in partnership with Lightbend.

The scenario is simple - Spawn about 10 tasks, each returning a string, and ultimately collect the results into a list.


A sequential version of this would be the following:

public void testSequentialScatterGather() throws Exception {
 List<String> list =
   IntStream.range(0, 10)


private String generateTask(int i) {
 return i + "-" + "test";

With CompletableFuture

A method can be made to return a CompletableFuture using a utility method called supplyAsync, I am using a variation of this method which accepts an explicit Executor to use, also I am deliberately throwing an exception for one of the inputs:

private CompletableFuture<String> generateTask(int i,
  ExecutorService executorService) {
 return CompletableFuture.supplyAsync(() -> {
  if (i == 5) {
   throw new RuntimeException("Run, it is a 5!");
  return i + "-" + "test";
 }, executorService);

Now to scatter the tasks:

List<CompletableFuture<String>> futures =
  IntStream.range(0, 10)
    .map(i -> this.generateTask(i, executors).exceptionally(t -> t.getMessage()))

At the end of scattering the tasks the result is a list of CompletableFuture. Now, to obtain the list of String from this is a little tricky, here I am using one of the solutions suggested in Stackoverflow:

CompletableFuture<List<String>> result = CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
  .thenApply(v -> futures.stream()

CompletableFuture.allOf method is being used here purely to compose the next action to take once all the scattered tasks are completed, once the tasks are completed the futures are again streamed and collected into a list of string.

The final result can then be presented asynchronously:

view source

result.thenAccept(l -> {

With Rx-java Observable

Scatter gather with Rx-java is relatively cleaner than the CompletableFuture version as Rx-java provides better ways to compose the results together, again the method which performs the scattered task:

view source

private Observable<String> generateTask(int i, ExecutorService executorService) {
    return Observable
            .<String>create(s -> {
                if ( i == 5) {
                    throw new RuntimeException("Run, it is a 5!");
                s.onNext( i + "-test");
            }).onErrorReturn(e -> e.getMessage()).subscribeOn(Schedulers.from(executorService));

and to scatter the tasks:

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

view source

Once more I have a List of Observable's, and what I need is a List of results, Observable provides a merge method to do just that:

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

view source

which can be subscribed to and the results printed when available:

  l -> logger.info(l.toString()));

view source

Strategies and techniques for building scalable and resilient microservices to refactor a monolithic application step-by-step, a free O'Reilly book. Brought to you in partnership with Lightbend.

java 8

Published at DZone with permission of Biju Kunjummen, DZone MVB. See the original article here.

Opinions expressed by DZone contributors are their own.


Dev Resources & Solutions Straight to Your Inbox

Thanks for subscribing!

Awesome! Check your inbox to verify your email so you can start receiving the latest in tech news and resources.


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

{{ parent.tldr }}

{{ parent.urlSource.name }}