Structured Concurrency

구조화된 동시성은 Java 25의 JEP 505 다섯 번째 프리뷰 API입니다.

StructuredTaskScope는 형제 하위 작업을 하나의 코드 범위에 묶고 소유 작업이 join한 뒤 함께 범위를 벗어나게 합니다.

JDK 24까지의 예제와 달리 Java 25 API는 public 생성자 대신 open() 팩터리와 Joiner를 사용합니다.

컴파일과 실행에는 모두 프리뷰 플래그가 필요합니다.

join 전 Subtask.get 호출

아래 하위 작업은 래치에서 기다리는데 소유자가 join() 전에 결과를 읽습니다.

소유자가 join() 전에 get()을 호출하므로 IllegalStateException 조건에 해당합니다. 래치는 하위 작업도 아직 완료하지 못하게 합니다.

preview/StructuredSubtaskBeforeJoinFailure.java

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());
            } finally {
                release.countDown();
                scope.join();
            }
        }
    }
}

소유자가 Subtask 결과를 읽는 조건
소유자의 상태	하위 작업 상태	get 결과
join 전	성공 여부와 무관	IllegalStateException
join 뒤	성공	결과 반환 · null 가능
join 뒤	미완료 또는 실패	IllegalStateException

join 전: 하위 작업 상태: 성공 여부와 무관
get 결과: IllegalStateException
join 뒤: 하위 작업 상태: 성공
get 결과: 결과 반환 · null 가능
join 뒤: 하위 작업 상태: 미완료 또는 실패
get 결과: IllegalStateException

실행 순서는 open → fork → join → 성공 결과 읽기 → close입니다.

범위의 소유권과 수명을 벗어나 하위 작업 핸들을 보관하지 않습니다.

javac --enable-preview --release 25 -Xlint:preview StructuredSubtaskBeforeJoinFailure.java
java --enable-preview StructuredSubtaskBeforeJoinFailure
# IllegalStateException: 소유자가 join 전에 get 호출

독립 조회의 요청 단위 결합

사용자와 게시글 조회가 모두 성공해야 응답을 만들 수 있다면 default 범위를 사용합니다.

하나가 실패하면 join()이 실패를 전파하고 나머지 작업을 취소하는 정책을 적용합니다.

preview/StructuredBoardResponse.java

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());
    }
}

try-with-resources는 close()를 호출합니다. 취소는 하위 작업에 인터럽트를 요청하며, close()는 작업이 실제로 끝날 때까지 기다립니다. 인터럽트에 반응하지 않는 작업은 종료를 지연시킬 수 있습니다.

fork()한 소유자가 join()하지 않고 닫으면 종료를 기다린 뒤 IllegalStateException을 던집니다. 범위의 잘못된 중첩도 별도로 검사합니다.

Joiner 완료 정책과 결과 형태

모든 결과가 필요한지, 성공 하나면 되는지, 단순 완료만 기다리는지에 따라 Joiner를 선택합니다.

아래는 두 미러 서버 중 먼저 성공한 값을 결과로 받습니다.

실패한 형제 작업의 처리와 취소는 Joiner 정책이 담당합니다.

preview/AnySuccessfulMirrorResult.java

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만 정상 결과를 만들 수 있습니다. sleep의 5ms·10ms가 실제 완료 순서를 보장하지는 않습니다.

두 작업이 모두 외부 부수 효과를 만들면 나머지 작업 취소만으로 이미 발생한 효과를 롤백할 수 없습니다.

구조화된 작업의 ScopedValue

Java 25의 정식 기능인 ScopedValue 바인딩은 범위를 open()할 때 포착됩니다. fork()는 현재 바인딩이 그때와 같은지 확인하고 하위 작업에 상속합니다.

가변 ThreadLocal 값을 복사하는 방식보다 상위·하위 작업의 코드 범위가 분명합니다.

preview/StructuredScopedContext.java

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);
    }
}

별도로 검증할 실패·취소 항목

위 정상 경로만으로 다음 항목을 검증한 것은 아닙니다.

하위 작업 예외가 join에서 어떤 래퍼와 원인으로 전파되는지 기록합니다.
형제 작업이 블로킹 호출에서 interrupt에 반응하는지 확인합니다.
상위 작업 interrupt가 범위와 하위 작업에 어떻게 전달되는지 봅니다.
시간 제한 구성의 기한 초과와 취소를 확인합니다. 이 기한은 close()의 종료 대기 상한이 아닙니다.
JFR·스레드 덤프에서 상위 작업과 하위 작업 관계가 어떻게 보이는지 관찰합니다.

프리뷰 API의 시그니처는 다음 릴리스에서 달라질 수 있으므로 래퍼 인터페이스 뒤에 실험 구현으로 둡니다.

운영 서비스의 public 시그니처에 StructuredTaskScope 타입을 직접 노출하지 않습니다.

StructuredTaskScope가 ExecutorService를 대체하나요?

아닙니다.

JEP의 목표도 모든 동시성 API를 대체하는 것이 아닙니다.

상위 요청의 코드 범위 안에서 관련 하위 작업을 병렬로 분기하고 join하는 문제에 적합합니다.

장기 백그라운드 큐, 예약 실행, 독립 작업 제출에는 기존 실행기나 메시징 시스템이 더 적합합니다.

연습 문제

기본 범위에서 집중 점수와 복습 점수를 fork하고 join 뒤 합계를 반환하세요.

하나가 실패하면 합계를 만들지 않아야 합니다.

해설 보기

preview/StructuredScoreSolution.java

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());
    }
}

컴파일·실행 모두 프리뷰 플래그가 필요합니다. 원문의 두 계산이 정상 완료하면 total=95이며, 이 main은 실패를 주입하지 않습니다.