CompletableFuture: 비동기 실행과 결과 조합

1. 개요

API는 보통 여러 서비스를 호출합니다: 고객 정보 조회, 주문 정보 조회, 배송료 조회. 순차적으로 호출하면 총 소요 시간은 각 호출의 지연 시간의 합이 됩니다. CompletableFuture는 독립적인 호출들을 병렬로 실행한 후, 모두 완료되었을 때 결과를 조합할 수 있게 해줍니다.

이 글을 읽고 나면 비동기 작업을 만드는 방법, 처리 단계를 연결하는 방법(thenApply, thenCompose), 결과를 조합하는 방법(thenCombine, allOf, anyOf)을 알게 될 것입니다. 또한 오류를 처리하고, 타임아웃을 설정하고, 비동기 코드를 테스트하는 방법도 배우게 됩니다.

2. 예제 준비

우리는 세 개의 모의 서비스를 사용할 것입니다. 각 서비스는 네트워크 지연을 시뮬레이션하기 위해 일시적으로 “대기”합니다:

import java.math.BigDecimal;
import java.util.List;

record Customer(long id, String name, String city) {}
record Order(long id, long customerId, BigDecimal total) {}

final class Delay {
    static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new IllegalStateException(e);
        }
    }
}

class CustomerService {
    Customer findById(long id) {
        Delay.sleep(300);
        return new Customer(id, "Lan", "Hue");
    }
}

class OrderService {
    List<Order> findByCustomer(long customerId) {
        Delay.sleep(400);
        return List.of(
            new Order(1, customerId, new BigDecimal("150000")),
            new Order(2, customerId, new BigDecimal("90000")));
    }
}

class ShippingService {
    BigDecimal feeFor(String city) {
        Delay.sleep(200);
        return "Hue".equals(city) ? new BigDecimal("30000") : new BigDecimal("45000");
    }
}

이 세 서비스는 JDK만 사용합니다. 이를 실행하기 위해 추가 라이브러리가 필요하지 않습니다.

3. 비동기 작업 생성

3.1. supplyAsync 및 runAsync

supplyAsync는 Supplier를 실행하고 CompletableFuture<T>를 반환합니다. runAsync는 결과가 없는 Runnable을 실행하고 CompletableFuture<Void>를 반환합니다:

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

public class CreateDemo {
    public static void main(String[] args) {
        var customerService = new CustomerService();

        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            CompletableFuture<Customer> customerFuture =
                CompletableFuture.supplyAsync(() -> customerService.findById(42L), executor);

            CompletableFuture<Void> auditFuture =
                CompletableFuture.runAsync(() -> System.out.println("접근 로그 기록"), executor);

            System.out.println("Main thread는 블로킹되지 않음");
            System.out.println(customerFuture.join());
            auditFuture.join();
        }
    }
}

Main thread는 블로킹되지 않음
접근 로그 기록
Customer[id=42, name=Lan, city=Hue]

처음 두 줄의 순서는 바뀔 수 있습니다. 서로 다른 스레드에서 실행되기 때문입니다. 다음 사항들에 주의해야 합니다:

• 두 번째 매개변수는 Executor입니다. 생략하면 작업은 ForkJoinPool.commonPool()에서 실행됩니다.
• Executors.newVirtualThreadPerTaskExecutor()(Java 21)는 각 작업마다 virtual thread를 생성합니다. 이 선택은 HTTP 호출이나 데이터베이스 쿼리 같은 I/O 작업에 적합합니다.
• ExecutorService는 Java 19부터 AutoCloseable입니다. try-with-resources 블록은 작업이 완료된 후에 executor를 닫습니다.
• join()은 결과를 얻을 때까지 현재 스레드를 블로킹합니다. 작업이 실패하면 join()은 CompletionException(미검사 예외)을 던집니다. 반면 get()은 ExecutionException(검사 예외)을 던집니다.

3.2. future를 수동으로 완료하기

때로는 결과가 lambda가 아닌 콜백에서 옵니다. 이 경우 빈 future를 만들고 수동으로 완료합니다:

CompletableFuture<String> paymentResult = new CompletableFuture<>();

// 어딘가에서, 예를 들어 결제 게이트웨이의 콜백에서:
paymentResult.complete("PAID");

System.out.println(paymentResult.join()); // PAID

오류를 보고하려면 completeExceptionally(new RuntimeException("카드가 거부됨"))을 호출합니다. 값이 이미 있으면 CompletableFuture.completedFuture(value)가 이미 완료된 future를 만드는 가장 간단한 방법입니다.

4. 처리 단계 연결

4.1. thenApply 및 thenAccept

thenApply는 결과를 변환하며, Stream의 map과 유사합니다. thenAccept는 결과를 소비하고 아무것도 반환하지 않습니다:

CompletableFuture<String> greeting = CompletableFuture
    .supplyAsync(() -> customerService.findById(42L), executor)
    .thenApply(Customer::name)
    .thenApply(name -> "안녕하세요, " + name);

greeting.thenAccept(System.out::println).join();

안녕하세요, Lan

각 thenApply 호출은 새로운 CompletableFuture를 반환합니다. 원래 future는 변경되지 않습니다.

4.2. 다음 단계도 비동기일 때 thenCompose

주문을 조회하려면 이전 단계의 customerId가 필요합니다. thenApply를 future를 반환하는 함수와 함께 사용하면 결과는 CompletableFuture<CompletableFuture<List<Order>>>가 됩니다. thenCompose는 이 중첩된 타입을 “평탄화”하며, flatMap과 유사합니다:

CompletableFuture<List<Order>> ordersFuture = CompletableFuture
    .supplyAsync(() -> customerService.findById(42L), executor)
    .thenCompose(customer -> CompletableFuture.supplyAsync(
        () -> orderService.findByCustomer(customer.id()), executor));

System.out.println(ordersFuture.join().size()); // 2

이 두 단계는 서로 의존하므로 여전히 순차적으로 실행됩니다. 총 소요 시간은 약 700ms입니다. thenCompose가 더 빠르게 만드는 것은 아닙니다. 이것은 스레드를 블로킹하지 않고 단계를 연결하는 데 도움이 됩니다.

5. 결과 조합

독립적인 작업들과 함께 CompletableFuture가 진정한 힘을 발휘합니다.

5.1. 두 개의 future를 위한 thenCombine

고객의 도시를 알고 있고 동시에 주문 목록과 배송료를 가져오고 싶다고 가정해봅시다:

long start = System.currentTimeMillis();

CompletableFuture<List<Order>> orders =
    CompletableFuture.supplyAsync(() -> orderService.findByCustomer(42L), executor);
CompletableFuture<BigDecimal> shippingFee =
    CompletableFuture.supplyAsync(() -> shippingService.feeFor("Hue"), executor);

CompletableFuture<BigDecimal> grandTotal = orders.thenCombine(shippingFee,
    (orderList, fee) -> orderList.stream()
        .map(Order::total)
        .reduce(BigDecimal.ZERO, BigDecimal::add)
        .add(fee));

System.out.println(grandTotal.join());
System.out.println("소요 시간: " + (System.currentTimeMillis() - start) + " ms");

270000
소요 시간: 405 ms

두 호출이 병렬로 실행됩니다. 총 소요 시간은 가장 느린 호출(400ms)에 근접하며, 순차적으로 호출했을 경우의 600ms가 아닙니다. thenCombine의 두 번째 매개변수에 있는 함수는 두 future가 모두 완료되었을 때만 실행됩니다.

5.2. 여러 개의 future를 위한 allOf

두 개 이상의 작업이 있을 때 allOf를 사용합니다. 이것은 CompletableFuture<Void>를 반환하므로 각 하위 future에서 결과를 수동으로 추출해야 합니다:

List<String> cities = List.of("Hue", "Hanoi", "Danang");

List<CompletableFuture<BigDecimal>> fees = cities.stream()
    .map(city -> CompletableFuture.supplyAsync(() -> shippingService.feeFor(city), executor))
    .toList();

CompletableFuture<List<BigDecimal>> allFees = CompletableFuture
    .allOf(fees.toArray(CompletableFuture[]::new))
    .thenApply(ignored -> fees.stream()
        .map(CompletableFuture::join)
        .toList());

System.out.println(allFees.join());

[30000, 45000, 45000]

thenApply 내에서 모든 하위 future가 완료되었습니다. 따라서 join()은 블로킹되지 않고 즉시 반환됩니다. 결과의 순서는 cities의 순서를 유지합니다.

5.3. 가장 빠른 결과를 얻는 anyOf

anyOf는 하나의 하위 future가 완료되면 즉시 완료됩니다. 이는 동일한 서비스의 여러 복제본을 호출할 때 유용합니다:

CompletableFuture<Object> fastest = CompletableFuture.anyOf(
    CompletableFuture.supplyAsync(() -> { Delay.sleep(500); return "replica-A"; }, executor),
    CompletableFuture.supplyAsync(() -> { Delay.sleep(100); return "replica-B"; }, executor));

System.out.println(fastest.join()); // replica-B

반환 타입은 CompletableFuture<Object>이므로 수동으로 캐스트해야 합니다. 추가 참고: 가장 빠른 future가 오류로 끝나면 anyOf도 해당 오류로 끝납니다.

6. 오류 처리 및 타임아웃

6.1. exceptionally 및 handle

어떤 단계에서든 발생한 예외는 후속 단계로 전파됩니다. 해당 단계들은 오류 처리 단계를 만날 때까지 건너뜁니다:

CompletableFuture<BigDecimal> fee = CompletableFuture
    .supplyAsync(() -> {
        throw new IllegalStateException("Shipping service down");
    }, executor)
    .thenApply(value -> (BigDecimal) value)          // 건너뛰어짐
    .exceptionally(ex -> {
        System.out.println("오류: " + ex.getMessage());
        return new BigDecimal("50000");              // 기본값
    });

System.out.println(fee.join());

오류: java.lang.IllegalStateException: Shipping service down
50000

매개변수 ex는 보통 원래 예외를 감싼 CompletionException입니다. 따라서 메시지에는 클래스 이름 접두사가 있습니다. 실제 오류를 얻으려면 ex.getCause()를 사용합니다.

handle은 결과와 오류를 모두 받습니다. 작업이 성공하든 실패하든 항상 호출됩니다:

CompletableFuture<String> status = CompletableFuture
    .supplyAsync(() -> orderService.findByCustomer(42L), executor)
    .handle((orderList, ex) -> ex == null
        ? orderList.size() + "개의 주문이 있습니다"
        : "주문을 로드할 수 없습니다");

또한 whenComplete도 있습니다. 결과와 오류를 모두 받지만 결과를 변경하지는 않습니다. 이 메서드는 로그를 기록하는 데 적합합니다.

6.2. orTimeout 및 completeOnTimeout

외부 서비스를 무한정 기다리지 마세요. Java 9+에는 이를 위한 두 가지 메서드가 있습니다:

import java.util.concurrent.TimeUnit;

CompletableFuture<BigDecimal> withDefault = CompletableFuture
    .supplyAsync(() -> { Delay.sleep(2000); return new BigDecimal("30000"); }, executor)
    .completeOnTimeout(new BigDecimal("45000"), 500, TimeUnit.MILLISECONDS);

System.out.println(withDefault.join()); // ~500ms 후 45000

CompletableFuture<BigDecimal> strict = CompletableFuture
    .supplyAsync(() -> { Delay.sleep(2000); return new BigDecimal("30000"); }, executor)
    .orTimeout(500, TimeUnit.MILLISECONDS);
// strict.join()은 원인이 TimeoutException인 CompletionException을 던집니다

completeOnTimeout은 시간이 초과되면 기본값을 반환합니다. orTimeout은 TimeoutException으로 future를 종료합니다.

참고: 타임아웃은 future를 완료하기만 합니다. 기본 작업을 중지하지는 않습니다. 스레드는 계속 2초를 “대기”합니다. cancel(true)도 실행 중인 스레드를 중단하지 않습니다. 실제로 취소하려면 작업 내에서 취소 플래그를 직접 확인하거나 HTTP 클라이언트의 타임아웃을 사용해야 합니다.

7. 테스트 작성

Spring Boot를 사용하면 spring-boot-starter-test에 JUnit 5와 AssertJ가 이미 포함되어 있습니다. 순수 Java 프로젝트의 경우 다음을 추가합니다:

<dependency>
    <groupId>org.junit.jupiter</groupId>
    <artifactId>junit-jupiter</artifactId>
    <version>5.11.3</version>
    <scope>test</scope>
</dependency>
<dependency>
    <groupId>org.assertj</groupId>
    <artifactId>assertj-core</artifactId>
    <version>3.26.3</version>
    <scope>test</scope>
</dependency>

먼저 테스트하기 쉽도록 로직을 클래스로 묶습니다:

class CheckoutService {
    private final OrderService orderService;
    private final ShippingService shippingService;
    private final Executor executor;

    CheckoutService(OrderService o, ShippingService s, Executor executor) {
        this.orderService = o;
        this.shippingService = s;
        this.executor = executor;
    }

    CompletableFuture<BigDecimal> grandTotal(long customerId, String city) {
        var orders = CompletableFuture.supplyAsync(() -> orderService.findByCustomer(customerId), executor);
        var fee = CompletableFuture.supplyAsync(() -> shippingService.feeFor(city), executor);
        return orders.thenCombine(fee, (list, f) -> list.stream()
            .map(Order::total)
            .reduce(BigDecimal.ZERO, BigDecimal::add)
            .add(f));
    }
}

그런 다음 성공과 실패 사례 모두에 대해 테스트를 작성합니다:

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executors;
import org.junit.jupiter.api.Test;

class CheckoutServiceTest {

    @Test
    void shouldCombineOrdersAndShippingFee() {
        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var service = new CheckoutService(new OrderService(), new ShippingService(), executor);

            assertThat(service.grandTotal(42L, "Hue"))
                .succeedsWithin(Duration.ofSeconds(2))
                .isEqualTo(new BigDecimal("270000"));
        }
    }

    @Test
    void shouldPropagateShippingFailure() {
        ShippingService broken = new ShippingService() {
            @Override
            BigDecimal feeFor(String city) {
                throw new IllegalStateException("down");
            }
        };

        try (var executor = Executors.newVirtualThreadPerTaskExecutor()) {
            var service = new CheckoutService(new OrderService(), broken, executor);

            assertThatThrownBy(() -> service.grandTotal(42L, "Hue").join())
                .isInstanceOf(CompletionException.class)
                .hasCauseInstanceOf(IllegalStateException.class);
        }
    }
}

succeedsWithin은 최대 2초를 기다린 후 값을 확인합니다. 이렇게 하면 코드에 버그가 있어도 테스트가 행(hanging)되지 않습니다. 두 번째 테스트에서 ShippingService의 오류는 thenCombine을 통해 전파되고 CompletionException으로 래핑됩니다.

테스트에서 Runnable::run을 Executor로 전달하여 모든 것을 테스트 스레드에서 동기적으로 실행할 수 있습니다. 이렇게 하면 테스트가 빠르고 결과가 안정적입니다.

8. 일반적인 실수

• commonPool에서 I/O 실행. 이 풀은 CPU 수에서 1을 뺀 수의 스레드를 가지며 전체 애플리케이션(병렬 스트림 포함)에서 공유됩니다. 느린 HTTP 호출 몇 개로 인해 전체가 막힐 수 있습니다. I/O의 경우 별도의 executor를 전달해야 하며, virtual thread가 가장 좋습니다.
• join()을 너무 일찍 호출. supplyAsync(a).join()을 쓰고 나서 supplyAsync(b).join()을 쓰면 여전히 순차적입니다. 모든 future를 먼저 시작한 후 조합하거나 기다려야 합니다.
• 예외를 삼키기. join(), get() 또는 exceptionally/whenComplete를 호출하지 않으면 오류가 흔적 없이 사라집니다.
• thenApply와 thenApplyAsync 혼동. thenApply는 이전 단계를 완료한 스레드에서 실행되거나 future가 이미 완료되었으면 호출하는 스레드에서 실행될 수 있습니다. 다음 단계가 무겁거나 차단하면 ...Async(fn, executor) 변형을 사용해야 합니다.
• 타임아웃 설정 안 함. 멈춘 서비스는 요청도 함께 멈추게 합니다.
• 프로그램이 너무 일찍 종료. commonPool의 스레드는 데몬 스레드입니다. 작업이 완료되기 전에 main이 종료되면 결과가 절대 인쇄되지 않습니다.

참고: 호출이 하나뿐이고 결과를 즉시 기다리는 경우 CompletableFuture를 사용해야 합니다. Java 21부터는 virtual thread에서 실행되는 코드에서 일반적인 동기 작성이 성능 손실 없이 더 읽기 쉬운 경우가 많습니다. CompletableFuture는 여러 독립적인 작업을 병렬로 실행해야 하거나 논-블로킹 파이프라인이 필요할 때만 사용할 가치가 있습니다.

9. 결론

CompletableFuture는 스레드를 직접 관리하지 않고도 독립적인 작업을 병렬로 실행하고 결과를 조합하는 데 도움이 됩니다. 빠른 요약:

• 작업 생성: supplyAsync/runAsync, 항상 적절한 executor를 전달합니다.
• 단계 연결: 값을 변환할 때 thenApply, 다음 단계가 future를 반환할 때 thenCompose.
• 조합: 두 개의 future에는 thenCombine, 여러 개의 future에는 allOf, 가장 빠른 결과를 얻으려면 anyOf.
• 오류 및 타임아웃: exceptionally, handle, orTimeout, completeOnTimeout.

테스트에서는 AssertJ의 succeedsWithin을 사용하여 테스트가 행(hanging)될 걱정 없이 결과를 확인합니다.