《Java 并发编程·线程与内存模型》15.3 从 ExecutorService 迁移的边界

    public void fanout(String memberId, String text) {
        ExecutorService pool = Executors.newVirtualThreadPerTaskExecutor();
        List<Future<String>> futures = new ArrayList<>();
        try {
            for (String name : CHANNELS) {
                StructuredNotifier.Channel channel = channels.get(name);
                futures.add(pool.submit(() -> {
                    channel.send(memberId, text);
                    return name;
                }));
            }
            for (Future<String> future : futures) {
                future.get();
            }
        } catch (ExecutionException e) {
            futures.forEach(f -> f.cancel(true));
            Throwable cause = e.getCause();
            if (cause instanceof LoanException loan) {
                throw loan;
            }
            throw new LoanException(ErrorCode.CONFLICT, "通知扇出失败: " + cause);
        } catch (InterruptedException e) {
            futures.forEach(f -> f.cancel(true));
            Thread.currentThread().interrupt();
            throw new LoanException(ErrorCode.INTERRUPTED, "通知扇出被中断");
        } finally {
            pool.shutdown();
        }
    }

    public void fanout(String memberId, String text) {
        try (var scope = StructuredTaskScope.<String, Stream<StructuredTaskScope.Subtask<String>>>open(
                StructuredTaskScope.Joiner.allSuccessfulOrThrow(),
                cfg -> cfg.withThreadFactory(Thread.ofVirtual().name("notify-", 0).factory()))) {

            for (String name : CHANNELS) {
                Channel channel = channels.get(name);
                scope.fork(() -> {
                    channel.send(memberId, text);
                    return name;
                });
            }

            scope.join().toList();        // 全部成功才会走到这里
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new LoanException(ErrorCode.INTERRUPTED, "通知扇出被中断");
        } catch (StructuredTaskScope.FailedException e) {
            Throwable cause = e.getCause();
            if (cause instanceof LoanException loan) {
                throw loan;
            }
            throw new LoanException(ErrorCode.CONFLICT, "通知扇出失败: " + cause);
        }
    }

  [push] + 145ms 503
  [sms ] + 827ms ok
  [mail] + 827ms ok
executor + 828ms code=CONFLICT msg=push 下游 503
  [push] + 135ms 503
  [mail] + 139ms INTERRUPTED
  [sms ] + 139ms INTERRUPTED
scope    + 140ms code=CONFLICT msg=push 下游 503

import java.time.Duration;
import java.util.concurrent.StructuredTaskScope;
import java.util.concurrent.StructuredTaskScope.Subtask;

public class ScopeBoundary {

    public static void main(String[] args) throws Exception {
        System.out.println("A. 非 owner 线程 fork");
        try (var scope = StructuredTaskScope.<String>open()) {
            Thread other = Thread.ofVirtual().start(() -> {
                try {
                    scope.fork(() -> "borrowed");
                } catch (RuntimeException e) {
                    System.out.println("  " + e.getClass().getName() + ": " + e.getMessage());
                }
            });
            other.join();
            scope.join();
        }

        System.out.println("B. 作用域关闭后读结果");
        Subtask<String> done;
        Subtask<String> cancelled;
        try (var scope = StructuredTaskScope.<String, String>open(
                StructuredTaskScope.Joiner.anySuccessfulResultOrThrow())) {
            done = scope.fork(() -> "fast");
            cancelled = scope.fork(() -> {
                Thread.sleep(Duration.ofSeconds(5));
                return "slow";
            });
            scope.join();
        }
        System.out.println("  done.state=" + done.state() + " done.get=" + done.get());
        System.out.println("  cancelled.state=" + cancelled.state());
        try {
            cancelled.get();
        } catch (RuntimeException e) {
            System.out.println("  cancelled.get -> " + e.getClass().getSimpleName() + ": " + e.getMessage());
        }
    }
}

A. 非 owner 线程 fork
  java.lang.WrongThreadException: Current thread not owner
B. 作用域关闭后读结果
  done.state=SUCCESS done.get=fast
  cancelled.state=UNAVAILABLE
  cancelled.get -> IllegalStateException: Result is unavailable or subtask did not complete successfully

# 故意不加 --enable-preview，看编译器怎么说
$JAVA_HOME/bin/javac --release 25 -d out MigrateDemo.java

MigrateDemo.java:4: 错误: StructuredTaskScope 是预览 API，默认情况下处于禁用状态。
import java.util.concurrent.StructuredTaskScope.Subtask;
                           ^
  （请使用 --enable-preview 以启用预览 API）

错误: 加载主类 MigrateDemo 时出现 LinkageError
	java.lang.UnsupportedClassVersionError: Preview features are not enabled for MigrateDemo (class file version 69.65535). Try running with '--enable-preview'

// 需要「任务边界」的请求路径
LoanService service = new StructuredLoanService(books, loans, members);

// 后台批处理：任务要比请求活得久，继续用执行器
try (ExecutorService batch = Executors.newVirtualThreadPerTaskExecutor()) {
    isbns.forEach(isbn -> batch.submit(() -> reindex(isbn)));
}

export JAVA_HOME=~/.jdks/jdk-25.0.5+8/Contents/Home
mvn -q compile -Dmaven.compiler.enablePreview=true
java --enable-preview --source 25 -cp target/classes com.plume.conc.core.DemoMigrate