结构化并发：让任务有作用域

API 形状与执行模型01

public class StructuredTaskScope<T> implements AutoCloseable {
    public <U extends T> Subtask<U> fork(Callable<? extends U> task);
    public StructuredTaskScope<T> join() throws InterruptedException;
    public StructuredTaskScope<T> joinUntil(Instant deadline)
            throws InterruptedException, TimeoutException;
    public void shutdown();
    public void close();
}

public interface Subtask<T> {
    State state();        // SUCCESS / FAILED / UNAVAILABLE
    T get();              // 仅 SUCCESS 时可用，否则抛 IllegalStateException
    Throwable exception(); // 仅 FAILED 时可用
}

• 默认情况下，fork 在虚拟线程上执行子任务；可以通过构造器传入自定义 ThreadFactory。
• join 只能由创建作用域的所有者线程调用；在子任务里调用会抛 IllegalStateException。
• close 会先 shutdown，再等待所有子任务终止；如果所有者线程被中断，close 仍会等待。
• 子任务的异常不会自动打印；必须通过 throwIfFailed()、result() 或 Subtask.exception() 显式获取。

取消与异常传播的细节02

1. 任一子任务失败，ShutdownOnFailure 调用 shutdown()，中断其他子任务。
2. 被中断的子任务如果响应中断（如阻塞 I/O、Thread.sleep），会尽快结束；不响应中断的任务可能让 close() 等待较久。
3. 所有者调用 throwIfFailed() 时，抛出第一个失败的 ExecutionException，原异常在 cause 中。
4. 如果所有子任务都成功，throwIfFailed() 什么也不做，可以安全读取 get()。

ScopedValue：不可变上下文03

ThreadLocal 是「每个线程一个可变副本」，在虚拟线程数量巨大时内存开销明显，而且容易忘记清理。ScopedValue 是「在某个作用域内绑定一个不可变值」，子任务自动继承，作用域结束即失效。

private static final ScopedValue<String> USER = ScopedValue.newInstance();

void handle() {
    ScopedValue.where(USER, "alice").run(() -> {
        System.out.println(USER.get()); // alice
        // fork 出的子任务也能读到 USER
    });
    // 作用域外 USER.isBound() == false
}

从 invokeAll 迁移到作用域01

// Before：超时与取消要自己拼
List<Callable<Result>> tasks = List.of(this::loadA, this::loadB);
List<Future<Result>> futures = executor.invokeAll(tasks, 1, TimeUnit.SECONDS);
for (Future<Result> f : futures) {
    if (f.isCancelled()) throw new TimeoutException();
    results.add(f.get());
}

// After：作用域负责生命周期，超时用 joinUntil
try (var scope = new StructuredTaskScope.ShutdownOnFailure()) {
    var a = scope.fork(this::loadA);
    var b = scope.fork(this::loadB);
    scope.joinUntil(Instant.now().plusSeconds(1)).throwIfFailed();
    return new Pair(a.get(), b.get());
}

• joinUntil 超时会抛 TimeoutException，随后 close() 取消未完成任务。
• 如果某个子任务失败，ShutdownOnFailure 会取消另一个，避免继续做无用功。
• 不要在外层再套一个线程池等待；作用域自己管理子任务线程。

可运行示例：并发加载与快速失败02

import java.util.concurrent.*;

public class StructuredDemo {
    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 = new StructuredTaskScope.ShutdownOnFailure()) {
            var user = scope.fork(StructuredDemo::loadUser);
            var order = scope.fork(StructuredDemo::loadOrder);
            scope.join().throwIfFailed();
            System.out.println(user.get() + " / " + order.get());
        }

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

javac --enable-preview --release 21 StructuredDemo.java
java --enable-preview StructuredDemo

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

第二个作用域的慢任务会被取消/中断，但示例中的 Thread.sleep 会响应中断；如果子任务捕获并吞掉中断，close() 可能等待它自然结束。

先猜再看01

费曼自检：生命周期与传播02