A preliminary exploration of CompletableFuture

Translated from www.baeldung.com/java-comple…

1. Introduction

This article focuses on some of the basic features and use of the Java 8 Concurrency API’s improved class, CompletableFuture.

Asynchronous computing in Java

Asynchronous computations are more difficult to express, and we usually expect any operation to be step-by-step, but in the case of asynchronous computations, actions like callback callbacks are often interspersed with code or nested calls. Things get even worse when we need to handle exceptions that might occur at one of these steps.

The Future interface was added to Java5 to handle asynchronous operations, but Future does not provide a way to combine asynchronous operations with exception handling.

In Java8, the CompletableFuture class is born. It implements not only the Future interface but also the CompletionStage interface. The CompletionStage interface defines the protocol for combining steps of asynchronous computation with other steps.

CompetableFuture provides more than 50 methods for composing,combining, performing asynchronous computing tasks, and handling exceptions.

While such a large API is disruptive, most of this can be demonstrated with a few clear cases.

3. Use CompletableFuture as a simple Future

First, the CompletanbleFuture class is an implementation of the Future interface, so you don’t need extra implementation logic to use it as a normal Future.

For example, you can create a class instance with a no-argument constructor to represent some asynchronous result, call the processing through the consumer, or use the complete method. The consumer can get the results using the GET method, but is blocked until the asynchronous results come out.

In the example below, we create an instance of CompletableFuture, fetch its calculation in another thread, and immediately return the Future. When the calculation is complete, the complete method provided by the Future can end the calculation.

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

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

    return completableFuture;
}

 Future<String> completableFuture = calculateAsync();
 String result = completableFuture.get();
 System.out.println("result="+result);
Copy the code

Results:

result=Hello
Copy the code

To implement asynchronous operations, we use the Executor API. But the creation and termination methods of CompletableFuture can be used by any concurrency mechanism or API that contains native threads.

Note: calculateAsync() returns a Future instance.

We could simply call the Future instance’s get method to get the result, but it would block. Also note that the GET () method throws exceptions such as ExecutionException, which occurs during runtime, and InterruptedException, which is interrupted when a thread executes a method.

If you have a result to compute, you can use the static method completedFuture(), which takes an argument to represent a result it computed. The Future’s get() method will never block, instead returning the result immediately.

Future<String> completableFuture = CompletableFuture.completedFuture("Hello");
String result = completableFuture.get();
System.out.println("result="+result);
Copy the code

Results:

result=Hello
Copy the code

If another scenario happens, like you want to cancel the Future. Suppose we don’t want to get the result and want to cancel the asynchronous task, this can be done using the Future’s cancel() method. This method accepts a Boolean argument, mayInterruptIfRunning, but has no effect in CompletableFuture. Because the CompletanleFuture doesn’t use interrupts to handle the process.

Here are the modified asynchronous methods:

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

    Executors.newCachedThreadPool().submit(() -> {
        Thread.sleep(500);
        completableFuture.cancel(false);
        return null;
    });

    return completableFuture;
}
Copy the code

When we use the Future’s get() method to block the result, we throw a CancellationException /

Future<String> future = calculateAsyncWithCancellation(); future.get(); // CancellationException  Exception in thread "main" java.util.concurrent.CancellationException at java.base/java.util.concurrent.CompletableFuture.cancel(CompletableFuture.java:2475) at main.java.com.yoyocheknow.java8.CompletableFutureTest.lambda$calculateAsyncWithCancellation$6(CompletableFutureTest.java :126) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) at java.base/java.lang.Thread.run(Thread.java:830)Copy the code

4.CompletableFuture encapsulation method

All of the above code is executed using concurrent mechanisms (such as thread pools), but what if we want to skip the useless template methods and simply execute some code changes asynchronously?

Static methods such as runAsync() and supplyAsync() both allow us to create an instance of CompletableFuture, both Runnable and Supplier instances accordingly.

Due to new features in Java8, both Runnable and Suplier allow passing their instances through lambda expressions.

The Runnable interface, like the old interface used in threads, does not allow a return value.

The Supplier interface is a generic method with no arguments that returns a parameterized value. It allows us to pass a lambda expression as a Supplier instance and return the result. The code is as follows:

CompletableFuture<String> future= CompletableFuture.supplyAsync(() -> "Hello");
assertEquals("Hello", future.get());
Copy the code

5. Process the result of asynchronous calculation

The most common way to process the result of a calculation is to pass in a method. The thenApply() method does just that: it takes a Function instance, uses it to process the result, and returns a Future containing the result.

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

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

assertEquals("Hello World", future.get());
Copy the code

If you don’t need to return a value under the Future chain, you can use an instance of the Consumer method interface. It takes an argument, but returns no value.

For cases where you don’t need a return value, there is an alternative: the -thenAccept() method, which takes a Consumer and passes her the result of the calculation, and eventually calls to future.get() will return a null value.

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

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

 System.out.print("result="+future.get());
Copy the code

Output:

Computation returned: Hello
result=null
Copy the code

Finally, if you neither need to evaluate the returned value nor want to get the return at the end, then you can pass a Runnable lambda expression to thenRun(). In the following exit, after the future.get() method is called, we simply print to the console.

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

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

System.out.print("result="+future.get());
Copy the code

Output:

Computation finished.
result=null
Copy the code

6. Merge features

The best feature of the CompletableFuture API is the ability to merge CompletableFuture instances in the chain of computed steps.

The result of the CompletableFuture chain is the ability to allow chaining and composition in a way that is unique among programming languages. And is often used as a single responsibility design pattern.

In the following example, we use thenCompose() to merge two Futures in turn. Notice that this method returns an instance of CompletableFuture. The arguments to this method are the results of the previous calculation step. This allows us to use this value in the lambda expression of the next CompletableFuture.

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

assertEquals("Hello World", completableFuture.get());
Copy the code

The thenCompose() and thenApply() methods implement the template for building a single pattern. They are particularly like the map and flatMap methods in the Stream and Optional classes in Java8.

Both methods take a function and apply it to the result. But the thenCompose(flatMap) method accepts a function of the same type. The structure of this function allows you to combine instances of these classes into a module.

If you want to execute two separate Futures and do something with their results, then use thenCombine() to accept a Future and a Funtion with two parameters to handle the results of two Futures. Such as:

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

assertEquals("Hello World", completableFuture.get());
Copy the code

You want to do something with the results of two Futures, and a simpler example is that you don’t need to pass any result values into the Future chain. Just use the thenAcceptBoth() method.

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

7. The difference between thenApply() and thenCompose()

In our previous section, we showed how to use thenApply() and thenCompose(). Two apis are used in the CompletableFuture call chain, but the two are used differently.

7.1 thenApply ()

This method is used to process the result of the previous call. However, a key point to remember is that the return values of all calls will be combined. So this method is useful when we want to transform a CompletableFuture result.

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

7.2 thenCompose ()

This method, like thenApply(), returns a new execution phase. ThenCompose () uses the previous phase as an argument. It will flatten out and return a Future with results, rather than nested results like thenApply().

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

So to render the CompletableFuture method as a chain, it’s best to use the thenCompose() method. Note also that the difference between these two methods is similar to the difference between map() and flatMap().

8. Run multiple Futures in parallel

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

The allOf static method of CompletableFuture allows waiting for all Futures to complete.

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());
Copy the code

Note that the return value of CompleTableFuture.allof () is a CompletableFuture. The limitation of this approach is that it cannot exploit the results of all Futures portfolios. Instead you have to get the results manually from Futures. Fortunately, the CompletableFeature.join() method and the Java 8 Streams API make it even easier:

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

assertEquals("Hello Beautiful World", combined);
Copy the code

The CompletableFuture.join() method is similar to the get() method. But when the Future doesn’t complete properly, it throws an exception. Using join in the stream.map () method makes this possible.

9. Handling errors

The throw/catch approach is a popular trend for error handling in asynchronous computations.

Instead of catching exceptions in a syntax block, CompletableFuture allows you to use the handle() method to handle exceptions. This method takes two arguments: the result of the calculation (if successful) and the exception thrown (if the calculation did not complete properly).

In the example below, we use the handle() method to get a default value when the asynchronous operation ends abnormally because no name is 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());Copy the code

Consider another scenario, where we want to manually terminate a Future with a return value, but also have the ability to handle exceptions. The completeExceptionally() method is exceptionally suitable. The completableFuture.get() method throws a RuntimeException in the following example

CompletableFuture<String> completableFuture = new CompletableFuture<>(); completableFuture.completeExceptionally( new RuntimeException("Calculation failed!" )); completableFuture.get(); // ExecutionExceptionCopy the code

In the above example we can use the handle() method to handle asynchronous exceptions, but with the GET () method we can use a more typical asynchronous exception handling approach.

10. Asynchronous methods

Most of the API for the CompletableFuture class has two variants ending in Async. These methods are usually used to run the execution steps of another thread.

Methods with the Async suffix perform the next phase of execution by calling a thread. Async methods that do not use an Executor thread pool parameter are executed using a common fork/join thread pool framework such as ForkJoinPool.commonPool(). Async methods with Executor arguments run the next step through the Executor thread pool.

Here is an example of using a Future instance to process the resulting calculation. The only visible difference is the thenApplyAsync() method. A parameter application of the following method is modified as a ForkJoinTask instance. This allows your calculations to be executed in parallel, making your system resources more efficient.

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

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

assertEquals("Hello World", future.get());
Copy the code

11. A summary

This article summarizes some of the methods and common uses of the CompletableFuture class. See the source code:

Github.com/eugenp/tuto…