[Java] Callable과 Future & CompletableFuture

개요

자바는 스레드를 통해 작업들을 병렬적으로 처리할 수 있다.

하지만 Runnable 인터페이스를 통해 작업을 제출하게 되면, 반환값을 받을 수 없다는 단점이 있다.

Executor 프레임워크는 이러한 문제를 해결하기 위해 Callable과 Future라는 인터페이스를 도입했다.

 

이전 글에 이어지는 내용이다.

 

 

Runnable vs Callable

public interface Runnable {
	void run();
}

Runnable 인터페이스는 반환 값이 없으며, 예외를 던질 수도 없다.

 

public interface Callable<V> {
	V call() throws Exception;
}

Callable 인터페이스는 값을 반환할 수 있으며, 예외를 던질 수 있다.

 

 

Callable과 Future 사용

class MyTask implements Callable<Integer> {
    @Override
    public Integer call() {
    	//난수 생성하는데 걸리는 시간
    	sleep(2000);
    	int value = new Random().nextInt(10);
    	System.out.println("난수 생성: " + value);
    	return value;
    }
}

난수를 생성하는 Callable 작업이 있다. 난수를 생성하는 데는 2초라는 시간이 걸린다고 가정하며, 작업이 완료되면 난수를 반환한다.

 

try (ExecutorService es = Executors.newFixedThreadPool(100)) {
    Future<Integer> future = es.submit(new MyTask());
    System.out.println(future.get());
}

Callable 작업을 ExecutorService 스레드 풀에 전달하면, Future 객체를 반환한다.

Future 객체의 get() 메서드를 통해 작업의 결과를 직접 받아올 수 있다!

 

정리한다면, Callable 객체는 결과를 반환하는 작업이고, Future 객체는 미래의 결과를 받을 수 있는 객체인 것이다.

 

 

동작 분석

Callable 객체를 통해 ExecutorService에 작업을 전달하게 되면, 작업을 보관하는 Queue에 작업을 감싼 Future 객체가 보관된다. 

 

작업이 스레드에게 할당되면, Future 객체를 즉시 반환하고 작업을 비동기적으로 수행한다.

 

이렇게 전달받은 Future 객체를 통해 요청 스레드는 작업의 결과값을 직접 조회할 수 있는 것이다. 

 

 

Future 사용 시 주의 사항

long startTime = System.currentTimeMillis();

Integer result = 0;

try(ExecutorService es = Executors.newFixedThreadPool(100)) {
    for (int i = 0; i < 10; i++) {
        Future<Integer> future = es.submit(new MyTask());
        //future.get() 호출
        result += future.get();
    }
}

long endTime = System.currentTimeMillis();

System.out.println("모든 작업 완료 시간 : " + (endTime - startTime));
System.out.println("결과 : " + result);

난수를 10개를 동시에 생성해서 더해주려 한다. 2초가 걸리는 난수를 10개 실행하지만, 스레드에 할당하기에 2초가 걸릴 것으로 예상된다. 

 

하지만 결과는 20초가 걸렸다..! 작업들이 동기적으로 수행된 것이다.

 

문제는 get() 메서드 호출시점이다.

get() 메서드는 스레드의 할당된 작업의 결과 값을 반환받는다. 만약 아직 결과가 나오지 않았다면, 메인 스레드를 블로킹 한 후 작업의 결과 값을 기다린다. 마치 Thread.join()과 같다. 

 

//작업 할당
Future<Integer> future = es.submit(new MyTask());
//할당 후 바로 작업의 결과 값 조회 시도
result += future.get();

위 코드에서 보면 작업을 할당 후 바로 결과 값을 시도한다. 작업의 과정을 모두 기다린 후 결과 값을 조회하고 다음 작업을 할당하는 것이므로 동기적으로 동작하는 것과 다름 없다. 

 

수정 사항

long startTime = System.currentTimeMillis();

//Future 객체 저장 리스트
List<Future<Integer>> results = new ArrayList<>();

Integer result = 0;

try(ExecutorService es = Executors.newFixedThreadPool(100)) {
    for (int i = 0; i < 10; i++) {
        Future<Integer> future = es.submit(new MyTask());
        
        //Future 객체를 바로 조회하는 대신 보관한다.
        results.add(future);
    }
}

//보관된 Future 객체의 작업들을 한 번에 조회한다.
for (Future<Integer> future : results) {
    result += future.get();
}

long endTime = System.currentTimeMillis();

System.out.println("모든 작업 완료 시간 : " + (endTime - startTime));
System.out.println("결과 : " + result);

Future 객체의 결과 값을 바로 조회하는 대신 리스트에 보관하였고, 모든 작업을 할당 후 

 

모든 작업이 비동기적으로 수행되었기에, 결과를 구하는데는 2초면 충분하다.

 

정리하자면, 필요한 모든 요청을 할당한 후에 작업의 결과 값이 사용되야 할 시점에 get() 메서드를 통해 결과 값을 조회하면 된다. 

 

 

Future 예외 결과 반환

class ExCallable implements Callable<Integer> {
    @Override
    public Integer call() {
        System.out.println("Callable 실행, 예외 발생");
        throw new IllegalStateException("ex!");
    }
}

무조건 예외를 발생시키는 Callable 객체가 있다.

 

ExecutorService es = Executors.newFixedThreadPool(100);

Future<Integer> future = es.submit(new ExCallable());

try {
    future.get();
} catch (ExecutionException e) {		// Future 객체 내에서 예외 발생 시, ExecutionException
    System.out.println("e = " + e);
    System.out.println("e.cause = " + e.getCause());
}

es.close();

 해당 작업을 할당 후, get() 메서드로 결과 값을 조회해보았다. 

 

Future의 작업 중 예외가 발생하면 FAILED로 상태가 변경되며, 요청 스레드에는 ExecutionException 예외를 던지게 된다.

ExecutionException 내부에는  Future 작업 중 발생한 원본 예외 내용이 저장되어 있다.

 

Future.get() 메서드는 작업의 결과 값 뿐만 아닌, 작업 중 발생한 예외도 조회할 수 있다

 

 

Future 메서드 정리

boolean cancel(boolean mayInterruptIfRunning)

Future 객체를 취소 상태로 변경한다. 

매개 변수가 true일 경우, 실행 중인 Future 객체라면 인터럽트를 통해 작업을 중단한다.

매개 변수가 false일 경우, 실행 중인 Future 객체는 중단하지 않는다.

 

 

boolean isCancelled()

Future 객체의 작업이 취소되었는지 확인한다.

 

boolean isDone()

Future 객체의 작업이 완료되었는지 확인한다.

 

T get()

작업이 완료될 때까지 대기한 후, 결과 혹은 예외를 반환한다.

 

T get(long timeout, TimeUnit unit)

주어진 시간 내 작업이 완료될 때까지 대기한 후, 결과 혹은 예외를 반환한다.

 

 

작업 컬렉션 처리

ExecutorService는 여러 작업을 한 번에 처리하는 invokeAll()과 invokeAny()를 제공한다.

 

//Future 객체 저장 리스트
List<Future<Integer>> results = new ArrayList<>();

Integer result = 0;

try(ExecutorService es = Executors.newFixedThreadPool(100)) {
    for (int i = 0; i < 10; i++) {
        Future<Integer> future = es.submit(new MyTask());
        
        //Future 객체를 바로 조회하는 대신 보관한다.
        results.add(future);
    }
}

//보관된 Future 객체의 작업들을 순차적으로 조회한다.
for (Future<Integer> future : results) {
    result += future.get();
}

우리는 위에서 여러 작업을 한 번에 할당하기 위해서 반복문을 사용하였다.

 

List<MyTask> tasks = new ArrayList<>();

List<Future<Integer>> results;

Integer result = 0;

try(ExecutorService es = Executors.newFixedThreadPool(100)) {
    //1. 작업을 컬렉션에 저장
    for (int i = 0; i < 10; i++) {
        tasks.add(new MyTask());
    }

    //2. 작업 컬렉션을 한 번에 처리
    results = es.invokeAll(tasks);
}

for (Future<Integer> future : results) {
    result += future.get();
}

invokeAll()과 invokeAny()를 통해 작업 컬렉션을 한 번에 처리할 수 있다. 

 

List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks)

invokeAll() 메서드는 컬렉션의 모든 Callable 작업을 제출하고, 모든 작업이 완료될 때까지 기다린다.

 

 

<T> T invokeAny(Collection<? extends Callable<T>> tasks)
        throws InterruptedException, ExecutionException;

invokeAny() 컬렉션의 Callable 작업 중 가장 먼저 완료된 작업의 결과를 반환한다.

 

 

CompletableFuture 개요

Future는 작업을 비동기로 수행하여 결과를 조회할 수 있는 좋은 라이브러리이다.

하지만 작업의 결과를 조회하여 다른 작업에 이용하기 위해서는 필수로 블로킹 메서드인 get() 함수를 호출하여야 한다.

 

CompletableFuture를 Future의 한계들을 보완하고 리액티브 프로그래밍 스타일을 가능하게 한다.

 

 

CompletableFuture 메서드

int makeRandom() {
    sleep(2000);
    int value = new Random().nextInt(10);
    System.out.println("난수 생성 : " + value);
    return value;
}

작업은 위에서 계속해서 진행했던 난수를 생성하는 작업을 예로 들겠다.

 

생성

ExecutorService es = Executors.newFixedThreadPool(100)

//결과 값을 반환하는 CompletableFuture 생성
CompletableFuture<Integer> task = CompletableFuture.supplyAsync(() -> makeRandom(), es);

supplyAsync(Supplier) - 결과를 반환하는 CompletableFuture 객체를 생성

 

//결과 값을 반환하지 않는 CompletableFuture 생성
CompletableFuture<Void> task = CompletableFuture.runAsync(() -> makeRandom(), es);

runAsync(Runnable) - 결과를 반환하지 않는 CompletableFuture 객체를 생성한다. 제네릭 타입이 Void인 것을 확인할 수 있다.

 

이 때, 
인자로 ExecutorService를 넘겨줄 수 있다. 넘겨주지 않으면 Default로 ForkJoinPool이 스레드 풀로 사용
된다. 

 

변환

CompletableFuture<String> task = CompletableFuture.supplyAsync(() -> makeRandom(), es)
.thenApply(value -> value.toString());    //생성한 난수를 String으로 변환

thenApply(Function) - 결과를 다른 값으로 단순 변환한다. 제네릭 타입이 String으로 변환된 것을 볼 수 있다.

 

CompletableFuture<String> task = CompletableFuture.supplyAsync(() -> makeRandom(), es)
.thenCompose(value -> CompletableFuture.supplyAsync(() -> String.valueOf(value)));
			//생성한 난수 -> 난수를 String으로 변환하는 CompletableFuture으로 변환

thenCompose(Function) - 결과를 또다른 비동기 작업(CompletableFuture)에 연결한다. 

 

.thenApplyAsync(value -> value.toString(), es);
.thenComposeAsync(value -> CompletableFuture.supplyAsync(() -> String.valueOf(value)), es);

thenApplyAsync(), thenComposeAsync - 변환하는 작업을 비동기로 처리할 수도 있다.

 

소비

CompletableFuture.supplyAsync(() -> makeRandom(), es)
.thenComposeAsync(value -> CompletableFuture.supplyAsync(() -> String.valueOf(value)), es)
.thenAccept(result -> System.out.println("결과값: " + result));  //위 작업들에서 처리한 결과를 소비한다.

thenAccept(Consumer) - 작업에 대한 결과를 사용한다.

 

CompletableFuture.supplyAsync(() -> makeRandom(), es)
.thenComposeAsync(value -> CompletableFuture.supplyAsync(() -> String.valueOf(value)), es)
.thenRun(() -> System.out.println("단순 작업 완료 로그 호출"));  //작업에 대한 결과를 신경쓰지 않고, 추가적인 작업을 실행한다.

thenRun(Runnable) - 작업에 대한 결과를 사용하지 않고 추가적인 작업을 실행한다.

 

조합

CompletableFuture<Integer> task1 = CompletableFuture.supplyAsync(() -> makeRandom(), es);
CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> makeRandom(), es);
CompletableFuture<Integer> result = task1
		//두 작업의 결과를 조합하여 새 결과 값을 생성
		.thenCombine(task2, (taskResult1, taskResult2) -> taskResult1 + taskResult2);

thenCombine(CompletableFuture, BiFunction) - 두 작업의 결과를 조합해 새 결과 값을 생성할 수 있다.

 

CompletableFuture<Integer> task1 = CompletableFuture.supplyAsync(() -> makeRandom(), es);
CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> makeRandom(), es);

//모든 작업을 실행하고, 모두 완료될 때까지 기다린다.
CompletableFuture.allOf(task1, task2)
                .join();

allOf(CompletableFuture..) - 모든 작업이 완료될 때까지 대기, 작업의 결과는 Void

 

CompletableFuture<Integer> task1 = CompletableFuture.supplyAsync(() -> makeRandom(), es);
CompletableFuture<Integer> task2 = CompletableFuture.supplyAsync(() -> makeRandom(), es);

//어느 작업 하나가 완료될 때까지 대기
CompletableFuture.anyOf(task1, task2)
                    .thenAccept(anyResult -> System.out.println(anyResult));

anyOf(CompletableFuture..) - 어느 작업 하나가 완료될 때까지 대기, 작업의 결과는 Object

 

예외 처리

CompletableFuture.supplyAsync(() -> {
	int random = makeRandom();
	if (random < 5) {
		throw new RuntimeException("난수가 5보다 작아서 예외가 발생");
	}

	return random;
}, es)

//예외가 발생할 경우 다른 값으로 처리
.exceptionally(ex -> {
	System.out.println(ex.getMessage());
	return 0;
});

exceptionally(Function) - 예외가 발생할 경우 호출되며, 결과 값을 변경한다.

 

 

 

CompletableFuture를 통한 Future 한계 보완

한계 보완 1. 블로킹 없이 작업의 결과 값을 다룰 수 있다.

//Future
Future<String> future = executor.submit(() -> fetchFromDB());

String result = future.get();  //결과를 받기위해 블로킹
process(result);

//CompletableFuture
CompletableFuture.supplyAsync(() -> fetchFromDB())
    .thenAccept(result -> process(result));  //논블로킹

Future의 경우 결과값에 추가적인 작업을 위해서, 블로킹 메서드를 통해 결과값을 조회한 후 진행해야 한다.

CompletableFuture는 완료된 작업의 결과값에 대해 논블로킹으로 추가적인 작업을 진행할 수 있다.

 

한계 보완 2. 작업 체이닝 가능

Future<User> userFuture = executor.submit(() -> getUser(1));
User user = userFuture.get();  //블로킹

Future<Order> orderFuture = executor.submit(() -> getOrder(user.getId()));
Order order = orderFuture.get();  //블로킹

Future<Product> productFuture = executor.submit(() -> getProduct(order.getProductId()));
Product product = productFuture.get();  //블로킹

Future의 경우 순차적 비동기 작업이 콜백 지옥처럼 물리게 된다.

 

CompletableFuture.supplyAsync(() -> getUser(1))
    .thenCompose(user -> CompletableFuture.supplyAsync(() -> getOrder(user.getId())))
    .thenCompose(order -> CompletableFuture.supplyAsync(() -> getProduct(order.getProductId())))
    .thenAccept(product -> System.out.println(product));

CompletableFuture를 사용하면 블로킹 없이 비동기 작업들을 논블로킹으로 체이닝하여 작업할 수 있다.

 

한계 보완 3. 여러 CompletableFuture 조합 가능

Future<Integer> price1 = executor.submit(() -> getPrice("쿠팡"));
Future<Integer> price2 = executor.submit(() -> getPrice("네이버"));

int p1 = price1.get();  //블로킹
int p2 = price2.get();  //블로킹  
int min = Math.min(p1, p2, p3);

Future는 각 작업을 조합할 수 없기에, 작업의 결과값을 각각 조회하여서 작업을 진행해야 한다.

 

CompletableFuture<Integer> price1 = CompletableFuture.supplyAsync(() -> getPrice("쿠팡"));
CompletableFuture<Integer> price2 = CompletableFuture.supplyAsync(() -> getPrice("네이버"));

price1.thenCombine(price2, (p1, p2) -> Math.min(p1, p2))
      .thenAccept(min -> System.out.println("최저가: " + min));

CompletableFuture는 작업의 두 결과를 조합하여 추가적인 작업을 블로킹 없이 진행할 수 있다.

 

한계 보완 4. 예외 처리 가능

Future<String> future = executor.submit(() -> {
    throw new RuntimeException("DB 연결 실패");
});

try {
    String result = future.get();
} catch (ExecutionException e) {
    Throwable cause = e.getCause();
}

Future는 작업 중 발생하는 예외를 try-catch 문으로 처리해야하며, 또한 원본 예외가 ExecutionException 포장되어 나온다.

 

CompletableFuture.supplyAsync(() -> {
    if (Math.random() > 0.5) throw new RuntimeException("실패!");
    return "성공";
})
.exceptionally(ex -> {
    System.out.println("에러 발생: " + ex.getMessage());
    return "대체값";
})
.thenAccept(System.out::println);

CompletableFuture의 경우, 작업 중 예외가 발생한다면 대체 값을 반환하도록 할 수 있다. 

 

한계 보완 5. 실행 스레드 제어 가능

ExecutorService myExecutor = Executors.newFixedThreadPool(10);

CompletableFuture.supplyAsync(() -> fetchData())
    //ForkJoinPool 사용
    .thenApply(data -> process(data))
    //스레드를 새로 생성하여 비동기 실행
    .thenApplyAsync(data -> heavyProcess(data))
    //스레드 풀의 스레드를 사용하여 비동기 실행
    .thenApplyAsync(data -> anotherProcess(data), myExecutor)
    .thenAccept(System.out::println);

Future의 경우 실행 스레드를 제어할 수 없지만, CompletableFuture의 경우 작업을 별도의 스레드로 할당시켜서 비동기적으로 진행할 수 있다.