Java 8 CompletableFuture Tutorial with Examples

Introduction

The problem with Future

• Blocking only: get() blocks the calling thread until the result is ready. There is no way to say "when the result arrives, do this with it".
• No chaining: a second step that needs the first result has to wait for it with get() and then submit another task by hand.
• No combining: waiting for two or ten futures, or for the first of several to finish, needs hand-written loops.
• No manual completion: a Future can only be completed by the task that created it.
• No error pipeline: an exception only appears, wrapped in an ExecutionException, when someone calls get().

What is CompletableFuture

Methods to create and run a CompletableFuture

• supplyAsync() runs a SupplierClick to read: Supplier on another thread and completes the returned future with its value.
• runAsync() runs a Runnable that returns nothing; its future completes with null when the task ends.
• Both have an overload taking an Executor as a second argument. Without it, tasks run in ForkJoinPool.commonPool(), the same shared pool that parallel streams use.
• completedFuture(value) returns a future that is already complete, and new CompletableFuture<>() creates an empty one to be completed later with complete(value).

Methods to attach the next step

• thenApply(Function<T,U>): transforms the result, returning CompletableFuture<U> - the CompletableFuture equivalent of map().
• thenAccept(Consumer<T>): uses the result without producing a new one, returning CompletableFuture<Void>.
• thenRun(Runnable): runs an action after completion without even looking at the result.
• join() and get(): wait for the result. join() throws an unchecked CompletionException on failure, so it is convenient in lambdas and streams; get() throws the checked InterruptedException and ExecutionException, and has a timeout overload.

Java 8 code example of CompletableFuture basics

//Employee.java
public class Employee {
  private final String name;
  private final double monthlySalary;

  public Employee(String name, double monthlySalary) {
    this.name = name;
    this.monthlySalary = monthlySalary;
  }

  public String getName() { return name; }
  public double getMonthlySalary() { return monthlySalary; }
}

//CompletableFutureBasics.java
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class CompletableFutureBasics {

  // Imitates a slow lookup, such as a database query or a remote call
  static Employee findEmployee(String id) {
    sleep(200);
    return new Employee("Tom Jones", 5200);
  }

  static void sleep(long millis) {
    try {
      Thread.sleep(millis);
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
    }
  }

  public static void main(String[] args) throws Exception {
    // 1. Start a task asynchronously and collect the result later
    CompletableFuture<Employee> lookup =
        CompletableFuture.supplyAsync(() -> findEmployee("E-1001"));
    System.out.println("1. Lookup started, done yet? " + lookup.isDone());
    System.out.println("   Found: " + lookup.join().getName());

    // 2. A pipeline: each stage runs when the previous one completes
    CompletableFuture<String> annual = CompletableFuture
        .supplyAsync(() -> findEmployee("E-1001"))
        .thenApply(e -> e.getMonthlySalary() * 12)
        .thenApply(salary -> String.format("Annual salary: %.2f", salary));
    System.out.println("2. " + annual.join());

    // 3. thenAccept and thenRun end a pipeline with side effects
    CompletableFuture<Void> done = CompletableFuture
        .supplyAsync(() -> findEmployee("E-1001"))
        .thenAccept(e -> System.out.println("3. Sending payslip to " + e.getName()))
        .thenRun(() -> System.out.println("   Payslip run finished"));
    done.join();

    // 4. A future completed by hand, from another thread
    CompletableFuture<String> approval = new CompletableFuture<>();
    new Thread(() -> {
      sleep(100);
      approval.complete("approved");
    }).start();
    System.out.println("4. Manager decision: " + approval.get(1, TimeUnit.SECONDS));

    // 5. Running the task in our own thread pool instead of the common pool
    ExecutorService payrollPool = Executors.newFixedThreadPool(2);
    try {
      String name = CompletableFuture
          .supplyAsync(() -> findEmployee("E-1001").getName(), payrollPool)
          .thenApply(String::toUpperCase)
          .join();
      System.out.println("5. With our own pool: " + name);
    } finally {
      payrollPool.shutdown();
    }

    // 6. An already completed future, useful for caches and tests
    CompletableFuture<String> cached = CompletableFuture.completedFuture("Tom Jones");
    System.out.println("6. Completed at once? " + cached.isDone()
        + ", value: " + cached.join());
  }
}

1. Lookup started, done yet? false
   Found: Tom Jones
2. Annual salary: 62400.00
3. Sending payslip to Tom Jones
   Payslip run finished
4. Manager decision: approved
5. With our own pool: TOM JONES
6. Completed at once? true, value: Tom Jones

Explanation of the code

• Part 1: supplyAsync() returns at once, while findEmployee() is still sleeping on a pool thread - so isDone() is false. join() then waits for the result.
• Part 2: two thenApply() calls turn a CompletableFuture<Employee> into a CompletableFuture<Double> and then into a CompletableFuture<String>. The whole pipeline is set up before any of it has run; the main thread only waits at join().
• Part 3: thenAccept() consumes the employee, and thenRun() runs after it. The resulting CompletableFuture<Void> is joined only to keep main() from ending before the messages are printed.
• Part 4: an empty CompletableFuture is completed by another thread with complete(). This is how callback-based APIs, such as a message listener, can be turned into a CompletableFuture. get(1, TimeUnit.SECONDS) waits at most one second.
• Part 5: passing an Executor to supplyAsync() runs the task in that pool. Use your own pool for slow or blocking work such as I/O, so that it does not tie up the common pool that parallel streams and other code share. A method reference, String::toUpperCase, works as a stage just like a lambda.
• Part 6: completedFuture() is complete from the start, which makes it easy to return a cached value from a method whose signature returns a CompletableFuture.

thenApply versus thenApplyAsync

• thenApply() runs the function in the thread that completed the previous stage, or, if that stage is already complete when thenApply() is called, in the calling thread. It is the cheapest option for short, non-blocking steps.
• thenApplyAsync() always hands the function to the common pool (or to the given executor). Use it when a step is slow, so that it does not hold up the thread that completed the previous stage.

Best practices

• Keep the pipeline non-blocking: call join() or get() only at the very end, where a result is really needed - not between stages.
• Use your own executor for blocking I/O, and size it for the I/O, not for the number of CPUs.
• Always shut down executors you create, as part 5 does, or the JVM may not exit.
• Handle failures: an exception in any stage skips the remaining stages. The follow-up tutorial shows exceptionally() and handle().

Summary