구조화된 동시성과 StructuredTaskScope

게시판 화면 하나를 그리려면 사용자 정보와 게시글 요약이 모두 필요하다고 해 봅시다. 두 조회는 서로 기다릴 이유가 없으니 동시에 실행하면 응답이 빨라집니다. ExecutorService로 작업 실행하기와 Callable과 Future로 작업 결과 받기에서 배운 방법이면 두 작업을 제출하고 Future.get()으로 결과를 모으면 됩니다. 그런데 둘 중 하나가 실패하면 어떻게 될까요? 이 글은 그 질문에서 출발해, JDK 25의 프리뷰 API인 StructuredTaskScope가 무엇을 해결하는지 살펴봅니다.

ExecutorService로 짜면 생기는 문제

다음 프로그램은 두 조회를 가상 스레드 실행기에 제출합니다. 사용자 조회는 50밀리초 뒤에 실패하고, 요약 조회는 300밀리초 뒤에 성공하도록 만들었습니다. 이 프로그램은 프리뷰 기능을 쓰지 않으므로 옵션 없이 실행됩니다.

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

public final class UnstructuredBoardResponse {
    public static void main(String[] args) throws InterruptedException {
        try (ExecutorService executor = Executors.newVirtualThreadPerTaskExecutor()) {
            Future<String> user =
                    executor.submit(
                            () -> {
                                Thread.sleep(Duration.ofMillis(50));
                                throw new IllegalStateException("user service down");
                            });
            Future<String> summary =
                    executor.submit(
                            () -> {
                                Thread.sleep(Duration.ofMillis(300));
                                System.out.println("summary finished anyway");
                                return "summary";
                            });
            try {
                System.out.println(user.get() + " / " + summary.get());
            } catch (ExecutionException e) {
                System.out.println("request failed: " + e.getCause().getMessage());
            }
        }
        System.out.println("executor closed");
    }
}

user.get()이 예외로 끝나면 catch가 원인 메시지를 찍고, 요약 작업은 300밀리초를 쉰 뒤 스스로 summary finished anyway를 찍습니다. 마지막 줄은 실행기의 close()가 끝난 뒤에 찍힙니다.

request failed: user service down
summary finished anyway
executor closed

첫 줄은 user.get()이 50밀리초쯤에 실패를 알려 준 결과입니다. 이 시점에 요청은 이미 실패로 끝났으니 요약 조회의 결과는 쓸 데가 없습니다. 그런데 둘째 줄을 보면 요약 작업은 그 뒤로도 250밀리초 동안 계속 실행되어 끝까지 갔습니다. 실패한 쪽과 요약 작업 사이에 아무 연결이 없어서, 아무도 요약 작업에 그만두라고 알려 주지 않았기 때문입니다.

셋째 줄이 마지막에 나온 이유도 같습니다. try-with-resources가 호출하는 ExecutorService.close()는 제출된 작업이 모두 끝날 때까지 기다리므로, 이미 실패한 요청이 쓸모없는 작업을 300밀리초까지 기다린 것입니다. 서버처럼 실행기를 여러 요청이 함께 쓰고 요청마다 닫지 않는다면 상황은 더 나쁩니다. 그때는 아무도 기다리지 않는 작업이 백그라운드에서 혼자 계속 돌게 됩니다.

이 문제를 직접 고치려면 실패를 잡은 곳에서 summary.cancel(true)를 부르고, 반대로 요약 쪽이 먼저 실패하는 경우와 요청 스레드 자체가 인터럽트되는 경우도 따로 처리해야 합니다. 작업이 늘어날수록 이런 코드는 금방 복잡해지고 빠뜨리기 쉽습니다. 근본 원인은 '이 두 작업은 이 요청에 속한다'는 관계가 개발자의 머릿속에만 있고 코드에는 드러나지 않는다는 데 있습니다.

구조화된 동시성의 아이디어

구조화된 동시성(structured concurrency)은 그 관계를 코드의 구조로 표현합니다. 하나의 작업이 여러 하위 작업(subtask)으로 나뉘면, 하위 작업은 그것을 시작한 코드 블록 안에서 시작하고 그 블록을 벗어나기 전에 모두 끝나야 한다는 규칙입니다. 메서드를 호출하면 그 메서드가 끝나야 다음 줄로 넘어가듯이, 동시에 실행되는 작업에도 시작과 끝을 코드 블록에 맞추는 것입니다. 이 규칙이 지켜지면 하위 작업이 블록 밖으로 새어 나가 혼자 실행되는 일이 없고, 실패와 취소도 블록 단위로 한꺼번에 처리할 수 있습니다.

JDK 25에서 이 개념을 구현한 API가 java.util.concurrent.StructuredTaskScope입니다. JDK 21에서 처음 프리뷰로 나왔고, JDK 25의 JEP 505가 다섯 번째 프리뷰입니다. JDK 25에서 API가 크게 바뀌어서, new StructuredTaskScope.ShutdownOnFailure()처럼 생성자로 범위를 만들던 JDK 24까지의 코드는 더 이상 컴파일되지 않습니다. 대신 정적 팩터리 메서드 open()으로 범위를 열고, 완료 정책은 Joiner로 정합니다.

프리뷰 API이므로 이 글의 예제는 앞 글에서 본 것처럼 컴파일과 실행 모두에 --enable-preview가 필요하고, JDK 25에서만 컴파일됩니다. 예를 들어 다음 절의 예제는 이렇게 컴파일하고 실행합니다. 이 글의 JDK 25 예제 출력은 API 문서에 정해진 동작을 기준으로 적었습니다.

javac --enable-preview --release 25 -Xlint:preview StructuredBoardResponse.java
java --enable-preview StructuredBoardResponse

open, fork, join, close의 순서

앞의 게시판 조회를 StructuredTaskScope로 다시 써 보겠습니다. 두 조회가 모두 성공해야 응답을 만들 수 있으므로, 인수 없는 open()으로 범위를 엽니다. 이 범위는 모든 하위 작업이 성공하거나 하나라도 실패할 때까지 기다리고, 하나가 실패하면 나머지를 취소합니다.

import java.time.Duration;
import java.util.concurrent.StructuredTaskScope;

public final class StructuredBoardResponse {
    private record User(long id, String name) {}

    private record Summary(int posts, int comments) {}

    private record Response(User user, Summary summary) {}

    private static User loadUser() throws InterruptedException {
        Thread.sleep(Duration.ofMillis(5));
        return new User(7, "min");
    }

    private static Summary loadSummary() throws InterruptedException {
        Thread.sleep(Duration.ofMillis(5));
        return new Summary(3, 120);
    }

    static Response load() throws InterruptedException {
        try (var scope = StructuredTaskScope.open()) {
            StructuredTaskScope.Subtask<User> user = scope.fork(StructuredBoardResponse::loadUser);
            StructuredTaskScope.Subtask<Summary> summary =
                    scope.fork(StructuredBoardResponse::loadSummary);
            scope.join();
            return new Response(user.get(), summary.get());
        }
    }

    public static void main(String[] args) throws InterruptedException {
        System.out.println(load());
    }
}

두 조회가 모두 성공하므로 출력은 레코드의 toString 형식으로 된 한 줄입니다. Response 안에 든 User와 Summary도 레코드이므로, 각자의 toString 형식으로 그 안에 찍힙니다.

Response[user=User[id=7, name=min], summary=Summary[posts=3, comments=120]]

load() 안에서 일어나는 일을 순서대로 정리하면 다음과 같습니다. open()을 호출한 스레드가 이 범위의 소유자(owner)가 되고, join과 close는 소유자만 호출할 수 있습니다.

중요한 차이는 실패했을 때 드러납니다. 예를 들어 loadUser()가 예외를 던지면 범위가 취소되고, 아직 실행 중인 loadSummary()의 스레드에는 인터럽트가 걸립니다. Thread.sleep은 인터럽트에 반응하므로 요약 작업은 그 자리에서 InterruptedException으로 끝납니다. 소유자의 join()은 비검사 예외인 StructuredTaskScope.FailedException을 던지는데, 그 원인(cause)에는 가장 먼저 실패한 하위 작업의 예외가 들어 있습니다. 그리고 close()가 하위 작업 스레드가 모두 끝나기를 기다린 뒤에야 예외가 load() 밖으로 나가므로, 앞의 ExecutorService 예제처럼 혼자 남아 도는 작업이 생기지 않습니다.

다만 취소는 인터럽트로 요청하는 것이지 스레드를 강제로 멈추는 것이 아닙니다. close()는 하위 작업이 실제로 끝날 때까지 기다리므로, 인터럽트를 무시하고 계속 계산하는 작업이 있으면 그만큼 범위가 늦게 닫힙니다. 하위 작업 코드는 interrupt 플래그와 InterruptedException 처리에서 본 대로 인터럽트에 반응하도록 작성해야 합니다.

fork()만 하고 join()을 부르지 않은 채 범위를 닫으면, close()는 남은 하위 작업을 취소하고 끝나기를 기다린 뒤 IllegalStateException을 던집니다. 결과를 기다리지도 않을 작업을 시작했다면 코드에 실수가 있다는 뜻이기 때문입니다. 범위 안에서 다른 범위를 열 때 안쪽 범위보다 바깥쪽 범위를 먼저 닫는 것처럼 중첩 순서가 어긋나도 StructureViolationException이 납니다.

join 전에는 결과를 읽을 수 없다

Subtask.get()에는 지켜야 할 순서가 있습니다. 소유자는 join()을 마친 뒤에만 결과를 읽을 수 있습니다. 다음 코드는 이 순서를 일부러 어깁니다. 하위 작업은 래치(CountDownLatch)가 풀리기를 기다리고 있으므로 아직 끝나지 않았고, 그 사이에 소유자가 join() 없이 get()을 호출합니다.

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.StructuredTaskScope;

public final class StructuredSubtaskBeforeJoinFailure {
    public static void main(String[] args) throws Exception {
        CountDownLatch release = new CountDownLatch(1);
        try (var scope = StructuredTaskScope.<String>open()) {
            StructuredTaskScope.Subtask<String> subtask =
                    scope.fork(
                            () -> {
                                release.await();
                                return "done";
                            });
            try {
                System.out.println(subtask.get());
            } catch (IllegalStateException e) {
                System.out.println("get before join: IllegalStateException");
                throw e;
            } finally {
                release.countDown();
                scope.join();
            }
        }
    }
}

subtask.get()은 IllegalStateException을 던지므로 첫 줄에는 get before join: IllegalStateException이 출력되고, 같은 예외가 다시 던져져 main 밖으로 나가면서 프로그램이 비정상 종료합니다. finally에서 래치를 풀고 join()을 부르는 것은 하위 작업을 정상적으로 끝내고 범위를 닫기 위한 정리 작업입니다. 이 join()이 성공한다고 해서 앞에서 난 예외가 사라지지는 않습니다.

도표의 첫 행이 지금 본 경우입니다. 소유자가 join()하기 전이라면 하위 작업이 이미 성공했더라도 get()은 예외를 던집니다. join() 뒤에는 상태가 SUCCESS인 하위 작업에서만 결과를 읽을 수 있고, 실패했거나 아직 끝나지 않은 하위 작업에 get()을 부르면 역시 IllegalStateException이 납니다. 실패한 하위 작업의 예외는 get() 대신 exception()으로 읽습니다.

규칙이 이렇게 엄격한 이유는 join()을 기준으로 '여기부터는 모든 하위 작업이 끝났다'고 확실히 말할 수 있게 하려는 것입니다. 같은 이유로 Subtask 핸들을 필드에 저장하거나 범위 밖으로 넘겨 나중에 쓰는 것도 피해야 합니다. 핸들은 그것을 만든 범위 안에서만 의미가 있습니다.

Joiner로 완료 정책 고르기

인수 없는 open()은 '모두 성공해야 한다'는 정책을 씁니다. 하지만 성공한 결과 하나만 있으면 되는 경우도 있고, 실패가 있어도 모든 작업을 끝까지 기다려야 하는 경우도 있습니다. 이런 완료 정책은 open(Joiner)에 넘기는 Joiner 객체로 정합니다. Joiner가 제공하는 정적 팩터리 메서드는 다음과 같습니다.

다음 예제는 같은 데이터를 가진 두 미러 서버에 동시에 요청하고, 먼저 성공한 응답을 씁니다. primary는 5밀리초 뒤에 실패하고, secondary는 10밀리초 뒤에 성공합니다.

import java.time.Duration;
import java.util.concurrent.StructuredTaskScope;

public final class AnySuccessfulMirrorResult {
    private static String mirror(String name, long delayMillis, boolean fail) throws Exception {
        Thread.sleep(Duration.ofMillis(delayMillis));
        if (fail) {
            throw new IllegalStateException(name + " unavailable");
        }
        return name + "-result";
    }

    public static void main(String[] args) throws InterruptedException {
        var joiner = StructuredTaskScope.Joiner.<String>anySuccessfulResultOrThrow();
        try (var scope = StructuredTaskScope.open(joiner)) {
            scope.fork(() -> mirror("primary", 5, true));
            scope.fork(() -> mirror("secondary", 10, false));
            String result = scope.join();
            System.out.println(result);
        }
    }
}

출력은 secondary-result 한 줄입니다. 이 정책에서 join()은 Joiner가 만든 결과, 즉 성공한 하위 작업의 결과를 바로 돌려주므로 fork()가 돌려준 핸들을 따로 보관할 필요도 없습니다.

여기서 '먼저 끝난 것'과 '먼저 성공한 것'은 다르다는 점에 주의하세요. 먼저 끝나는 쪽은 primary지만 실패로 끝났으므로 결과가 될 수 없습니다. 또 sleep의 5밀리초와 10밀리초는 실제로 끝나는 순서를 보장하지 않습니다. 이 예제의 출력이 언제나 같은 이유는 순서 덕분이 아니라, 성공할 수 있는 작업이 secondary 하나뿐이기 때문입니다. 두 작업이 모두 실패하면 join()은 그중 한 작업의 예외를 원인으로 담은 FailedException을 던집니다.

취소에는 한계도 있습니다. 두 작업이 모두 데이터베이스에 기록하는 것처럼 바깥에 흔적을 남긴다면, 나머지 작업을 취소해도 이미 일어난 일은 되돌려지지 않습니다. 이런 작업을 경쟁시킬 때는 중복 실행되어도 괜찮도록 설계해야 합니다. 표에 없는 정책이 필요하면 Joiner 인터페이스를 직접 구현할 수 있습니다. 하위 작업이 시작될 때와 끝날 때 각각 onFork와 onComplete가 호출되고, 이 메서드가 true를 돌려주면 범위가 취소됩니다.

하위 작업에 ScopedValue 물려주기

요청 ID처럼 요청 하나에 속한 정보를 여러 계층에 전달할 때는 가상 스레드와 ScopedValue에서 본 ScopedValue를 씁니다. ScopedValue는 JDK 25에서 정식 기능이 되었고, StructuredTaskScope와 함께 쓰면 소유자 스레드의 바인딩이 하위 작업 스레드로 그대로 이어집니다.

import java.util.concurrent.StructuredTaskScope;

public final class StructuredScopedContext {
    private static final ScopedValue<String> REQUEST_ID = ScopedValue.newInstance();

    static String gather() throws InterruptedException {
        try (var scope = StructuredTaskScope.open()) {
            var first = scope.fork(() -> REQUEST_ID.get() + ":user");
            var second = scope.fork(() -> REQUEST_ID.get() + ":summary");
            scope.join();
            return first.get() + " | " + second.get();
        }
    }

    public static void main(String[] args) throws Exception {
        String result =
                ScopedValue.where(REQUEST_ID, "req-37").call(StructuredScopedContext::gather);
        System.out.println(result);
    }
}

main은 REQUEST_ID를 "req-37"에 묶은 채로 gather()를 호출합니다. 두 하위 작업은 새 가상 스레드에서 실행되지만 소유자의 바인딩을 물려받기 때문에, REQUEST_ID.get()이 둘 다 req-37을 돌려줍니다. 그래서 출력은 다음 한 줄입니다.

req-37:user | req-37:summary

바인딩은 범위를 open()할 때 기록됩니다. fork()는 그 순간의 바인딩이 open() 때와 같은지 확인하고, 다르면 StructureViolationException을 던집니다. 예를 들어 범위를 연 뒤에 ScopedValue.where(...)로 새 바인딩을 만들고 그 안에서 fork()를 부르면, 하위 작업이 범위와 다른 문맥을 보게 되므로 거부됩니다. ScopedValue에는 값을 바꾸는 메서드가 없어서 한 하위 작업이 다른 작업이 보는 값을 바꿀 수도 없으므로, 어느 하위 작업이 어떤 값을 보는지가 코드 블록만 보고도 분명합니다. 값이 복사된 뒤 각 스레드에서 따로 바뀔 수 있는 InheritableThreadLocal과 다른 점입니다.

ScopedValue는 정식 기능이지만, 이 프로그램은 StructuredTaskScope를 쓰므로 여전히 --enable-preview가 필요합니다.

시간 제한 걸기

외부 서비스를 부르는 작업이라면 무한정 기다릴 수는 없습니다. open()의 두 번째 인수로 설정을 바꾸는 함수를 넘기면 범위에 시간 제한을 걸 수 있습니다.

try (var scope = StructuredTaskScope.open(
        StructuredTaskScope.Joiner.<String>allSuccessfulOrThrow(),
        config -> config.withTimeout(Duration.ofSeconds(2)))) {
    // fork와 join은 앞의 예제와 같습니다
}

시간은 범위를 연 순간부터 잽니다. join()이 끝나기 전에 시간이 다 되면 범위가 취소되어 하위 작업 스레드에 인터럽트가 걸리고, join()은 StructuredTaskScope.TimeoutException을 던집니다. 이 시간 제한은 join()이 기다리는 시간의 상한이지, close()가 하위 작업이 끝나기를 기다리는 시간의 상한은 아닙니다. 인터럽트에 반응하지 않는 하위 작업이 있으면 시간이 지난 뒤에도 close()는 그 작업이 끝날 때까지 기다립니다.

실패와 취소를 직접 확인해 보기

이 글의 예제는 대부분 성공하는 경우만 보여 줍니다. JDK 25가 있다면 다음 실험으로 실패와 취소가 실제로 어떻게 동작하는지 확인해 보세요.

• 하위 작업 하나가 예외를 던지게 하고, join()이 던진 FailedException의 getCause()가 그 예외인지 확인합니다.
• 형제 작업이 sleep 같은 블로킹 호출 중에 인터럽트를 받고 바로 끝나는지 확인합니다.
• 소유자 스레드가 join()에서 기다리는 동안 인터럽트되면 어떻게 되는지 확인합니다.
• 시간 제한을 짧게 걸어 TimeoutException과 취소가 일어나는지 확인합니다.
• JSON 형식의 스레드 덤프(jcmd 명령의 Thread.dump_to_file -format=json)에서 범위와 하위 작업 스레드가 어떻게 보이는지 확인합니다.

더 깊이: 프리뷰 API를 감싸 두는 이유

StructuredTaskScope는 이미 한 번 API가 크게 바뀌었고, 다음 프리뷰에서도 시그니처가 달라질 수 있습니다. 그래서 실험 코드에서도 StructuredTaskScope를 직접 쓰는 부분을 작은 클래스 하나에 모으고, 나머지 코드는 직접 정의한 인터페이스로 그 클래스를 부르게 하는 편이 좋습니다. 특히 운영 서비스의 public 메서드 시그니처에 StructuredTaskScope나 Subtask 타입이 드러나면, API가 바뀔 때 그 메서드를 쓰는 모든 코드를 고쳐야 합니다. 이런 감싸기와 프리뷰 실험을 정리하는 방법은 프리뷰 실험을 이어 갈지 끝낼지 정하기에서 다룹니다.

정리

• ExecutorService와 Future만으로는 '이 작업들이 이 요청에 속한다'는 관계가 코드에 드러나지 않아, 하나가 실패해도 나머지가 계속 실행됩니다.
• 구조화된 동시성은 하위 작업의 시작과 끝을 코드 블록에 맞춥니다. JDK 25에서는 open(), fork(), join(), 결과 읽기, close() 순서로 씁니다.
• 소유자는 join()을 마친 뒤에만 성공한 하위 작업의 결과를 get()으로 읽을 수 있습니다.
• 완료 정책은 Joiner로 고르고, 취소는 인터럽트로 이루어지므로 하위 작업은 인터럽트에 반응해야 합니다.
• ScopedValue 바인딩은 하위 작업으로 이어지며, fork() 시점의 바인딩은 open() 때와 같아야 합니다.

연습 문제

기본 범위에서 집중 점수와 복습 점수를 각각 하위 작업으로 fork하고, join 뒤에 두 점수의 합계를 돌려주는 total() 메서드를 작성하세요. 둘 중 하나라도 실패하면 합계를 만들지 않아야 합니다.

집중 점수는 70점, 복습 점수는 25점입니다. 먼저 합계를 손으로 계산하고, 두 값을 20점과 35점으로 바꾼 경우의 합계도 구해 보세요. 이어서 복습 점수 계산이 예외를 던지면 합계를 계산하는 문장까지 실행이 도달하는지 예상해 보세요.

import java.util.concurrent.StructuredTaskScope;

public final class StructuredScoreSolution {
    private static int focusScore() {
        return 70;
    }

    private static int reviewScore() {
        return 25;
    }

    static int total() throws InterruptedException {
        try (var scope = StructuredTaskScope.open()) {
            var focus = scope.fork(StructuredScoreSolution::focusScore);
            var review = scope.fork(StructuredScoreSolution::reviewScore);
            scope.join();
            return focus.get() + review.get();
        }
    }

    public static void main(String[] args) throws InterruptedException {
        System.out.println("total=" + total());
    }
}

다음 글

기본형 패턴과 switch에서는 JDK 25의 또 다른 프리뷰 기능인 기본형 패턴으로, 값이 정보 손실 없이 더 작은 타입에 들어가는지 검사하고 long, double, boolean 값으로 switch를 쓰는 방법을 봅니다.