结构化并发与 Scoped Values

// 典型写法：Future 游离，生命周期不可控
public Response handle(Request req) {
    Future<User> user = pool.submit(() -> loadUser(req));
    Future<Orders> orders = pool.submit(() -> loadOrders(req));
    try {
        return new Response(user.get(), orders.get());
    } catch (Exception e) {
        // 一个失败，另一个还在跑；取消逻辑要手写
        throw new RuntimeException(e);
    }
}

结构化并发的核心承诺：
  1. 子任务的生命周期 ≤ 父作用域
  2. 作用域退出前，所有子任务必须结束（正常/取消/失败）
  3. 取消沿作用域树自动向下传播
  4. 失败沿作用域树向上冒泡，不丢失

# 预览期需要显式开启
javac --release 21 --enable-preview App.java
java  --enable-preview App

JDK 21 后的 API 形状：
  try (var scope = StructuredTaskScope.open(joiner)) {
      var a = scope.fork(taskA);
      var b = scope.fork(taskB);
      scope.join();          // 等待并聚合
      return combine(a.get(), b.get());
  }
  // 退出 try：作用域关闭，所有子任务已结束

Response handle(Request req) throws Exception {
    try (var scope = StructuredTaskScope.open(
            Joiner.<Response>awaitAllSuccessfulOrThrow())) {
        var user   = scope.fork(() -> loadUser(req));
        var orders = scope.fork(() -> loadOrders(req));
        scope.join();                    // 任一失败 → 抛异常并取消其余
        return new Response(user.get(), orders.get());
    }
}

语义要点：
  awaitAllSuccessfulOrThrow()
    - 全部成功 → join() 正常返回
    - 任一失败 → 取消其他子任务，join() 抛出失败原因
    - 抛出的异常类型是 StructuredTaskScope.FailedException，
      其 cause 是真正的业务异常

String fetchFastest(String key) throws Exception {
    try (var scope = StructuredTaskScope.open(
            Joiner.<String>anySuccessfulResultOrThrow())) {
        scope.fork(() -> queryCache(key));
        scope.fork(() -> queryReplicaA(key));
        scope.fork(() -> queryReplicaB(key));
        return scope.join();             // 返回第一个成功结果
    }
}

竞速模式的注意点：
  1. 被取消的子任务要能响应中断（阻塞调用需可中断）
  2. 结果不确定 —— 三个副本数据不一致时以最快为准
  3. 取消不是「立刻停」，是设置中断标志，配合可中断 API 生效

// 收集成功结果，容忍部分失败
class CollectJoiner<T> implements StructuredTaskScope.Joiner<T, List<T>> {
    private final List<T> results = new ArrayList<>();

    @Override
    public boolean onComplete(Subtask<T> subtask) {
        if (subtask.state() == Subtask.State.SUCCESS) {
            results.add(subtask.get());
        }
        return false;              // false = 继续等待其余子任务
    }

    @Override
    public List<T> result() {
        return List.copyOf(results);
    }
}

List<Price> bestEffort(List<Supplier> suppliers) throws Exception {
    try (var scope = StructuredTaskScope.open(new CollectJoiner<Price>())) {
        suppliers.forEach(s -> scope.fork(() -> s.quote()));
        return scope.join();       // 返回成功结果集合，忽略失败
    }
}

onComplete 返回值的含义：
  true  → 立即关闭作用域，取消剩余子任务
  false → 继续等待其他子任务完成
  该方法可能被多个线程并发调用，实现必须线程安全

static final ScopedValue<User> CURRENT_USER = ScopedValue.newInstance();

void handle(Request req) {
    User user = authenticate(req);
    ScopedValue.where(CURRENT_USER, user).run(() -> {
        // 作用域内任意深度都能读到，无需逐层传参
        doBusiness();
    });
    // 出了作用域，CURRENT_USER.isBound() == false
}

void doBusiness() {
    User u = CURRENT_USER.get();   // 隐式读取
    audit(u);
}

ScopedValue vs ThreadLocal：
  不可变     —— 作用域内只读，无 set，数据流单向
  自动清理   —— 作用域退出即解绑，无泄漏风险
  零拷贝继承 —— 子线程/子任务直接读同一份，不复制
  虚拟线程友好 —— 百万虚拟线程共享一个绑定，内存恒定

Response handle(Request req) {
    User user = authenticate(req);
    return ScopedValue.where(CURRENT_USER, user).call(() -> {
        try (var scope = StructuredTaskScope.open(Joiner.<Data>awaitAllSuccessfulOrThrow())) {
            // fork 出的子任务自动继承 CURRENT_USER 绑定
            var a = scope.fork(() -> loadWithAudit(req));
            scope.join();
            return new Response(a.get());
        }
    });
}

Pinning 的两大来源：
  1. synchronized 块内的阻塞（ReentrantLock 不会 pin）
  2. 本地方法（JNI）阻塞

// 反例：synchronized 内做阻塞 IO，会 pin 住载体线程
synchronized (lock) {
    var data = httpClient.send(request, BodyHandlers.ofString());  // 危险
}

// 正例：用 ReentrantLock 替代 synchronized
lock.lock();
try {
    var data = httpClient.send(request, BodyHandlers.ofString());
} finally {
    lock.unlock();
}

# JDK 21+ 诊断 pinning
-Djdk.tracePinnedThreads=full        # JDK 21 打印固定堆栈
-Djdk.tracePinnedThreads=short

# JDK 24+ 用 JFR 事件观察
jfr print --events jdk.VirtualThreadPinned recording.jfr

取消传播的实现：
  1. scope.close() 对每个未完成子任务调用 cancel()
  2. cancel() 触发 Thread.interrupt()
  3. 子任务需在可中断点（阻塞 IO、sleep、await）响应
  4. 不响应中断的忙循环会拖慢作用域关闭

迁移检查清单：
  1. 把「提交 N 个任务再 join」的代码识别出来 → 改造成作用域
  2. 手写的取消/超时逻辑 → 交给 Joiner 与作用域生命周期
  3. 线程池的并发上限 → 改用 Semaphore 或结构化限流
  4. 逐层传参的上下文 → 改用 ScopedValue
  5. 保留线程池做「资源池化」（如数据库连接）而非「并发控制」

// 用 Semaphore 限制并发度（结构化作用域没有池的概念）
private static final Semaphore PERMITS = new Semaphore(64);

Response handle(Request req) throws Exception {
    PERMITS.acquire();
    try (var scope = StructuredTaskScope.open(Joiner.<Data>awaitAllSuccessfulOrThrow())) {
        var data = scope.fork(() -> load(req));
        scope.join();
        return new Response(data.get());
    } finally {
        PERMITS.release();
    }
}

观测要点：
  1. 线程转储：虚拟线程以「结构化树」呈现，父任务可见子任务
  2. 每个作用域可命名：StructuredTaskScope.open(joiner, cfg -> cfg.withName("load-order"))
  3. JFR 事件：jdk.VirtualThreadStart/End、jdk.VirtualThreadPinned
  4. 与 Micrometer 结合，记录作用域内子任务的成功/失败/取消计数
  5. 超时统一在作用域外层用 orTimeout 或 Joiner 的超时配置

// 给作用域命名，便于线程转储与追踪定位
try (var scope = StructuredTaskScope.open(
        Joiner.<Data>awaitAllSuccessfulOrThrow(),
        cfg -> cfg.withName("order-aggregation"))) {
    // ...
}

// 正确读取失败原因（含被抑制的异常）
try {
    scope.join();
} catch (StructuredTaskScope.FailedException e) {
    Throwable cause = e.getCause();
    for (Throwable s : cause.getSuppressed()) {
        log.warn("子任务失败", s);
    }
}