Write non-blocking async pipelines with CompletableFuture, thenApply, thenCompose, and allOf.
Published February 14, 2025
CompletableFuture<T> (Java 8+) is a composable async computation. Unlike Future, it supports non-blocking callbacks, chaining, and combining multiple async operations.
// Run async, no return value
CompletableFuture<Void> cf1 = CompletableFuture.runAsync(() -> {
System.out.println("async task");
});
// Async with return value
CompletableFuture<String> cf2 = CompletableFuture.supplyAsync(() -> {
return fetchUserFromDB("u1"); // runs in ForkJoinPool.commonPool()
});
// With custom executor
ExecutorService pool = Executors.newFixedThreadPool(4);
CompletableFuture<String> cf3 = CompletableFuture.supplyAsync(
() -> fetchUser("u1"), pool
);
CompletableFuture.supplyAsync(() -> "user_123")
.thenApply(userId -> fetchUser(userId)) // transform: String → User
.thenApply(user -> user.getEmail()) // transform: User → String
.thenAccept(email -> sendEmail(email)) // consume: String → void
.thenRun(() -> log.info("Email sent")) // run after: void → void
.exceptionally(e -> {
log.error("Failed", e);
return null;
});
// thenApply wraps the result: CompletableFuture<CompletableFuture<User>>
// thenCompose flattens it: CompletableFuture<User>
CompletableFuture<User> future = CompletableFuture
.supplyAsync(() -> "user_123")
.thenCompose(userId -> fetchUserAsync(userId)); // returns CF<User>
// Wait for all to complete
CompletableFuture<String> f1 = CompletableFuture.supplyAsync(() -> fetchName());
CompletableFuture<Integer> f2 = CompletableFuture.supplyAsync(() -> fetchAge());
CompletableFuture<String> combined = f1.thenCombine(f2,
(name, age) -> name + " is " + age);
// Wait for a list of futures
List<CompletableFuture<String>> futures = userIds.stream()
.map(id -> CompletableFuture.supplyAsync(() -> fetchUser(id)))
.toList();
CompletableFuture<Void> allDone = CompletableFuture.allOf(
futures.toArray(new CompletableFuture[0])
);
// Collect results after all complete
allDone.thenApply(v -> futures.stream()
.map(CompletableFuture::join)
.toList());
// Complete when the FIRST one finishes
CompletableFuture<Object> anyDone = CompletableFuture.anyOf(
futures.toArray(new CompletableFuture[0])
);
CompletableFuture.supplyAsync(() -> riskyOperation())
.exceptionally(ex -> {
log.error("Operation failed", ex);
return defaultValue(); // recovery value
})
.handle((result, ex) -> {
// Called whether succeeded or failed
if (ex != null) return handleError(ex);
return transform(result);
});
public UserProfileDto getProfile(String userId) {
CompletableFuture<User> userFuture =
CompletableFuture.supplyAsync(() -> userService.findById(userId));
CompletableFuture<List<Order>> ordersFuture =
CompletableFuture.supplyAsync(() -> orderService.findByUser(userId));
CompletableFuture<UserStats> statsFuture =
CompletableFuture.supplyAsync(() -> statsService.getStats(userId));
// Wait for all three in parallel
return CompletableFuture.allOf(userFuture, ordersFuture, statsFuture)
.thenApply(v -> new UserProfileDto(
userFuture.join(),
ordersFuture.join(),
statsFuture.join()
))
.join(); // block for the final result
}
thenApply (synchronous transform in callback thread) and thenApplyAsync (transform in a new thread).join() is like get() but throws unchecked exceptions — preferred in streams.CompletableFuture.allOf() returns CompletableFuture<Void> — you must call join() on each individual future to get results.