SupplyAsync wait for all CompletableFutures to finish

user_4632971 0
Viewed 3581

I'm running some async tasks below and need to wait until they all finish. I'm not sure why but the join() isn't forcing a wait for all tasks and the code continues executing without waiting. Is there a reason why the stream of joins isn't working as expected?

The CompletableFutures list is just a stream that maps supplyAsync

List<Integer> items = Arrays.asList(1, 2, 3);

List<CompletableFuture<Integer>> futures = items
                .stream()
                .map(item -> CompletableFuture.supplyAsync(() ->  {

                    System.out.println("processing");
                    // do some processing here
                    return item;

                }))
                .collect(Collectors.toList());

And I wait for the futures like.

CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
                .thenApply(ignored -> futures.stream()
                        .map(CompletableFuture::join)
                        .collect(Collectors.toList()));

I'm able to get the wait working with futures.forEach(CompletableFuture::join); but I wanted to know why my stream approach wasn't working.

completable-future java-8 asynchronous java
1 Answers

This code:

CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
                .thenApply(ignored -> futures.stream()
                        .map(CompletableFuture::join)
                        .collect(Collectors.toList()));

Does not wait for all futures in futures to complete. What it does is create a new future that will wait for all async executions in futures to complete before it itself completes (but will not block until all those futures complete). When this allOf future completes, then your thenApply code runs. But allOf() will return immediately without blocking.

This means futures.stream().map(CompletableFuture::join).collect(Collectors.toList()) in your code only runs after all async executions complete, which defeats your purpose. The join() calls will all return immediately. But this is not the bigger problem. Your challenge is that allOf().thenApply() will not wait for async executions to complete. It will just create another future that won't block.

The easiest solution is to use the first pipeline and map to an Integer list:

List<Integer> results = items.stream()
    .map(item -> CompletableFuture.supplyAsync(() -> {

        System.out.println("processing " + item);
        // do some processing here
        return item;

    }))
    .collect(Collectors.toList()) //force-submit all
    .stream()
    .map(CompletableFuture::join) //wait for each
    .collect(Collectors.toList());

If you want to use something like your original code, then your second snippet would have to be changed to:

List<Integer> reuslts = futures.stream()
    .map(CompletableFuture::join)
    .collect(Collectors.toList());

That's because CompletableFuture.allOf does not wait, it just combines all futures into a new one that completes when all complete:

Returns a new CompletableFuture that is completed when all of the given CompletableFutures complete.

Alternatively, you could still use allOf() with join(), then run your current thenApply() code:

//wrapper future completes when all futures have completed
CompletableFuture.allOf(futures.toArray(new CompletableFuture[futures.size()]))
        .join(); 

//join() calls below return immediately
List<Integer> result = futures.stream()
        .map(CompletableFuture::join) 
        .collect(Collectors.toList());

The join() calls in the last statement return immediately, becaues the join() call on the wrapper (allOf()) future will have waited for all futures passed to it to complete. This is why I don't see the reason to do this when you can use the first approach.

user_57615580
Related