Łączenie future: allOf i anyOf

Dlaczego łączyć future?

allOf: Oczekiwanie na wszystkie future

CompletableFuture<String> f1 = fetchUserAsync();
CompletableFuture<List<Order>> f2 = fetchOrdersAsync();
CompletableFuture<Double> f3 = fetchBalanceAsync();
CompletableFuture<Void> all = CompletableFuture.allOf(f1, f2, f3);
all.join(); // wait for all three
String user   = f1.join();
List<Order> orders = f2.join();
Double balance = f3.join();

Zbieranie wyników allOf

List<CompletableFuture<String>> futures = List.of(
    fetchAsync("url1"), fetchAsync("url2"), fetchAsync("url3"));
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0]))
    .thenApply(v -> futures.stream()
        .map(CompletableFuture::join)
        .collect(Collectors.toList()))
    .thenAccept(results -> results.forEach(System.out::println))
    .join();

anyOf: Wygrywa pierwszy ukończony

CompletableFuture<Object> first = CompletableFuture.anyOf(
    fetchFromCacheAsync(),
    fetchFromDbAsync(),
    fetchFromApiAsync()
);
System.out.println("Fastest result: " + first.join());

anyOf jako rozwiązanie awaryjne dla limitu czasu

CompletableFuture<String> real    = fetchDataAsync();
CompletableFuture<String> timeout = CompletableFuture
    .supplyAsync(() -> {
        try { Thread.sleep(2000); } catch (InterruptedException e) {}
        return "timeout-default";
    });
String result = (String) CompletableFuture.anyOf(real, timeout).join();

orTimeout (Java 9+)

CompletableFuture<String> result = fetchDataAsync()
    .orTimeout(3, TimeUnit.SECONDS)
    .exceptionally(ex -> "timeout-default");
System.out.println(result.join());

Równoległe wywołania HTTP za pomocą allOf

List<String> urls = List.of(url1, url2, url3);
List<CompletableFuture<String>> futures = urls.stream()
    .map(url -> CompletableFuture.supplyAsync(() -> httpGet(url)))
    .collect(Collectors.toList());
List<String> responses = CompletableFuture
    .allOf(futures.toArray(new CompletableFuture[0]))
    .thenApply(v -> futures.stream().map(CompletableFuture::join).collect(Collectors.toList()))
    .join();

Obsługa błędów w allOf

List<CompletableFuture<String>> safe = futures.stream()
    .map(f -> f.exceptionally(ex -> "ERROR: " + ex.getMessage()))
    .collect(Collectors.toList());
CompletableFuture.allOf(safe.toArray(new CompletableFuture[0])).join();

allOf z różnymi typami

CompletableFuture<User> userFuture   = fetchUser(id);
CompletableFuture<Config> configFuture = fetchConfig();
CompletableFuture.allOf(userFuture, configFuture).join();
User   user   = userFuture.join();
Config config = configFuture.join();

Łączenie allOf i thenApply

CompletableFuture<String> summary =
    CompletableFuture.allOf(f1, f2, f3)
        .thenApply(v -> "User: " + f1.join() +
                       ", Orders: " + f2.join().size() +
                       ", Balance: " + f3.join());
System.out.println(summary.join());

Kwestie dotyczące puli wątków

// With virtual threads (Java 21+):
ExecutorService vt = Executors.newVirtualThreadPerTaskExecutor();
CompletableFuture<String> f = CompletableFuture.supplyAsync(() -> httpGet(url), vt);

Propagowanie anulowania

CompletableFuture<Object> first = CompletableFuture.anyOf(f1, f2, f3);
first.thenRun(() -> Stream.of(f1, f2, f3).forEach(f -> f.cancel(true)));

Szybkie sprawdzenie

Podsumowanie