CompletableFuture

the picture

ordering food at a counter that hands you a buzzer. a plain Future is a buzzer you can only stare at: get() means standing at the counter until it goes off, doing nothing else. a CompletableFuture is a buzzer you can program: "when it buzzes, put ketchup on it, then when the drinks buzzer goes too, bring both to table 4". you walk away. the instructions run when the food is ready, whoever happens to be standing there at the time.

a pipeline

each method returns a new future for the next stage, so a pipeline reads top to bottom:

import java.util.concurrent.CompletableFuture;
 
CompletableFuture<String> lookupTitle(String name) {
    return CompletableFuture.supplyAsync(() -> "dr. " + name);   // async, returns a future
}
 
void main() {
    CompletableFuture<Integer> age = CompletableFuture.supplyAsync(() -> 36);
 
    String result = CompletableFuture
            .supplyAsync(() -> "ada")                     // runs on a pool thread
            .thenApply(String::toUpperCase)               // a plain function: map
            .thenCompose(name -> lookupTitle(name))       // a function returning a future: flatMap
            .thenCombine(age, (who, a) -> who + ", " + a) // wait for both, then merge
            .join();                                      // block main until it's all done
 
    IO.println(result);
}

dr. ADA, 36

the ones you'll actually use:

a graph, step by step

real pipelines fan out and back in. here's a product page: fetch the user and the price in parallel, derive the name, chain an orders lookup, combine, recover, and wait for everything.

two things fall out of that. total time is the slowest branch, not the sum of all of them, which is the whole point. measured, with three fake calls that sleep 100, 150 and 200 ms:

the other thing: every stage you didn't make Async runs on some thread you didn't choose.

which thread runs it?

this trips everyone up at least once. a stage without Async runs on one of two threads:

• if its input is not complete yet: the thread that eventually completes the input.
• if its input is already complete: the thread that's attaching the stage, right now, inline.

import java.util.concurrent.CompletableFuture;
 
String where() { return Thread.currentThread().getName(); }
 
void main() {
    CompletableFuture<String> done = CompletableFuture.completedFuture("x");
    done.thenApply(s -> { IO.println("thenApply on a finished future: " + where()); return s; });
    done.thenApplyAsync(s -> { IO.println("thenApplyAsync: common pool? " + where().startsWith("ForkJoinPool.commonPool")); return s; }).join();
}

thenApply on a finished future: main
thenApplyAsync: common pool? true

usually that's fine. it bites when a "quick" callback is actually slow, and runs on a thread you needed for something else: a UI thread, an I/O event loop, or the common pool.

thenApply vs thenCompose

same shape as map vs flatMap on streams and optionals. if your function itself returns a future, thenApply gives you a future of a future:

CompletableFuture<CompletableFuture<List<Order>>> nested =
        user.thenApply(u -> fetchOrders(u));

CompletableFuture<List<Order>> orders =
        user.thenCompose(u -> fetchOrders(u));

when things fail

an exception thrown inside a stage doesn't crash anything. it's stored in that stage's future, and every stage that depends on it is skipped and completes exceptionally too, until something handles it:

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
 
void main() {
    CompletableFuture<Integer> price = CompletableFuture.supplyAsync(() -> {
        throw new IllegalStateException("price service down");
    });
    CompletableFuture<String> page = price.thenApply(p -> "price: " + p);  // never runs
 
    IO.println(page.exceptionally(t -> "fallback, because " + t.getCause().getMessage()).join());
    IO.println(page.handle((v, t) -> t == null ? v : "handle saw " + t.getClass().getSimpleName()).join());
    try {
        page.join();
    } catch (CompletionException e) {
        IO.println("join threw CompletionException, cause: " + e.getCause());
    }
}

fallback, because price service down
handle saw CompletionException
join threw CompletionException, cause: java.lang.IllegalStateException: price service down

notice the wrapping: the handlers see a CompletionException with your exception as its cause, so t.getCause() is the thing you threw. join() throws that CompletionException (unchecked); the old get() throws a checked ExecutionException instead, also wrapping your exception as its cause.

int slowAdd(int a, int b) {
    try { Thread.sleep(100); } catch (InterruptedException e) { }
    return a + b;
}
 
void main() {
    CompletableFuture.supplyAsync(() -> slowAdd(1, 2))
            .thenAccept(IO::println);
}

timeouts and cancellation

• orTimeout(2, SECONDS) java 9+: complete exceptionally with a TimeoutException if it isn't done in time.
• completeOnTimeout(fallback, 2, SECONDS): complete with a default value instead.
• cancel(true) completes the future with a CancellationException, but it does not interrupt the thread running the task. the work keeps going, nobody listens to its result. for real cancellation that propagates to running work, look at structured concurrency.