CompletableFuture介绍

1. Introduction

This tutorial is a guide to the functionality and use cases of the CompletableFuture class that was introduced as a Java 8 Concurrency API improvement.

2. Asynchronous Computation in Java

Asynchronous computation is difficult to reason about. Usually, we want to think of any computation as a series of steps, but in the case of asynchronous computation, actions represented as callbacks tend to be either scattered across the code or deeply nested inside each other. Things get even worse when we need to handle errors that might occur during one of the steps.

The Future interface was added in Java 5 to serve as a result of an asynchronous computation, but it did not have any methods to combine these computations or handle possible errors.

Java 8 introduced the CompletableFuture class. Along with the Future interface, it also implemented the CompletionStage interface. This interface defines the contract for an asynchronous computation step that we can combine with other steps.

CompletableFuture is at the same time, a building block and a framework, with about 50 different methods for composing, combining, and executing asynchronous computation steps and handling errors.

Such a large API can be overwhelming, but these mostly fall into several clear and distinct use cases.

3. Using CompletableFuture as a Simple Future

First, the CompletableFuture class implements the Future interface so that we can use it as a Future implementation but with additional completion logic.

For example, we can create an instance of this class with a no-arg constructor to represent some future result, hand it out to the consumers, and complete it at some time in the future using the complete method. The consumers may use the get method to block the current thread until this result is provided.

In the example below, we have a method that creates a CompletableFuture instance, then spins off some computation in another thread and returns the Future immediately.

When the computation is done, the method completes the Future by providing the result to the complete method:

public Future<String> calculateAsync() throws InterruptedException {
    CompletableFuture<String> completableFuture = new CompletableFuture<>();

    Executors.newCachedThreadPool().submit(() -> {
        Thread.sleep(500);
        completableFuture.complete("Hello");
        return null;
    });

    return completableFuture;
}

To spin off the computation, we use the Executor API. This method of creating and completing a CompletableFuture can be used with any concurrency mechanism or API, including raw threads.

Notice that the calculateAsync method returns a Future instance.

We simply call the method, receive the Future instance, and call the get method on it when we’re ready to block for the result.

Also, observe that the get method throws some checked exceptions, namely ExecutionException (encapsulating an exception that occurred during a computation) and InterruptedException (an exception signifying that a thread was interrupted either before or during an activity):

Future<String> completableFuture = calculateAsync();

// ... 

String result = completableFuture.get();
assertEquals("Hello", result);

If we already know the result of a computation, we can use the static completedFuture method with an argument that represents the result of this computation. Consequently, the get method of the Future will never block, immediately returning this result instead:

Future<String> completableFuture = 
  CompletableFuture.completedFuture("Hello");

// ...

String result = completableFuture.get();
assertEquals("Hello", result);

As an alternative scenario, we may want to cancel the execution of a Future.

4. CompletableFuture with Encapsulated Computation Logic

The code above allows us to pick any mechanism of concurrent execution, but what if we want to skip this boilerplate and execute some code asynchronously?

Static methods runAsync and supplyAsync allow us to create a CompletableFuture instance out of Runnable and Supplier functional types correspondingly.

Runnable and Supplier are functional interfaces that allow passing their instances as lambda expressions thanks to the new Java 8 feature.

The Runnable interface is the same old interface used in threads and does not allow to return a value.

The Supplier interface is a generic functional interface with a single method that has no arguments and returns a value of a parameterized type.

This allows us to provide an instance of the Supplier as a lambda expression that does the calculation and returns the result. It is as simple as:

CompletableFuture<String> future
  = CompletableFuture.supplyAsync(() -> "Hello");

// ...

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

5. Processing Results of Asynchronous Computations

The most generic way to process the result of a computation is to feed it to a function. The thenApply method does exactly that; it accepts a Function instance, uses it to process the result, and returns a Future that holds a value returned by a function:

CompletableFuture<String> completableFuture
  = CompletableFuture.supplyAsync(() -> "Hello");

CompletableFuture<String> future = completableFuture
  .thenApply(s -> s + " World");

assertEquals("Hello World", future.get());

If we don’t need to return a value down the Future chain, we can use an instance of the Consumer functional interface. Its single method takes a parameter and returns void.

There’s a method for this use case in the CompletableFuture. The thenAccept method receives a Consumer and passes it the result of the computation. Then the final future.get() call returns an instance of the Void type:

CompletableFuture<String> completableFuture
  = CompletableFuture.supplyAsync(() -> "Hello");

CompletableFuture<Void> future = completableFuture
  .thenAccept(s -> System.out.println("Computation returned: " + s));

future.get();

Finally, if we neither need the value of the computation nor want to return some value at the end of the chain, then we can pass a Runnable lambda to the thenRun method. In the following example, we simply print a line in the console after calling the future.get():

CompletableFuture<String> completableFuture 
  = CompletableFuture.supplyAsync(() -> "Hello");

CompletableFuture<Void> future = completableFuture
  .thenRun(() -> System.out.println("Computation finished."));

future.get();

6. Combining Futures

The best part of the CompletableFuture API is the ability to combine CompletableFuture instances in a chain of computation steps.

The result of this chaining is itself a CompletableFuture that allows further chaining and combining. This approach is ubiquitous in functional languages and is often referred to as a monadic design pattern.

In the following example, we use the thenCompose method to chain two Futures sequentially.

Notice that this method takes a function that returns a CompletableFuture instance. The argument of this function is the result of the previous computation step. This allows us to use this value inside the next CompletableFuture‘s lambda:

CompletableFuture<String> completableFuture 
  = CompletableFuture.supplyAsync(() -> "Hello")
    .thenCompose(s -> CompletableFuture.supplyAsync(() -> s + " World"));

assertEquals("Hello World", completableFuture.get());

The thenCompose method, together with thenApply, implements the basic building blocks of the monadic pattern. They closely relate to the map and flatMap methods of Stream and Optional classes, also available in Java 8.

Both methods receive a function and apply it to the computation result, but the thenCompose (flatMap) method receives a function that returns another object of the same type. This functional structure allows composing the instances of these classes as building blocks.

If we want to execute two independent Futures and do something with their results, we can use the thenCombine method that accepts a Future and a Function with two arguments to process both results:

CompletableFuture<String> completableFuture 
  = CompletableFuture.supplyAsync(() -> "Hello")
    .thenCombine(CompletableFuture.supplyAsync(
      () -> " World"), (s1, s2) -> s1 + s2));

assertEquals("Hello World", completableFuture.get());

A simpler case is when we want to do something with two Futures‘ results but don’t need to pass any resulting value down a Future chain. The thenAcceptBoth method is there to help:

CompletableFuture future = CompletableFuture.supplyAsync(() -> "Hello")
  .thenAcceptBoth(CompletableFuture.supplyAsync(() -> " World"),
    (s1, s2) -> System.out.println(s1 + s2));

7. Difference Between thenApply() and thenCompose()

In our previous sections, we’ve shown examples regarding thenApply() and thenCompose(). Both APIs help chain different CompletableFuture calls, but the usage of these two functions are different.

7.1. thenApply()

We can use this method to work with the result of the previous call. However, a key point to remember is that the return type will be combined of all calls.

So this method is useful when we want to transform the result of a CompletableFuture call:

CompletableFuture<Integer> finalResult = compute().thenApply(s-> s + 1);

7.2. thenCompose()

The thenCompose() is similar to thenApply() in that both return a new CompletionStage. However, thenCompose() uses the previous stage as the argument. It will flatten and return a Future with the result directly, rather than a nested future as we observed in thenApply():

CompletableFuture<Integer> computeAnother(Integer i){
    return CompletableFuture.supplyAsync(() -> 10 + i);
}
CompletableFuture<Integer> finalResult = compute().thenCompose(this::computeAnother);

So if the idea is to chain CompletableFuture methods, then it’s better to use thenCompose().

Also, note that the difference between these two methods is analogous to the difference between map() and flatMap().

8. Running Multiple Futures in Parallel

When we need to execute multiple Futures in parallel, we usually want to wait for all of them to execute and then process their combined results.

The CompletableFuture.allOf static method allows to wait for the completion of all of the Futures provided as a var-arg:

CompletableFuture<String> future1  
  = CompletableFuture.supplyAsync(() -> "Hello");
CompletableFuture<String> future2  
  = CompletableFuture.supplyAsync(() -> "Beautiful");
CompletableFuture<String> future3  
  = CompletableFuture.supplyAsync(() -> "World");

CompletableFuture<Void> combinedFuture 
  = CompletableFuture.allOf(future1, future2, future3);

// ...

combinedFuture.get();

assertTrue(future1.isDone());
assertTrue(future2.isDone());
assertTrue(future3.isDone());

Notice that the return type of the CompletableFuture.allOf() is a CompletableFuture. The limitation of this method is that it does not return the combined results of all Futures. Instead, we have to get results from Futures manually. Fortunately, CompletableFuture.join() method and Java 8 Streams API makes it simple:

String combined = Stream.of(future1, future2, future3)
  .map(CompletableFuture::join)
  .collect(Collectors.joining(" "));

assertEquals("Hello Beautiful World", combined);

The CompletableFuture.join() method is similar to the get method, but it throws an unchecked exception in case the Future does not complete normally. This makes it possible to use it as a method reference in the Stream.map() method.

9. Handling Errors

For error handling in a chain of asynchronous computation steps, we have to adapt the throw/catch idiom in a similar fashion.

Instead of catching an exception in a syntactic block, the CompletableFuture class allows us to handle it in a special handle method. This method receives two parameters: a result of a computation (if it finished successfully) and the exception thrown (if some computation step did not complete normally).

In the following example, we use the handle method to provide a default value when the asynchronous computation of a greeting was finished with an error because no name was provided:

String name = null;

// ...

CompletableFuture<String> completableFuture  
  =  CompletableFuture.supplyAsync(() -> {
      if (name == null) {
          throw new RuntimeException("Computation error!");
      }
      return "Hello, " + name;
  }).handle((s, t) -> s != null ? s : "Hello, Stranger!");

assertEquals("Hello, Stranger!", completableFuture.get());

As an alternative scenario, suppose we want to manually complete the Future with a value, as in the first example, but also have the ability to complete it with an exception. The completeExceptionally method is intended for just that. The completableFuture.get() method in the following example throws an ExecutionException with a RuntimeException as its cause:

CompletableFuture<String> completableFuture = new CompletableFuture<>();

// ...

completableFuture.completeExceptionally(
  new RuntimeException("Calculation failed!"));

// ...

completableFuture.get(); // ExecutionException

In the example above, we could have handled the exception with the handle method asynchronously, but with the get method, we can use the more typical approach of synchronous exception processing.

10. Async Methods

Most methods of the fluent API in the CompletableFuture class have two additional variants with the Async postfix. These methods are usually intended for running a corresponding execution step in another thread.

The methods without the Async postfix run the next execution stage using a calling thread. In contrast, the Async method without the Executor argument runs a step using the common fork/join pool implementation of Executor that is accessed with the ForkJoinPool.commonPool(), as long as parallelism > 1. Finally, the Async method with an Executor argument runs a step using the passed Executor.

Here’s a modified example that processes the result of a computation with a Function instance. The only visible difference is the thenApplyAsync method, but under the hood, the application of a function is wrapped into a ForkJoinTask instance (for more information on the fork/join framework, see the article “Guide to the Fork/Join Framework in Java”). This allows us to parallelize our computation even more and use system resources more efficiently:

CompletableFuture<String> completableFuture  
  = CompletableFuture.supplyAsync(() -> "Hello");

CompletableFuture<String> future = completableFuture
  .thenApplyAsync(s -> s + " World");

assertEquals("Hello World", future.get());

11. JDK 9 CompletableFuture API

Java 9 enhances the CompletableFuture API with the following changes:

• New factory methods added
• Support for delays and timeouts
• Improved support for subclassing

and new instance APIs:

• Executor defaultExecutor()
• CompletableFuture newIncompleteFuture()
• CompletableFuture copy()
• CompletionStage minimalCompletionStage()
• CompletableFuture completeAsync(Supplier<? extends T> supplier, Executor executor)
• CompletableFuture completeAsync(Supplier<? extends T> supplier)
• CompletableFuture orTimeout(long timeout, TimeUnit unit)
• CompletableFuture completeOnTimeout(T value, long timeout, TimeUnit unit)

We also now have a few static utility methods:

• Executor delayedExecutor(long delay, TimeUnit unit, Executor executor)
• Executor delayedExecutor(long delay, TimeUnit unit)
• CompletionStage completedStage(U value)
• CompletionStage failedStage(Throwable ex)
• CompletableFuture failedFuture(Throwable ex)

Finally, to address timeout, Java 9 has introduced two more new functions:

• orTimeout()
• completeOnTimeout()

Here’s the detailed article for further reading: Java 9 CompletableFuture API Improvements.

12. Conclusion

In this article, we’ve described the methods and typical use cases of the CompletableFuture class.

The source code for the article is available over on GitHub.