Java 8: Definitive Guide to CompletableFuture

Java 8 is coming so it's time to study new features. While Java 7 and Java 6 were rather minor releases, version 8 will be a big step forward. Maybe even too big? Today I will give you a thorough explanation of new abstraction in JDK 8 - CompletableFuture<T>. As you all know Java 8 will hopefully be released in less than a year, therefore this article is based on JDK 8 build 88 with lambda support. CompletableFuture<T> extendsFuture<T> by providing functional, monadic (!) operations and promoting asynchronous, event-driven programming model, as opposed to blocking in older Java. If you openedJavaDoc of CompletableFuture<T> you are surely overwhelmed. About fifty methods(!), some of them being extremely cryptic and exotic, e.g.:

public <U,V> CompletableFuture<V> thenCombineAsync(
CompletableFuture<? extends U> other,
BiFunction<? super T,? super U,? extends V> fn,
Executor executor)

Don't worry, but keep reading. 

 collects all the features ofListenableFuture in Guava with SettableFuture. Moreover built-in lambda support brings it closer to Scala/Akka futures. Sounds too good to be true, but keep reading.

 has two major areas superior to good ol' 

 - asynchronous callback/transformations support and the ability to set value of

 from any thread at any point in time. 

Extract/modify wrapped value

Typically futures represent piece of code running by other thread. But that's not always the case. Sometimes you want to create a 

 representing some event that you know will occur, e.g. JMS message arrival. So you have 

 but there is no asynchronous job underlying this future. You simply want to complete (resolve) that future when JMS message arrives, and this is driven by an event. In this case you can simply create 

, return it to your client and whenever you think your results are available, simply 

 the future and unlock all clients waiting on that future.

For starters you can simply create new 

 out of thin air and give it to your client:

public CompletableFuture<String> ask() {
final CompletableFuture<String> future = new CompletableFuture<>();
//...
return future;
}

Notice that this future is not associated wtih any 

, no thread pool, no asynchronous job. If now the client code calls 

 it will block forever. If it registers some completion callbacks, they will never fire. So what's the point? Now you can say:

...and at this very moment all clients blocked on 

 will get the result string. Also completion callbacks will fire immediately. This comes quite handy when you want to represent a task in the future, but not necessarily computational task running on some thread of execution. 

 can only be called once, subsequent invocations are ignored. But there is a back-door called

 which overrides previous value of the

 with new one. Use with caution.

Sometimes you want to signal failure. As you know 

 objects can handle either wrapped result or exception. If you want to pass some exception further, there is

 (and 

evil brother that overrides the previous exception). 

 also unlock all waiting clients, but this time throwing an exception from 

. Speaking of

, there is also 

 method with some subtle changes in error handling. But in general they are the same. And finally there is also

 method that doesn't block but if the

 is not completed yet, returns default value. Useful when building robust systems where we don't want to wait too much.

Last 

 utility method is 

 that returns already completed

 object. Might be useful for testing or when writing some adapter layer.

Creating and obtaining CompletableFuture

OK, so is creating 

 manually our only option? Not quite. Just as with normal 

s we can wrap existing task with 

 using the following family of factory methods:

static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier);
static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor);
static CompletableFuture<Void> runAsync(Runnable runnable);
static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor);

Methods that do not take an 

 as an argument but end with 

 will useForkJoinPool.commonPool() (global, general purpose pool introduces in JDK 8). This applies to most methods in 

 class. 

 is simple to understand, notice that it takes 

, therefore it returns

 as 

 doesn't return anything. If you need to process something asynchronously and return result, use Supplier<U>:

final CompletableFuture<String> future = CompletableFuture.supplyAsync(new Supplier<String>() {
@Override
public String get() {
//...long running...
return "42";
}
}, executor);

But hey, we have lambdas in Java 8!

final CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> {
//...long running...
return "42";
}, executor);

or even:

final CompletableFuture<String> future =
CompletableFuture.supplyAsync(() -> longRunningTask(params), executor);

This article is not about project Lambda, but I will be using lambdas quite extensively.

Transforming and acting on one CompletableFuture(thenApply)

So I said that 

 is superior to 

 but you haven't yet seen why? Simply put, it's because 

 is a monad and a functor. Not helping I guess? Both Scala and JavaScript allow registering asynchronous callbacks when future is completed. We don't have to wait and block until it's ready. We can simply say: run this function on a result, when it arrives. Moreover, we can stack such functions, combine multiple futures together, etc. For example if we have a function from 

 to 

we can turn 

 to 

 without unwrapping it. This is achieved with 

 family of methods:

<U> CompletableFuture<U> thenApply(Function<? super T,? extends U> fn);
<U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn);
<U> CompletableFuture<U> thenApplyAsync(Function<? super T,? extends U> fn, Executor executor);

As stated before 

 versions are provided for most operations on

 thus I will skip them in subsequent sections. Just remember that first method will apply function within the same thread in which the future completed while the remaining two will apply it asynchronously in different thread pool.

Let's see how 

 works:

CompletableFuture<String> f1 = //...
CompletableFuture<Integer> f2 = f1.thenApply(Integer::parseInt);
CompletableFuture<Double> f3 = f2.thenApply(r -> r * r * Math.PI);

Or in one statement:

CompletableFuture<Double> f3 =
f1.thenApply(Integer::parseInt).thenApply(r -> r * r * Math.PI);

You see a sequence of transformations here. From 

 to 

 and then to

. But what's most important, these transformations are neither executed immediately nor blocking. They are simply remembered and when original 

 completes they are executed for you. If some of the transformations are time-consuming, you can supply your own 

 to run them asynchronously. Notice that this operation is equivalent to monadic 

 in Scala.

Running code on completion (thenAccept/thenRun)

CompletableFuture<Void> thenAccept(Consumer<? super T> block);
CompletableFuture<Void> thenRun(Runnable action);

These two methods are typical "final" stages in future pipeline. They allow you to consume future value when it's ready. While 

 provides the final value, 

executes 

 which doesn't even have access to computed value. Example:

future.thenAcceptAsync(dbl -> log.debug("Result: {}", dbl), executor);
log.debug("Continuing");

 variants are available as well for both methods, with implicit and explicit executor. I can't emphasize this enough: 

/

 methods do not block (even without explicit 

). Treat them like an event listener/handler that you attach to a future and that will execute some time in the future. 

 message will appear immediately, even if 

 is not even close to completion.

Error handling of single CompletableFuture

So far we only talked about result of computation. But what about exceptions? Can we handle them asynchronously as well? Sure!

CompletableFuture<String> safe =
future.exceptionally(ex -> "We have a problem: " + ex.getMessage());

 takes a function that will be invoked when original future throws an exception. We then have an opportunity to recover by transforming this exception into some value compatible with 

's type. Further transformations of 

 will no longer yield an exception but instead a 

 returned from supplied function.

A more flexible approach is 

 that takes a function receiving either correct result or exception:

CompletableFuture<Integer> safe = future.handle((ok, ex) -> {
if (ok != null) {
return Integer.parseInt(ok);
} else {
log.warn("Problem", ex);
return -1;
}
});

 is called always, with either result or exception argument being not-

. This is a one-stop catch-all strategy.

Combining two CompletableFuture together

Asynchronous processing of one 

 is nice but it really shows its power when multiple such futures are combined together in various ways.

Combining (chaining) two futures (thenCompose())

Sometimes you want to run some function on future's value (when it's ready). But this function returns future as well. 

 should be smart enough to understand that the result of our function should now be used as top-level future, as opposed to 

. Method 

is thus equivalent to 

 in Scala:

<U> CompletableFuture<U> thenCompose(Function<? super T,CompletableFuture<U>> fn);

 variations are available as well. Example below, look carefully at the types and the difference between 

 (

) and 

 (

) when applying a 

 function returning 

:

CompletableFuture<Document> docFuture = //...
CompletableFuture<CompletableFuture<Double>> f =
docFuture.thenApply(this::calculateRelevance);
CompletableFuture<Double> relevanceFuture =
docFuture.thenCompose(this::calculateRelevance);
//...
private CompletableFuture<Double> calculateRelevance(Document doc)  //...

 is an essential method that allows building robust, asynchronous pipelines, without blocking or waiting for intermediate steps.

Transforming values of two futures (thenCombine())

While 

 is used to chain one future dependent on the other,

 combines two independent futures when they are both done:

 variations are available as well. Imagine you have two 

s, one that loads 

 and other that loads nearest 

. They are completely independent from each other, but when both of them are completed, you want to use their values to calculate 

. Here is a stripped example:

CompletableFuture<Customer> customerFuture = loadCustomerDetails(123);
CompletableFuture<Shop> shopFuture = closestShop();
CompletableFuture<Route> routeFuture =
customerFuture.thenCombine(shopFuture, (cust, shop) -> findRoute(cust, shop));
//...
private Route findRoute(Customer customer, Shop shop) //...

Notice that in Java 8 you can replace 

with simple 

 method reference:

customerFuture.thenCombine(shopFuture, this::findRoute);

So you get the idea. We have 

 and 

. Then 

wraps them and "waits" for both to complete. When both of them are ready, it runs our supplied function that combines results (

). Thus 

 will complete when two underlying futures are resolved and

 is done.

Waiting for both CompletableFutures to complete

If instead of producing new 

 combining both results we simply want to be notified when they finish, we can use 

/

 family of methods (

 variations are available as well). They work similarly to

 and 

 but wait for two futures instead of one:

extends U> other, BiConsumer<? super T,? super U> block)
CompletableFuture<Void> runAfterBoth(CompletableFuture<?> other, Runnable action)

Imagine that in the example above, instead of producing new

 you simply want send some event or refresh GUI immediately. This can be easily achieved with 

:

customerFuture.thenAcceptBoth(shopFuture, (cust, shop) -> {
final Route route = findRoute(cust, shop);
//refresh GUI with route
});

I hope I'm wrong but maybe some of you are asking themselves a question: why can't I simply block on these two futures? Like here:

Future<Customer> customerFuture = loadCustomerDetails(123);
Future<Shop> shopFuture = closestShop();
findRoute(customerFuture.get(), shopFuture.get());

Well, of course you can. But the whole point of 

 is to allow asynchronous, event driven programming model instead of blocking and eagerly waiting for result. So functionally two code snippets above are equivalent, but the latter unnecessarily occupies one thread of execution.

Waiting for first CompletableFuture to complete

Another interesting part of the 

 API is the ability to wait for first (as opposed to all) completed future. This can come handy when you have two tasks yielding result of the same type and you only care about response time, not which task resulted first. API methods (

 variations are available as well):

CompletableFuture<Void> acceptEither(CompletableFuture<? extends T> other, Consumer<? super T> block)
CompletableFuture<Void> runAfterEither(CompletableFuture<?> other, Runnable action)

As an example say you have two systems you integrate with. One has smaller average response times but high standard deviation. Other one is slower in general, but more predictable. In order to take best of both worlds (performance and predictability) you call both systems at the same time and wait for the first one to complete. Normally it will be the first one, but in case it became slow, second one finishes in an acceptable time:

CompletableFuture<String> fast = fetchFast();
CompletableFuture<String> predictable = fetchPredictably();
fast.acceptEither(predictable, s -> {
System.out.println("Result: " + s);
});

 represents 

 reply either from 

 or from 

. We neither know nor care.

Transforming first completed

 is an older brother of 

. While the latter simply calls some piece of code when faster of two futures complete, 

 will return a new future. This future will complete when first of the two underlying futures complete. API is a bit similar (

 variations are available as well):

<U> CompletableFuture<U> applyToEither(CompletableFuture<? extends T> other, Function<? super T,U> fn)

The extra 

 function is invoked on the result of first future that completed. I am not really sure what's the purpose of such a specialized method, after all one could simply use:

. Since we are stuck with this API but we don't really need extra function application, I will simply useFunction.identity() placeholder:

CompletableFuture<String> fast = fetchFast();
CompletableFuture<String> predictable = fetchPredictably();
CompletableFuture<String> firstDone =
fast.applyToEither(predictable, Function.<String>identity());

 future can then be passed around. Notice that from the client perspective the fact that two futures are actually behind 

 is hidden. Client simply waits for future to complete and 

 takes care of notifying the client when any of the two finish first.

Combining multiple CompletableFuture together

So we now know how to wait for two futures to complete (using 

) and for the first one to complete (

). But can it scale to arbitrary number of futures? Sure, using 

 helper methods:

static CompletableFuture<Void> allOf(CompletableFuture<?>... cfs)
static CompletableFuture<Object> anyOf(CompletableFuture<?>... cfs)

 takes an array of futures and returns a future that completes when all of the underlying futures are completed (barrier waiting for all). 

 on the other hand will wait only for the fastest of the underlying futures. Please look at the generic type of returned futures. Not quite what you would expect? We will take care of this issue in the next article.

Summary

We explored pretty much whole CompletableFuture API. I'm sure this was quite overwhelming so in the next article shortly we will develop yet another implementation of simple web crawling program, taking advantage of 

 functionalities and Java 8 lambdas. We will also look at disadvantages and shortcomings of

.