퓨처 결합: allOf와 anyOf

퓨처를 결합하는 이유

allOf: 모든 퓨처 대기

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();

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: 먼저 완료된 작업의 승리

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

시간 초과 대체 처리를 위한 anyOf

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());

allOf를 사용한 병렬 HTTP 호출

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();

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

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

allOf와 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());

스레드 풀 고려 사항

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

취소 전파

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

빠른 확인

복습