Java——CompletableFuture 异步编程实战

CompletableFuture 是 JDK 8 引入的异步编排工具（java.util.concurrent，类上 @since 1.8），同时实现了 Future<T> 与 CompletionStage<T>。
它解决的问题是：把「查用户、查订单、算总价」这类有依赖关系或可以并行执行的步骤，写成一条声明式的异步链，而不是一堆层层嵌套的回调或者手动管理线程。
本文逐条讲清每个 API 的语义边界：谁有返回值、谁吞异常、回调在哪个线程跑、默认线程池有什么坑、取消到底能不能打断正在执行的任务，以及它与仍在预览中的结构化并发（StructuredTaskScope）的差别。
文中所有输出都来自 JDK 25（LTS）在 macOS 上的实测，示例代码可以原样编译运行。

示例里复用两个类型与一批工具方法，避免每个代码块都重复声明。先看这两段公共代码，后面的例子都直接引用它们。

import java.util.List;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

/** 全文共用的示例数据与工具方法。 */
final class DemoData {

    record User(String id, String name) {}

    record Order(String id, int amount) {}

    static void sleep(long millis) {
        try {
            Thread.sleep(millis);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }

    static User findUser(String id) {
        sleep(80);
        return new User(id, "张三");
    }

    static List<Order> findOrders(String userId) {
        sleep(120);
        return List.of(new Order("o-1", 100), new Order("o-2", 200));
    }

    static int totalAmount(List<Order> orders) {
        return orders.stream().mapToInt(Order::amount).sum();
    }

    /** 给池线程起个能看懂的名字，否则调试时只有 pool-1-thread-1。 */
    static ThreadFactory namedFactory(String prefix) {
        AtomicInteger seq = new AtomicInteger();
        return runnable -> new Thread(runnable, prefix + "-" + seq.incrementAndGet());
    }

    private DemoData() {
    }
}

import java.time.Duration;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

/** 等待全部任务并保留每个任务的结果与异常。 */
final class Futures {

    record Outcome<T>(String name, T value, Throwable error) {
        boolean ok() {
            return error == null;
        }
    }

    /** 把「成功或失败」都收敛成正常的返回值，这样 allOf 永远不会因为某个任务失败而短路。 */
    static <T> CompletableFuture<Outcome<T>> settle(String name, CompletableFuture<T> future) {
        return future.handle((value, error) -> new Outcome<>(name, value, error));
    }

    /** 等所有任务结束，再按输入顺序取出每个任务的结果。 */
    static <T> List<Outcome<T>> awaitAll(List<CompletableFuture<Outcome<T>>> settled) {
        CompletableFuture.allOf(settled.toArray(CompletableFuture[]::new)).join();
        return settled.stream().map(CompletableFuture::join).toList();
    }

    /** 剥掉 CompletionException / ExecutionException 的包装，拿到任务里真正抛出的异常。 */
    static Throwable rootCause(Throwable error) {
        Throwable current = error;
        while ((current instanceof CompletionException || current instanceof ExecutionException)
                && current.getCause() != null) {
            current = current.getCause();
        }
        return current;
    }

    /** 提交时捕获上下文，执行时恢复，执行后清理，避免脏值留在池线程上。 */
    static Executor propagating(ThreadLocal<String> context, Executor delegate) {
        String captured = context.get();
        return task -> delegate.execute(() -> {
            String previous = context.get();
            context.set(captured);
            try {
                task.run();
            } finally {
                if (previous == null) {
                    context.remove();
                } else {
                    context.set(previous);
                }
            }
        });
    }

    /** 到点还没完成就取消，下游拿到 CancellationException。 */
    static <T> CompletableFuture<T> withDeadline(CompletableFuture<T> future,
                                                 Duration timeout,
                                                 ScheduledExecutorService scheduler) {
        ScheduledFuture<?> deadline = scheduler.schedule(
                () -> future.cancel(false), timeout.toNanos(), TimeUnit.NANOSECONDS);
        future.whenComplete((value, error) -> deadline.cancel(false));
        return future;
    }

    /** 全部成功时把结果收集成 List，任一失败则整体失败。 */
    static <T> CompletableFuture<List<T>> allSuccessful(List<CompletableFuture<T>> futures) {
        return CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new))
                .thenApply(ignored -> futures.stream().map(CompletableFuture::join).toList());
    }

    private Futures() {
    }
}

创建异步任务

supplyAsync 与 runAsync

supplyAsync 接收一个 Supplier<T>，任务执行完能返回一个值；runAsync 接收 Runnable，任务执行完没有任何值，返回的 future 是 CompletableFuture<Void>。二者都有一个不带 Executor 的重载和一个带 Executor 的重载。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ForkJoinPool;

public class CreateDemo {

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2, DemoData.namedFactory("biz"));

        System.out.println("commonPool parallelism = " + ForkJoinPool.getCommonPoolParallelism());
        System.out.println("CPU = " + Runtime.getRuntime().availableProcessors());

        CompletableFuture<String> onDefaultPool =
                CompletableFuture.supplyAsync(() -> Thread.currentThread().getName());
        System.out.println("默认执行器: " + onDefaultPool.join());

        CompletableFuture<String> onCustomPool =
                CompletableFuture.supplyAsync(() -> Thread.currentThread().getName(), pool);
        System.out.println("自定义执行器: " + onCustomPool.join());

        CompletableFuture<Void> run = CompletableFuture.runAsync(
                () -> System.out.println("runAsync 线程: " + Thread.currentThread().getName()), pool);
        System.out.println("runAsync join = " + run.join());

        pool.shutdown();
    }
}

运行输出（本机是 10 核，availableProcessors() - 1 正好是 9）：

commonPool parallelism = 9
CPU = 10
默认执行器: ForkJoinPool.commonPool-worker-1
自定义执行器: biz-1
runAsync 线程: biz-2
runAsync join = null

completedFuture 与 failedFuture

不需要真的去另一个线程跑任务，只是想「先占个位置」的时候，用 completedFuture 构造一个已经正常完成的 future，用 failedFuture 构造一个已经异常完成的 future（后者 @since 9）。

CompletableFuture<String> ready = CompletableFuture.completedFuture("已就绪");
System.out.println("completedFuture: isDone=" + ready.isDone() + ", join=" + ready.join());

CompletableFuture<String> failed =
        CompletableFuture.failedFuture(new IllegalStateException("构造即失败"));
System.out.println("failedFuture: isCompletedExceptionally=" + failed.isCompletedExceptionally());

1 2	completedFuture: isDone=true, join=已就绪 failedFuture: isCompletedExceptionally=true

这两个工厂方法在单元测试里尤其好用：你可以先「假造」一个已经完成的下游结果，去验证组合逻辑本身，而不必真的拉起线程。

默认执行器是公共 ForkJoinPool

不带 Executor 的 supplyAsync / runAsync / 各种 xxxAsync 变体，默认都提交到公共 ForkJoinPool。这个池的并行度是 max(1, Runtime.getRuntime().availableProcessors() - 1)（ForkJoinPool 源码第 3081 行附近），它有两个麻烦：

它是 JVM 全局共享的。并行流（parallelStream()）、以及任何没有显式传 Executor 的 CompletableFuture 都在抢同一个池。一个服务的突发流量或阻塞任务，会拖累整个 JVM 里所有用默认执行器的代码。
你没法按业务给它单独配线程数、队列或拒绝策略。调优 commonPool 要么靠全局系统属性，要么不现实。

JDK 9 起可以通过 CompletableFuture.defaultExecutor() 显式拿到这个默认执行器（源码里它返回 ForkJoinPool.asyncCommonPool()，也就是同一个公共池），但在生产代码里，它的作用是让你看清自己默认跑在哪个池上，不是拿来用的。

为什么生产代码要显式传 Executor

只要涉及阻塞 IO（数据库、HTTP 调用、文件读写），就不要用默认池。公共 ForkJoinPool 的 worker 是有限的，阻塞 IO 会把它们占住，后面的任务只能干等；而且这个池被设计来跑可分解的 CPU 密集任务，不是用来扛阻塞等待的。生产代码要么维护自己的业务线程池，要么在 JDK 21（JEP 444，虚拟线程转正）之后用 Executors.newVirtualThreadPerTaskExecutor() 这种「一任务一线程」的池，把阻塞 IO 交给虚拟线程去等。

链式变换：thenApply / thenAccept / thenRun

三者的签名差别

这三个方法最容易混，区别只在「有没有入参、有没有返回值」：

方法	入参	返回值	得到的 future
`thenApply(Function<T,U>)`	有，上游结果 `T`	有，`U`	`CompletableFuture<U>`
`thenAccept(Consumer<T>)`	有，上游结果 `T`	无	`CompletableFuture<Void>`
`thenRun(Runnable)`	无	无	`CompletableFuture<Void>`

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class StageKindDemo {

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2, DemoData.namedFactory("biz"));
        System.out.println("当前线程 = " + Thread.currentThread().getName());

        CompletableFuture<DemoData.User> user =
                CompletableFuture.supplyAsync(() -> DemoData.findUser("u-1"), pool);

        CompletableFuture<Integer> nameLength = user.thenApply(u -> u.name().length());
        CompletableFuture<Void> printed = nameLength.thenAccept(len -> System.out.println("名字长度 = " + len));
        CompletableFuture<Void> tail = printed.thenRun(() -> System.out.println("收尾动作"));
        System.out.println("thenApply 结果 = " + nameLength.join());
        System.out.println("thenAccept 结果 = " + printed.join());
        System.out.println("thenRun 结果 = " + tail.join());

        pool.shutdown();
    }
}

当前线程 = main
名字长度 = 2
收尾动作
thenApply 结果 = 2
thenAccept 结果 = null
thenRun 结果 = null

注意 thenAccept 和 thenRun 返回的 future join() 出来是 null，它们只是「副作用」阶段，价值在于触发打印、落库、发消息这些不产生下游数据的动作。

回调在哪个线程执行

这是一条很多人踩坑的规则：不带 Async 后缀的方法，回调不一定在调用线程执行，而是看「上游 future 完成的那一刻」。

上游 future 已经完成时，回调直接在当前（调用 thenApply 的）线程同步执行。
上游 future 还没完成时，回调在「完成上游的那个线程」里执行（通常就是池线程）。

带 Async 后缀的方法则总是把回调丢给 Executor（不传就默认丢给公共池）。

CompletableFuture<String> done = CompletableFuture.completedFuture("值");
done.join();
System.out.println("上游已完成, thenApply 线程 = "
        + done.thenApply(v -> Thread.currentThread().getName()).join());
System.out.println("上游已完成, thenApplyAsync 线程 = "
        + done.thenApplyAsync(v -> Thread.currentThread().getName()).join());
System.out.println("上游已完成, thenApplyAsync(pool) 线程 = "
        + done.thenApplyAsync(v -> Thread.currentThread().getName(), pool).join());

CompletableFuture<String> pending = CompletableFuture.supplyAsync(() -> {
    DemoData.sleep(100);
    return "值";
}, pool);
System.out.println("上游未完成, thenApply 线程 = "
        + pending.thenApply(v -> Thread.currentThread().getName()).join());

上游已完成, thenApply 线程 = main
上游已完成, thenApplyAsync 线程 = ForkJoinPool.commonPool-worker-1
上游已完成, thenApplyAsync(pool) 线程 = biz-2
上游未完成, thenApply 线程 = biz-1

这条规则的工程含义：如果你在链里做阻塞 IO，别用裸的 thenApply 以为「它一定在别的线程」，上游一完成它可能就在调用线程（比如某个请求线程）里把你堵死；反过来，一段必须快速返回的轻量转换，用 thenApply 让它在完成线程里顺带执行，比再往池里排一次队更划算。

thenCompose：把嵌套拍平

当「下一步的输入依赖上一步的输出，而且下一步本身又是一个异步调用」时，用 thenCompose。它的参数是 Function<? super T, ? extends CompletionStage<U>>，返回 CompletableFuture<U>，而不是 CompletableFuture<CompletableFuture<U>>。

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ComposeDemo {

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2, DemoData.namedFactory("biz"));

        CompletableFuture<List<DemoData.Order>> flat =
                CompletableFuture.supplyAsync(() -> DemoData.findUser("u-1"), pool)
                        .thenCompose(user -> CompletableFuture.supplyAsync(
                                () -> DemoData.findOrders(user.id()), pool));
        System.out.println("thenCompose 结果 = " + DemoData.totalAmount(flat.join()));

        CompletableFuture<Integer> chained = flat.thenApply(DemoData::totalAmount);
        System.out.println("扁平之后继续串 = " + chained.join());

        pool.shutdown();
    }
}

1 2	thenCompose 结果 = 300 扁平之后继续串 = 300

thenCompose 拿到的是 List<Order>，所以还能继续 .thenApply(totalAmount)，一层层往下串。这正是「扁平化」的意义：链上的类型一直是 CompletableFuture<T>，不会越套越深。

thenApply 嵌套的经典错误

如果用 thenApply 返回一个 future，就会得到 CompletableFuture<CompletableFuture<List<Order>>>，拿到订单列表需要连续 join() 两次，而且类型在链上再也拼不回去。

CompletableFuture<CompletableFuture<List<DemoData.Order>>> nested =
        CompletableFuture.supplyAsync(() -> DemoData.findUser("u-1"), pool)
                .thenApply(user -> CompletableFuture.supplyAsync(
                        () -> DemoData.findOrders(user.id()), pool));
System.out.println("thenApply 嵌套要 join 两次 = " + DemoData.totalAmount(nested.join().join()));

1	thenApply 嵌套要 join 两次 = 300

thenApply 管「同步的纯函数映射」，thenCompose 管「异步的、依赖上一步输出的下一步」。如果你发现自己要在 lambda 里再 supplyAsync 一个 future 然后返回它，那就是 thenCompose 的场景。

组合多个任务

thenCombine / thenAcceptBoth / runAfterBoth

当两个 future 彼此独立、可以并行，但要等两个都到了才能往下走时，用这三个：

thenCombine(other, BiFunction)：拿两个结果，算出一个新值。
thenAcceptBoth(other, BiConsumer)：拿两个结果，做副作用，无返回值。
runAfterBoth(other, Runnable)：只用「两个都结束」这个信号，不碰结果。

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CombineDemo {

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(4, DemoData.namedFactory("biz"));

        CompletableFuture<DemoData.User> user =
                CompletableFuture.supplyAsync(() -> DemoData.findUser("u-1"), pool);
        CompletableFuture<List<DemoData.Order>> orders =
                CompletableFuture.supplyAsync(() -> DemoData.findOrders("u-1"), pool);

        CompletableFuture<String> summary = user.thenCombine(orders,
                (u, list) -> u.name() + " 有 " + list.size() + " 笔订单");
        System.out.println("thenCombine = " + summary.join());

        CompletableFuture<Void> both = user.thenAcceptBoth(orders,
                (u, list) -> System.out.println("thenAcceptBoth = " + u.name() + "/" + list.size()));
        System.out.println("thenAcceptBoth join = " + both.join());

        CompletableFuture<Void> after = user.runAfterBoth(orders,
                () -> System.out.println("runAfterBoth = 两个任务都结束了"));
        System.out.println("runAfterBoth join = " + after.join());

        pool.shutdown();
    }
}

thenCombine = 张三 有 2 笔订单
thenAcceptBoth = 张三/2
thenAcceptBoth join = null
runAfterBoth = 两个任务都结束了
runAfterBoth join = null

allOf 的返回值和它「不告诉你」的事

allOf(CompletableFuture<?>... cfs) 等所有任务都结束，返回 CompletableFuture<Void>——结果要你自己去每个 future 里取，它本身不携带任何业务数据。

1 2	CompletableFuture<Void> all = CompletableFuture.allOf(user, orders); System.out.println("allOf join = " + all.join());

1	allOf join = null

再验证一次「等全部」这个语义：三个分别耗时 200ms、300ms、100ms 的任务并行提交，allOf 的完成时间由最慢的那个决定。

long started = System.nanoTime();
CompletableFuture<String> a = CompletableFuture.supplyAsync(() -> {
    DemoData.sleep(200);
    return "a";
}, pool);
CompletableFuture<String> b = CompletableFuture.supplyAsync(() -> {
    DemoData.sleep(300);
    return "b";
}, pool);
CompletableFuture<String> c = CompletableFuture.supplyAsync(() -> {
    DemoData.sleep(100);
    return "c";
}, pool);
System.out.println("allOf join = " + CompletableFuture.allOf(a, b, c).join()
        + ", 耗时 " + (System.nanoTime() - started) / 1_000_000 + "ms");

1	allOf join = null, 耗时 309ms

关于异常，allOf 的语义要点是这三条：

只要有任一任务异常完成，allOf 返回的 future 也会异常完成，但只携带其中一个任务的异常，你无法从这一个 future 知道是哪个任务、还有没有别的任务也失败了。
它是等所有任务都结束才完成（不是谁先失败谁立刻返回），所以最慢的任务决定它的完成时间——上面那段 309ms 就是证据。
更隐蔽的坑是「没人 join 就没异常」。如果只调用 allOf(...) 而不 join() / get()，任何一个任务失败都只会静默地沉没，日志里什么都没有。

下面这条链只会告诉你「有一个任务失败了」，不会告诉你是哪个、其余的是成功还是失败：

CompletableFuture<Void> failed = CompletableFuture.allOf(
        CompletableFuture.supplyAsync(() -> DemoData.findUser("u-2"), pool),
        CompletableFuture.failedFuture(new IllegalStateException("库存服务挂了")));
try {
    failed.join();
} catch (RuntimeException e) {
    System.out.println("allOf 异常类型 = " + e.getClass().getSimpleName()
            + ", 真实原因 = " + Futures.rootCause(e));
}

1	allOf 异常类型 = CompletionException, 真实原因 = java.lang.IllegalStateException: 库存服务挂了

正确等待全部并收集每一个异常

如果你需要「等所有任务结束，然后逐个汇报谁成功、谁失败」，正确做法是先把每个 future 用 handle 收敛成「永远正常完成」的 Outcome，再对这批收敛后的 future 做 allOf。因为 handle 无论成功失败都会返回一个正常值，这批 future 永远不会把 allOf 带进异常短路。

import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class AwaitAllDemo {

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(4, DemoData.namedFactory("biz"));
        long started = System.nanoTime();

        List<Futures.Outcome<String>> outcomes = Futures.awaitAll(List.of(
                Futures.settle("查用户", CompletableFuture.supplyAsync(
                        () -> DemoData.findUser("u-1").name(), pool)),
                Futures.settle("查库存", CompletableFuture.failedFuture(
                        new IllegalStateException("库存服务超时"))),
                Futures.settle("查优惠", CompletableFuture.supplyAsync(() -> "满减券", pool))));

        for (Futures.Outcome<String> outcome : outcomes) {
            System.out.println(outcome.name() + " -> "
                    + (outcome.ok() ? "成功: " + outcome.value()
                                    : "失败: " + Futures.rootCause(outcome.error())));
        }
        System.out.println("耗时 " + (System.nanoTime() - started) / 1_000_000 + "ms");

        pool.shutdown();
    }
}

查用户 -> 成功: 张三
查库存 -> 失败: java.lang.IllegalStateException: 库存服务超时
查优惠 -> 成功: 满减券
耗时 93ms

反过来，如果语义就是「全部成功才算成功，一个失败整体失败」，那不需要逐个收集，直接：

List<CompletableFuture<String>> all = List.of(
        CompletableFuture.supplyAsync(() -> "a", pool),
        CompletableFuture.supplyAsync(() -> "b", pool));
System.out.println("全部成功时收集结果 = " + Futures.allSuccessful(all).join());

1	全部成功时收集结果 = [a, b]

Futures.allSuccessful 里的 futures.stream().map(CompletableFuture::join) 之所以安全，是因为此时 allOf 已经正常完成，说明每个 future 都已经完成且没有失败，join() 不会阻塞也不会抛。

anyOf：第一个完成的结果（包括异常）

anyOf(CompletableFuture<?>... cfs) 返回 CompletableFuture<Object>，在任意一个任务完成的瞬间就完成，携带那个任务的结果。注意两点：

返回类型是 Object，因为静态签名无法知道你传进来的各种 future 各自是什么类型。
「完成」包括异常完成。如果第一个完成的任务是失败的，anyOf 也会失败，即使后面有任务成功。

import java.util.concurrent.CompletableFuture;

public class AnyOfDemo {

    public static void main(String[] args) {
        long started = System.nanoTime();
        CompletableFuture<String> slow = CompletableFuture.supplyAsync(() -> {
            DemoData.sleep(400);
            return "镜像 A";
        });
        CompletableFuture<String> fast = CompletableFuture.supplyAsync(() -> {
            DemoData.sleep(100);
            return "镜像 B";
        });

        CompletableFuture<Object> any = CompletableFuture.anyOf(slow, fast);
        System.out.println("anyOf = " + any.join()
                + ", 耗时 " + (System.nanoTime() - started) / 1_000_000 + "ms"
                + ", 运行时类型 = " + any.join().getClass().getSimpleName());

        CompletableFuture<Object> anyFailed = CompletableFuture.anyOf(
                CompletableFuture.failedFuture(new IllegalStateException("第一个完成的失败了")),
                CompletableFuture.supplyAsync(() -> {
                    DemoData.sleep(200);
                    return "晚到的成功";
                }));
        try {
            anyFailed.join();
        } catch (RuntimeException e) {
            System.out.println("第一个完成的是异常: " + Futures.rootCause(e));
        }
    }
}

1 2	anyOf = 镜像 B, 耗时 108ms, 运行时类型 = String 第一个完成的是异常: java.lang.IllegalStateException: 第一个完成的失败了

另一个容易忽略的问题是「赢了的是谁」。anyOf 只给你结果，不给你来源。多个镜像竞争时，你通常还想知道是哪个镜像先回来了，这时要先把每个结果打上标签再丢进 anyOf：

record Tagged(String source, String value) {}

CompletableFuture<Object> tagged = CompletableFuture.anyOf(
        CompletableFuture.supplyAsync(() -> new Tagged("镜像 A", "A")),
        CompletableFuture.supplyAsync(() -> new Tagged("镜像 B", "B")));
Tagged winner = (Tagged) tagged.join();
System.out.println("先返回的是 " + winner.source() + " = " + winner.value());

1	先返回的是镜像 A = A

异常处理三件套

贯穿本节的一个事实（JDK 25 源码可见）：一个任务抛出的异常，在进入 future 内部时会被 encodeThrowable 包装成 CompletionException（CancellationException 除外，取消走的是另一条路径）。所以你在 exceptionally / handle / whenComplete 回调里直接拿到的 Throwable，往往是 CompletionException，业务异常本身要 getCause()（可能还要再剥一层，所以前面写了 Futures.rootCause）。

exceptionally：把异常换成兜底值

exceptionally(Function<Throwable, ? extends T>) 只在「上游失败」时触发，返回一个同类型的正常值作为兜底。它只影响它返回的这个新 future，原 future 仍然是失败状态。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;

public class ExceptionDemo {

    public static void main(String[] args) {
        CompletableFuture<String> source = CompletableFuture.supplyAsync(() -> {
            throw new IllegalStateException("远程调用失败");
        });

        CompletableFuture<String> recovered = source.exceptionally(error -> "兜底值");
        System.out.println("exceptionally = " + recovered.join());
        System.out.println("原始 future 仍然是失败的: " + source.isCompletedExceptionally());
        System.out.println("exceptionally 回调里看到的异常类型 = "
                + source.exceptionally(error -> error.getClass().getSimpleName()).join());
    }
}

1
2
3

exceptionally = 兜底值
原始 future 仍然是失败的: true
exceptionally 回调里看到的异常类型 = CompletionException

如果兜底函数本身又抛异常，那返回的 future 也会异常完成。另外 JDK 12 起还有 exceptionallyCompose / exceptionallyAsync（@since 12），前者适合兜底逻辑本身又是异步调用（类似 exceptionally 版的 thenCompose），后者强制把兜底丢到别的线程。

handle：成功和失败都能拿到

handle(BiFunction<? super T, Throwable, ? extends U>) 无论成功失败都会执行：成功时第二个参数是 null，失败时第一个参数是 null。它还可以把类型从 T 换成 U，这是它比 exceptionally 强的地方。

1
2
3

CompletableFuture<String> handled = source.handle(
        (value, error) -> error == null ? "成功: " + value : "失败: " + Futures.rootCause(error));
System.out.println("handle = " + handled.join());

1	handle = 失败: java.lang.IllegalStateException: 远程调用失败

whenComplete：只看不改，但小心回调自己抛异常

whenComplete(BiConsumer<? super T, ? super Throwable>) 就是「观察者」：它拿到的结果和异常原样透传给返回的 future，返回类型仍然是 CompletableFuture<T>。

CompletableFuture<String> ok = CompletableFuture.supplyAsync(() -> "原始值");
CompletableFuture<String> observed = ok.whenComplete(
        (value, error) -> System.out.println("whenComplete 看到 value=" + value + ", error=" + error));
System.out.println("whenComplete 透传 = " + observed.join());

1 2	whenComplete 看到 value=原始值, error=null whenComplete 透传 = 原始值

但它有一个经典陷阱：whenComplete 回调里如果自己抛异常，返回的 future 就会以这个新异常异常完成，原来那个结果/异常被丢弃。所以别把可能失败的业务逻辑塞进 whenComplete，它只适合打日志、记指标这种「理论上不该失败」的动作。

try {
    CompletableFuture.supplyAsync(() -> "原始值")
            .whenComplete((value, error) -> {
                throw new IllegalStateException("回调自己炸了");
            })
            .join();
} catch (CompletionException e) {
    System.out.println("whenComplete 里抛异常后下游看到 = " + e.getCause());
}

1	whenComplete 里抛异常后下游看到 = java.lang.IllegalStateException: 回调自己炸了

谁吞掉异常，谁把异常传下去

把三件套并列看：

exceptionally：吞掉异常，返回兜底值；从此下游看到的是一个正常完成的 future。异常链在这里终结。
handle：吞掉异常，返回新值；成功失败都接管，是「把异常翻译成结果」的入口。
whenComplete：不吞异常，原样透传；但如果回调自己抛异常，就会用新异常替换掉原结果。

选择口诀：只想兜底就用 exceptionally；想根据成败产出不同类型的结果就用 handle；只想在旁边看着（记日志）就用 whenComplete，并且保证回调不抛。

最常见的静默失败

下面这行代码里任务抛出的异常没有任何人 join() / get()，也没有 exceptionally，于是它既不打印、也不打断流程，悄悄就没了：

CompletableFuture.<String>supplyAsync(() -> {
    throw new RuntimeException("没人管的异常，不写任何日志就消失了");
});
System.out.println("静默失败演示：上一行没有任何输出");

1	静默失败演示：上一行没有任何输出

每一条 CompletableFuture 链，要么有人最终 join()/get()，要么在链尾挂一个 exceptionally 记日志。没有兜底的失败就是静默失败。

等待结果：join、get 与超时

join 与 get 的差别

join() 不声明受检异常，失败时抛非受检的 CompletionException（cause 是底层异常）。
get() 声明 InterruptedException 和 ExecutionException 两个受检异常，失败时抛 ExecutionException。
底层失败原因一致，差别只在包装类型和要不要被迫处理受检异常。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.ExecutionException;

public class JoinVsGet {

    public static void main(String[] args) throws Exception {
        CompletableFuture<String> failed =
                CompletableFuture.failedFuture(new IllegalStateException("失败原因"));
        try {
            failed.join();
        } catch (CompletionException e) {
            System.out.println("join -> " + e.getClass().getSimpleName()
                    + ", cause=" + e.getCause().getClass().getSimpleName());
        }
        try {
            failed.get();
        } catch (ExecutionException e) {
            System.out.println("get -> " + e.getClass().getSimpleName()
                    + ", cause=" + e.getCause().getClass().getSimpleName());
        }
    }
}

1 2	join -> CompletionException, cause=IllegalStateException get -> ExecutionException, cause=IllegalStateException

工程上多数用 join()，因为异步链的失败通常是「运行期故障」，用非受检异常更顺手；get() 适合想在调用点显式处理 InterruptedException / ExecutionException 的场合。

get(timeout) 只放弃了等待

get(long, TimeUnit) 到点还没结果就抛 TimeoutException。它只是调用方不等了，任务本身还在跑，将来仍然会完成。这跟「取消」完全是两回事。

ExecutorService pool = Executors.newFixedThreadPool(2, DemoData.namedFactory("biz"));
CompletableFuture<String> slow = CompletableFuture.supplyAsync(() -> {
    DemoData.sleep(300);
    return "慢结果";
}, pool);
try {
    slow.get(100, TimeUnit.MILLISECONDS);
} catch (TimeoutException e) {
    System.out.println("get(timeout) -> TimeoutException, 任务还在跑, isDone=" + slow.isDone());
}
System.out.println("稍后仍然拿到了结果 = " + slow.join());

1 2	get(timeout) -> TimeoutException, 任务还在跑, isDone=false 稍后仍然拿到了结果 = 慢结果

orTimeout 与 completeOnTimeout（JDK 9 起）

这两个方法在 JDK 9 引入（源码里 @since 9，属于 Java 9 对 java.util.concurrent 的增量更新，没有独立 JEP），它们让 future 本身在超时后「自动」进入终态，而不需要调用方在外面做 get(timeout)。

orTimeout(timeout, unit)：到点仍没完成，就把这个 future 异常完成为 TimeoutException。
completeOnTimeout(value, timeout, unit)：到点仍没完成，就正常完成为给定的兜底值。

两者都返回 this（修改当前 future，而不是像大多数方法那样返回新阶段）；如果 future 已经完成，它们不会安排任何超时。

CompletableFuture<String> orTimedOut = CompletableFuture.supplyAsync(() -> {
    DemoData.sleep(300);
    return "慢结果";
}, pool).orTimeout(100, TimeUnit.MILLISECONDS);
DemoData.sleep(150);
System.out.println("orTimeout: isCompletedExceptionally=" + orTimedOut.isCompletedExceptionally()
        + ", isCancelled=" + orTimedOut.isCancelled());
try {
    orTimedOut.join();
} catch (CompletionException e) {
    System.out.println("orTimeout 的 cause = " + e.getCause());
}

1 2	orTimeout: isCompletedExceptionally=true, isCancelled=false orTimeout 的 cause = java.util.concurrent.TimeoutException

completeOnTimeout 则给你一个正常完成的兜底结果：

CompletableFuture<String> fallback = CompletableFuture.supplyAsync(() -> {
    DemoData.sleep(300);
    return "慢结果";
}, pool).completeOnTimeout("兜底值", 100, TimeUnit.MILLISECONDS);
System.out.println("completeOnTimeout = " + fallback.join()
        + ", isCancelled=" + fallback.isCancelled());

1	completeOnTimeout = 兜底值, isCancelled=false

超时之后任务还在跑

orTimeout 和 completeOnTimeout 都是「超时后给你一个结果」，但它们不会取消或打断底层任务：那个 sleep(300) 的 supplier 照样跑完，只是它的结果没地方投了。如果底层任务持有连接、锁、占用线程，超时并不会释放它们。它们和 cancel / 结构化并发的差别就在这里，下一节和「取消语义」会展开。

线程池怎么选

默认池的共同问题

前面说过默认池是公共 ForkJoinPool。这里再强调它的两类典型坑：

阻塞 IO 占满 worker。ForkJoinPool 靠工作窃取来调度，worker 数等于并行度（通常 CPU-1）。一旦几个 worker 卡在 sleep / 网络等待 / 数据库调用上，其他所有默认提交者的任务都跟着排队，吞吐直接塌。
无法隔离。JVM 里所有用 parallelStream() 和默认 CompletableFuture 的代码共享一个池，一个组件打爆它，别的组件遭殃，还查不出是谁干的。

所以生产里凡是会阻塞的任务，都要显式指定执行器。CPU 密集的小任务可以用自定义的 ForkJoinPool，IO 密集的用业务 ThreadPoolExecutor，或者在 JDK 21（JEP 444）之后用虚拟线程池把阻塞等一等这件事交给廉价的虚拟线程。

无界队列与有界队列的不同表现

同样 6 个耗时 100ms 的任务，丢给「2 线程 + 无界队列」和「1~2 线程 + 有界队列」的池，表现完全不同。

import java.util.List;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;

public class PoolDemo {

    public static void main(String[] args) {
        ThreadPoolExecutor unbounded = new ThreadPoolExecutor(2, 2, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(), DemoData.namedFactory("unbounded"));
        long started = System.nanoTime();
        List<CompletableFuture<String>> futures = IntStream.rangeClosed(1, 6)
                .mapToObj(i -> CompletableFuture.supplyAsync(() -> {
                    DemoData.sleep(100);
                    return "任务" + i;
                }, unbounded))
                .toList();
        CompletableFuture.allOf(futures.toArray(CompletableFuture[]::new)).join();
        System.out.println("6 个 100ms 任务 / 2 线程 = "
                + (System.nanoTime() - started) / 1_000_000 + "ms（队列无界，只会越排越久）");

        ThreadPoolExecutor bounded = new ThreadPoolExecutor(1, 2, 0L, TimeUnit.MILLISECONDS,
                new ArrayBlockingQueue<>(1), DemoData.namedFactory("bounded"));
        try {
            for (int i = 1; i <= 6; i++) {
                CompletableFuture.supplyAsync(() -> {
                    DemoData.sleep(200);
                    return "x";
                }, bounded);
            }
        } catch (RejectedExecutionException e) {
            System.out.println("队列满 + 线程满时，supplyAsync 在调用线程同步抛出 "
                    + e.getClass().getSimpleName());
        }

        unbounded.shutdown();
        bounded.shutdown();
    }
}

1 2	6 个 100ms 任务 / 2 线程 = 323ms（队列无界，只会越排越久）队列满 + 线程满时，supplyAsync 在调用线程同步抛出 RejectedExecutionException

（毫秒数是实测值，随机器浮动；关键是这个量级——2 个线程跑 6 个 100ms 任务必然是 3 轮，约 300ms，而不是 100ms。）

有界队列 + 默认拒绝策略（AbortPolicy）的关键细节：supplyAsync 内部是 e.execute(...) 直接提交，所以拒绝时 RejectedExecutionException 是在调用 supplyAsync 的那个线程同步抛出来的，而不是出现在返回的 future 里。想在 future 里 exceptionally 接住它是不可能的：要在调用点 try/catch，或者换成会回退的拒绝策略（比如 CallerRunsPolicy 把任务交给调用线程跑，天然限流）。

无界队列则相反：永不拒绝，但队列越排越长，请求的延迟随堆积水涨船高。生产里要么给队列设上限并配好拒绝策略，要么用 Semaphore 这类信号量在源头限流。

与 Spring @Async 的关系

Spring 的 @Async 方法在容器里由配置的 TaskExecutor 执行。Spring Boot 在没有自定义 Executor 时会自动配置 applicationTaskExecutor：默认是 ThreadPoolTaskExecutor（官方文档写的是「8 个核心线程，可按负载伸缩」，且只有把 spring.task.execution.pool.queue-capacity 设成有界值，max-size 才会生效——默认队列是无界的，所以默认配置下池基本不会超过 8 线程）；如果开启 spring.threads.virtual.enabled=true（Java 21+），则会换成基于虚拟线程的 SimpleAsyncTaskExecutor。

几个和 CompletableFuture 直接相关的点：

返回类型是 CompletableFuture<T> 的 @Async 方法，调用方拿到的就是一个可继续编排的 future，异常会进入这个 future（join/get 时抛出来），而不是吞掉。
返回 void 的 @Async 方法抛出的异常不会进入任何 future，默认只由 AsyncUncaughtExceptionHandler 处理（默认实现只记日志）。
自调用不走代理：在同一个类里 this.asyncMethod() 不会异步，直接同步执行，这也是 @Async 的经典坑。
你在 CompletableFuture 链里手动传的 Executor，和 @Async 用的 applicationTaskExecutor 是两套东西，别指望它们共享限流；要么统一，要么各自独立配好容量。

delayedExecutor

CompletableFuture.delayedExecutor(delay, unit[, executor])（@since 9）返回一个「延迟后才把任务交给基底执行器」的 Executor，适合「先等一个固定时间再开始干活」这种调度，比如重试前的退避。

Executor delayed = CompletableFuture.delayedExecutor(200, TimeUnit.MILLISECONDS, pool);
long before = System.nanoTime();
CompletableFuture<String> delayedTask = CompletableFuture.supplyAsync(() -> "延迟结果", delayed);
System.out.println("delayedExecutor = " + delayedTask.join()
        + ", 耗时 " + (System.nanoTime() - before) / 1_000_000 + "ms");
System.out.println("delayedExecutor 默认基底线程 = "
        + CompletableFuture.supplyAsync(() -> Thread.currentThread().getName(),
                CompletableFuture.delayedExecutor(0, TimeUnit.MILLISECONDS)).join());

1 2	delayedExecutor = 延迟结果, 耗时 205ms delayedExecutor 默认基底线程 = ForkJoinPool.commonPool-worker-1

不传基底 Executor 的重载，延迟结束后还是落回公共池（所以第二个输出是 ForkJoinPool.commonPool-worker-1），生产里要传基底执行器，尤其是延迟后要跑阻塞 IO 的时候。

上下文传递

ThreadLocal 为什么会丢

ThreadLocal 是「按线程隔离」的。你在请求线程里 REQUEST_ID.set("req-1")，然后 supplyAsync 把任务丢给池里的另一个线程，那个线程里 get() 自然是 null。SLF4J 的 MDC 底层就是一个 ThreadLocal 的 Map，所以「异步后日志丢了 traceId」是同一个原因。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class ContextDemo {

    static final ThreadLocal<String> REQUEST_ID = new ThreadLocal<>();

    static final InheritableThreadLocal<String> INHERITED_ID = new InheritableThreadLocal<>();

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2, DemoData.namedFactory("biz"));

        REQUEST_ID.set("req-1");
        System.out.println("主线程 REQUEST_ID = " + REQUEST_ID.get());
        System.out.println("commonPool 里 REQUEST_ID = "
                + CompletableFuture.supplyAsync(REQUEST_ID::get).join());
        System.out.println("业务池里 REQUEST_ID = "
                + CompletableFuture.supplyAsync(REQUEST_ID::get, pool).join());

        pool.shutdown();
    }
}

1
2
3

主线程 REQUEST_ID = req-1
commonPool 里 REQUEST_ID = null
业务池里 REQUEST_ID = null

InheritableThreadLocal 救不了线程池

InheritableThreadLocal 只能在「创建子线程的那一刻」把父线程的值复制过去。线程池里的线程是复用很久的：要么在 set 之前就创建好了（继承不到，得到 null），要么在某次任务里继承过一次旧值，之后父线程改了值它也看不到（读到过期脏值）。

ExecutorService fresh = Executors.newSingleThreadExecutor(DemoData.namedFactory("fresh"));
INHERITED_ID.set("req-A");
System.out.println("InheritableThreadLocal 建线程时的值 = "
        + CompletableFuture.supplyAsync(INHERITED_ID::get, fresh).join());
INHERITED_ID.set("req-B");
System.out.println("改值之后再提交 = "
        + CompletableFuture.supplyAsync(INHERITED_ID::get, fresh).join());
System.out.println("在 set 之前就建好的池线程 = "
        + CompletableFuture.supplyAsync(INHERITED_ID::get, pool).join());
fresh.shutdown();

1
2
3

InheritableThreadLocal 建线程时的值 = req-A
改值之后再提交 = req-A
在 set 之前就建好的池线程 = null

所以 InheritableThreadLocal 只适合「一次性派生子线程」的场景，和线程池复用天然冲突。

三种可行的传法

显式传参。把 requestId 当普通参数传给任务方法，最笨也最可靠，上下文就是方法入参，不依赖任何线程魔法。缺点是要改签名、层层透传。
包装 Executor。在提交时捕获上下文，执行时恢复，执行完清理，把「上下文传递」收敛到一个包装器里。这是框架（如 Spring 的 TaskDecorator、Micrometer 的 Context 传播）最常用的做法。前面 Futures.propagating 就是一个最小实现：

1
2
3

System.out.println("包装后 REQUEST_ID = "
        + CompletableFuture.supplyAsync(REQUEST_ID::get,
                Futures.propagating(REQUEST_ID, pool)).join());

1	包装后 REQUEST_ID = req-1

注意 finally 里的清理：不 remove() 的话，这个值会残留在池线程上，下一次复用时泄漏到不相关的任务里，这比「丢失」更危险。

ScopedValue（JDK 25 起正式）。java.lang.ScopedValue 在 JDK 25 随 JEP 506 转正，它是 ThreadLocal 的现代替代：不可变、无 set、生命周期由代码块（where(...).run(...)）界定，且能被 StructuredTaskScope 派生的子任务继承。但它解决的是「作用域内传播」的问题，不是线程池任务的上下文传递——提交给池的任务不属于这个作用域的子线程。请求级上下文的传递，在 CompletableFuture 场景里仍然要用显式参数或包装 Executor。

thenApply 与 thenApplyAsync 的执行线程差别

这节已经在「链式变换」里验证过了，放到上下文传递这里是因为它和上下文传播直接相关：回调在哪个线程跑，决定了它能不能看到你在提交线程里设的 ThreadLocal。不带 Async 的回调可能跑在完成上游的池线程上，也可能跑在调用线程上；带 Async 的一定跑在 Executor 上。上下文要么在提交时就显式捕获好，要么靠包装 Executor 在任务边界上恢复，不能赌「回调恰好跑在设置上下文的那个线程」。

取消语义

cancel 做了什么

CompletableFuture.cancel(boolean mayInterruptIfRunning) 的语义和 Future 完全不同。它的 javadoc 明确写着：mayInterruptIfRunning 这个参数在这个实现里没有任何效果，因为中断不被用来控制处理过程。它做的事情等价于：如果还没完成，就把 future 异常完成为 CancellationException，并让下游尚未完成的依赖阶段跟着异常完成。

public boolean cancel(boolean mayInterruptIfRunning) {
    boolean cancelled = (result == null) &&
        internalComplete(new AltResult(new CancellationException()));
    postComplete();
    return cancelled || isCancelled();
}

（这是 JDK 25 源码的原文。返回值的含义是「现在是否处于已取消状态」，所以对已经取消过的 future 再 cancel 仍返回 true，对已经正常完成的返回 false。）

对已经开始的任务无效

因为 cancel 不打断线程，一个已经在跑的任务不会被停掉：任务体照常执行到最后，只是它的结果被丢弃，下游拿到的是 CancellationException。真正能被 cancel 阻止执行的，是「提交了但还没开始跑」的任务。

import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class CancelDemo {

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2, DemoData.namedFactory("biz"));

        CompletableFuture<String> running = CompletableFuture.supplyAsync(() -> {
            DemoData.sleep(200);
            System.out.println("任务体在 cancel 之后仍然跑完");
            return "值";
        }, pool);
        DemoData.sleep(50);
        System.out.println("对已开始的任务 cancel(true) 返回 = " + running.cancel(true));
        System.out.println("isCancelled = " + running.isCancelled());
        try {
            running.join();
        } catch (CancellationException e) {
            System.out.println("取消之后 join -> CancellationException, isDone=" + running.isDone());
        }

        ThreadPoolExecutor single = new ThreadPoolExecutor(1, 1, 0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(), DemoData.namedFactory("single"));
        single.submit(() -> {
            DemoData.sleep(300);
            return "占住唯一的线程";
        });
        CompletableFuture<String> queued = CompletableFuture.supplyAsync(() -> {
            System.out.println("排队中被取消的任务体不应该执行");
            return "值";
        }, single);
        System.out.println("排队中 cancel 返回 = " + queued.cancel(false));
        DemoData.sleep(350);
        System.out.println("已取消的 future 再 cancel = " + queued.cancel(false)
                + ", 正常完成的 future 再 cancel = "
                + CompletableFuture.completedFuture("已完成").cancel(false));

        single.shutdown();
        pool.shutdown();
    }
}

对已开始的任务 cancel(true) 返回 = true
isCancelled = true
取消之后 join -> CancellationException, isDone=true
排队中 cancel 返回 = true
任务体在 cancel 之后仍然跑完
已取消的 future 再 cancel = true, 正常完成的 future 再 cancel = false

关键对照：queued 那个任务被取消时还没开始跑（单线程池被占住），所以它的 supplier 从头到尾没执行（没有打印）；而 running 那个任务体在 cancel 之后照样打印。至于资源清理（关连接、释放锁），只能靠任务体自己在检查中断标志或捕获 CancellationException 后处理，cancel 不会替你收尾。

用 isCancelled 配合超时做整体放弃

orTimeout 给的是 TimeoutException，isCancelled() 是 false，语义是「超时降级」；而如果你要的是「超时后整体放弃这组任务」，就要用 cancel，让下游以 CancellationException 结束，并且用 isCancelled() 来判断「这组是不是被我们主动放弃的」。前面 Futures.withDeadline 就是这个模式：到点没完成就 cancel，完成或失败就撤销定时器。

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;

public class DeadlineDemo {

    public static void main(String[] args) {
        ExecutorService pool = Executors.newFixedThreadPool(2, DemoData.namedFactory("biz"));
        ScheduledExecutorService scheduler =
                Executors.newSingleThreadScheduledExecutor(DemoData.namedFactory("deadline"));

        CompletableFuture<String> future = Futures.withDeadline(
                CompletableFuture.supplyAsync(() -> {
                    DemoData.sleep(500);
                    return "永远等不到的结果";
                }, pool),
                java.time.Duration.ofMillis(100),
                scheduler);
        try {
            future.join();
        } catch (RuntimeException e) {
            System.out.println("超时后 = " + e.getClass().getSimpleName()
                    + ", isCancelled=" + future.isCancelled());
        }

        pool.shutdown();
        scheduler.shutdown();
    }
}

1	超时后 = CancellationException, isCancelled=true

需要整组放弃时，把同一批 future 放进一个列表，定时器到点后逐个 cancel(false) 即可。cancel 只把 future 置为取消态、阻断下游，不会打断已经在跑的底层任务；要回收底层资源，要靠任务体对中断/取消的配合，或者换用结构化并发里那种「短路即中断并等待子任务结束」的机制。

与结构化并发对比

StructuredTaskScope 的现状

java.util.concurrent.StructuredTaskScope 是 Project Loom 提出的结构化并发 API，它的目标正是解决 CompletableFuture 的几处软肋：任务生命周期无约束、失败不会自动短路、cancel 不打断运行中的任务、线程转储看不出任务父子关系。

需要先把它的成熟度说清楚（这是版本敏感信息，以下均核实过）：

JDK 19 / 20 以孵化模块形式交付（JEP 428、JEP 437）。
JDK 21 转成预览 API（JEP 453），fork 的返回类型从 Future 改为 Subtask。
JDK 22、23、24 连续再预览（JEP 462、JEP 480、JEP 499）。
JDK 25 第五次预览（JEP 505），把公开构造器换成静态工厂方法 StructuredTaskScope.open(...)，并引入 Joiner 完成策略。
JDK 26 第六次预览（JEP 525），给 Joiner 增加 onTimeout()，allSuccessfulOrThrow() 改为返回结果列表，anySuccessfulResultOrThrow() 改名 anySuccessfulOrThrow()。
JDK 27 第七次预览（JEP 533），给 StructuredTaskScope 与 Joiner 增加第三个类型参数 R_X 以显式编码 join() 可抛的异常类型，标准 Joiner 从抛 FailedException 改为抛 ExecutionException，移除 awaitAll()，onTimeout() 改为 timeout()。

截至 JDK 27（2026 年 9 月），它仍是预览特性，API 仍在变动，编译运行都需要 --enable-preview（javac --release 27 --enable-preview / java --enable-preview）。 生产代码直接依赖它要冒 API 变更的风险；相比之下 CompletableFuture 自 JDK 8 起就是正式 API。

下面这段在 JDK 25 上以 --enable-preview 编译运行，展示了三种用法与失败短路：

import java.util.List;
import java.util.concurrent.StructuredTaskScope;
import java.util.concurrent.StructuredTaskScope.Joiner;
import java.util.concurrent.StructuredTaskScope.Subtask;

/** 需要 JDK 25 + --enable-preview 才能编译运行。 */
public class StructuredConcurrencyDemo {

    static String findStock(String sku) {
        DemoData.sleep(120);
        return sku + " 有货";
    }

    public static void main(String[] args) throws Exception {
        try (var scope = StructuredTaskScope.open()) {
            Subtask<DemoData.User> user = scope.fork(() -> DemoData.findUser("u-1"));
            Subtask<List<DemoData.Order>> orders = scope.fork(() -> DemoData.findOrders("u-1"));
            scope.join();
            System.out.println("默认策略 = " + user.get().name() + " / " + DemoData.totalAmount(orders.get()));
        }

        try (var scope = StructuredTaskScope.open(Joiner.<String>anySuccessfulResultOrThrow())) {
            scope.fork(() -> {
                DemoData.sleep(300);
                return "镜像 A";
            });
            scope.fork(() -> {
                DemoData.sleep(100);
                return "镜像 B";
            });
            System.out.println("第一个成功 = " + scope.join());
        }

        try (var scope = StructuredTaskScope.open(Joiner.<String>allSuccessfulOrThrow())) {
            scope.fork(() -> findStock("sku-1"));
            scope.fork(() -> findStock("sku-2"));
            System.out.println("全部成功 = " + scope.join().map(Subtask::get).toList());
        }

        try (var scope = StructuredTaskScope.open()) {
            scope.fork(() -> {
                DemoData.sleep(500);
                return "慢任务";
            });
            scope.fork(() -> {
                DemoData.sleep(50);
                throw new IllegalStateException("库存服务失败");
            });
            scope.join();
        } catch (StructuredTaskScope.FailedException e) {
            System.out.println("子任务失败 = " + e.getClass().getSimpleName() + ", cause=" + e.getCause());
        }
    }
}

默认策略 = 张三 / 300
第一个成功 = 镜像 B
全部成功 = [sku-1 有货, sku-2 有货]
子任务失败 = FailedException, cause=java.lang.IllegalStateException: 库存服务失败

对比几个关键维度：

维度	CompletableFuture	StructuredTaskScope
任务生命周期	无约束，可跨方法、跨请求传递	限定在 `try-with-resources` 块内，块退出即等待收尾
失败传播	只影响依赖该 future 的下游阶段	短路：任一子任务失败 → 中断并取消同级子任务，`join` 抛异常
取消	`cancel` 不打断运行中的任务	短路/`close` 会中断子任务并等待它们结束
执行线程	依赖 Executor（默认公共池）	默认每个子任务一个虚拟线程
结果收集	手动（`allOf` + 逐个 `join`）	`Subtask.get()` / `Joiner` 直接给结果
可观察性	线程转储看不出任务父子关系	线程转储体现父子层级
成熟度	JDK 8 起正式	截至 JDK 27 仍为预览（JEP 533）

什么时候仍然该用 CompletableFuture

结构化并发解决的是「一组短命任务在同一段代码里 fork/join」的场景，尤其是基于虚拟线程、失败需要短路的那一类。CompletableFuture 在下面这些场景依然更合适：

任务要跨方法、跨对象、甚至跨请求边界传递，生命周期无法收敛到某个代码块时。
需要「先把 future 存起来，稍后再接回调」这类自由组合时。
你依赖的是 JDK 8+ 的稳定 API，不想引入预览特性。
编排粒度很细、回调很多，纯异步链的表达力比 try-with-resources + fork/join 更贴合。

二者也不是二选一：在结构化作用域内部，子任务本身仍可以是 CompletableFuture 的编排；反过来，把结构化并发塞进 CompletableFuture 的链子里通常没有意义。选型的关键是「任务之间是否共享一个明确的、词法上的生命周期」。

总结

这篇文章按「创建 → 编排 → 组合 → 异常 → 等待/超时 → 线程池 → 上下文 → 取消 → 结构化并发」的顺序，把 CompletableFuture 的语义边界过了一遍：

默认执行器是全局共享的公共 ForkJoinPool（并行度 CPU-1），阻塞 IO 必须显式传 Executor。
thenApply/thenAccept/thenRun 的差别只在入参与返回值；不带 Async 的回调跑在哪个线程由「上游何时完成」决定。
thenCompose 把异步依赖拍平，避免 CompletableFuture<CompletableFuture<T>>。
allOf 返回 Void、只给一个异常、且「没人 join 就没异常」；要逐个收集成功失败，先用 handle 收敛再 allOf。
anyOf 返回 Object，第一个完成的若是失败，整体也失败，要「知道谁赢了」得自己打标签。
exceptionally 兜底、handle 接管、whenComplete 透传但回调抛异常会替换原结果；链尾必须有人 join 或记日志，否则异常静默消失。
join 抛非受检 CompletionException，get 抛受检异常；orTimeout/completeOnTimeout（JDK 9 起）让 future 自己超时，但不取消底层任务。
ThreadLocal 在池线程里丢失，InheritableThreadLocal 只在建线程时继承；可靠的传法是显式参数或包装 Executor，ScopedValue（JDK 25 正式）适合作用域内传播而非线程池传参。
cancel 不打断运行中的任务，只阻断下游；「整体放弃」要用 cancel + isCancelled 而非 orTimeout。
结构化并发 StructuredTaskScope 截至 JDK 27 仍是预览（JEP 533，第七次预览），失败短路、子任务取消、线程转储父子关系是它相对 CompletableFuture 的核心优势。