Devin.KR
로그인

자바 동시성 패키지 - ExecutorService 스레드 풀 CompletableFuture (자바 고급 18단원)

개발자 조회 1

이 단원에서 배우는 것

17단원에서는 new Thread() 로 스레드를 직접 만들고 synchronized 로 재고를 지켰다. 동작은 했지만 문제가 둘 남았다. 첫째, 스레드를 요청마다 만들면 생성 비용과 메모리(스택 기본 1MB)가 그대로 부하가 된다. 둘째, 작업의 결과를 받아 오려면 공유 필드에 써 놓고 join() 후에 읽는 식의 조립이 필요했다. java.util.concurrent 패키지는 이 둘을 각각 스레드 풀과 Future/CompletableFuture 로 해결한다. 이번 단원은 같은 재고·주문 도메인을 스레드 풀 위에 올리고, 여러 외부 호출을 병렬로 묶는 파이프라인까지 만든다.

  • ThreadPoolExecutor 의 파라미터가 실제 장애와 어떻게 연결되는지 이해하고 직접 튜닝한다
  • ConcurrentHashMap 등 동시성 컬렉션이 synchronized 래핑보다 나은 이유와 그 한계를 안다
  • CompletableFuture 로 병렬 호출을 조합하고 타임아웃·예외·폴백을 설계한다

왜 필요한가

주문 상세 화면 하나를 그리는 데 재고 API, 가격 API, 배송비 API 세 곳을 호출한다고 하자. 각각 200ms면 순차 호출은 600ms다. 세 개는 서로 의존하지 않으므로 동시에 부르면 300ms(가장 느린 것)면 끝난다. 이걸 new Thread() 로 짜면 스레드 세 개를 만들고, 결과를 배열에 담고, join() 하고, 예외가 나면 어느 스레드에서 났는지 따로 챙겨야 한다. 스레드 하나가 던진 예외는 기본적으로 호출한 쪽에 전달되지 않고 스택트레이스만 찍히고 사라진다. 그 결과 "화면 일부가 비어 있는데 로그에는 아무 문제가 없어 보이는" 상황이 만들어진다.

거기에 트래픽이 튀면 스레드가 무한정 만들어져 OutOfMemoryError: unable to create native thread 로 죽는다. 결국 스레드 수의 상한과 결과·예외를 값처럼 다루는 수단이 필요하다. 그게 이 패키지의 존재 이유다.

문법과 예제

1. ExecutorService — Executors 편의 메서드의 함정부터

교재는 보통 Executors.newFixedThreadPool(10) 으로 시작한다. 짧아서 좋지만 실무에서는 이 한 줄이 장애의 원인이 된다.

팩터리실제 구성부하가 몰리면
newFixedThreadPool(n)스레드 n개 + 무한 큐큐에 작업이 무한정 쌓여 힙을 먹고 OOM. 응답 지연이 끝없이 늘어난다
newCachedThreadPool()스레드 무제한 + 큐 없음스레드가 수천 개까지 늘어 OS 자원이 고갈된다
newSingleThreadExecutor()스레드 1개 + 무한 큐순서 보장은 되지만 한 작업이 막히면 전부 정지

그래서 운영 코드에서는 ThreadPoolExecutor 를 직접 만든다. 특히 큐를 유한하게 잡고 거부 정책을 정하는 것이 핵심이다. 무한 큐는 문제를 숨기고, 유한 큐는 문제를 즉시 드러낸다.

import java.util.List;
import java.util.concurrent.*;

public class PoolDemo {
    public static void main(String[] args) throws Exception {
        ThreadPoolExecutor pool = new ThreadPoolExecutor(
                4, 4,                                   // 코어 = 최대 = 4
                0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(100),          // 큐를 반드시 막아 둔다
                new ThreadFactory() {
                    private int n = 0;
                    @Override public Thread newThread(Runnable r) {
                        return new Thread(r, "order-worker-" + (n++));   // 이름을 준다
                    }
                },
                new ThreadPoolExecutor.CallerRunsPolicy());  // 넘치면 호출자가 직접 처리

        List<Callable<String>> tasks = List.of(
                () -> lookup("SKU-1"),
                () -> lookup("SKU-2"),
                () -> lookup("SKU-3"));

        List<Future<String>> futures = pool.invokeAll(tasks, 2, TimeUnit.SECONDS);
        for (Future<String> f : futures) {
            System.out.println(f.get());
        }

        pool.shutdown();
        if (!pool.awaitTermination(10, TimeUnit.SECONDS)) {
            pool.shutdownNow();
        }
        System.out.println("종료됨 = " + pool.isTerminated());
    }

    static String lookup(String sku) throws InterruptedException {
        Thread.sleep(300);
        return Thread.currentThread().getName() + " -> " + sku + " 재고 10";
    }
}
order-worker-0 -> SKU-1 재고 10
order-worker-1 -> SKU-2 재고 10
order-worker-2 -> SKU-3 재고 10
종료됨 = true

세 가지가 이 코드에서 의도적으로 들어간 부분이다.

  • 스레드 이름: 이름을 안 주면 스레드 덤프에 pool-1-thread-3 으로만 찍힌다. 풀이 다섯 개쯤 되면 어느 풀이 막혔는지 알 수 없다. 이름 한 줄이 장애 분석 시간을 몇 시간 줄인다.
  • 거부 정책: 기본값은 AbortPolicy(RejectedExecutionException)다. CallerRunsPolicy 는 호출한 스레드가 직접 작업을 처리하게 해서 자연스럽게 유입 속도를 늦춘다(배압, backpressure). 반대로 DiscardPolicy 는 작업을 조용히 버리므로 로그 수집처럼 유실이 허용되는 곳에만 쓴다.
  • 종료 절차: shutdown() 은 새 작업만 막고 진행 중인 것은 기다린다. shutdownNow() 는 대기 작업을 회수하고 실행 중 스레드에 인터럽트를 건다. 둘 다 "즉시 끝난다"는 뜻이 아니라서 awaitTermination 이 반드시 따라와야 한다.

풀 크기 산정에는 하나의 기준만 기억하면 된다. CPU를 쓰는 작업이면 코어 수 근처, I/O를 기다리는 작업이면 그보다 훨씬 크게. 계산과 I/O를 섞은 작업을 한 풀에 넣으면 둘 다 나빠지므로, 성격이 다르면 풀을 분리한다.

2. 동시성 컬렉션

17단원의 InventoryService 는 HashMap 전체를 synchronized 로 감쌌다. 읽기가 압도적으로 많은 재고 조회에서는 이것이 그대로 병목이 된다. ConcurrentHashMap 은 버킷 단위로만 잠그므로 읽기는 사실상 락 없이 진행된다.

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.LongAdder;

class StockCache {
    private final Map<String, LongAdder> hits = new ConcurrentHashMap<>();

    void hit(String sku) {
        hits.computeIfAbsent(sku, k -> new LongAdder()).increment();
    }

    long of(String sku) {
        LongAdder a = hits.get(sku);
        return a == null ? 0 : a.sum();
    }
}

computeIfAbsent 는 "없으면 만들어 넣는다"를 한 번의 원자적 연산으로 처리한다. if (!map.containsKey(k)) map.put(k, new LongAdder()); 로 쓰면 두 스레드가 각각 LongAdder 를 만들어 하나가 덮어써지고, 그 사이 센 값이 사라진다. 자주 쓰는 원자적 메서드는 다음과 같다.

메서드용도
putIfAbsent(k, v)값이 이미 만들어져 있을 때 한 번만 넣기
computeIfAbsent(k, fn)값 생성 비용이 클 때(캐시 적재). fn 안에서 같은 맵을 건드리면 데드락
merge(k, v, fn)누적 합계·카운트
compute(k, fn)기존 값을 읽어 새 값으로 바꾸기

선택 기준을 정리하면 이렇다.

상황선택이유
읽기 많고 쓰기 적음, 키-값ConcurrentHashMap버킷 단위 락, 읽기 무잠금
거의 읽기만 하고 가끔 전체 교체CopyOnWriteArrayList쓰기마다 배열 복사. 쓰기가 잦으면 최악
생산자-소비자ArrayBlockingQueue크기 상한이 곧 배압
지연 실행·재시도 큐DelayQueue시간이 된 것만 꺼내진다

Collections.synchronizedMap 은 모든 메서드를 하나의 락으로 감싼 것이라 Hashtable 과 성능 특성이 같고, 순회할 때는 여전히 바깥에서 직접 synchronized 를 걸어야 한다. 새 코드에서 쓸 이유가 거의 없다.

3. CompletableFuture — 호출을 조립한다

Future.get() 은 블로킹이라, 세 개를 병렬로 부르고 다시 조합하는 코드를 쓰면 결국 get() 이 줄줄이 늘어선다. CompletableFuture 는 "끝나면 이걸 해라"를 값으로 표현한다.

import java.util.concurrent.*;

public class OrderPipeline {
    static final ExecutorService IO_POOL = Executors.newFixedThreadPool(8);

    public static void main(String[] args) {
        CompletableFuture<Integer> stock =
                CompletableFuture.supplyAsync(() -> callStock("SKU-1"), IO_POOL);

        CompletableFuture<Integer> price =
                CompletableFuture.supplyAsync(() -> callPrice("SKU-1"), IO_POOL);

        CompletableFuture<String> quote = stock
                .thenCombine(price, (s, p) -> "재고 " + s + "개, 단가 " + p + "원")
                .orTimeout(2, TimeUnit.SECONDS)                       // Java 9+
                .exceptionally(e -> "조회 실패: " + e.getClass().getSimpleName());

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

        // 실패했을 때 폴백값으로 대체하기
        CompletableFuture<String> broken = CompletableFuture
                .<String>supplyAsync(() -> { throw new IllegalStateException("재고 API 500"); }, IO_POOL)
                .handle((v, e) -> e == null ? v : "폴백값");
        System.out.println(broken.join());

        IO_POOL.shutdown();
    }

    static int callStock(String sku) { sleep(200); return 10; }
    static int callPrice(String sku) { sleep(300); return 12_900; }
    static void sleep(long ms) {
        try { Thread.sleep(ms); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }
}
재고 10개, 단가 12900원
폴백값

메서드 이름 규칙만 알면 나머지는 조합이다.

메서드하는 일
thenApply결과를 변환한다(값 → 값)
thenCompose결과로 다음 비동기 호출을 이어 붙인다(값 → Future). 중첩을 편다
thenCombine독립적인 두 Future의 결과를 합친다
allOf / anyOf여러 개를 모두 기다리기 / 가장 빠른 하나만
exceptionally / handle실패를 값으로 되돌린다
이름에 Async 가 붙은 것콜백을 지정한 풀에서 실행한다

allOf 는 CompletableFuture<Void> 를 돌려주므로 결과는 각 Future에서 다시 꺼낸다.

List<CompletableFuture<Integer>> calls = skus.stream()
        .map(sku -> CompletableFuture.supplyAsync(() -> callStock(sku), IO_POOL))
        .toList();

int total = CompletableFuture.allOf(calls.toArray(new CompletableFuture[0]))
        .thenApply(v -> calls.stream().mapToInt(CompletableFuture::join).sum())
        .join();

여기서 .toList() 로 먼저 리스트를 만드는 것이 중요하다. 스트림이 지연 평가라서 map 안에서 바로 join() 을 하면 하나씩 순차 실행되어 병렬화가 통째로 사라진다. 먼저 전부 띄우고, 그다음에 전부 기다린다가 원칙이다.

실무에서 자주 틀리는 것

1. 풀을 지정하지 않고 supplyAsync 를 쓴다

CompletableFuture.supplyAsync(task) 는 ForkJoinPool.commonPool() 에서 돈다. 이 풀의 크기는 기본이 CPU 코어 수 − 1 이고 JVM 전체가 공유한다. 4코어 서버라면 스레드가 3개다. 여기에 200ms짜리 HTTP 호출을 밀어 넣으면 세 개만 실행되고 나머지는 줄을 선다. 게다가 parallelStream() 도 같은 풀을 쓰므로 서로를 굶긴다. I/O 작업에는 반드시 전용 풀을 두 번째 인자로 넘긴다.

2. 예외를 어디서도 받지 않는다

executor.submit(task) 이 돌려준 Future 를 버리면, 작업 안에서 난 예외는 로그에도 안 남고 사라진다. execute() 는 최소한 스택트레이스라도 찍히지만 submit() 은 예외를 Future 안에 담아 두기 때문이다. 결과가 필요 없어도 Future 는 받아서 확인하거나, 작업 본문 자체를 try/catch 로 감싸 로깅해야 한다. CompletableFuture 도 마찬가지로 exceptionally / handle / whenComplete 중 하나가 체인 끝에 반드시 있어야 한다.

또 하나. CompletableFuture 안에서 던져진 예외는 join() 시점에 CompletionException 으로, get() 시점에는 ExecutionException 으로 감싸여 나온다. 원인을 보려면 getCause() 를 벗겨야 한다. 이걸 모르고 catch (IllegalStateException e) 를 써 두면 절대 안 잡힌다.

3. 타임아웃이 없다

future.get() 을 인자 없이 호출하면 영원히 기다린다. 외부 API가 응답을 안 주면 요청 스레드가 하나씩 잠기다가 톰캣 스레드 풀이 고갈되고, 서비스 전체가 멈춘다. 항상 get(2, TimeUnit.SECONDS) 나 orTimeout(...) 을 쓴다. 다만 orTimeout 은 대기를 포기할 뿐 실제 HTTP 연결을 끊지는 않는다는 점을 알아야 한다. 커넥션까지 회수하려면 HTTP 클라이언트 자체의 connect/read 타임아웃을 따로 설정해야 한다.

4. ThreadLocal 값이 사라진 것을 버그로 착각한다

MDC 로그 추적 ID, 스프링 시큐리티 인증 정보, 트랜잭션 컨텍스트는 모두 ThreadLocal 에 담긴다. 작업을 풀에 넘기는 순간 다른 스레드에서 실행되므로 이 값들이 전부 비어 있다. 로그에서 추적 ID가 끊기거나 SecurityContextHolder.getContext().getAuthentication() 이 null 이 되는 원인이 대부분 이것이다. 필요한 값은 작업을 만들 때 캡처해서 인자로 넘기는 것이 가장 안전하다. 반대로 풀 스레드는 재사용되므로 ThreadLocal 에 값을 넣고 안 지우면 다음 요청에 남의 데이터가 보이는 사고로 이어진다. try/finally 로 remove() 를 보장해야 한다.

5. 풀을 종료하지 않는다

ExecutorService 의 스레드는 기본이 논데몬이라, 종료하지 않으면 JVM이 안 끝난다. 배치 프로그램이 "다 끝났는데 프로세스가 안 죽는" 증상은 열에 아홉 이것이다. 스프링에서는 @Bean(destroyMethod = "shutdown") 이나 ThreadPoolTaskExecutor 의 setWaitForTasksToCompleteOnShutdown(true) 로 처리한다.

스스로 확인하기

  1. 4코어 서버에서 외부 API를 호출하는 작업 100개를 CompletableFuture.supplyAsync(task)(풀 미지정)로 띄웠다. 각 호출은 200ms다. 전체는 대략 얼마나 걸리며, 왜 그런가?
  2. 아래 코드는 무한 큐 때문에 위험하다. 어떤 증상으로 나타나며, 어떻게 고치는가? 
    ExecutorService pool = Executors.newFixedThreadPool(10);
while (queueFromKafka.hasNext()) {
    pool.submit(() -> save(queueFromKafka.next()));
}
  3. 재고 API가 실패하면 캐시된 값을 쓰고, 캐시도 없으면 0으로 처리하려 한다. CompletableFuture 체인으로 어떻게 표현하는가?

정답

  1. 대략 7초 근처다. 풀을 지정하지 않으면 ForkJoinPool.commonPool() 을 쓰고 그 크기는 코어 수 − 1, 즉 3이다. 100개를 3개씩 나눠 처리하면 34 라운드 × 200ms ≈ 6.8초가 된다. CPU는 거의 놀고 있는데도 그렇다. 크기 32~64짜리 전용 풀을 넘기면 1초 이하로 떨어진다. 이 계산은 "I/O 대기 작업의 풀 크기는 코어 수와 무관하게 잡는다"는 원칙의 근거이기도 하다.
  2. 카프카에서 읽는 속도가 save() 보다 빠르면 작업이 무한 큐에 계속 쌓인다. 증상은 (a) 힙 사용량이 계속 늘다가 OutOfMemoryError: Java heap space, (b) 죽기 전까지는 지연 시간만 늘고 CPU는 한가해 보임, (c) 프로세스를 죽이면 큐에 있던 작업이 통째로 유실됨이다. 고치는 방법은 ArrayBlockingQueue 로 큐 크기를 제한하고 CallerRunsPolicy 를 주는 것이다. 그러면 큐가 차는 순간 읽는 스레드가 직접 저장 작업을 하게 되어 소비 속도가 자동으로 억제된다.
  3. exceptionally 를 두 번 겹치는 대신 폴백 단계를 명시적으로 나눈다. 
    CompletableFuture<Integer> stock =
    CompletableFuture.supplyAsync(() -> api.stockOf(sku), IO_POOL)
        .orTimeout(500, TimeUnit.MILLISECONDS)
        .exceptionally(e -> cache.stockOf(sku))   // 1차 폴백: 캐시
        .exceptionally(e -> 0);                   // 2차 폴백: 캐시도 실패
    첫 exceptionally 안에서 캐시 조회가 또 예외를 던지면 그 예외가 다음 exceptionally 로 흘러 0이 된다. 다만 캐시 조회가 null 을 반환하는 경우는 예외가 아니므로 걸러지지 않는다. 그래서 실제로는 thenApply(v -> v == null ? 0 : v) 를 마지막에 붙이거나, 폴백 로직을 handle 안에서 if 로 명시하는 편이 읽기에도 낫다.