CompletableFuture for Async Workflows
`CompletableFuture` is Java's promise-style API for composing async work. This snippet covers `supplyAsync` to start a task, `thenApply` for synchronous transforms, `thenCompose` for chaining another async step, and `allOf` to fan out parallel tasks. Use it for any workflow that mixes I/O calls or CPU work without blocking the calling thread.
433 views
3
import java.util.concurrent.CompletableFuture;
public class Main {
public static void main(String[] args) throws Exception {
CompletableFuture<String> hello = CompletableFuture
.supplyAsync(() -> {
sleep(50);
return "hello";
})
.thenApply(s -> s + ", world")
.thenApply(String::toUpperCase);
System.out.println(hello.get()); // HELLO, WORLD
}
static void sleep(long ms) {
try { Thread.sleep(ms); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); }
}
}supplyAsync runs the supplier on the common ForkJoinPool and returns a future that completes when the supplier finishes. thenApply chains a synchronous transformation; the lambda runs on whichever thread completed the previous stage. get() blocks the caller until the chain finishes, so use it sparingly (only at the top of main or in tests). For real services, return the CompletableFuture itself and let the framework or caller decide when to wait.
import java.util.concurrent.CompletableFuture;
public class Main {
static void sleep(long ms) {
try { Thread.sleep(ms); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); }
}
static CompletableFuture<Integer> fetchUserId(String name) {
return CompletableFuture.supplyAsync(() -> {
sleep(30);
return name.length(); // pretend this is an ID lookup
});
}
static CompletableFuture<String> fetchProfile(int userId) {
return CompletableFuture.supplyAsync(() -> {
sleep(30);
return "profile#" + userId;
});
}
public static void main(String[] args) throws Exception {
CompletableFuture<String> chain = fetchUserId("ada")
.thenCompose(Main::fetchProfile);
System.out.println(chain.get()); // profile#3
}
}Use thenCompose (NOT thenApply) when the next step itself returns a CompletableFuture. With thenApply you would end up with CompletableFuture<CompletableFuture<String>> and have to call .join() to flatten it; thenCompose handles that flattening for you. The mental model is flatMap for futures, the same way you would use flatMap over Optional or Stream. Reserve thenApply for pure transforms that return a plain value.
import java.util.concurrent.CompletableFuture;
public class Main {
static void sleep(long ms) {
try { Thread.sleep(ms); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); }
}
static CompletableFuture<Integer> work(int i) {
return CompletableFuture.supplyAsync(() -> {
sleep(20);
return i * i;
});
}
public static void main(String[] args) throws Exception {
java.util.List<CompletableFuture<Integer>> futures = new java.util.ArrayList<>();
for (int i = 1; i <= 5; i++) futures.add(work(i));
CompletableFuture<Void> all = CompletableFuture.allOf(
futures.toArray(new CompletableFuture[0]));
long start = System.nanoTime();
all.get();
long ms = (System.nanoTime() - start) / 1_000_000;
java.util.List<Integer> results = new java.util.ArrayList<>();
for (CompletableFuture<Integer> f : futures) results.add(f.join());
System.out.println("results=" + results + " took~" + ms + "ms");
}
}allOf returns a future that completes when EVERY input future completes; it does not, however, give you the results directly (its type is CompletableFuture<Void>). The standard idiom is allOf(futures).thenApply(v -> futures.stream().map(CompletableFuture::join).collect(...)) or, if you have already awaited, just iterate and call join() since each future is now done. The wall-clock time is roughly max(individual durations) rather than their sum, which is the whole point of fanning out. For early-fail semantics use anyOf plus your own error checking.