JEP 543은 여러 스레드에서 실행되는 연관 작업을 하나의 작업 단위로 다루는 구조적 동시성(Structured Concurrency) API를 JDK 28에서 정식 기능으로 확정하자고 제안한다. 이 API는 하위 작업의 오류 처리와 취소를 자동화하고, 스레드 덤프에서 작업 간 관계를 확인할 수 있게 한다.
제안 배경과 적용 범위
구조적 동시성은 JEP 428을 통해 JDK 19에서 인큐베이터 모듈로 처음 제공됐고, JEP 437을 통해 JDK 20에서도 제공됐다. 이후 JEP 453으로 JDK 21에서 프리뷰 API가 됐으며, 이때 fork 메서드는 Future 대신 Subtask를 반환하도록 바뀌었다. 이 형태는 JEP 462(JDK 22), JEP 480(JDK 23), JEP 499(JDK 24)에서 다시 프리뷰됐다.
JEP 505(JDK 25)는 API를 몇 가지 바꾸고, 특히 StructuredTaskScope의 공개 생성자를 정적 팩토리 메서드로 대체했다. JEP 525(JDK 26)는 소폭 변경을, JEP 533(JDK 27)는 join()이 던지는 예외의 타입 매개변수 추가를 반영했다. 이번 제안은 추가 변경 없이 JDK 28에서 정식화하는 것이다.
목표는 취소와 종료 과정에서 발생하는 스레드 누수나 취소 지연 같은 위험을 줄이는 동시성 프로그래밍 방식을 지원하고, 동시 실행 코드의 관찰 가능성을 높이는 것이다. 기존 java.util.concurrent 패키지의 ExecutorService나 Future를 대체하거나, 모든 자바 프로그램에 적용할 단일 구조적 동시성 API를 만들려는 것은 아니다. 데이터 스트림을 스레드 간에 공유하는 채널이나 기존 스레드 인터럽트를 대체할 새로운 취소 메커니즘도 이번 목표에 포함되지 않는다. 이런 기능은 향후 제안될 수 있다.
비구조적 동시성에서 생기는 문제
작업을 서로 독립적인 하위 작업으로 나누고 동시에 실행하면 지연 시간을 줄일 수 있다. 예를 들어 여러 입출력 작업의 결과를 조합하는 일은 각 입출력 작업을 별도 스레드에서 실행하면 빨라질 수 있다. JEP 444의 가상 스레드는 입출력 작업마다 스레드를 두는 비용을 낮추지만, 늘어난 스레드를 관리하는 문제는 남는다.
자바 5에 도입된 ExecutorService는 하위 작업을 동시에 실행할 수 있다. 다음 handle()은 사용자 정보와 주문 정보를 각각 가져온 뒤 결과를 합친다. submit은 각 작업의 Future를 반환하고, get()은 결과를 기다린다.
Response handle() throws ExecutionException, InterruptedException {
Future<String> user = executor.submit(() -> findUser());
Future<Integer> order = executor.submit(() -> fetchOrder());
String theUser = user.get(); // Join findUser
int theOrder = order.get(); // Join fetchOrder
return new Response(theUser, theOrder);
}
두 하위 작업은 각각 성공하거나 예외를 던질 수 있다. 하지만 한 작업이 실패해도 다른 작업의 스레드가 계속 실행될 수 있다. findUser()가 예외를 던지면 user.get()에서 handle()이 실패하지만 fetchOrder()는 계속 돌아 스레드 누수를 일으킬 수 있다. handle() 실행 스레드가 인터럽트돼도 그 인터럽트가 하위 작업에 전파되지 않아 두 스레드가 계속 실행될 수 있다. 또 findUser()가 오래 걸리는 동안 fetchOrder()가 실패하면, 코드는 먼저 user.get()에서 기다리므로 실패를 늦게 처리하고 불필요하게 기다릴 수 있다.
이 문제는 프로그램에는 작업과 하위 작업의 논리적 관계가 있지만, ExecutorService와 Future가 이를 추적하거나 강제하지 않는 데서 비롯된다. 작업을 제출한 스레드와 결과를 기다리는 스레드가 달라도 되고, Future를 참조하는 어느 코드든 get()을 호출할 수 있다. 따라서 한 작업의 실패를 다른 작업의 자동 취소로 연결할 수 없다. 취소를 직접 구현하려면 try-finally에서 다른 Future의 cancel(boolean)을 호출하고, 취소된 작업이 끝날 때까지 기다릴 수 있도록 try-with-resources로 ExecutorService를 관리하는 방법 등을 고려해야 한다. 작업 간 관계를 수동으로 관리하면 실수하기 쉽고 코드의 의도도 흐려질 수 있다.
일반적인 단일 스레드 코드에서는 메서드가 호출한 하위 메서드가 호출자에게 돌아오거나 예외를 던져야 한다. 하위 작업은 부모 작업보다 오래 살아남지 않으며, 호출 스택에 작업의 계층과 실행 위치가 드러난다. 구조적 동시성은 이 관계를 여러 스레드에도 적용해, 코드 블록 구조가 하위 작업의 생명주기를 정하고 실행 중에도 부모·자식 관계를 표현하도록 한다. 그러면 오류 전파와 취소를 자동화하고 관찰 도구에서 작업의 계층을 확인할 수 있다.
구조적 동시성이라는 용어는 마틴 수스트리크(Martin Sústrik)가 만들었고, 네이셔니얼 J. 스미스(Nathaniel J. Smith)가 널리 알렸다. 오류 처리 설계에는 얼랭의 계층형 감독자 같은 다른 언어의 아이디어도 반영됐다. 자바 플랫폼에는 병렬 스트림의 실행 엔진인 ForkJoinPool도 있지만, 이 API는 입출력 작업보다는 계산 집약적 작업을 위해 설계됐다.
`StructuredTaskScope` 사용
핵심 API는 java.util.concurrent 패키지의 StructuredTaskScope다. 스코프 안에서 하위 작업을 각각 fork로 시작하고, join으로 한꺼번에 기다린다. 작업의 시작, 결과 대기, 오류 처리, 결과 조합은 명확한 어휘적 범위 안에서 이뤄진다.
앞선 예제를 다음과 같이 바꾸면 두 하위 작업의 생명주기는 try 블록으로 제한된다.
Response handle() throws ExecutionException, InterruptedException {
try (var scope = StructuredTaskScope.open()) {
Subtask<String> user = scope.fork(() -> findUser());
Subtask<Integer> order = scope.fork(() -> fetchOrder());
scope.join(); // Join subtasks, propagating exceptions
// Both subtasks have succeeded, so compose their results
return new Response(user.get(), order.get());
}
}
한 하위 작업이 예외로 실패하면 아직 끝나지 않은 다른 작업을 취소한다. handle() 실행 스레드가 join() 전이나 도중 인터럽트돼도 스코프를 벗어날 때 하위 작업을 취소한다. 스레드 덤프에는 findUser()와 fetchOrder()를 실행하는 스레드가 스코프의 자식으로 표시된다.
StructuredTaskScope에는 하위 작업 결과 타입 T, join()의 반환 타입 R, join()이 던질 예외 타입 R_X라는 세 타입 매개변수가 있다.
public sealed interface StructuredTaskScope<T, R, R_X extends Throwable>
extends AutoCloseable
{
public static <T> StructuredTaskScope<T, Void, ExecutionException> open();
public static <T, R, R_X extends Throwable> StructuredTaskScope<T, R, R_X> open(
Joiner<? super T, ? extends R, R_X> joiner);
public <U extends T> Subtask<U> fork(Callable<? extends U> task);
public Subtask<? extends T> fork(Runnable task);
public R join() throws R_X, InterruptedException;
public void close();
}
일반적인 사용 흐름은 정적 open 메서드로 스코프를 열고, fork로 하위 작업을 시작한 다음 join으로 모두 기다리는 것이다. 결과를 처리한 뒤 스코프를 닫는다. 보통 try-with-resources가 닫기를 맡는다. 스코프가 아직 취소되지 않았다면 닫을 때 취소하고, 남은 하위 작업을 취소한 뒤 종료될 때까지 기다린다.
매개변수가 없는 open()은 하위 작업 하나라도 실패하면 실패하는 기본 완료 정책을 사용한다. 다른 정책은 Joiner를 전달해 선택할 수 있다. fork를 호출할 때마다 기본적으로 가상 스레드가 하위 작업을 실행한다. 하위 작업은 다시 스코프를 열어 하위 작업을 만들 수 있다. 스코프가 닫히면 모든 하위 작업 스레드가 종료되므로 블록을 벗어난 뒤 남는 스레드가 없다.
join()은 스코프 소유 스레드가 스코프 안에서 호출해야 한다. join() 전에 블록이 끝나면 스코프가 취소되고, 소유 스레드는 close()에서 하위 작업이 모두 끝날 때까지 기다린 뒤 예외를 던진다. fork가 반환한 Subtask의 get()은 join() 전에 호출하면 예외를 던진다.
소유 스레드가 join() 전이나 도중 인터럽트되면 join()은 예외를 던진다. try-with-resources가 스코프를 취소하고 하위 작업이 종료될 때까지 기다리므로 취소가 하위 작업에 전파된다. 따라서 하위 작업은 인터럽트에 가능한 한 빨리 반응하도록 작성해야 한다. 인터럽트할 수 없는 메서드에서 대기하는 등 인터럽트에 반응하지 않는 하위 작업은 스코프 종료를 무기한 늦출 수 있다. close()는 취소 여부와 상관없이 하위 작업 스레드가 끝날 때까지 기다린다.
스코프에서 시작한 하위 작업은 바인딩된 ScopedValue의 값을 상속한다. 소유 스레드가 바인딩된 값을 읽으면 각 하위 작업도 같은 값을 읽는다. 런타임은 구조적 사용도 검사한다. 소유자가 아닌 스레드가 fork를 호출하면 예외가 발생하며, try-with-resources 밖에서 스코프를 사용하거나 close() 호출의 중첩 순서를 어기면 메서드가 StructureViolationException을 던질 수 있다. StructuredTaskScope는 비구조적 사용이 흔한 ExecutorService나 Executor를 구현하지 않지만, 구조가 필요한 기존 코드를 이 API로 옮길 수 있다.
완료 정책과 결과 처리
기본 정책에서는 하위 작업 하나라도 실패하면 join()이 예외를 던지고 스코프를 취소한다. 모두 성공하면 join()은 정상 반환하며 결과는 null이다. 다른 정책은 StructuredTaskScope.Joiner로 지정할 수 있다. 조인에 따라 join()은 결과, 목록 또는 다른 객체를 반환할 수 있고, 실패 때 던지는 예외도 달라진다.
anySuccessfulOrThrow()는 성공한 하위 작업 하나의 결과를 반환하고 모두 실패하면 ExecutionException을 던진다. 하나가 성공하는 즉시 스코프를 취소해 미완료 작업을 취소하므로, 여러 중복 서비스 가운데 하나의 결과만 필요한 서버 작업에 쓸 수 있다.
<T> T race(Collection<Callable<T>> tasks)
throws ExecutionException, InterruptedException
{
try (var scope = StructuredTaskScope.open(Joiner.<T>anySuccessfulOrThrow())) {
tasks.forEach(scope::fork);
return scope.join();
}
}
allSuccessfulOrThrow()는 모든 하위 작업이 성공하면 결과 목록을 반환한다. 하나 이상 실패하면 실패한 작업 중 하나의 예외를 원인으로 담은 ExecutionException을 던진다.
<T> List<T> runConcurrently(Collection<Callable<T>> tasks)
throws ExecutionException, InterruptedException {
try (var scope = StructuredTaskScope.open(Joiner.<T>allSuccessfulOrThrow())) {
tasks.forEach(scope::fork);
return scope.join();
}
}
이 정책은 매개변수 없는 open()의 기본 정책과 완료 조건은 같지만, join()이 null 대신 결과 목록을 반환한다. 같은 타입의 결과를 모두 얻고 fork가 반환한 Subtask 객체는 쓰지 않는 경우에 적합하다.
그 밖에도 awaitAllSuccessfulOrThrow()는 모든 작업이 성공적으로 끝날 때까지 기다린다. anySuccessfulOrThrow(Function), allSuccessfulOrThrow(Function), awaitAllSuccessfulOrThrow(Function)은 각각 대응하는 매개변수 없는 팩토리 메서드와 같지만, 실패 결과일 때 주어진 예외 공급 함수가 만든 예외를 join()이 던진다. allUntil(Predicate<? super Subtask<T>> isDone)은 모든 작업이 끝나거나 완료된 작업에 대한 조건식이 true가 되면 스코프를 취소하고 모든 하위 작업 목록을 반환한다. Joiner는 스코프마다 새로 만들어야 하며, 다른 스코프에서 재사용하거나 스코프를 닫은 뒤 다시 사용해서는 안 된다.
Joiner를 직접 구현하면 사용자 정의 완료 정책을 만들 수 있다. T는 하위 작업의 결과 타입, R은 join()의 반환 타입, R_X는 그 메서드가 던지는 예외 타입이다.
public interface Joiner<T, R, R_X extends Throwable> {
public default boolean onFork(Subtask<T> subtask);
public default boolean onComplete(Subtask<T> subtask);
public R result() throws R_X;
public R timeout() throws R_X;
}
onFork는 하위 작업을 시작할 때, onComplete는 하위 작업이 끝날 때 호출된다. 두 메서드는 스코프를 취소할지 나타내는 boolean을 반환한다. result는 join()의 결과나 예외를 만들며, 스코프에 제한 시간이 설정돼 있고 join() 전이나 도중 시간이 만료되면 timeout이 호출된다.
다음 구현은 성공한 작업의 결과를 모으고 실패한 작업은 무시한다. onComplete는 여러 스레드에서 동시에 호출될 수 있으므로 스레드 안전하게 작성해야 한다. 제한 시간이 만료되면 timeout은 CompletionException을 던진다.
class CollectingJoiner<T> implements Joiner<T, List<T>, CompletionException> {
private final Queue<T> results = new ConcurrentLinkedQueue<>();
public boolean onComplete(Subtask<T> subtask) {
if (subtask.state() == Subtask.State.SUCCESS) {
results.add(subtask.get());
}
return false;
}
public List<T> result() {
return List.copyOf(results);
}
public List<T> timeout() {
throw new CompletionException(new CancelledByTimeoutException());
}
}
이 정책은 다음처럼 사용할 수 있다.
<T> List<T> allSuccessful(List<Callable<T>> tasks) throws InterruptedException {
try (var scope = StructuredTaskScope.open(new CollectingJoiner<T>())) {
tasks.forEach(scope::fork);
return scope.join();
}
}
예외와 설정
예외 처리는 사용 방식에 따라 달라진다. 스코프가 실패한 것으로 판단되면 join()은 R_X 타입의 예외를 던진다. 앞의 handle() 예제에서는 하위 작업의 예외를 원인으로 담은 ExecutionException이 발생한다. 스코프가 닫힌 뒤 예외를 처리하려면 다음처럼 catch를 둘 수 있다.
try (var scope = StructuredTaskScope.open()) {
...
} catch (ExecutionException e) {
Throwable cause = e.getCause();
switch (cause) {
case IOException ioe -> ..
default -> ..
}
}
예외 처리 코드는 패턴 매칭과 함께 instanceof 연산자를 사용해 원인별로 처리할 수 있다(JEP 394). 특정 하위 작업의 예외가 기본값 반환으로 이어져야 한다면, 소유 스레드가 처리하기보다 하위 작업 안에서 예외를 잡고 기본값을 결과로 반환하는 편이 적절할 수 있다.
open에는 앞서 소개한 두 정적 메서드 외에 두 메서드가 더 있다. open(UnaryOperator)는 기본 설정 객체에 연산자를 적용해 모니터링·관리용 스코프 이름, 제한 시간, fork가 스레드를 만들 때 쓸 스레드 팩토리를 설정한다. open(Joiner, UnaryOperator)는 조인과 설정 연산자를 모두 받는다.
다음 예제는 스레드 팩토리와 제한 시간을 설정한다.
<T> List<T> runConcurrently(Collection<Callable<T>> tasks,
ThreadFactory factory,
Duration timeout)
throws ExecutionException, InterruptedException
{
try (var scope = StructuredTaskScope.open(Joiner.<T>allSuccessfulOrThrow(),
cf -> cf.withThreadFactory(factory)
.withTimeout(timeout))) {
tasks.forEach(scope::fork);
return scope.join();
}
}
이 스코프의 fork는 지정한 스레드 팩토리를 사용해 각 하위 작업의 스레드를 만든다. 스레드 이름이나 다른 속성을 설정할 때 쓸 수 있다. 제한 시간은 java.time.Duration으로 지정한다. join() 전에 또는 기다리는 도중 시간이 만료되면 스코프와 미완료 하위 작업이 취소되고, join()은 원인이 CancelledByTimeoutException인 ExecutionException을 던진다.
스레드 덤프와 대안
JEP는 가상 스레드에 추가된 JSON 스레드 덤프 형식을 확장해 StructuredTaskScope가 스레드를 어떻게 계층으로 묶는지 보여준다. 다음 명령으로 덤프를 만들 수 있다.
$ jcmd <pid> Thread.dump_to_file -format=json <file>
각 스코프의 JSON 객체에는 그 스코프에서 시작한 스레드 배열과 스택 추적이 들어간다. 소유 스레드는 보통 하위 작업을 기다리며 join()에서 대기한다. 덤프는 하위 작업 스레드를 트리로 보여주며, 스코프 객체에는 부모 스코프 참조도 있어 프로그램 구조를 재구성할 수 있다. com.sun.management.HotSpotDiagnosticsMXBean으로도 덤프를 만들 수 있다. 직접 호출하거나 플랫폼 MBeanServer와 로컬·원격 JMX 도구를 통해 사용할 수 있다.
대안으로 ExecutorService를 구조적 동시성에 맞게 확장하는 방안도 시제품으로 검토했지만, JDK와 생태계에서 ExecutorService와 상위 인터페이스 Executor는 대개 비구조적으로 사용된다. 더 제한적인 개념에 같은 API를 재사용하면 혼란을 일으킬 수 있고, 기존 메서드에 구조적 ExecutorService를 전달했을 때 대부분의 상황에서 예외가 발생할 수 있다.
인큐베이터 단계에서는 fork가 Future를 반환했다. 기존 ExecutorService::submit과 비슷해 익숙하다는 장점이 있었지만, 구조적 사용을 의도한 StructuredTaskScope와 비구조적 사용이 가능한 ExecutorService의 차이를 흐렸다. 구조적 스코프에서는 join()이 끝난 뒤 하위 작업이 완료된 것을 확인하고 결과를 처리해야 한다. 따라서 대기하는 Future.get() 대신 비차단 메서드인 resultNow()를 써야 했고, CompletableFuture의 미완료 작업용 기능도 이 맥락에서는 이점을 주지 않았다. 구조적 동시성은 여러 스레드의 관련 작업을 하나의 작업 단위로 다루지만, Future는 작업을 각각 다룰 때 유용하다. 현 API의 Subtask::get()은 인큐베이터 API에서 Future::resultNow()가 하던 것과 같은 역할을 한다.
부록: 원문 예시
Response handle() throws IOException {
String theUser = findUser();
int theOrder = fetchOrder();
return new Response(theUser, theOrder);
}
댓글 (0)
로그인하면 이 기사에 내 생각을 남길 수 있어요