How to run multiple service calls in parallel using CompletableFuture?

Question

I am returning following response to the user

class FinalResponseDTO {
    List<Service1ResponseDTO> service1ResponseDTO;
    Long totalCount;
    List<Service2ResponseDTO> service2ResponseDTO;
}

As of now I am making three sequential calls to compute this FinalResponseDTO, each of this call can be run independent of others. I tried making three different CompletableFuture as:

CompletableFuture<List<Service1ResponseDTO> future1 = CompletableFuture.supplyAsync(() -> service1.callMethod1());
CompletableFuture<Long> future2 = CompletableFuture.supplyAsync(() -> service2.callMethod2());
CompletableFuture<Service2ResponseDTO> future3 = CompletableFuture.supplyAsync(() -> service3.callMethod3());

If I do CompletableFuture.allOf(future1, future2, future3).join(); or Should I call CompletableFuture.allOf(future1, future2, future3).get();? Even If I call either of these join or get then how should I construct FinalResponseDTO from it. I am new to the Java 8 concurrency features such as CompletableFuture and I am confused, as each return type of each of these future is different, how should I get combined responses of all such futures and then construct my final output?

M A · Accepted Answer · 2022-01-14T11:35:35.973

From the Javadocs of CompletableFuture.allOf():

Returns a new CompletableFuture that is completed when all of the given CompletableFutures complete. If any of the given CompletableFutures complete exceptionally, then the returned CompletableFuture also does so, with a CompletionException holding this exception as its cause. Otherwise, the results, if any, of the given CompletableFutures are not reflected in the returned CompletableFuture, but may be obtained by inspecting them individually.

So when the combining CompletableFuture completes, you can inspect the values and construct the final response object using a simple constructor by applying a function that constructs the object:

CompletableFuture.allOf(future1, future2, future3).thenApply(v -> 
    new FinalResponseDTO(future1.join(), future2.join(), future3.join())
);

Don't forget to mention that for having the List type again you have to get it: future1.get() otherwise it will of the type: CompletableFuture — Lucas Favaro Borsatto, Jul 20 '22 at 18:09

Torino · Answer 2 · 2022-01-13T13:22:51.230

    CompletableFuture<List<Service1ResponseDTO> future1 = 
        CompletableFuture.supplyAsync(() -> service1.callMethod1());
    CompletableFuture<Long> future2 = 
        CompletableFuture.supplyAsync(() -> service2.callMethod2());
    CompletableFuture<List<Service2ResponseDTO>> future3 = 
        CompletableFuture.supplyAsync(() -> service3.callMethod3());

    CompletableFuture.allOf(future1, future2, future3).get();

    return new FinalResponseDTO(future1.join(), future2.join(), future3.join());

Note: supplyAsync runs on ForkJoinPool.commonPool() so it would be a good choice to provide your own executor e.g. Executors.newCachedThreadPool():

CompletableFuture.supplyAsync(() -> action, executor);

This blocks the caller. Putting your `join()` statements into a `thenApply` allows you to preserve the “pipe”. — Boris the Spider, Jan 13 '22 at 18:56

score 2 · Answer 3 · answered Jan 15 '22 at 19:20

Instead of using CompletableFuture.allOf, you could fall back to the somewhat hidden Applicative of the CompletableFuture:

static <T, R> CompletableFuture<R> alsoApply(CompletableFuture<T> future,     CompletableFuture<Function<T, R>> f) {
    return f.thenCompose(future::thenApply);
}

With this helper function you can execute futures in parallel threads:

CompletableFuture<String> future = alsoApply(
    CompletableFuture.supplyAsync(() -> "a"),
    CompletableFuture.supplyAsync(() -> "b")
.thenApply(b -> a -> a + b));

assertEquals("ab", future.get());

See this question and answer for where this is coming from and why and how it works.

How to run multiple service calls in parallel using CompletableFuture?

3 Answers3

Linked