Understanding Java 8 CompletionStage and CompletableFuture with Practical Examples

This article introduces the Java 8 CompletionStage API and its standard library implementation CompletableFuture. It first explains the contract of CompletionStage as a stage in a computation pipeline that can be completed synchronously or asynchronously, and shows how multiple stages can be chained.

It then presents a series of practical examples, each illustrating a specific method of CompletableFuture:

1. Create a completed CompletableFuture

static void completedFutureExample() {
    CompletableFuture cf = CompletableFuture.completedFuture("message");
    assertTrue(cf.isDone());
    assertEquals("message", cf.getNow(null));
}

2. Run a simple asynchronous stage

static void runAsyncExample() {
    CompletableFuture cf = CompletableFuture.runAsync(() -> {
        assertTrue(Thread.currentThread().isDaemon());
        randomSleep();
    });
    assertFalse(cf.isDone());
    sleepEnough();
    assertTrue(cf.isDone());
}

3. Apply a function to the previous stage

static void thenApplyExample() {
    CompletableFuture cf = CompletableFuture.completedFuture("message").thenApply(s -> {
        assertFalse(Thread.currentThread().isDaemon());
        return s.toUpperCase();
    });
    assertEquals("MESSAGE", cf.getNow(null));
}

4. Apply a function asynchronously

static void thenApplyAsyncExample() {
    CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {
        assertTrue(Thread.currentThread().isDaemon());
        randomSleep();
        return s.toUpperCase();
    });
    assertNull(cf.getNow(null));
    assertEquals("MESSAGE", cf.join());
}

5. Apply a function with a custom Executor

static ExecutorService executor = Executors.newFixedThreadPool(3, new ThreadFactory() {
    int count = 1;
    @Override
    public Thread newThread(Runnable runnable) {
        return new Thread(runnable, "custom-executor-" + count++);
    }
});
static void thenApplyAsyncWithExecutorExample() {
    CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(s -> {
        assertTrue(Thread.currentThread().getName().startsWith("custom-executor-"));
        assertFalse(Thread.currentThread().isDaemon());
        randomSleep();
        return s.toUpperCase();
    }, executor);
    assertNull(cf.getNow(null));
    assertEquals("MESSAGE", cf.join());
}

6. Consume the result of a previous stage

static void thenAcceptExample() {
    StringBuilder result = new StringBuilder();
    CompletableFuture.completedFuture("thenAccept message")
        .thenAccept(s -> result.append(s));
    assertTrue("Result was empty", result.length() > 0);
}

7. Consume the result asynchronously

static void thenAcceptAsyncExample() {
    StringBuilder result = new StringBuilder();
    CompletableFuture cf = CompletableFuture.completedFuture("thenAcceptAsync message")
        .thenAcceptAsync(s -> result.append(s));
    cf.join();
    assertTrue("Result was empty", result.length() > 0);
}

8. Complete a computation exceptionally

static void completeExceptionallyExample() {
    CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
        CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));
    CompletableFuture exceptionHandler = cf.handle((s, th) -> th != null ? "message upon cancel" : "");
    cf.completeExceptionally(new RuntimeException("completed exceptionally"));
    assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally());
    try { cf.join(); fail("Should have thrown an exception"); }
    catch (CompletionException ex) { assertEquals("completed exceptionally", ex.getCause().getMessage()); }
    assertEquals("message upon cancel", exceptionHandler.join());
}

9. Cancel a computation

static void cancelExample() {
    CompletableFuture cf = CompletableFuture.completedFuture("message").thenApplyAsync(String::toUpperCase,
        CompletableFuture.delayedExecutor(1, TimeUnit.SECONDS));
    CompletableFuture cf2 = cf.exceptionally(throwable -> "canceled message");
    assertTrue("Was not canceled", cf.cancel(true));
    assertTrue("Was not completed exceptionally", cf.isCompletedExceptionally());
    assertEquals("canceled message", cf2.join());
}

10‑20. Various combinators – The article continues with examples of applyToEither, acceptEither, runAfterBoth, thenAcceptBoth, thenCombine, thenCombineAsync, thenCompose, anyOf, allOf, and allOfAsync. Each snippet shows how to combine, compose, or wait for multiple CompletableFuture instances, both synchronously and asynchronously, and how to handle results or exceptions.

Real‑world scenario

cars().thenCompose(cars -> {
    List<CompletionStage> updatedCars = cars.stream()
        .map(car -> rating(car.manufacturerId).thenApply(r -> { car.setRating(r); return car; }))
        .collect(Collectors.toList());
    CompletableFuture done = CompletableFuture.allOf(updatedCars.toArray(new CompletableFuture[updatedCars.size()]));
    return done.thenApply(v -> updatedCars.stream().map(CompletionStage::toCompletableFuture)
        .map(CompletableFuture::join).collect(Collectors.toList()));
}).whenComplete((cars, th) -> {
    if (th == null) { cars.forEach(System.out::println); }
    else { throw new RuntimeException(th); }
}).toCompletableFuture().join();

The example demonstrates fetching a list of cars asynchronously, enriching each car with a rating obtained from another asynchronous call, and using allOf to wait for all enrichments before printing the results, illustrating how CompletableFuture can replace manual thread coordination for better performance.

Throughout the article, the author interleaves explanatory text with code, highlights important behaviors (e.g., methods ending with Async run on the ForkJoinPool by default), and provides tips such as using a custom executor or handling cancellation via completeExceptionally. The final sections also contain promotional messages for a WeChat public account, but the core technical content remains a comprehensive tutorial on Java 8 asynchronous programming.