CompletableFuture for the impatient

CompletableFuture<T> is the JDK’s most capable async primitive — a Future you can compose, chain, fork and join without ever calling Thread.start yourself. This tutorial gets you to “I can ship this” in ten minutes.

Two ways to start

You always begin with one of two factory methods on CompletableFuture:

// No result expected — pass a Runnable
CompletableFuture<Void> a =
    CompletableFuture.runAsync(() -> System.out.println("started"));

// A result is expected — pass a Supplier<T>
CompletableFuture<String> b =
    CompletableFuture.supplyAsync(() -> "hello");

Both run on the common ForkJoinPool unless you hand them an explicit Executor (we’ll get to that). The call returns immediately; the work happens in the background.

Four ways to end

Once you have a CompletableFuture<T>, you need to pull the value out. Four common terminators:

// 1. join() — blocks. Throws CompletionException (RuntimeException) on failure.
String x = b.join();

// 2. get() — blocks. Declares InterruptedException + ExecutionException.
String y = b.get();
String z = b.get(1, TimeUnit.SECONDS);          // with timeout

// 3. getNow(default) — doesn't block. Returns default if not yet complete.
String maybe = b.getNow("not ready yet");

// 4. thenAccept(consumer) — non-blocking; runs the consumer when ready.
CompletableFuture<Void> done = b.thenAccept(System.out::println);

In day-to-day code, thenAccept is what you want. It keeps the call site non-blocking and slots the consumer into the same pipeline.

CompletableFuture
    .supplyAsync(() -> "hello reactive world")
    .thenAccept(System.out::println);

Sync vs Async on every method

Every then* method has an *Async cousin: thenAccept / thenAcceptAsync, thenApply / thenApplyAsync, thenCompose / thenComposeAsync, etc.

Non-Async variant — the next stage runs on whatever thread completed the previous one.
*Async variant — the next stage is scheduled onto the common ForkJoinPool, freeing the completer immediately.
*Async(..., Executor) — same, but onto an executor you provide.

Rule of thumb: use *Async whenever the next stage is non-trivial or its concurrency profile differs from the previous stage’s. Passing your own Executor is how you isolate CPU-bound work from blocking I/O.

ExecutorService ioPool   = Executors.newFixedThreadPool(32);
ExecutorService cpuPool  = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

CompletableFuture
    .supplyAsync(() -> blockingFetch(), ioPool)        // I/O on ioPool
    .thenApplyAsync(this::heavyParse,    cpuPool)      // CPU on cpuPool
    .thenAccept(System.out::println);

Building a multi-stage pipeline

The interesting workflows have several inputs from different sources, combined into a result. The recipe is supplyAsync for each input and thenCombineAsync to merge two of them:

ExecutorService waitPool = Executors.newFixedThreadPool(8);
ExecutorService workPool = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

CompletableFuture<String> result =
    CompletableFuture.supplyAsync(() -> fetchA(), waitPool)
        .thenCombineAsync(
            CompletableFuture.supplyAsync(() -> fetchB(), waitPool),
            (a, b) -> a + " + " + b,                       // operator one
            workPool)
        .thenCombineAsync(
            CompletableFuture.supplyAsync(() -> fetchC(), waitPool),
            (ab, c) -> ab + " - " + c,                     // operator two
            workPool);

result.thenAccept(System.out::println);

Notice the separation of concerns: I/O lives on waitPool, transformations on workPool. Each thenCombineAsync waits for both inputs before firing.

Many pipelines in parallel

A Stream of CompletableFutures lets you fan out hundreds of independent pipelines and collect them at the end:

IntStream.range(0, 1_000)
    .parallel()
    .mapToObj(i -> CompletableFuture
        .supplyAsync(() -> fetchA(), waitPool)
        .thenCombineAsync(CompletableFuture.supplyAsync(() -> fetchB(), waitPool),
                          (a, b) -> a + " + " + b, workPool)
        .thenCombineAsync(CompletableFuture.supplyAsync(() -> fetchC(), waitPool),
                          (ab, c) -> ab + " - " + c, workPool)
        .thenAcceptAsync(System.out::println, workPool))
    .forEach(CompletableFuture::join);

Stream.parallel() walks the index range concurrently; each iteration kicks off an independent pipeline; the final forEach(join) waits for all of them to finish.

For 1000 elements this beats a thread-per-task design comfortably and stays comprehensible — no StampedLock, no manual synchronization.

Failure handling

When a stage throws, the failure propagates down the chain as a CompletionException. Two methods let you intervene:

// handleAsync — gets (value, throwable); decides what to return next
CompletableFuture<String> safe =
    CompletableFuture.supplyAsync(() -> Integer.parseInt(input))
        .handleAsync((value, ex) -> ex == null
            ? "got " + value
            : "failed: " + ex.getMessage());

// exceptionally — only fires on failure
CompletableFuture<Integer> fallback =
    CompletableFuture.supplyAsync(() -> Integer.parseInt(input))
        .exceptionally(ex -> -1);

If you want the failure to travel as a value instead of an exception, swap handleAsync for a Result-producing form:

import com.svenruppert.functional.model.Result;

CompletableFuture<Result<Integer>> r =
    CompletableFuture.supplyAsync(() -> Integer.parseInt(input))
        .handleAsync((value, ex) -> ex == null
            ? Result.success(value)
            : Result.failure(ex.getMessage()));

Now every downstream stage sees a Result<Integer> and can short-circuit cleanly. Pair this with CheckedFunction and the handleAsync becomes a one-liner.

What’s next

You now have everything you need to build async pipelines by hand. As they grow, the per-stage scaffolding starts to repeat. CompletableFutureQueue encapsulates the pattern so you can build pipelines as values and apply them later — see Async pipelines with CompletableFutureQueue.

Recap

runAsync (no result) and supplyAsync (result) start work.
thenAccept ends non-blocking pipelines; join / get block.
Every then* method has an *Async cousin that re-schedules onto an Executor.
thenCombineAsync merges two futures with a BiFunction.
handleAsync / exceptionally handle failure; convert to Result<T> for cleaner downstream code.

Async pipelines with CompletableFutureQueue — pipelines as values.
Error handling with Result — what every failure stage should produce.