StructuredTaskScope
StructuredTaskScope (구조적 동시성 지원 기본 API)
구조적 동시성(structured concurrency)을 위한 기본 API예요. 작업이 여러 동시 서브태스크로 나뉘고, 메인 작업이 계속 진행되기 전에 그 서브태스크들이 반드시 끝나야 하는 경우를 지원해요. 동시 작업의 수명을 하나의 문법 블록 안에 가둘 수 있게 해줘요.
본문
StructuredTaskScope는 public 생성자 중 하나로 만들어져요. 서브태스크를 실행할 쓰레드를 시작하는 fork 메서드, 모든 서브태스크가 끝날 때까지 기다리는 join 메서드, 태스크 스코프를 닫는 close 메서드를 정의해요. try-with-resources와 함께 쓰도록 설계됐어요.
기본 사용법은 try 블록 안에서 fork로 서브태스크를 실행하고, join으로 끝나길 기다린 뒤 결과를 처리하는 거예요. fork는 성공적으로 끝난 결과나 실패 시 예외를 얻을 수 있는 Subtask를 돌려줘요.
Callable<String> task1 = ...;
Callable<Integer> task2 = ...;
try (var scope = new StructuredTaskScope<Object>()) {
Subtask<String> subtask1 = scope.fork(task1);
Subtask<Integer> subtask2 = scope.fork(task2);
scope.join();
// 결과/예외 처리
} // close
join과 close는 소유자(태스크 스코프를 연 쓰레드)만 호출할 수 있어요. fork 후 join을 호출하지 않고 닫으면 예외가 던져져요.
shutdown 메서드는 닫지 않고 태스크 스코프를 종료해요. 미완료 서브태스크를 인터럽트로 취소하고 새 쓰레드 시작을 막아요. 소유자가 join에서 기다리고 있다면 깨워줘요.
공통 정책을 구현한 두 서브클래스가 제공돼요.
ShutdownOnSuccess: 첫 번째로 성공한 서브태스크의 결과를 잡아두고 태스크 스코프를 종료해요. "아무 결과나 좋다(invoke any)" 경우에 사용해요.ShutdownOnFailure: 첫 번째로 실패한 서브태스크의 예외를 잡아두고 종료해요. 모든 서브태스크의 결과가 필요한 "invoke all" 경우에 사용해요.
try (var scope = new StructuredTaskScope.ShutdownOnSuccess<String>()) {
scope.fork(() -> fetch(left));
scope.fork(() -> fetch(right));
scope.join();
String result = scope.result(e -> new WebApplicationException(e));
}
Instant deadline = ...;
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
Supplier<String> supplier1 = scope.fork(() -> query(left));
Supplier<String> supplier2 = scope.fork(() -> query(right));
scope.joinUntil(deadline);
scope.throwIfFailed(e -> new WebApplicationException(e));
String result = Stream.of(supplier1, supplier2)
.map(Supplier::get)
.collect(Collectors.joining(", ", "{ ", " }"));
}
StructuredTaskScope를 확장하고 handleComplete를 재정의하면 다른 정책을 구현할 수 있어요. 예를 들어 성공한 서브태스크의 결과만 모으는 구현은 이렇게 생겼어요.
class CollectingScope<T> extends StructuredTaskScope<T> {
private final Queue<Subtask<? extends T>> subtasks = new LinkedTransferQueue<>();
@Override protected void handleComplete(Subtask<? extends T> subtask) {
if (subtask.state() == Subtask.State.SUCCESS) subtasks.add(subtask);
}
@Override public CollectingScope<T> join() throws InterruptedException {
super.join(); return this;
}
public Stream<Subtask<? extends T>> completedSuccessfully() {
super.ensureOwnerAndJoined(); return subtasks.stream();
}
}
태스크 스코프는 트리 구조를 이루며, 부모-자식 관계는 새 태스크 스코프를 열 때 암묵적으로 생겨요. 이 트리는 스코프 값의 상속과 한정(confinement) 검사를 지원해요.
private static final ScopedValue<String> USERNAME = ScopedValue.newInstance();
ScopedValue.runWhere(USERNAME, "duke", () -> {
try (var scope = new StructuredTaskScope<String>()) {
scope.fork(() -> childTask());
}
});
String childTask() {
String name = USERNAME.get(); // "duke"
...
}
이 클래스의 생성자나 메서드에 null 인자를 넘기면 별다른 명시가 없는 한 NullPointerException이 던져져요. 코드와 시그니처는 원문 그대로 보존돼요.