[Java] Callable과 Future 및 Executor, ExecutorService, Executors

Callable이 등장한 이유

자바에서 멀티스레딩을 사용할 때 가장 기본적인 방법은 Runnable 인터페이스를 활용하는 것이다. 그러나 Runnable 인터페이스는 다음과 같은 한계를 가지고 있다.

• 반환값을 가질 수 없음: Runnable의 run() 메서드는 반환값이 없는 void 타입이다. 따라서 실행 결과를 얻으려면 별도의 공유 변수 또는 콜백을 사용해야 한다.
• 예외 처리가 어려움: Runnable은 checked exception을 명시적으로 던질 수 없다. run() 메서드는 throws 절을 가질 수 없으며, 내부에서 발생한 예외는 잡아서 처리해야 한다.

이러한 문제를 해결하기 위해 Java 5에서 Callable<T> 인터페이스가 도입되었다. java.util.concurrent.Callable 인터페이스는 별도의 스레드에서 실행할 수 있는 비동기 작업을 나타낸다. 예를 들어, Callable 객체를 ExecutorService에 submit 하면 비동기적으로 실행 후 반환값을 받아올 수 있다.

Callable 인터페이스

Callable<T>는 멀티쓰레드 환경에서 쓰레드의 실행 결과값을 반환하기 위한 인터페이스다. java.util.concurrent 패키지에 속하며, Runnable과 유사하게 멀티스레드 환경에서 비동기 작업을 실행하기 위해 사용되지만, 다음과 같은 개선점을 제공한다.

• 제네릭을 사용해 반환값을 가질 수 있다.
• call() 메서드는 checked exception을 던질 수 있다.

내부 구현

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

• Callable은 함수형 인터페이스로, call() 메서드 하나만 가지고 있다.
• call() 메서드는 비동기 작업을 실행하는 데 사용되며, 결과값을 반환할 수 있고, 작업 실행 중 오류가 발생하면 예외를 던질 수 있다.
• call() 메서드가 비동기적으로 실행될 경우, 결과값은 Java Future 객체를 통해 작업을 생성한 측에 전달된다.

Callable 과 Runnable 은 각각 언제 사용해야 할까?

Callable 과 Runnable은 모두 하나의 추상 메서드만 가지는 함수형 인터페이스이고, 별도의 스레드에서 실행될 작업을 나타낸다는 공통점이 있지만, 다음과 같은 차이점을 가진다.

• Runnable 의 run() 메서드는 반환값이 없으며, 예외를 던질 수도 없다.
• Callable은 call() 은 결과를 반환할 수 있으며, 예외 처리도 가능하다.

따라서 Callable은 단일 작업을 실행하고 결과를 반환해야 하는 경우 적합하며, Runnable은 결과값 반환 없이 장시간 실행되는 프로세스를 사용해야 할 경우 적합하다.

Future 인터페이스

Future 인터페이스는 비동기 작업의 결과를 나타내는 인터페이스다. 제공되는 메서드를 통해 작업이 완료되었는지 확인하고, 완료될 때까지 대기하며, 작업 결과를 가져올 수 있다.

Callable 에서 구현한 작업은 별도 쓰레드에서 비동기적으로 실행되기 때문에 언제 완료되어 값을 반환받을지 알 수 없다. 실행결과를 바로 받을 수 없으므로 미래에 완료될 Callable 의 반환값을 얻기 위해 Future 가 사용된다.

• 비동기 작업이 생성되면 Future 객체가 반환된다.
• 비동기 작업이 완료되면 작업이 시작될 때 반환된 Future 객체를 통해 결과를 받아온다.

내부 구현

public interface Future<V> {

    boolean cancel(boolean mayInterruptIfRunning);

    boolean isCancelled();

    boolean isDone();

    V get() throws InterruptedException, ExecutionException;

    V get(long timeout, TimeUnit unit)
        throws InterruptedException, ExecutionException, TimeoutException;
}

get()

• 연산이 완료될 때까지 블로킹 후 결과를 반환
• 타임아웃 설정 가능
• 취소된 경우 CancellationException, 예외 발생 시 ExecutionException을 던짐

isDone()

• 작업의 완료 여부를 반환
• 작업이 취소되어 끝난 경우, 정상적으로 완료된 경우, 예외로 종료된 경우

isCancelled()

• cancel()이 호출된 후 작업이 성공적으로 취소되었는지 확인
• 작업이 아직 실행 중이거나 정상 종료되었을 경우 false 반환

cancel()

• 작업을 취소시키며, 취소 가능 여부를 boolean으로 반환함
• boolean mayInterruptIfRunning 파라미터에 true를 전달하면 스레드를 interrupt 시켜 InterrepctException를 발생시켜 스레드가 인터럽트된다. false를 전달하면 진행중인 작업이 끝날 때까지 기다린다.
• 작업이 이미 정상적으로 완료되어 취소할 수 없는 경우에는 false를, 그렇지 않으면 true를 반환한다. 그 외에도 작업이 이미 취소되었거나 취소가 불가능한 경우에도 false가 반환될 수 있다.
• 반환 값은 작업이 실제로 취소되었는지를 보장하지 않기 때문에 작업이 취소되었는지 확인하려면 isCancelled()을 사용

Callable, Future 예시

public class Main {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(2);

        int initValue = 10;
        System.out.println("초기값 = " + initValue);
        Callable<Integer> task = new CallableImpl(10);
        Future<Integer> future = executor.submit(task);

        System.out.println("작업 실행 중...");
        Integer result = future.get();
        System.out.println("결과 = " + result);
        executor.shutdown();
    }

}

class CallableImpl implements Callable<Integer>{
    private final int val;

    CallableImpl(int val) {
        this.val = val;
    }

    @Override
    public Integer call() throws Exception {
        Thread.sleep(2000);

        return val * 100;
    }
}

Callable는 ExecutorService에 submit 되어 스레드 풀에서 작업을 수행하고 Future<T> 로 실행결과를 반환한다. future.get()을 호출하면 해당 작업이 완료될 때까지 현재 스레드가 블로킹(blocking) 상태가 되고, 작업이 완료되면 결과를 받을 수 있다.

Callable 또한 Runnable 과 마찬가지로 함수형 인터페이스이므로 동일한 작업을 람다식으로도 작성할 수 있다.

int initValue = 10;

Callable<Integer> task = () -> {
    Thread.sleep(2000);
    return initValue * 100;
};

Executor 인터페이스

스레드를 필요할때마다 생성하는 것은 비효율적이다. 그래서 스레드를 미리 만들어놓고 사용하는 스레드 풀(thread pool)이 등장하게 되는데, Excecutor인터페이스는 스레드 풀 구현을 위한 인터페이스이다.

스레드는 크게 작업 등록과 작업 실행으로 나누어진다. 이중에서 Executor는 작업 실행하는 책임을 가진다. 그래서 내부적으로도 전달받은 작업을 실행하는 execute() 메서드만 가지고 있다.

public interface Executor {
    void execute(Runnable command);
}

• 어떻게 실행할지(새 스레드/기존 스레드/호출 스레드)는 구현체가 결정한다.
• 스레드 생성 로직을 감추고, 스레드 풀 같은 최적화 전략 사용 가능하다.

ExecutorService 인터페이스

ExecutorService 는 Executor 인터페이스를 확장하여 작업 실행과 등록을 모두 할 수 있는 인터페이스로 실무에서 가장 많이 사용된다.

대표적으로 ThreadPoolExecutor가 ExecutorService 의 구현체인데, ThreadPoolExecutor 내부에 있는 블로킹 큐에 작업들을 등록해둔다. 그리고 각 작업들을 Thread Pool의 사용 가능한 쓰레드들에 할당해서 작업을 수행한다. 만약에 사용가능한 쓰레드가 없다면 작업은 큐에 대기하게 되고, 쓰레드가 작업이 끝나면 큐에 있는 다음 작업을 할당받는다.

쓰레드를 매번 새로 만드는 비용을 줄이고, 동시 작업 개수를 제한하여 시스템 자원을 효율적으로 관리할 수 있다.

public interface ExecutorService extends Executor {
    // 작업 제출
    Future<?> submit(Runnable task);
    <T> Future<T> submit(Callable<T> task);
    
    // 종료 관련
    void shutdown();
    List<Runnable> shutdownNow();
    boolean isShutdown();
    boolean isTerminated();
    boolean awaitTermination(long timeout, TimeUnit unit);
    
    // 여러 작업 실행
    <T> List<Future<T>> invokeAll(Collection<? extends Callable<T>> tasks);
    <T> T invokeAny(Collection<? extends Callable<T>> tasks);
}

sumit()

• 작업 제출과 결과를 받는다.
• Runnable뿐만 아니라 결과값을 돌려주는 Callable 작업도 제출할 수 있다.
• Runnable은 반환값을 받을 수 없는데도 Future를 반환하는 이유는 Future에서는 작업 완료여부/취소 등의 작업도 할 수 있기 때문이다. Future.get()을 하면 null을 반환하지만 작업이 끝날때까지 블로킹된다.

shutdown()

• 실행중인 스레드를 안전하게 종료할 수 있다.
• ExecutorService를 만들어 작업을 실행하면 shutdown()이 실행될때까지 다음 작업을 기다린다. 따라서 작업이 완료되면 반드시 명시적으로 shutdown()을 해야한다.

invokeAny() / invokeAll()

List<Callable<String>> tasks = Arrays.asList(
    () -> { Thread.sleep(1000); return "작업1"; },
    () -> { Thread.sleep(2000); return "작업2"; },
    () -> { Thread.sleep(500); return "작업3"; }
);

// 모든 작업 완료 대기
List<Future<String>> futures = executor.invokeAll(tasks);

// 가장 빨리 완료된 작업의 결과만 반환
String result = executor.invokeAny(tasks); // "작업3" (500ms)

• 여러 작업들을 비동기적으로 실행한다.

invokeAny()

• 모든 작업을 제출한다.
• Future가 아닌 가장 빨리 완료된 작업의 결과값을 반환한다.
• 나머지 실행중인 작업은 자동으로 취소한다.

invokeAll()

• 모든 작업을 제출한다.
• 모든 작업이 완료될 때까지 블로킹한다.
• 각 작업의 Future를 담은 List 반환한다.
• 반환된 Future들은 이미 완료된 상태이다. (isDone이 true. 하지만 항상 작업이 성공한 상태는 아님.)

Executors

ExecutorService를 쉽게 생성할 수 있는 팩토리 메서드 모음 (유틸리티 클래스)이다. 고정된 쓰레드 개수를 가진 풀, 작업량에 따라 늘어나는 풀 등 다양한 형태의 ExecutorService를 한 줄의 코드로 생성해준다.

newFixedThreadPool(int n)

• 고정된 개수(n)의 스레드를 가진 스레드 풀을 만들어준다.
• 작업이 많으면 큐에 대기한다.
• CPU 집약적 작업, 안정적인 성능 필요한 경우 사용한다.

newCachedThreadPool()

• 필요하면 새 스레드 생성하고 60초동안 작업이 없으면(idle이면) 제거한다.
• 스레드 수에 제한이 없다.
• 짧고 많은 비동기 작업, I/O 작업에 적합하다.
• 작업이 폭증하면 스레드를 무한 생성 가능하므로 Out Of Memory 에러가 발생할 수 있어 주의해야한다.

newSingleThreadExecutor()

• 스레드를 하나만 만들어 순서대로 작업을 처리한다.
• 순서 보장이 필요한 작업, 이벤트 처리에 적합하다.

예시 코드

public class Main {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        ExecutorService executor = Executors.newFixedThreadPool(10);

        int initValue = 10;

        List<CallableImpl> taskList = new ArrayList<>();
        for(int i=1; i<=initValue; i++){
            taskList.add(new CallableImpl(i));
        }

        System.out.println("작업 실행 중...");
        List<Future<String>> futures = executor.invokeAll(taskList);
        System.out.println("작업 종료...");
        System.out.println("결과출력...");
        for(Future<String> future: futures){
            System.out.println(future.get());
        }
        executor.shutdown();
    }

}

class CallableImpl implements Callable<String>{
    private final int idx;

    CallableImpl(int val) {
        this.idx = val;
    }

    @Override
    public String call() throws Exception {
        LocalTime startTime = LocalTime.now();
        System.out.println("Thread: " + Thread.currentThread().getName() + ", call idx: " + idx + ", startTime: " + startTime);
        LocalTime endTime = LocalTime.now();
        return String.valueOf("idx: " + idx + ", duration: " + Duration.between(startTime, endTime).toMillis());
    }
}

실무에서는 Executors.newCachedThreadPool()이나 newFixedThreadPool()을 사용할 때 주의해야 한다. 내부적으로 사용하는 작업 큐(Queue)의 크기가 무제한일 경우, 요청이 폭주하면 Out of Memory(OOM) 에러가 발생할 수 있다. 따라서 정교한 자원 관리가 필요한 운영 환경에서는 ThreadPoolExecutor 클래스를 직접 생성하여 쓰레드 개수와 큐 크기를 엄격하게 제한하는 것이 권장된다.

+) TIP
스레드 풀 크기가 너무 크면 생성된 스레드들이 cpu를 과도하게 점유하게 되어 과부하게 생길수 있다. 반면 스레드 풀 크기가 너무 작으면 스레드 풀 크기보다 더 많은 요청이 있을 경우 대기 상태에 들어가 응답이 지연된다. 또한 cpu 자원이 충분히 활용되지 못하여 낭비된다는 단점도 있다.

출처: https://mangkyu.tistory.com/259 [MangKyu's Diary:티스토리]

---

Thread와 Runnable은 반환값을 받을 수 없기 때문에 공유 자원을 사용하여 우회를 해야했다. 또한 예외를 던질수 없어서 무조건 run()내부에서 처리해야했다. 이런 이유 때문에 자바5 버전에는 반환 값을 가질 수 있는 새로운 인터페이스들이 등장했다. Callable은 멀티 스레드 환경에서 스레드의 실행 결과값을 받을 수 있는 인터페이스이다. 제네릭을 사용해서 반환값을 받을 수 있고 call() 메서드에서 checked exception을 던질 수 있다. Future는 비동기 작업의 결과를 나타내는 인터페이스이다. Callable 에서 구현한 작업은 별도 쓰레드에서 비동기적으로 실행되기 때문에 언제 완료되어 값을 반환받을지 알 수 없다. 실행결과를 바로 받을 수 없으므로 미래에 완료될 Callable 의 반환값을 얻기 위해 Future 가 사용된다. 비동기 작업이 생성되면 Future 객체가 반환된다. 비동기 작업이 완료되면 작업이 시작될 때 반환된 Future 객체를 통해 결과를 받아온다. 그 다음 Executor, ExecutorService, Executors가 있다. 이 인터페이스들은 작업과 실행을 분리하여 스레드 풀을 구현할 수 있다. Executor는 최상위 인터페이스로, 전달받은 작업을 실행하는 역할만 수행한다. 스레드 생성로직을 감추고 스레드 풀같은 최적화 전략을 사용할 수 있다. ExecutorService는 Executor를 확장하여 작업을 등록하고 실행까지 할 수 있다. 여러 작업을 비동기로 실행시켜 결과를 Future로 받을 수 있다. Executors는 ExecutorService를 쉽게 생성할수 있는 팩토리 메서드를 가지고 있다. 고정된 개수의 스레드 풀이나 작업량에 따라 늘어나는 풀을 생성할 수 있다.
Callable의 call() 메소드는 개발자가 설정한 제네릭 타입의 실제 결과값을 반환한다. 하지만 이 작업을 ExecutorService에 제출하면, 작업이 완료될 때까지 기다리지 않고 즉시 Future 객체를 반환받게 된다. 여기서 Future는 비동기 작업의 결과를 담는 '핸들' 혹은 '지연 완료 객체' 역할을 하며, 나중에 get() 메소드를 통해 call()이 반환한 실제 값을 꺼낼 수 있다