Structured Concurrency：第五次预览的新 API

状态：第五次预览，API 与 21 不兼容01

结构化并发从 JDK 21 开始预览，JDK 25 是第五次预览（JEP 505）。这一版把 StructuredTaskScope 改成接口，用静态 open(...) 创建，用 Joiner 表达完成策略；21 的 ShutdownOnFailure / ShutdownOnSuccess 已被移除。

import java.util.concurrent.*;

public class OldApi {
    void fetch() throws Exception {
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
            scope.fork(() -> "a");
            scope.join().throwIfFailed();
        }
    }
}

javac --release 25 --enable-preview OldApi.java

OldApi.java:5: error: cannot find symbol
        try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
                                                ^
  symbol:   class ShutdownOnFailure
  location: interface StructuredTaskScope
Note: OldApi.java uses preview features of Java SE 25.
Note: Recompile with -Xlint:preview for details.
1 error

默认策略：全成功，或首个失败02

无参 open() 使用默认策略：等待所有子任务成功；任一子任务失败就取消其余任务，join() 抛 FailedException，原始异常在 cause 中。全部成功时 join() 返回 null（R 是 Void）。

import java.util.concurrent.*;

public class BasicScope {
    record User(String id) {}
    record Order(String id) {}

    static User loadUser() throws Exception {
        Thread.sleep(100);
        return new User("u-1");
    }

    static Order loadOrder() throws Exception {
        Thread.sleep(150);
        return new Order("o-1");
    }

    public static void main(String[] args) throws Exception {
        try (var scope = StructuredTaskScope.open()) {
            var user = scope.fork(BasicScope::loadUser);
            var order = scope.fork(BasicScope::loadOrder);
            scope.join();
            System.out.println(user.get() + " / " + order.get());
        }
    }
}

javac --release 25 --enable-preview BasicScope.java
java --enable-preview BasicScope

User[id=u-1] / Order[id=o-1]

• fork 默认在虚拟线程上执行子任务；也可以传自定义 ThreadFactory。
• join() 只能由打开作用域的所有者线程调用；它等待所有子任务结束或被取消。
• Subtask.get() 只能在 join 之后调用；提前调用抛 IllegalStateException。
• try-with-resources 的 close() 会取消未完成任务并等待它们终止。

Joiner：把完成策略变成对象03

Joiner 负责在子任务完成时决定是否取消作用域，并产出 join 的结果。常用工厂：allSuccessfulOrThrow()、anySuccessfulResultOrThrow()、awaitAll()、awaitAllSuccessfulOrThrow()、allUntil(predicate)。

try (var scope = StructuredTaskScope.open(
        StructuredTaskScope.Joiner.<String>allSuccessfulOrThrow())) {
    scope.fork(() -> "a");
    scope.fork(() -> "b");
    scope.join().forEach(s -> System.out.println(s.get()));
}

a
b

注意：Subtask 流的顺序不是 API 契约；本机这次恰好输出 a、b，换成有真实耗时的任务后顺序可能变化。需要稳定顺序时，按结果字段排序或改用 Map 关联。

try (var scope = StructuredTaskScope.open(
        StructuredTaskScope.Joiner.<String>anySuccessfulResultOrThrow())) {
    scope.fork(() -> { Thread.sleep(300); return "slow"; });
    scope.fork(() -> { Thread.sleep(100); return "fast"; });
    System.out.println("winner=" + scope.join());
}

winner=fast

失败、超时与关闭04

失败传播是短路的：一个子任务抛异常，作用域取消其余子任务（通过中断），join() 抛 FailedException。取消是协作式的，子任务必须响应中断；close() 会一直等到所有子任务真正终止。

try (var scope = StructuredTaskScope.open()) {
    var slow = scope.fork(() -> { Thread.sleep(5000); return "slow"; });
    scope.fork(() -> { throw new IllegalStateException("boom"); });
    try {
        scope.join();
    } catch (StructuredTaskScope.FailedException e) {
        System.out.println("failed cause=" + e.getCause());
    }
    System.out.println("slow state=" + slow.state());
}

failed cause=java.lang.IllegalStateException: boom
slow state=UNAVAILABLE

try (var scope = StructuredTaskScope.open(
        StructuredTaskScope.Joiner.<String>allSuccessfulOrThrow(),
        cfg -> cfg.withTimeout(java.time.Duration.ofMillis(100)))) {
    scope.fork(() -> { Thread.sleep(1000); return "late"; });
    try {
        scope.join();
    } catch (StructuredTaskScope.TimeoutException e) {
        System.out.println("timeout=" + e.getClass().getSimpleName());
    }
}

timeout=TimeoutException

• 超时抛的是 StructuredTaskScope.TimeoutException（unchecked），不是 FailedException。
• 所有者线程被中断时，join() 抛 InterruptedException；close 会取消并等待子任务。
• 子任务吞掉中断或不响应中断，会让 close 等待很久，结构化并发也无法强制杀死线程。
• 作用域逃逸、嵌套不当或未在词法范围内关闭，可能抛 StructureViolationException。

运行时规则：owner、fork 与 Subtask05

结构化并发在运行时强制「结构」：只有打开作用域的 owner 线程能 fork/join；Subtask 的结果只能在 join 后读取；作用域不能在词法块之外继续使用。

import java.util.concurrent.*;

public class ScopeRules {
    public static void main(String[] args) throws Exception {
        try (var scope = StructuredTaskScope.open()) {
            var subtask = scope.fork(() -> 1);
            try {
                subtask.get();
            } catch (Exception e) {
                System.out.println("get before join=" + e.getClass().getSimpleName());
            }

            var nested = scope.fork(() -> {
                try {
                    scope.fork(() -> 2);
                    return "forked";
                } catch (Exception e) {
                    return e.getClass().getSimpleName();
                }
            });
            scope.join();
            System.out.println("subtask state=" + subtask.state());
            System.out.println("fork from subtask=" + nested.get());
        }
    }
}

get before join=IllegalStateException
subtask state=SUCCESS
fork from subtask=WrongThreadException

版本边界BOUNDARY

这张表比 API 本身更值得记。写代码时先确认你的运行时版本。

费曼自检SELF CHECK

先合上页面，用自己的话回答，再展开对照。能讲出因果和反例，才算真正掌握。