invokeAll과 CompletableFuture로 결과 모으기

주문이 하나 들어오면 배송 시스템에 알리고, 회계 시스템에 기록하고, 재고를 예약해야 한다고 해 봅시다. 세 호출이 각각 0.1초 안팎 걸리고 서로의 결과를 기다릴 필요가 없다면, 하나씩 차례로 호출할 이유가 없습니다. 차례로 호출하면 세 시간을 모두 더한 만큼 걸리지만, 동시에 호출하면 가장 느린 호출 하나만큼만 기다리면 됩니다.

이렇게 독립적인 작업 여러 개를 동시에 출발시키는 것을 팬아웃(fan-out), 흩어진 결과를 다시 한곳에 모으는 것을 팬인(fan-in)이라고 합니다. 이 글에서는 팬아웃과 팬인을 구현하는 두 가지 도구를 배웁니다. ExecutorService.invokeAll은 작업 목록을 한꺼번에 제출하고 모두 끝나기를 기다립니다. CompletableFuture는 결과가 나오면 이어서 할 일을 미리 연결해 두는 방식으로, 결과를 변환하고 합치고 실패를 복구하는 단계를 사슬처럼 이어 붙입니다. 두 도구 모두 Future 취소와 시간 초과, 실패 구분하기에서 본 정상 완료·실패·취소의 구분 위에서 동작합니다.

독립된 호출을 차례로 기다리면

먼저 비교 기준이 될 순차 버전입니다. call은 외부 시스템 호출을 흉내 내는 메서드로, 주어진 시간만큼 잠든 뒤 결과 문자열을 반환합니다. 배송은 150ms, 회계는 50ms, 재고는 100ms가 걸린다고 정했습니다.

public final class SequentialOrderFanout {
    static String call(String name, long millis) throws InterruptedException {
        Thread.sleep(millis);
        return name + "-ok";
    }

    public static void main(String[] args) throws Exception {
        long start = System.nanoTime();
        System.out.println(call("shipping", 150));
        System.out.println(call("accounting", 50));
        System.out.println(call("inventory", 100));
        System.out.println("millis=" + (System.nanoTime() - start) / 1_000_000);
    }
}

shipping-ok
accounting-ok
inventory-ok
millis=311

세 호출이 차례로 실행되므로 걸린 시간은 150 + 50 + 100 = 300ms에 출력과 스레드 스케줄링 비용이 더해진 값입니다. millis는 실행할 때마다 조금씩 달라서, 이 글의 환경에서는 여러 번 실행하는 동안 310대의 값이 나왔습니다. 결과 세 줄은 코드에 적은 호출 순서 그대로 나옵니다.

동시에 실행하기 전에 확인할 것이 두 가지 있습니다. 첫째, 세 호출이 정말 서로 독립인지입니다. 재고 예약에 성공해야만 회계에 기록한다는 규칙이 있다면 두 호출에는 순서가 있으므로 동시에 보내면 안 됩니다. 둘째, 상대 시스템이 동시 요청을 얼마나 받아 줄 수 있는지입니다. 동시에 보낸다는 것은 상대에게 한꺼번에 부하를 건다는 뜻이므로, 실행기의 스레드 수로 동시 요청 수의 상한을 정해 둡니다.

invokeAll로 한꺼번에 제출하고 모두 기다리기

invokeAll은 Callable 목록을 받아 모두 제출하고, 모든 작업이 끝날 때까지 기다린 다음 Future 목록을 돌려줍니다. 다음 예제는 앞의 세 호출을 스레드 세 개짜리 풀에서 동시에 실행합니다. 작업이 끝나는 순서를 보려고 call이 끝날 때 finished: 줄을 출력하게 했습니다.

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;

public final class InvokeAllOrderTasks {
    static String call(String name, long millis) throws InterruptedException {
        Thread.sleep(millis);
        System.out.println("finished: " + name);
        return name + "-ok";
    }

    public static void main(String[] args) throws Exception {
        var executor = Executors.newFixedThreadPool(3);
        try {
            List<Callable<String>> tasks = List.of(
                    () -> call("shipping", 150),
                    () -> call("accounting", 50),
                    () -> call("inventory", 100));
            long start = System.nanoTime();
            var futures = executor.invokeAll(tasks);
            for (var future : futures) {
                System.out.println(future.get());
            }
            System.out.println("millis=" + (System.nanoTime() - start) / 1_000_000);
        } finally {
            executor.shutdown();
        }
    }
}

finished: accounting
finished: inventory
finished: shipping
shipping-ok
accounting-ok
inventory-ok
millis=153

처음 세 줄은 작업자 스레드가 작업을 마칠 때 출력한 것이라, 가장 짧은 회계(50ms)부터 재고(100ms), 배송(150ms) 순으로 나왔습니다. 그런데 다음 세 줄, 즉 main이 Future 목록을 차례로 읽은 결과는 입력 목록의 순서인 배송, 회계, 재고 순입니다. invokeAll은 작업이 끝난 순서와 상관없이 언제나 입력 목록과 같은 순서로 Future를 돌려주기 때문입니다. 덕분에 결과 목록의 몇 번째가 어느 작업인지 따로 기록하지 않아도 됩니다.

걸린 시간은 153ms로, 순차 버전의 절반쯤입니다. 세 작업이 동시에 시작해서 가장 느린 배송이 끝나는 150ms에 모두 끝났기 때문이고, 여러 번 실행해도 150을 조금 넘는 값이 나왔습니다. 이것은 스레드가 작업 수만큼 있어서 가능한 일입니다. 스레드가 둘뿐이라면 세 번째 작업은 앞의 작업 하나가 끝날 때까지 기다려야 하므로 시간이 더 걸립니다. 연습 문제 3에서 이 경우를 계산해 봅니다.

invokeAll이 돌려준 Future는 모두 끝난 상태이므로 get()은 기다리지 않고 바로 반응합니다. 다만 끝났다는 것이 성공했다는 뜻은 아니어서, 작업이 예외를 던졌다면 get()은 ExecutionException을 던집니다. 이 예제의 작업은 실패하지 않으므로 그 처리를 생략했지만, 실제 코드에서는 다음 예제처럼 Future마다 결과를 따로 처리해야 합니다. 그래야 한 작업의 실패 때문에 나머지 결과까지 버리지 않습니다.

시간 제한이 있는 invokeAll

invokeAll(작업 목록, 시간, 단위)는 모든 작업이 끝나거나 제한 시간이 지나면 돌아옵니다. 제한 시간이 지나 돌아올 때 아직 끝나지 않은 작업은 취소되므로, 이때도 돌려받은 Future는 모두 끝난 상태입니다. 다음 예제는 회계 호출만 500ms가 걸리게 하고, 전체를 200ms로 제한합니다.

import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.CancellationException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public final class InvokeAllWithTimeout {
    static String call(String name, long millis) throws InterruptedException {
        Thread.sleep(millis);
        return name + "-ok";
    }

    public static void main(String[] args) throws Exception {
        var executor = Executors.newFixedThreadPool(3);
        try {
            List<Callable<String>> tasks = List.of(
                    () -> call("shipping", 50),
                    () -> call("accounting", 500),
                    () -> call("inventory", 50));
            var futures = executor.invokeAll(tasks, 200, TimeUnit.MILLISECONDS);
            for (var future : futures) {
                try {
                    System.out.println(future.get());
                } catch (CancellationException e) {
                    System.out.println("cancelled");
                } catch (ExecutionException e) {
                    System.out.println("failed: " + e.getCause());
                }
            }
        } finally {
            executor.shutdown();
        }
    }
}

shipping-ok
cancelled
inventory-ok

배송과 재고는 50ms 만에 끝나 결과를 냈고, 200ms가 지나도 끝나지 않은 회계는 취소돼 get()이 CancellationException을 던졌습니다. 이때의 취소는 실행 중인 작업에 인터럽트를 거는 취소라서, 회계 작업의 sleep도 그 자리에서 InterruptedException으로 깨어납니다. 이처럼 Future마다 정상 값, CancellationException, ExecutionException을 구분해 처리하면, 일부 작업이 늦거나 실패해도 나머지 결과는 쓸 수 있습니다.

invokeAny와 CompletionService

invokeAll 말고도 여러 작업의 결과를 모으는 방법이 있습니다. invokeAny는 작업 목록을 받아 그중 성공한 작업 하나의 결과만 돌려주고, 돌아올 때 아직 끝나지 않은 나머지 작업을 취소합니다. 같은 데이터를 가진 서버 여러 대에 동시에 물어보고 가장 먼저 성공한 응답만 쓰는 경우에 맞습니다. 결과를 끝난 순서대로 하나씩 처리하고 싶다면 ExecutorCompletionService를 씁니다. 작업이 끝나는 대로 그 Future를 내부 큐에 넣어 주므로, take()나 poll()로 먼저 끝난 결과부터 꺼낼 수 있습니다.

CompletableFuture로 다음 단계 이어 붙이기

invokeAll은 '모두 끝날 때까지 기다린다'는 한 가지 일만 합니다. 그런데 실제로는 결과를 받은 뒤에 할 일이 더 있는 경우가 많습니다. 가격과 할인 금액을 각각 조회한 다음 둘을 빼고, 거기에 세금을 붙이는 식입니다. Future로 이런 흐름을 만들려면 get()으로 기다렸다가 다음 계산을 하고, 그다음 작업을 또 제출해야 합니다. CompletableFuture는 결과가 나오면 할 일을 미리 연결해 두는 방식으로, 이 흐름을 중간에 기다리지 않고 표현합니다.

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;

public final class CompletableBasics {
    public static void main(String[] args) {
        var executor = Executors.newFixedThreadPool(2);
        try {
            CompletableFuture<Integer> price =
                    CompletableFuture.supplyAsync(() -> 12_000, executor);
            CompletableFuture<Integer> coupon =
                    CompletableFuture.supplyAsync(() -> 2_000, executor);
            CompletableFuture<Integer> total = price
                    .thenCombine(coupon, (p, c) -> p - c)
                    .thenApply(amount -> amount + amount / 10);
            System.out.println("total=" + total.join());
        } finally {
            executor.shutdown();
        }
    }
}

total=11000

supplyAsync(공급자, 실행기)는 공급자 람다를 실행기에서 실행하도록 맡기고, 결과를 담을 CompletableFuture를 즉시 돌려줍니다. 그래서 가격 조회와 할인 조회는 풀의 두 스레드에서 동시에 실행될 수 있습니다. thenCombine은 두 Future가 모두 끝나면 두 결과로 함수를 호출하는 새 Future를 만들고, thenApply는 앞 단계의 결과를 변환하는 새 Future를 만듭니다. 계산은 12,000 − 2,000 = 10,000에 그 10%를 더한 11,000입니다. 마지막의 join()은 get()처럼 결과를 기다리는데, 실패하면 검사 예외인 ExecutionException 대신 비검사 예외인 CompletionException을 던집니다.

supplyAsync에 실행기를 넘긴 데는 이유가 있습니다. 실행기를 넘기지 않으면 작업은 ForkJoinPool.commonPool(), 즉 JVM 전체가 함께 쓰는 공용 풀에서 실행됩니다. 공용 풀은 CPU 계산을 위해 코어 수에 맞춰 작게 만들어진 풀이고, 병렬 스트림 같은 다른 코드도 이 풀을 씁니다. 여기에 외부 호출처럼 오래 기다리는 작업을 넣으면 공용 풀의 스레드가 기다리느라 묶여서, 관계없는 다른 코드까지 느려집니다. 그래서 기다림이 섞인 작업에는 이 예제처럼 전용 실행기를 만들어 넘기고, 다 쓰면 직접 종료합니다.

thenApply나 thenCombine처럼 이름이 Async로 끝나지 않는 메서드에 넘긴 함수는 어느 스레드에서 실행될지 정해져 있지 않습니다. CompletableFuture 문서에 따르면 앞 단계를 완료시킨 스레드가 실행할 수도 있고, 완료 메서드를 호출한 다른 스레드가 실행할 수도 있습니다. 예를 들어 앞 단계가 이미 끝난 뒤에 thenApply를 호출하면, 호출한 스레드(main일 수도 있습니다)가 그 자리에서 함수를 실행합니다. 가벼운 변환이라면 상관없지만, 무거운 계산이나 기다림이 들어간 단계라면 thenApplyAsync(함수, 실행기)처럼 실행기를 받는 Async 메서드를 씁니다.

실패를 결과 값으로 바꾸는 handle

이제 주문 예제로 돌아갑니다. 이번에는 회계 호출이 IllegalStateException을 던져 실패하도록 만들고, 그 실패가 다른 결과까지 망가뜨리지 않도록 각 호출의 결과를 Result 레코드로 바꿉니다. Result에는 시스템 이름, 성공 여부, 그리고 성공했을 때의 값이나 실패했을 때의 설명을 담습니다.

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.function.Supplier;

public final class CompletableOrderFanout {
    record Result(String system, boolean success, String detail) {}

    static CompletableFuture<Result> call(String system, Supplier<String> action,
                                          ExecutorService executor) {
        return CompletableFuture.supplyAsync(action, executor)
                .handle((value, error) -> error == null
                        ? new Result(system, true, value)
                        : new Result(system, false, error.getClass().getSimpleName()));
    }

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        try {
            var futures = List.of(
                    call("shipping", () -> "notified", executor),
                    call("accounting", () -> {
                        throw new IllegalStateException("closed");
                    }, executor),
                    call("inventory", () -> "reserved", executor));
            CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();
            System.out.println(futures.stream().map(CompletableFuture::join).toList());
        } finally {
            executor.shutdown();
        }
    }
}

call의 핵심은 handle입니다. handle에 넘긴 함수는 앞 단계가 정상 완료하든 예외 완료하든 호출되고, 정상이면 value에 결과가, 실패면 error에 예외가 들어옵니다. 이 함수가 어느 쪽이든 Result를 반환하므로, handle이 만든 Future는 언제나 정상 완료합니다. main은 세 Future를 allOf로 묶어 모두 끝나기를 기다립니다. allOf는 배열을 받으므로 toArray로 목록을 배열로 바꿔 넘기고, 마지막 줄에서는 스트림으로 각 Future의 join() 결과를 목록에 모아 출력합니다.

[Result[system=shipping, success=true, detail=notified], Result[system=accounting, success=false, detail=CompletionException], Result[system=inventory, success=true, detail=reserved]]

회계 호출은 실패했지만 출력에는 Result 세 개가 모두 있습니다. handle이 예외를 success=false인 값으로 바꿨으므로 회계의 Future도 정상 완료했고, 덕분에 allOf와 join()이 예외 없이 끝났습니다. 대신 이제는 Future가 정상 완료했다는 것이 업무가 성공했다는 뜻이 아닙니다. 결과를 이어서 쓰는 코드는 반드시 success를 확인해야 합니다. 출력 순서는 futures 목록의 순서이므로, 작업이 끝난 순서와는 관계가 없습니다.

그런데 회계의 detail이 이상합니다. 작업이 던진 것은 IllegalStateException인데 CompletionException이 기록됐고, 메시지 closed도 사라졌습니다. supplyAsync의 공급자가 던진 예외는 CompletionException으로 감싸여 저장되고, 다음 단계인 handle에는 이 감싼 예외가 전달되기 때문입니다. 원래 예외는 감싼 예외의 getCause()로 꺼낼 수 있습니다. 이 코드는 감싼 예외의 클래스 이름만 남겼으므로, 출력만 봐서는 실패 원인을 알 수 없습니다.

원인을 보존하려면 call을 다음처럼 고칩니다. 예외가 CompletionException이면 원인을 꺼내고, 클래스 이름만이 아니라 예외 클래스와 메시지를 함께 남깁니다. 파일 위쪽에 import java.util.concurrent.CompletionException;도 추가합니다.

    static CompletableFuture<Result> call(String system, Supplier<String> action,
                                          ExecutorService executor) {
        return CompletableFuture.supplyAsync(action, executor)
                .handle((value, error) -> {
                    if (error == null) {
                        return new Result(system, true, value);
                    }
                    Throwable cause = error instanceof CompletionException && error.getCause() != null
                            ? error.getCause()
                            : error;
                    return new Result(system, false, cause.toString());
                });
    }

[Result[system=shipping, success=true, detail=notified], Result[system=accounting, success=false, detail=java.lang.IllegalStateException: closed], Result[system=inventory, success=true, detail=reserved]]

이제 회계의 detail에 원래 예외와 메시지가 남습니다. 실제 서비스라면 문자열 대신 예외 객체나 실패 종류를 나타내는 필드를 Result에 두고, 호출에 걸린 시간도 함께 기록하면 장애를 분석하기가 쉬워집니다.

allOf가 기다리는 것과 기다리지 않는 것

allOf는 넘겨받은 Future가 모두 끝나면 완료되는 새 Future를 만듭니다. 반환 타입이 CompletableFuture<Void>라서 정상 완료해도 join()의 결과는 null이고, 실제 값은 앞 예제처럼 각 Future에서 따로 읽어야 합니다.

handle 없이 실패한 Future를 allOf에 넣으면 어떻게 될까요? allOf는 하나가 실패해도 바로 끝나지 않고, 나머지가 모두 끝날 때까지 기다린 다음 예외 완료합니다. 이때 join()은 실패 원인을 담은 CompletionException을 던집니다. 또 allOf는 실패를 알게 돼도 다른 Future를 취소하지 않으므로, 나머지 작업은 끝까지 실행되고 그 결과도 각 Future에 그대로 남습니다. 하나가 실패하면 나머지를 바로 멈추는 정책이 필요하다면 그 코드를 직접 작성해야 합니다. 연습 문제 2에서 이 동작을 따라가 봅니다.

thenApply와 thenCompose

다음 단계의 함수가 보통 값을 반환하면 thenApply를 쓰고, 또 다른 CompletableFuture를 반환하면 thenCompose를 씁니다. 사용자 ID를 비동기로 조회한 다음, 그 ID로 주문 목록을 다시 비동기로 조회하는 경우가 thenCompose에 해당합니다. 이런 함수에 thenApply를 쓰면 결과가 Future 안에 Future가 든 두 겹이 됩니다.

CompletableFuture<Integer> start = CompletableFuture.completedFuture(2);
CompletableFuture<CompletableFuture<Integer>> nested =
        start.thenApply(x -> CompletableFuture.completedFuture(x * 10));
CompletableFuture<Integer> flat =
        start.thenCompose(x -> CompletableFuture.completedFuture(x * 10));
System.out.println(nested.join().join() + " " + flat.join()); // 20 20

두 방법 모두 최종 값은 20이지만, nested는 join()을 두 번 불러야 값에 닿습니다. thenCompose는 안쪽 Future가 끝날 때 완료되는 하나의 Future로 펼쳐 주므로, 이어지는 단계도 계속 한 겹으로 연결할 수 있습니다. 다만 thenCompose는 계산을 병렬로 만드는 옵션이 아닙니다. 앞 결과를 받아야 시작할 수 있는 다음 비동기 작업을 사슬에 잇는 도구입니다.

시간 제한과 취소는 작업을 멈추지 않는다

CompletableFuture에는 Java 9부터 시간 제한 메서드가 있습니다. orTimeout(시간, 단위)은 그 시간 안에 끝나지 않으면 TimeoutException으로 예외 완료시키고, completeOnTimeout(값, 시간, 단위)은 대신 정해 둔 값으로 정상 완료시킵니다. 다음 예제는 500ms가 걸리는 호출에 두 메서드로 100ms 제한을 겁니다. CompletableTimeout::slowCall은 () -> slowCall()과 같은 뜻의 메서드 참조입니다.

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public final class CompletableTimeout {
    static String slowCall() {
        try {
            Thread.sleep(500);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
        System.out.println("slowCall finished");
        return "slow-ok";
    }

    public static void main(String[] args) {
        var executor = Executors.newFixedThreadPool(2);
        try {
            String value = CompletableFuture.supplyAsync(CompletableTimeout::slowCall, executor)
                    .completeOnTimeout("fallback", 100, TimeUnit.MILLISECONDS)
                    .join();
            System.out.println("value=" + value);
            try {
                CompletableFuture.supplyAsync(CompletableTimeout::slowCall, executor)
                        .orTimeout(100, TimeUnit.MILLISECONDS)
                        .join();
            } catch (CompletionException e) {
                System.out.println("failed: " + e.getCause());
            }
        } finally {
            executor.shutdown();
        }
    }
}

value=fallback
failed: java.util.concurrent.TimeoutException
slowCall finished
slowCall finished

첫 호출은 100ms 안에 끝나지 않아 fallback으로 완료됐고, 두 번째 호출은 TimeoutException으로 예외 완료돼 join()이 그 예외를 원인으로 담은 CompletionException을 던졌습니다. 그런데 마지막 두 줄을 보면 slowCall은 두 번 모두 끝까지 실행됐습니다. 시간 제한은 CompletableFuture의 결과를 먼저 정해 버릴 뿐, 공급자를 실행하는 스레드에는 아무것도 알리지 않기 때문입니다. 실행기의 작업자는 데몬 스레드가 아니므로, 프로세스도 두 slowCall이 끝난 뒤에야 종료됩니다.

cancel도 마찬가지입니다. CompletableFuture의 cancel(true)는 Future를 CancellationException으로 완료시킬 뿐, 인자가 true여도 작업자 스레드를 인터럽트하지 않습니다. 문서에도 이 구현은 인터럽트로 처리를 제어하지 않으므로 이 인자는 효과가 없다고 적혀 있습니다. 다음 예제는 300ms 동안 잠드는 공급자를 50ms 뒤에 취소합니다.

import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public final class CompletableCancelDemo {
    public static void main(String[] args) throws Exception {
        var executor = Executors.newSingleThreadExecutor();
        var future = CompletableFuture.supplyAsync(() -> {
            try {
                Thread.sleep(300);
                System.out.println("supplier finished");
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.out.println("supplier interrupted");
            }
            return "value";
        }, executor);
        Thread.sleep(50);
        System.out.println("cancel=" + future.cancel(true));
        try {
            future.join();
        } catch (CancellationException e) {
            System.out.println("join: CancellationException");
        }
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.SECONDS);
    }
}

cancel=true
join: CancellationException
supplier finished

취소는 성공했고 join()은 곧바로 CancellationException을 던졌지만, 공급자는 인터럽트를 받지 않고 300ms를 다 잔 뒤 supplier finished를 출력했습니다. 같은 람다를 ExecutorService.submit으로 제출하고 그 Future를 cancel(true)로 취소했다면, 인터럽트가 sleep을 깨워 supplier interrupted가 나왔을 것입니다. 이처럼 CompletableFuture에서는 시간 제한과 취소가 결과만 정할 뿐 실행 중인 작업을 멈추지 않습니다. 외부 요청을 실제로 중단해야 한다면 작업 쪽에서 확인할 수 있는 취소 신호를 따로 두거나, 요청을 보내는 클라이언트가 제공하는 시간 제한을 함께 설정해야 합니다.

여러 단계에 시간 제한을 걸 때는 전체 마감 시각을 먼저 정하는 것이 좋습니다. 단계마다 100ms씩 제한하면 세 단계짜리 흐름은 최대 300ms까지 걸릴 수 있기 때문입니다. 요청을 받은 시각에 마감 시각을 한 번 계산하고 각 단계에는 남은 시간만 넘기면, 전체가 정한 시간을 크게 넘지 않습니다. Future.get을 부르는 시점과 병렬 합산의 collect가 같은 방식이었습니다. 물론 남은 시간을 지켜도 스케줄링이나 정리에 드는 시간 때문에 실제 응답은 마감보다 조금 늦을 수 있습니다.

부분 실패를 어떻게 다룰지 먼저 정하기

팬아웃 코드를 쓰기 전에 정해야 할 것은 도구보다 정책입니다. 세 호출 중 하나가 실패했을 때 주문 전체를 실패로 볼지, 나머지 결과로 처리를 이어 갈지는 업무가 결정할 문제입니다. 전체를 실패로 봐야 한다면 handle로 예외를 삼키지 말고 그대로 전파해 allOf의 join()이 실패하게 둡니다. 부분 성공을 허용한다면 앞의 Result처럼 성공과 실패를 같은 결과 타입에 담습니다.

어느 쪽이든 피해야 할 것은 실패를 '데이터 없음'으로 바꿔 버리는 것입니다. 실패한 호출의 결과를 빈 목록이나 0으로 바꾸면, 결과를 받은 쪽은 상대 시스템에 장애가 난 것인지 정말 데이터가 없는 것인지 구별할 수 없습니다. 실패한 시스템 이름과 실패 종류를 결과에 남겨 두면, 재시도할지 사용자에게 일부 정보만 보여 줄지를 나중에 고를 수 있습니다. 팬아웃을 설계할 때 확인할 점을 정리하면 다음과 같습니다.

• 작업들이 서로의 결과를 기다리지 않는 독립된 단위인가
• 상대 시스템별 동시 요청 한도에 맞게 실행기 크기를 정했는가
• 전체 마감 시각을 정하고 각 단계에 남은 시간만 넘기는가
• 일부 실패를 허용한다면 성공과 실패를 같은 결과 타입에 담는가
• handle이나 exceptionally에서 실패 원인을 버리지 않는가
• 모든 작업이 끝난 뒤 전용 실행기를 종료하는가

정리

• 서로 독립적인 호출을 동시에 실행하면 전체 시간이 합계가 아니라 가장 느린 호출 정도로 줄어듭니다. 단, 독립성과 상대 시스템의 동시 요청 한도를 먼저 확인합니다.
• invokeAll은 모든 작업이 끝날 때까지 기다리고 입력 순서대로 Future를 돌려줍니다. 시간 제한 버전은 끝나지 않은 작업을 취소합니다. 돌려받은 Future마다 정상 값, 실패, 취소를 따로 처리합니다.
• CompletableFuture는 thenApply, thenCombine, thenCompose로 다음 단계를 연결합니다. 기다림이 섞인 작업은 공용 풀 대신 전용 실행기에서 실행합니다.
• handle로 예외를 값으로 바꾸면 Future는 정상 완료하므로 업무 성공 여부를 따로 확인해야 합니다. supplyAsync가 실행한 작업의 예외는 CompletionException으로 감싸여 전달되므로 getCause()로 원인을 꺼냅니다.
• allOf는 결과를 모아 주지 않고, 실패한 작업이 있어도 다른 작업을 취소하지 않습니다. CompletableFuture의 시간 제한과 cancel은 결과만 정할 뿐 실행 중인 작업을 멈추지 않습니다.

연습 문제

1. 두 공급자 중 최저가 고르기

두 공급자에게 가격을 묻는 작업을 전용 실행기에 동시에 제출하세요. 둘 다 성공하면 더 낮은 가격을, 한쪽만 성공하면 그 가격을 쓰고, 둘 다 실패하면 예외를 던집니다.

import java.util.OptionalInt;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.function.IntSupplier;

public final class BestPriceSolution {
    static OptionalInt safe(IntSupplier supplier) {
        try {
            return OptionalInt.of(supplier.getAsInt());
        } catch (RuntimeException e) {
            return OptionalInt.empty();
        }
    }

    public static void main(String[] args) {
        var executor = Executors.newFixedThreadPool(2);
        try {
            var a = CompletableFuture.supplyAsync(() -> safe(() -> 120), executor);
            var b = CompletableFuture.supplyAsync(() -> safe(() -> {
                throw new IllegalStateException("provider down");
            }), executor);
            int best = a.thenCombine(b, (left, right) -> {
                if (left.isPresent() && right.isPresent()) {
                    return Math.min(left.getAsInt(), right.getAsInt());
                }
                if (left.isPresent()) return left.getAsInt();
                if (right.isPresent()) return right.getAsInt();
                throw new IllegalStateException("all providers failed");
            }).join();
            System.out.println("best=" + best);
        } finally {
            executor.shutdown();
        }
    }
}

best=120

2. 실패한 Future가 섞인 allOf

CompletableFuture<Integer> A는 이미 IllegalStateException("closed")로 예외 완료했고, B는 아직 끝나지 않았습니다. handle 없이 allOf(A, B)를 만든 뒤 나중에 B를 90으로 완료합니다. B가 완료되기 전과 후에 allOf가 만든 Future는 각각 어떤 상태이고, 그 Future의 join()은 어떻게 될까요? B의 값 90은 남아 있을까요?

3. 작업자가 둘뿐일 때 걸리는 시간

서로 독립인 작업 A, B, C가 각각 정확히 120ms, 80ms, 60ms 걸리고, 작업자 스레드는 둘입니다. 시각 0에 A, B, C 순서로 한꺼번에 제출하면 두 작업자가 A와 B를 바로 시작하고, C는 먼저 일이 끝난 작업자가 이어서 실행합니다. 제출과 스케줄링 비용은 없다고 할 때, 모든 작업이 끝나는 시각을 구하고 차례로 실행할 때와 비교하세요.

다음 글

ExecutorService를 안전하게 종료하기에서는 이 글의 예제들이 finally에서 호출한 shutdown이 정확히 무엇을 하는지 살펴봅니다. 실행 중인 작업과 큐에 남은 작업을 정리하면서 실행기를 끝내는 방법도 배웁니다.