Futures in Java: CompletableFuture
Futures in Java: Asynchronous Programming using CompletableFuture
Java remains the most popular primary programming language. We all are, of course enjoying the automatic memory management unlike in C and C++ since couple of years. Also Java went through and is going through some major changes which changes the way we essentially think and write code in Java.
Java 8 which was released in March-2014, was a revolutionary one and it was a huge upgrade to the Java programming model with the introduction of Lambda expressions, Streams, Functional Interfaces, Default methods, Optional and CompletableFuture.
Typically when writing code, we write it to execute synchronously, meaning one instruction must complete before the next one is executed. classic example is that synchronization is a blocking way to prevent race conditions
For example whenever we make a database call, we wait until we get results from the database and the corresponding thread is blocked until we get the results. This is blocking mode of operation and doing things in a synchronous way. Essentially blocking means that a thread has to wait to execute a task or to access a resource.
Imagine if the query we want to execute gets into a deadlock and we will be waiting for the results forever and that thread will be blocked indefinitely. But we want threads to be as busy as possible.
This is where the asynchronous programming comes into picture.
I will try to illustrate how we can build pipeline of tasks in an asynchronous way, error handling while building these data pipelines and how to build such pipelines in a performant way.
We will see what are the components that can be used to make our code asynchronous.
1) Future InterfaceJava has a Future interface for this. A Future represents the result of an asynchronous operation. Long running methods, external service calls and database hits are the good candidates for asynchronous operations.
Future<T> future = database.getResultSet();
future.get();It is interesting that this get() is a blocking call, waits if necessary for the computation to complete and then retrieves the result.
We can also set a timeLimit on how long this asynchronous task can take by passing time and timeUnit in the get() and it throws TimeOutException along with the InterruptedException and ExecutionException.
try {
Future<T> future = someService.getResult();
future.get(200, TimeUnit.MILLISECONDS);
} catch(InterruptedException | ExecutionException |TimeoutException e) {
// handle exceptions here
}The Future interface also has methods called isDone() which is used to check the completion of the asynchronous operation, isCancelled() as the name suggests to check the task is cancelled, and cancel() which attempts to cancel the task and it fails if the task has already completed, cancelled or could not cancel the task.
Also, it’s worth taking a note that these asynchronous tasks usually gets executed by separate thread pool and the future object acts a bridge between the main thread and the thread executing the asynchronous task.
Lets see how this works:
ExecutorService service = Executors.newSingleThreadExecutorService();Runnable runnable = () -> { do something... };Future future = service.submit(runnable);so the main thread creates another threadPool, submits the task which is runnable here and gets the future back from the service thread. While the service thread executes the submitted runnable task, main thread can continue doing the work.
2) CompletionStage InterfaceCompletionStage interface represents possibly an asynchronous operation in a pipeline of computations. you can imagine a bunch of tasks chained together and the computation of each task depends on the previous task/asynchronous operation’s result.
The computation stage may be expressed as a Function, Consumer or Runnable. Usually, the completion of one stage triggers the execution of others. The CompletionStage interface has several methods which can be employed to chain several computations or tasks to achieve end result, to name a few, thenAccept(), thenApply(), thenCombine(), thenRun(), whenComplete(). Implementations of CompletionStage may provide means of achieving such effects, as appropriate.
3) CompletableFutureCompletableFuture was introduced in Java 8, as an extension of Future’s API. You may wonder why we needed CompletableFuture when there already exists Future’s API which was introduces in Java 5. well, it just represents the result of an asynchronous operation and on top of that
1. There is no way to complete the future. we can only attempt to cancel the task.
2. The get() method in the Future Interface is blocking operation
3. No support for exception handling
4. Multiple futures cannot be chained together.So all these concerns are addressed in Java 8 with the introduction of CompletableFuture.
Let’s see some example code: below code illustrates creation of simple CompletableFuture object.
CompletableFuture<String> completableFuture = new CompletableFuture<>();CompletableFuture doesn’t work with callable’s. but it does with runnable’s and supplier functions. but we have to be careful that supplier functions doesn’t throw checked exceptions.
Let us explore the different methods associated with the CompletableFuture class and how we can utilize them to build a pipeline of asynchronous operation or data pipelines.
Triggered by a single previous stage:
1) Complete(): Let’s you manually complete the Future with the given value
boolean result = completableFuture.complete("Future result value");2) thenApply(): Takes function and apply it on to the result of the previous stage. Remember that thenApply() is a synchronous mapping function. It returns a CompletionStage holding the result of the function.
CompletableFuture<String> f1 = //...
CompletableFuture<Integer> f2 = f1.thenApply(Integer::parseInt);3) thenAccept(): Takes a consumer and returns CompletionStage<Void>. thenAccept() is usually called during the final stages of pipeline and returns a final value.
CompletableFuture<String> f1 = //...
CompletableFuture<Void> f2 =f1.theAccept(Integer::parseInt);4) thenRun(): Executes a Runnable and returns CompletionStage<Void>. It will not even be having access to final value in the pipeline.
CompletableFuture<String> f1 = //...
CompletableFuture<Void> f2 = f1.thenRun(() -> System.out.println("Computation finished."));Triggered by both of previous stages: Each of these methods take another CompletionStage and one of: BiFunction, BiConsumer or Runnable.
5) thenCombine(): Takes BiFunction and acts upon the result of the previous two stages in the pipeline. It returns CompletionStage holding the result of the Function.
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> "Hello").thenCombine(CompletableFuture.supplyAsync(() -> " associates! Its thenCombined"), (value1, value2) -> value1 + value2); completableFuture.thenAccept(System.out::println); // Hello associates! Its thenCombined
6) thenAcceptBoth(): Takes a BiConsumer and returns CompletionStage<Void>. thenAccept() is usually called during the final stages of pipeline and returns a final value.
String original = "message";
StringBuilder result = new StringBuilder();CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).thenAcceptBoth(CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),(s1, s2) -> result.append(s1 + s2));
When both the previous stages result in an exception, then it’s not specified which of the two exceptions gets propagated to the dependent stage.
7) runAfterBoth(): Returns a new CompletionStage that, when this and the other given stage both complete normally, executes the given action.
String original = "message";
StringBuilder result = new StringBuilder();CompletableFuture.completedFuture(original).thenApply(String::toUpperCase).runAfterBoth(CompletableFuture.completedFuture(original).thenApply(String::toLowerCase),() -> result.append("done"));
Exception Handling: Instead of catching an exception in a syntactic block, the CompletableFuture class allows you 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).
8) completeExceptionally(): Operation can be completed exceptionally, indicating a failure in the computation. Separate “handler” stage, exceptionHandler, that handles any exception by returning another message "message upon cancel".Next, we explicitly complete the second stage with an exception.
CompletableFuture<String> f1 = CompletableFuture.completedFuture("message");f1.completeExceptionally(new RuntimeException("completed exceptionally"));
9) exceptionally(): Returns a new CompletableFuture that is completed when this CompletableFuture completes, with the result of the given function of the exception triggering this CompletableFuture’s completion when it completes exceptionally; otherwise, if this CompletableFuture completes normally.
CompletableFuture<String> f1 = CompletableFuture.completedFuture("message");CompletableFuture<String> f2 = f1.exceptionally(throwable -> "canceled message");
CompletableFuture’s provides us 50 different methods for composing, combining, executing asynchronous computation steps and handling errors. and most of the above methods have async variants, for example thenApplyAsync, thenCombineAsync, thenComposeAsync, thenRunAsync etc.
Also while executing the async functions, we have the flexibility to mention the callback executors as well.
Summary: The introduction of CompletableFuture is a big leap to the way we build pipelines of asynchronous computations. It allows us to attach callbacks and much more, in fact, it allows you to build a pipeline of steps which can each be executed asynchronously, each step depending on the result of one or more previous steps.
Interesting reads:1. Load Balancers: A deep dive2. Java Lombok: do we need getters and setters
Happy learning.!!
Follow me on Twitter.
Resources:
