Callable과 Future로 작업 결과 받기

앞 글에서는 execute로 작업을 맡겼습니다. execute가 받는 Runnable은 run()의 반환 타입이 void라서 작업의 결과를 돌려줄 방법이 없습니다. 그래서 결과가 필요하면 작업이 공유 객체에 값을 써 두고, 맡긴 쪽이 종료를 기다린 뒤 그 값을 읽어야 했습니다. 작업에서 난 예외는 표준 오류에 출력되고 끝이었습니다.

이 글에서는 결과를 돌려주는 작업인 Callable을 만들어 submit으로 맡기고, submit이 돌려주는 Future로 결과와 예외를 받습니다. 여러 작업의 결과를 어떤 순서로 모을지, Future를 받은 쪽이 무엇을 챙겨야 하는지도 함께 봅니다.

끝나기 전에 결과를 읽는 문제

먼저 결과를 공유 필드로 전달하는 방식이 왜 위험한지 다시 확인해 보겠습니다. 다음 프로그램은 작업자 스레드가 필드 result에 42를 쓰고, main은 스레드를 시작하자마자 그 필드를 읽습니다.

public final class SharedResultWithoutJoin {
    private int result;

    void calculate() { result = 21 * 2; }

    public static void main(String[] args) {
        SharedResultWithoutJoin task = new SharedResultWithoutJoin();
        Thread.ofPlatform().start(task::calculate);
        System.out.println("result=" + task.result);
    }
}

result=0

이 글을 쓰며 20번 실행했을 때 17번은 result=0, 3번은 result=42가 나왔습니다. main이 필드를 읽는 순간 작업자가 아직 값을 쓰지 않았다면 기본값 0을 읽습니다. 그렇다고 42가 나온 실행이 올바른 것도 아닙니다. 작업이 끝나기를 기다리는 코드가 없으니 언제 읽게 될지는 운에 달려 있고, 가시성과 동기화에서 본 것처럼 작업자가 쓴 값이 main에 반드시 보인다고 할 수도 없습니다.

필드에 volatile을 붙이면 쓴 값이 보이지 않는 문제는 해결되지만, main이 작업자보다 먼저 읽는 순서 문제는 그대로 남습니다. 필요한 것은 작업이 끝날 때까지 기다리는 장치입니다. 스레드를 직접 다룬다면 join을 쓰고, 실행기에 맡긴 작업이라면 이 글에서 볼 Future를 씁니다.

Callable: 값을 돌려주는 작업

Callable<V> 인터페이스에는 V call() throws Exception 메서드 하나가 있습니다. Runnable의 run()과 비교하면 두 가지가 다릅니다. 반환 타입이 V이므로 작업의 결과를 돌려줄 수 있고, throws Exception이 선언돼 있으므로 검사 예외(checked exception)도 그대로 던질 수 있습니다.

검사 예외 쪽의 차이는 컴파일러가 바로 보여 줍니다. Thread.sleep은 검사 예외인 InterruptedException을 던지므로, 다음처럼 Runnable 람다 안에서 그냥 호출하면 컴파일되지 않습니다.

executor.execute(() -> Thread.sleep(100));

error: unreported exception InterruptedException; must be caught or declared to be thrown

반면 값을 돌려주는 람다는 Callable로 해석되므로, 다음 코드는 try-catch 없이도 컴파일됩니다. 람다가 값을 돌려주면 컴파일러는 submit(Callable)을 고르고, System.out.println(…) 호출처럼 값 없이 끝나는 람다라면 submit(Runnable)을 고릅니다. 예외를 던지기만 하는 람다는 양쪽에 모두 맞는데, 이때는 submit(Callable)이 선택됩니다.

Future<Integer> future = executor.submit(() -> {
    Thread.sleep(100);
    return 42;
});

submit은 작업을 실행기에 맡기고 곧바로 Future를 돌려줍니다. Future는 아직 끝나지 않았을 수도 있는 작업의 결과를 나중에 받을 수 있게 해 주는 객체입니다. 음식점에서 주문하면 음식 대신 진동벨을 받고 나중에 그것으로 음식을 받는 것과 비슷합니다. 다만 진동벨과 달리 Future는 완료를 알려 주지 않으므로 결과가 필요한 쪽이 get()을 호출해 기다려야 하고, 음식 대신 작업이 실패했다는 소식을 받을 수도 있습니다.

get()이 하는 일은 작업의 상태에 따라 다음과 같습니다.

get()으로 받은 결과는 공유 필드와 달리 안전하게 읽을 수 있습니다. Future의 공식 문서는 작업이 한 일이 get()이 돌아온 뒤의 코드보다 먼저 일어난다고(happens-before) 정해 두었기 때문입니다. 그래서 돌려받은 값은 물론이고, 작업이 다른 객체에 써 둔 값도 get() 다음에는 빠짐없이 보입니다.

결과와 예외를 받아 보기

다음 프로그램은 작업 두 개를 맡깁니다. 첫 작업은 주식 시세를 나타내는 Quote 레코드를 돌려주고, 두 번째 작업은 모르는 종목이라며 IllegalArgumentException을 던집니다. main은 두 Future에서 차례로 결과를 받습니다.

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public final class CallableResultDemo {
    record Quote(String symbol, int price) {}

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executor = Executors.newFixedThreadPool(2);
        Future<Quote> ok = executor.submit(() -> new Quote("JAVA", 25));
        Future<Quote> broken = executor.submit(() -> {
            throw new IllegalArgumentException("unknown symbol");
        });
        try {
            System.out.println(ok.get());
            broken.get();
        } catch (ExecutionException e) {
            System.out.println("cause=" + e.getCause().getClass().getSimpleName());
        } finally {
            executor.shutdownNow();
        }
    }
}

Quote[symbol=JAVA, price=25]
cause=IllegalArgumentException

두 submit은 작업을 맡기자마자 각각 Future를 돌려주고, 두 작업은 작업자 두 개에서 실행됩니다. ok.get()은 첫 작업이 돌려준 Quote를 받고, 레코드의 toString()이 첫 줄을 만듭니다. 두 번째 작업이 던진 IllegalArgumentException은 작업자 스레드에서 사라지지 않고 Future에 보관돼 있다가, broken.get()을 호출하는 순간 ExecutionException에 싸여 던져집니다. catch가 getCause()로 원래 예외를 꺼내 둘째 줄을 출력합니다.

두 줄의 순서는 실행할 때마다 같습니다. 두 번째 작업이 먼저 끝나더라도 main은 ok.get()부터 호출하기 때문입니다. main 선언에 throws InterruptedException이 붙은 이유는 get()이 기다리는 동안 인터럽트될 수 있어서입니다. finally의 shutdownNow()는 실행기 정리를 요청할 뿐이며, 이 호출만으로 실행기가 종료됐다고 확인할 수는 없습니다. 이 예제에서는 두 작업이 이미 끝났으므로 중단할 작업도 없습니다.

get()이 원래 예외를 그대로 던지지 않고 ExecutionException으로 감싸 주는 데에는 이점이 있습니다. get()은 자기 자신의 사정으로도 InterruptedException이나 CancellationException을 던집니다. 작업이 던진 예외를 그대로 내보내면, 작업 안에서 난 InterruptedException과 기다리던 쪽이 인터럽트된 경우를 구별할 수 없습니다. 감싸 두면 ExecutionException은 작업이 실패했다는 뜻, 나머지는 기다리는 쪽에 일이 생겼다는 뜻으로 깔끔하게 나뉩니다.

그래서 ExecutionException 자체만 기록하면 실제로 무엇이 실패했는지가 가려집니다. getCause()로 원인을 꺼내 종류에 따라 처리해야 합니다. 예를 들어 잘못된 입력은 그 항목만 실패로 처리하고, 외부 서비스 장애는 다시 시도할 후보로, 프로그램의 버그는 바로 알려야 할 문제로 나눌 수 있습니다.

끝난 것과 성공한 것은 다르다

Future에는 작업이 끝났는지 묻는 isDone()도 있습니다. 그런데 이 메서드는 작업이 값을 돌려주고 끝났을 때뿐 아니라, 예외를 던지고 끝났을 때와 취소됐을 때도 true를 돌려줍니다. 그러니 isDone()이 true라고 결과를 꺼내 써도 된다고 판단하면 안 되고, 결과는 여전히 get()으로 받으면서 ExecutionException과 CancellationException을 구별해야 합니다.

기다리는 시간을 제한하는 get(시간, 단위)도 있습니다. 정한 시간 안에 작업이 끝나지 않으면 TimeoutException을 던지는데, 이것은 이번 호출이 기다리기를 그만뒀다는 뜻일 뿐 작업을 취소하지는 않습니다. 실제로 20밀리초 만에 시간이 초과된 직후 확인해 보면 isCancelled()도 isDone()도 false이고, 작업이 끝난 뒤 같은 Future에서 get()을 다시 호출하면 결과를 정상적으로 받습니다. 시간이 초과되면 다시 기다릴지, 작업 취소를 요청할지는 호출한 쪽이 정해야 하며, 취소는 Future 취소와 시간 초과, 실패 구분하기에서 다룹니다.

더 깊이: Java 19에 추가된 state와 resultNow

Java 19부터는 Future에 결과를 기다리지 않고 상태를 확인하는 메서드가 생겼습니다. state()는 RUNNING, SUCCESS, FAILED, CANCELLED 가운데 하나를 돌려주므로 isDone()만으로는 알 수 없던 성공과 실패를 구별해 줍니다. resultNow()는 성공한 작업의 결과를, exceptionNow()는 실패한 작업의 예외를 기다리지 않고 꺼냅니다.

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public final class FutureStateDemo {
    public static void main(String[] args) {
        Future<Integer> ok;
        Future<Integer> failed;
        try (ExecutorService executor = Executors.newFixedThreadPool(2)) {
            ok = executor.submit(() -> 42);
            failed = executor.submit(() -> {
                throw new IllegalStateException("boom");
            });
        }
        System.out.println(ok.state() + " " + ok.resultNow());
        System.out.println(failed.state() + " " + failed.exceptionNow());
    }
}

SUCCESS 42
FAILED java.lang.IllegalStateException: boom

try 블록을 벗어날 때 close()가 두 작업이 끝나기를 기다렸으므로, 그 뒤의 두 Future는 이미 완료 상태입니다. exceptionNow()는 ExecutionException으로 감싸지 않은 원래 예외를 돌려줍니다. 다만 두 메서드는 상태가 맞지 않으면 기다리는 대신 IllegalStateException을 던집니다. 예를 들어 아직 실행 중이거나 실패한 작업에 resultNow()를 호출하면 예외가 나므로, 완료가 확실한 Future에만 쓰거나 state()를 먼저 확인해야 합니다.

여러 결과를 제출 순서대로 모으기

작업이 여러 개라면 Future를 목록에 모아 두었다가 차례로 get()을 호출하면 됩니다. 다음 프로그램은 2, 3, 4의 제곱을 각각 작업으로 맡기고 결과를 모읍니다.

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public final class CallableBatch {
    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(3);
        try {
            List<Future<Integer>> futures = new ArrayList<>();
            for (int value : List.of(2, 3, 4)) {
                futures.add(executor.submit(() -> value * value));
            }
            List<Integer> results = new ArrayList<>();
            for (Future<Integer> future : futures) {
                results.add(future.get());
            }
            System.out.println(results);
        } finally {
            executor.shutdown();
        }
    }
}

[4, 9, 16]

람다 안에서 value를 그대로 쓸 수 있는 이유는, 향상된 for 문의 변수가 반복할 때마다 새로 만들어지고 본문에서 바뀌지 않기 때문입니다. 앞 글에서 일반 for 문의 task를 int id = task;로 복사해야 했던 것과 다른 점입니다.

결과 목록은 작업이 끝난 순서와 상관없이 언제나 제출한 순서를 따릅니다. futures에는 2, 3, 4 순서로 Future가 들어 있고, 두 번째 반복문은 첫 Future부터 get()을 호출하기 때문입니다. 예를 들어 3의 제곱이 가장 먼저 끝나고 4, 2의 제곱이 차례로 끝났다고 가정하면 반복문은 다음과 같이 움직입니다.

이 표는 수집 순서를 설명하려는 가정이며 실제 완료 시각을 잰 것은 아닙니다. 9가 일찍 준비돼 있어도 첫 결과 4를 기다리는 동안에는 처리되지 않는다는 점이 중요합니다. 앞 작업 하나가 느리면 이미 끝난 뒤 작업의 결과까지 함께 늦어지는 것입니다.

입력 순서대로 결과를 내야 하는 기능이라면 이 방식이 맞습니다. 반대로 끝난 결과부터 처리하고 싶다면 ExecutorCompletionService를 써서 완료된 Future가 쌓이는 큐에서 take()로 꺼내거나, CompletableFuture로 각 결과에 이어서 할 일을 연결합니다. 둘 다 invokeAll과 CompletableFuture로 결과 모으기에서 다룹니다. 어느 순서를 고를지는 성능 기법이라기보다, 결과를 받는 쪽이 무엇을 먼저 볼 수 있어야 하는지에 관한 결정입니다.

submit의 반환값을 버리면 예외도 사라진다

submit으로 맡긴 작업이 예외를 던졌는데 아무도 get()을 호출하지 않으면 어떻게 될까요? 다음 프로그램은 예외를 던지는 작업 두 개를 맡기고, 두 번째 작업의 Future만 확인합니다.

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public final class SubmitHidesFailure {
    public static void main(String[] args) throws InterruptedException {
        try (ExecutorService executor = Executors.newSingleThreadExecutor()) {
            executor.submit(() -> {
                throw new IllegalStateException("lost");
            });
            Future<?> kept = executor.submit(() -> {
                throw new IllegalStateException("kept");
            });
            try {
                kept.get();
            } catch (ExecutionException e) {
                System.out.println("caught " + e.getCause());
            }
        }
    }
}

caught java.lang.IllegalStateException: kept

두 작업 모두 예외를 던졌지만 출력은 한 줄뿐입니다. 앞 글에서 execute로 맡긴 작업은 예외를 던지면 스택 트레이스라도 출력됐는데, submit으로 맡긴 첫 작업은 아무 흔적도 없습니다. ThreadPoolExecutor 같은 표준 실행기는 submit으로 받은 작업을 FutureTask로 감싸 실행하고, FutureTask는 작업이 던진 예외를 잡아 자기 안에 보관하기 때문입니다. 예외가 작업자 스레드까지 올라가지 않으니 JVM의 기본 처리기도 호출되지 않고, get()을 호출해야만 예외가 드러납니다.

결과가 필요 없는 작업이라도 submit이 돌려준 Future는 보관했다가 get()으로 실패를 확인해야 합니다. 값을 돌려주지 않는 작업의 Future는 성공하면 get()이 null을 돌려줍니다. 실패를 확인할 계획이 없다면 차라리 execute로 맡기고, 앞 글처럼 작업을 감싸 예외를 잡거나 스레드 팩터리에 잡히지 않은 예외의 처리기(UncaughtExceptionHandler)를 지정합니다.

Future를 받은 쪽이 챙길 일

Future를 돌려주는 메서드는 작업의 뒷정리를 호출한 쪽에 넘기는 셈입니다. 호출한 쪽은 결과를 언제 받을지, 결과가 더는 필요 없을 때 취소할지, 얼마나 기다릴지를 정해야 합니다. 그래서 Future<T>를 돌려주는 메서드를 만든다면, 결과를 반드시 받아야 하는지와 메서드가 끝난 뒤에도 작업이 계속 실행될 수 있는지를 문서에 적어 두는 것이 좋습니다.

결과를 기다리다 InterruptedException을 받았는데 그 예외를 위로 다시 던질 수 없는 자리라면, Thread.currentThread().interrupt()로 인터럽트 상태를 되살려 둡니다. interrupt 플래그와 InterruptedException 처리에서 본 것처럼 이 예외가 던져질 때 인터럽트 상태가 지워지기 때문입니다. 그다음 남은 Future에 취소를 요청하고 결과 모으기를 멈추는 것이 일반적인 처리입니다. 취소를 요청했다고 작업이 끝났다는 뜻은 아니라는 점도 기억해 두세요.

끝난 Future를 목록에 계속 들고 있으면 그 결과 객체도 메모리에서 해제되지 않습니다. 작업이 아주 많다면 결과를 처리하는 대로 목록에서 지우고, 마지막에 입력 개수가 성공·실패·취소 개수의 합과 맞는지 확인합니다. 그래야 빠진 결과를 정상 완료로 착각하지 않습니다.

정리

• Callable은 값을 돌려주고 검사 예외도 던질 수 있는 작업입니다. submit으로 맡기면 결과를 나중에 받을 Future를 곧바로 돌려받습니다.
• get()은 작업이 끝날 때까지 기다린 뒤 값을 돌려주고, 작업이 던진 예외는 ExecutionException으로 감싸 던집니다. 원래 예외는 getCause()로 꺼내 종류별로 처리합니다.
• isDone()은 성공뿐 아니라 실패와 취소에도 true입니다. get(시간, 단위)의 TimeoutException은 작업을 취소하지 않습니다.
• Future 목록을 차례로 get()하면 결과는 제출 순서를 따르고, 앞 작업이 느리면 뒤 결과도 늦어집니다.
• submit의 반환값을 버리면 작업의 예외가 아무 흔적 없이 사라집니다.

연습 문제

1. 문자열 세 개 "10", "bad", "30"을 각각 정수로 바꾸는 Callable을 맡기세요. 변환에 실패해도 예외를 밖으로 던지지 말고, 성공과 실패를 모두 표현하는 sealed 결과 타입으로 돌려줘 세 결과를 입력 순서대로 출력하세요.
2. 1번 풀이의 입력을 "10", "2147483648", "-4"로 바꾸면 세 결과는 무엇인가요? 범위를 넘는 값 때문에 get()이 ExecutionException을 던지나요? 둘째 작업이 가장 먼저 끝난다면 출력 순서가 바뀌나요?
3. 작업 하나가 IllegalStateException을 던지고 끝난 뒤 그 Future의 isDone()이 true인 것을 확인했습니다. 이제 결과 값을 꺼내 써도 되나요? get()은 무엇을 던지며, 같은 실행기에서 성공한 다른 작업의 결과에도 영향이 있나요?
4. 어떤 작업이 신호를 받을 때까지 기다렸다가 42를 돌려준다고 합시다. 신호를 보내기 전에 get(20, TimeUnit.MILLISECONDS)를 호출하면 무엇을 던지나요? 그 뒤 신호를 보내 작업이 끝났을 때, 같은 Future에 get()을 다시 호출하면 결과는 무엇인가요?

import java.util.List;
import java.util.concurrent.Executors;

public final class TypedParseResultSolution {
    sealed interface ParseResult permits Success, Failure {}
    record Success(int value) implements ParseResult {}
    record Failure(String input) implements ParseResult {}

    static ParseResult parse(String input) {
        try { return new Success(Integer.parseInt(input)); }
        catch (NumberFormatException e) { return new Failure(input); }
    }

    public static void main(String[] args) throws Exception {
        var executor = Executors.newFixedThreadPool(2);
        try {
            var futures = List.of("10", "bad", "30").stream()
                    .map(text -> executor.submit(() -> parse(text)))
                    .toList();
            for (var future : futures) {
                System.out.println(future.get());
            }
        } finally {
            executor.shutdown();
        }
    }
}

Success[value=10]
Failure[input=bad]
Success[value=30]

다음 글

Future.get을 부르는 시점과 병렬 합산에서는 submit과 get을 어떤 순서로 부르느냐에 따라 작업이 실제로 동시에 실행되는지가 갈린다는 것을 보고, 여러 결과를 하나의 기한 안에서 모으는 방법을 익힙니다.