1. thenCompose vs. thenApply: 차이점과 사용 시점
Java의 비동기 프로그래밍(CompletableFuture)에서는 작업을 체인으로 연결하는 경우가 많습니다. 이를 위해 유사한 메서드 두 가지가 있는데, 바로 thenApply와 thenCompose입니다. 하지만 이 둘은 동작이 다릅니다!
thenApply
thenApply는 다음 단계가 새로운 비동기 작업을 시작하지 않고 값을 단순 변환하는 경우에 사용합니다. 이전 단계의 결과를 받아 처리하고, 새로운 값(CompletableFuture가 아님)을 반환합니다.
Stream API에 익숙하다면, thenApply는 map처럼 동작한다고 볼 수 있습니다. 결과를 받아 함수를 적용하고 변환된 값을 반환합니다.
예시:
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "42");
CompletableFuture<Integer> lengthFuture = cf.thenApply(s -> s.length());
// lengthFuture에는 2가 들어 있습니다(문자열 "42"의 길이)
한마디로, thenApply는 “결과가 준비되면 그 결과로 이렇게 해라”라고 말하는 방법입니다.
thenCompose
- 다음 단계가 또 다른 비동기 작업일 때 사용합니다(CompletableFuture를 반환).
- 중첩된 CompletableFuture를 “펼칠” 수 있습니다(flatMap과 유사).
- thenApply에 비동기 함수를 쓰면 CompletableFuture<CompletableFuture<T>>가 되어 불편합니다!
예시:
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> "user42");
// 예를 들어, 사용자 이름으로 주문 목록을 비동기로 가져와야 한다고 합시다
CompletableFuture<List<Order>> ordersFuture = cf.thenCompose(username -> fetchOrdersAsync(username));
// fetchOrdersAsync는 CompletableFuture<List<Order>>를 반환합니다
시각적으로:
- thenApply: CF<String> → thenApply(s -> s.length()) → CF<Integer>
- thenCompose: CF<User> → thenCompose(u -> fetchOrdersAsync(u.id)) → CF<List<Order>>
언제 무엇을 사용할까?
- 함수가 일반 값을 반환 — thenApply를 사용하세요.
- 함수가 CompletableFuture를 반환 — thenCompose를 사용하세요.
실수 예시:
cf.thenApply(username -> fetchOrdersAsync(username)); // CF<CF<List<Order>>>가 됩니다
cf.thenCompose(username -> fetchOrdersAsync(username)); // CF<List<Order>>를 얻게 됩니다
2. 스레드 풀 관리 (Executor): 왜, 그리고 어떻게 자체 Executor를 사용할까
기본값: ForkJoinPool.commonPool()
CompletableFuture.supplyAsync(...) 또는 thenApplyAsync(...)에서 Executor를 지정하지 않으면 Java는 공용 스레드 풀 — ForkJoinPool.commonPool()을 사용합니다. 편리하지만 항상 적합하진 않습니다:
- 오래 걸리거나 블로킹되는 작업(네트워크 요청, 파일 I/O)이 많다면 공용 풀이 “막혀” 모든 작업이 대기할 수 있습니다.
- 작업의 우선순위를 분리하거나 동시에 실행되는 스레드 수를 제한해야 할 때가 있습니다.
자체 Executor가 필요한 경우?
- 오래 걸리는 블로킹 작업(예: DB 질의, HTTP 요청, 파일 읽기).
- 작업 분리: 사용자 작업이 시스템 작업에 영향을 주지 않도록.
- 자원 제한: 예를 들어 동시 다운로드를 10개 이하로 제한.
자체 Executor 생성 방법
일반적으로 ThreadPoolExecutor 또는 Executors의 팩토리를 사용합니다:
ExecutorService myExecutor = Executors.newFixedThreadPool(10);
CompletableFuture에서 자체 Executor 사용하기
- supplyAsync, runAsync, thenApplyAsync, thenComposeAsync 등에서 두 번째 인자로 Executor를 전달할 수 있습니다.
예시:
CompletableFuture<String> cf = CompletableFuture.supplyAsync(
() -> loadDataFromNetwork(), myExecutor
);
cf.thenApplyAsync(data -> processData(data), myExecutor)
.thenAcceptAsync(result -> System.out.println(result), myExecutor);
중요: Executor를 지정하지 않으면 ForkJoinPool.commonPool()이 사용됩니다.
기본 Executor로 충분한 경우?
- 짧고 CPU에만 의존하는 작업(간단한 계산).
- 어떤 스레드에서 실행되는지가 중요하지 않을 때.
3. 타임아웃 처리: orTimeout과 completeOnTimeout
비동기 작업은 멈추거나 너무 오래 걸릴 수 있습니다(예: 서버가 응답하지 않는 경우). 영원히 기다리지 않기 위해 CompletableFuture에는 타임아웃을 다루는 메서드가 있습니다.
orTimeout
- 지정된 시간 안에 완료되지 않으면 TimeoutException으로 CompletableFuture를 실패 완료시킵니다.
- 실제 실행 중인 작업을 취소하지는 않지만, 이후 체인은 오류를 받습니다.
문법:
cf.orTimeout(3, TimeUnit.SECONDS)
.exceptionally(ex -> {
System.out.println("타임아웃: " + ex);
return null;
});
예시:
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
Thread.sleep(5000); // 오래 걸리는 작업을 흉내냅니다
return "OK";
});
cf.orTimeout(2, TimeUnit.SECONDS)
.exceptionally(ex -> {
System.out.println("오류: " + ex);
return "TIMEOUT";
});
결과:
2초 후 TimeoutException이 발생하고 exceptionally가 오류를 처리합니다.
completeOnTimeout
- 지정된 시간 안에 완료되지 않으면 CompletableFuture를 지정한 값으로 완료합니다.
- 예외를 던지지 않고 “대체” 값을 반환합니다.
문법:
cf.completeOnTimeout("DEFAULT", 2, TimeUnit.SECONDS);
예시:
CompletableFuture<String> cf = CompletableFuture.supplyAsync(() -> {
Thread.sleep(5000);
return "OK";
});
cf.completeOnTimeout("TIMEOUT", 2, TimeUnit.SECONDS)
.thenAccept(System.out::println); // 2초 후 "TIMEOUT"을 출력합니다
orTimeout과 completeOnTimeout 비교
| 메서드 | 타임아웃 시 동작 | 이후 처리 방식 |
|---|---|---|
|
TimeoutException으로 완료 | exceptionally/handle로 처리 가능 |
|
지정한 값으로 완료 | thenAccept/thenApply가 그 값을 받음 |
4. 실습: thenCompose, 커스텀 Executor와 타임아웃 예제
과제:
- id로 사용자를 가져오기(비동기, 지연 포함).
- 그 다음 비동기로 사용자 주문 목록 가져오기(지연 포함).
- 커스텀 Executor 사용.
- 주문 조회에 타임아웃 추가.
import java.util.concurrent.*;
import java.util.*;
public class AsyncDemo {
static ExecutorService ioExecutor = Executors.newFixedThreadPool(4);
// 비동기 사용자 가져오기 시뮬레이션
static CompletableFuture<String> fetchUserAsync(int userId) {
return CompletableFuture.supplyAsync(() -> {
sleep(1000);
return "user" + userId;
}, ioExecutor);
}
// 비동기 주문 조회 시뮬레이션
static CompletableFuture<List<String>> fetchOrdersAsync(String username) {
return CompletableFuture.supplyAsync(() -> {
sleep(3000); // 오래 걸리는 작업!
return List.of("order1", "order2");
}, ioExecutor);
}
static void sleep(long ms) {
try { Thread.sleep(ms); } catch (InterruptedException ignored) {}
}
public static void main(String[] args) {
fetchUserAsync(42)
.thenCompose(username ->
fetchOrdersAsync(username)
.orTimeout(2, TimeUnit.SECONDS) // 주문 조회에 대한 타임아웃
.exceptionally(ex -> {
System.out.println("주문을 가져오지 못했습니다: " + ex);
return List.of();
})
)
.thenAccept(orders -> System.out.println("주문: " + orders))
.join(); // 전체 체인이 끝날 때까지 대기
ioExecutor.shutdown();
}
}
동작 흐름:
- 사용자를 가져옴(1초).
- 주문을 가져옴(3초 소요, 타임아웃 2초).
- 기한 내 완료되지 않으면 TimeoutException을 받고 빈 리스트를 반환.
- 모든 작업은 커스텀 Executor에서 수행.
결과:
주문을 가져오지 못했습니다: java.util.concurrent.TimeoutException
주문: []
fetchOrdersAsync의 지연을 1_000ms로 줄이면 실제 주문을 보게 됩니다.
5. 자주 하는 실수와 유의사항
오류 №1: thenCompose 대신 thenApply를 비동기 작업에 사용하는 경우.
함수가 CompletableFuture를 반환하는데 thenApply를 적용하면 CompletableFuture<CompletableFuture<T>>라는 중첩 타입이 됩니다. 이는 체인을 복잡하게 만들고 불필요한 래핑을 유발합니다. 해결: thenCompose를 사용해 결과를 CompletableFuture<T>로 “평탄화”하세요.
오류 №2: 자체 Executor 없이 오래 걸리는 작업 또는 IO 작업을 실행하는 경우.
기본적으로 작업은 ForkJoinPool.commonPool()에서 실행됩니다. 이를 과부하시키면 지연이 증가하고 애플리케이션의 다른 작업도 느려질 수 있습니다. 해결: 자체 ExecutorService를 만들고 supplyAsync/thenApplyAsync에 전달하세요.
오류 №3: orTimeout이 작업 실행을 취소해 준다고 기대하는 경우.
orTimeout은 타임아웃 시 CompletableFuture를 예외로만 완료할 뿐, 작업 자체는 백그라운드에서 계속 실행됩니다. 해결: 실행을 중단하려면 cancel(true) 또는 자체 인터럽트 메커니즘을 사용하세요.
오류 №4: 타임아웃의 적용 범위를 잘못 이해하는 경우.
orTimeout과 completeOnTimeout은 체인의 특정 단계에만 적용되고, 전체 체인에는 적용되지 않습니다. 해결: 전체 체인에 대한 공통 타임아웃이 필요하다면 체인을 별도의 CompletableFuture로 감싸고 그 객체에 타임아웃을 적용하세요.
오류 №5: ExecutorService를 닫지 않는 경우.
작업이 끝난 뒤 ExecutorService의 shutdown()/shutdownNow()를 호출하지 않으면 스레드가 계속 살아 있어 프로그램이 “멈춘” 것처럼 보일 수 있습니다. 해결: 항상 finally에서 ExecutorService를 종료하거나 Java 21+에서는 try-with-resources를 사용하세요.
GO TO FULL VERSION