Concurrency

55. CompletableFuture

CompletableFuture enables non-blocking async pipelines: thenApply transforms the result, thenCompose chains async steps, allOf awaits multiple futures, and exceptionally handles failures.

Java 8+ (CompletableFuture); Java 21 (core language)

Explanation

CompletableFuture<T> is a Future that you can complete programmatically and chain operations on. It supports a fluent API where each step returns a new CompletableFuture, enabling non-blocking pipelines.

Key operations: supplyAsync(supplier) runs asynchronously in ForkJoinPool.commonPool() (or a custom executor). thenApply(fn) transforms the result when available (like map). thenCompose(fn) chains to another async operation (like flatMap). thenAccept(consumer) consumes without returning. thenCombine(other, fn) merges two futures. allOf(futures...) waits for all. anyOf(futures...) waits for the first. exceptionally(fn) recovers from a failure.

Async vs non-async: thenApply(fn) runs on the completing thread (or the calling thread if already done). thenApplyAsync(fn) always dispatches to the executor. Use Async variants in server code to prevent blocking event loops.

Example

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;

public class CompletableFutureDemo {
    static String fetchUser(long id) throws InterruptedException {
        Thread.sleep(100); return "User" + id;
    }
    static String fetchEmail(String user) throws InterruptedException {
        Thread.sleep(50); return user + "@example.com";
    }

    public static void main(String[] args) throws Exception {
        // Basic pipeline
        CompletableFuture<String> email = CompletableFuture
            .supplyAsync(() -> { try { return fetchUser(1L); } catch (Exception e) { throw new RuntimeException(e); } })
            .thenCompose(user -> CompletableFuture.supplyAsync(() -> {
                try { return fetchEmail(user); } catch (Exception e) { throw new RuntimeException(e); }
            }))
            .thenApply(String::toUpperCase);

        System.out.println("Email: " + email.get()); // USER1@EXAMPLE.COM

        // Combining two independent async results
        var f1 = CompletableFuture.supplyAsync(() -> "Hello");
        var f2 = CompletableFuture.supplyAsync(() -> " World");
        String combined = f1.thenCombine(f2, (a, b) -> a + b).get();
        System.out.println("Combined: " + combined);

        // allOf — wait for all
        var futures = new CompletableFuture<?>[]{
            CompletableFuture.runAsync(() -> System.out.println("Task A")),
            CompletableFuture.runAsync(() -> System.out.println("Task B")),
            CompletableFuture.runAsync(() -> System.out.println("Task C"))
        };
        CompletableFuture.allOf(futures).get();

        // Error handling
        CompletableFuture<String> withError = CompletableFuture
            .supplyAsync(() -> { throw new RuntimeException("oops"); })
            .exceptionally(ex -> "Recovered: " + ex.getMessage());
        System.out.println(withError.get());

        // Timeout (Java 9+)
        CompletableFuture<String> slow = CompletableFuture
            .supplyAsync(() -> { try { Thread.sleep(5000); } catch (Exception e) {} return "slow"; })
            .orTimeout(200, TimeUnit.MILLISECONDS)
            .exceptionally(ex -> "Timed out!");
        System.out.println(slow.get());
    }
}

Sources