[Java] Callable, Feature 이해 및 사용예시

Callable 이해하기

[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을 던질 수 있다.

[Callable 인터페이스 정의]

Java의 Callable 인터페이스는 매우 간단하다. 단 하나의 메서드인 call()을 포함하고 있으며, 인터페이스의 구조는 다음과 같다.

 

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

 

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

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

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

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

Feature 이해하기

[Feature 인터페이스란?]

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

 

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

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

[Featue 인터페이스 정의]

Future 인터페이스에는 작업의 상태를 관리하고 결과를 가져오거나 취소할 수 있는 메서드들이 존재한다.

package java.util.concurrent;

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을 던짐
• get(long timeout, TimeUnit unit)
  • timeout까지 기다렸다가 결과 반환 (초과 시 TimeoutException 발생)

• isDone(),
  • 작업이 완료되었는지 여부를 확인
  • 작업이 취소되어 끝난 경우, 정상적으로 완료된 경우, 예외로 종료된 경우
•  isCancelled()
  • cancel()이 호출된 후 작업이 성공적으로 취소되었는지 확인
  • 작업이 아직 실행 중이거나 정상 종료되었을 경우 false 반환

• cancel(boolean mayInterruptIfRunning)
  • 작업을  취소하려고 시도
  • 작업이 이미 완료되었거나 취소된 경우 또는 다른 이유로 인해 취소할 수 없는 경우 아무 동작도 하지 않음
  • 작업이 이미 실행 중인 경우, mayInterruptIfRunning 매개변수는 현재 실행 중인 스레드를 인터럽트할지 여부를 결정
    • true일 경우: 작업을 실행하는 스레드가 인터럽트됨 (스레드를 중단하려고 시도)
    • false일 경우: 진행 중인 작업은 그대로 두고 완료될 때까지 실행
  • 작업이 실행되기 전 cancel 을 호출하면 해당 작업은 실행되지 않는다.
  • 반환 값은 작업이 실제로 취소되었는지를 보장하지 않기 때문에 작업이 취소되었는지 확인하려면 isCancelled()을 사용

해당 내용은 테스트를 통해 확인해보았다.

class FutureTest {

    ExecutorService executorService;

    @BeforeEach
    void setUp() {
        executorService = Executors.newSingleThreadExecutor();
    }

    @AfterEach
    void tearDown() {
        executorService.shutdown();
    }

    @Test
    void 작업이_지정된_시간_이내에_완료되지_않으면_TimeoutException이_발생한다() {
        ExecutorService executor = Executors.newSingleThreadExecutor();

        Future<String> future = executor.submit(() -> {
            Thread.sleep(5000);
            return "작업 완료";
        });

        assertThatThrownBy(() -> future.get(2, TimeUnit.SECONDS))
                .isInstanceOf(TimeoutException.class);

        executor.shutdown();
    }

    @Test
    void 취소된_작업에서_get을_호출하면_CancellationException이_발생한다() {
        Future<String> future = executorService.submit(callable());
        future.cancel(true);

        assertThatThrownBy(future::get)
                .isInstanceOf(CancellationException.class);
    }

    @Test
    void 초기_상태에서는_isCancelled가_false를_반환한다() {
        Future<String> future = executorService.submit(callable());

        assertThat(future.isCancelled()).isFalse();
    }

    @Test
    void 작업을_취소한_경우_isCancelled가_true를_반환한다() {
        Future<String> future = executorService.submit(callable());
        future.cancel(true);

        assertThat(future.isCancelled()).isTrue(); // 취소 후 isCancelled()는 true
    }

    @Test
    void 작업이_진행중이면_isDone이_false를_반환한다() {
        Future<String> future = executorService.submit(callable());
        assertThat(future.isDone()).isFalse(); // 실행 중이므로 false
    }

    @Test
    void 작업이_완료된_경우_isDone이_true를_반환한다() throws ExecutionException, InterruptedException {
        Future<String> future = executorService.submit(callable());
        future.get(); // 완료될 때까지 대기
        assertThat(future.isDone()).isTrue(); // 완료된 경우 true
    }

    @Test
    void 작업이_시작되기_전에_cancel_true를_호출하면_작업이_실행되지_않는다() throws InterruptedException {
        AtomicBoolean isRunning = new AtomicBoolean(false);

        Future<String> future = executorService.submit(() -> {
            isRunning.set(true); // 실행되었는지 확인용
            Thread.sleep(3000);
            return "Executed";
        });

        boolean isCanceled = future.cancel(true);

        Thread.sleep(500); // 작업이 실행되지 않았음을 보장하기 위해 대기
        assertThat(isCanceled).isTrue();
        assertThat(future.isCancelled()).isTrue();
        assertThat(future.isDone()).isTrue();
        assertThat(isRunning.get()).isFalse(); // 실행되지 않았음을 확인
    }

    @Test
    void 작업이_시작되기_전에_cancel_false를_호출하면_작업이_실행되지_않는다() throws InterruptedException {
        AtomicBoolean isRunning = new AtomicBoolean(false);

        Future<String> future = executorService.submit(() -> {
            isRunning.set(true); // 실행되었는지 확인용
            Thread.sleep(3000);
            return "Executed";
        });

        boolean isCanceled = future.cancel(false);

        Thread.sleep(500); // 작업이 실행되지 않았음을 보장하기 위해 대기
        assertThat(isCanceled).isTrue();
        assertThat(future.isCancelled()).isTrue();
        assertThat(future.isDone()).isTrue();
        assertThat(isRunning.get()).isFalse(); // 실행되지 않았음을 확인
    }

    @Test
    void 실행_중인_작업에_cancel_true를_호출하면_인터럽트된다() throws InterruptedException {
        AtomicBoolean isRunning = new AtomicBoolean(false);
        AtomicBoolean wasInterrupted = new AtomicBoolean(false);

        Future<String> future = executorService.submit(() -> {
            try {
                isRunning.set(true);
                while (!Thread.currentThread().isInterrupted()) {
                    Thread.sleep(200);
                }
            } catch (InterruptedException e) {
                wasInterrupted.set(true); // 인터럽트 감지
            }
            return "Executed";
        });

        Thread.sleep(500); // 작업이 시작되도록 대기
        future.cancel(true);
        Thread.sleep(500);

        assertThat(isRunning.get()).isTrue(); // 작업이 실행되었음을 확인
        assertThat(wasInterrupted.get()).isTrue(); // 인터럽트가 발생했는지 확인
        assertThat(future.isDone()).isTrue();
        assertThat(future.isCancelled()).isTrue();
    }

    @Test
    void 실행_중인_작업에_cancel_false를_호출하면_작업이완료될때까지_실행된다() throws InterruptedException {
        AtomicBoolean isRunning = new AtomicBoolean(false);
        AtomicBoolean wasInterrupted = new AtomicBoolean(false);

        Future<String> future = executorService.submit(() -> {
            try {
                isRunning.set(true);
                Thread.sleep(200);
            } catch (InterruptedException e) {
                wasInterrupted.set(true);
            }
            return "Executed";
        });

        Thread.sleep(500); // 작업이 시작되도록 대기
        future.cancel(false); // 실행 중이지만 취소 시도 (인터럽트 없음)
        Thread.sleep(500);

        assertThat(isRunning.get()).isTrue();
        assertThat(wasInterrupted.get()).isFalse();
        assertThat(future.isDone()).isTrue();
        assertThat(future.isCancelled()).isFalse();
    }

    Callable<String> callable() {
        return () -> {
            Thread.sleep(200);
            return "작업 완료";
        };
    }
}

 

새롭게 알게 된 내용으로, cancel(false) 의 경우 실행중인 작업을 취소하지 않고 완료될 때까지 실행하므로, 실행중인 작업에 cancel(true) 를 호출했다 하더라도 isCancelled() 가 false를 반환한다. cancel(false) 는 작업이 시작되기 전에 취소해야 isCancelled() 가 true 를 반환한다.

Callable 과 Feature 사용 예시

public class CallableSample implements Callable<Integer> {
    private final int value;

    public CallableSample(int value) {
        this.value = value;
    }

    @Override
    public Integer call() throws Exception {
        Thread.sleep(2000); // 2초 동안 대기
        return value * 100;
    }
}

Callable을 구현하려면 call() 메서드를 오버라이드하면 된다. 2초 동안 대기한 후, 전달받은 값에 100 을 곱한 값을 반환하도록 구현했다.

ExecutorService executor = Executors.newFixedThreadPool(2);

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

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

 

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

 

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

int initValue = 10;

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