《Java 并发编程·线程与内存模型》13.1 CompletableFuture 的创建与组合

承接的问题

第 12 章的分治是「一个任务自己裂成很多小任务」；这一章反过来，是调用方手里有三个互不依赖的结果，要把它们拼成一个结果。

借阅系统里最典型的一段：会员预约成功之后，要同时给 mail、sms、push 三个下游各发一条通知。三件事没有先后依赖，但都必须做完才能给用户回「已通知」。用 ExecutorService 写出来的版本通常是这样：

Future<String> mail = pool.submit(() -> gateway.send("mail", memberId, msg));
Future<String> sms = pool.submit(() -> gateway.send("sms", memberId, msg));
Future<String> push = pool.submit(() -> gateway.send("push", memberId, msg));
// 然后三个 get()，每个都可能阻塞，异常还得逐个 try/catch

能跑，但有三个问题：get() 是阻塞取值，拿不到结果就没有任何后续动作可以注册；三路结果要手工收集，异常处理是三个重复的 catch；最关键的是不能组合——「拿到结果之后再做一步异步操作」这件事，Future 表达不出来。

CompletableFuture 解决的就是这个：它既是 Future，也是 CompletionStage，可以往一个还没完成的结果上挂「完成之后做什么」。

创建：默认执行器到底是谁

先把创建方式列全（表中 CF 代指 CompletableFuture）。这一节的实验都在 JDK 25 上跑，--release 25。

「默认执行器」是最容易被含糊过去的一点。实测：

== 默认执行器 ==
availableProcessors = 10
commonPool.parallelism = 9
  supplyAsync 线程 = ForkJoinPool.commonPool-worker-1，daemon = true

四件事可以直接从这几行读出来：

1. 不传执行器时用的是 ForkJoinPool.commonPool()，线程名以 ForkJoinPool.commonPool-worker- 开头。
2. 这台机器 10 核，commonPool 的并行度是 9，不是 10。规则是 max(1, availableProcessors - 1)：留一个核给调用线程。
3. commonPool 的线程是 daemon，不会阻止 JVM 退出；这也意味着进程退出时未完成的异步任务会被直接丢掉。
4. 如果 commonPool 的并行度小于 2（单核机器，或把 java.util.concurrent.ForkJoinPool.common.parallelism 设成 1），supplyAsync 会退化成「每个任务新建一个线程」，行为完全不同。

commonPool 还有两个不能忽略的性质：它是全 JVM 共享的，parallelStream() 用的也是它（12.2 已经踩过这个坑）；它的大小由系统属性在首次使用时决定，之后不可调；遇到 ManagedBlocker 时会有临时补偿线程，但不保证。

结论很直接：业务里的 IO 型异步任务不要用默认执行器，传自己的池。13.3 会把这个池和上下文传递一起处理。

组合：thenApply、thenCompose 与 thenCombine

三个方法的名字很像，差别在「下一步返回什么」：

// 同步转换：T -> U
CompletableFuture<Integer> len = cf.thenApply(String::length);

// 异步接续：T -> CompletionStage<U>，自动摊平
CompletableFuture<Integer> flat = cf.thenCompose(s -> CompletableFuture.supplyAsync(() -> s.length()));

// 两路合并：(T, U) -> V
CompletableFuture<String> both = cfA.thenCombine(cfB, (a, b) -> a + b);

thenApply 与 thenCompose 的差别不是风格问题。如果 thenApply 的函数返回了一个 CompletableFuture，得到的是嵌套类型 CompletableFuture<CompletableFuture<T>>：

== thenApply 返回嵌套 ==
  nested 类型 = CompletableFuture
  nested.join().join() = quota:member-42
  compose 结果 = quota:member-42

nested.join() 拿到的是一个 CompletableFuture 对象本身——它不会替你等第二层。用 thenCompose 才是「等第一步完成、再执行第二步、最后得到第二步的值」。判断标准只有一条：下一步的动作是否依赖上一步的值，并且它本身也是异步的。是，就用 thenCompose；只是纯计算，用 thenApply。

thenCombine 则是两路已经独立的异步结果做归并。实测两个任务分别 80ms 和 60ms：

== thenCombine ==
  member-42/quota=5，耗时 91ms（两路并行）

91ms 接近较慢那一路的 80ms，而不是 140ms。

回调在哪个线程上跑

这是 CompletableFuture 最容易被误解的地方：thenApply 不是提交任务，它是注册回调。谁来执行回调取决于 future 的状态：

已完成 future 上的 thenApply 线程 = main
已完成 future 上的 thenApplyAsync 线程 = ForkJoinPool.commonPool-worker-1
未完成 future 的 thenApply 由 = planner-finisher 执行

三种情况对应三条规则：

• future 已经完成：thenApply 直接在调用 thenApply 的线程上同步执行，这里是 main。
• future 尚未完成：回调被压进该 future 的栈，由把它完成的那个线程执行，这里是 planner-finisher。如果它是 commonPool worker 完成的，回调就跑在 commonPool 上。
• thenApplyAsync：不管当前状态，一律提交到执行器；不传执行器就是 commonPool。

所以 thenApply 里不能写阻塞调用。你在里面 Thread.sleep 或者发一个同步 HTTP 请求，阻塞的是「完成者线程」——可能是一个 commonPool worker，也可能是别的业务线程。要阻塞就用 thenApplyAsync 并传自己的池。

取结果：join 与 get

组合链的最后总要有一个终点。join() 与 get() 的差别不只是受检异常：

cf.join();                          // 非受检，抛 CompletionException；不可中断
cf.get();                           // 抛 InterruptedException / ExecutionException
cf.get(200, TimeUnit.MILLISECONDS); // 再加 TimeoutException

• join() 不响应中断：等待期间被 interrupt 不会抛 InterruptedException，中断标志会被留到返回之后再体现。适合「必须拿到结果」的场景。
• get() 可中断、可带超时，代价是每个调用点都要处理受检异常。
• 两者都会解开一层 CompletionException 包装，所以 catch 里拿到的 e.getCause() 通常是原始业务异常。但在回调内部（exceptionally/handle）看到的形态不一定——13.2 会实测这个差异。

扇出：allOf 与 anyOf

三路下游的扇出是 allOf 的标准用法：

List<CompletableFuture<String>> futures = new ArrayList<>();
for (int i = 0; i < channels.size(); i++) {
    String ch = channels.get(i);
    long cost = costs[i];
    futures.add(CompletableFuture.supplyAsync(() -> send(ch, cost), pool));
}
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
List<String> results = futures.stream().map(CompletableFuture::join).toList();

allOf 返回的是 CF<Void>，它不携带结果。结果仍然在原来那三个 future 里，要自己 join() 收集。这是 API 设计上的取舍：allOf 的元素类型可以各不相同，没法用一个泛型表达。

实测（三路耗时 120ms / 90ms / 60ms，串行一次、并行两轮）：

== 三路扇出 allOf ==
  串行: mail:ok sms:ok push:ok，耗时 290ms
  并行 round1: [mail:ok, sms:ok, push:ok]，耗时 136ms（max=120ms）
  并行 round2: [mail:ok, sms:ok, push:ok]，耗时 120ms（max=120ms）

串行是 290ms（三路之和加调度开销），并行是 120–136ms（约等于最慢那一路的 120ms 加调度抖动）。扇出的收益就是「总耗时从求和变成求最大值」，前提是池里有足够的线程。

池容量决定扇出的上限

同一个扇出，只换池的大小：

池大小 1 -> 290ms / 307ms
池大小 2 -> 160ms / 161ms
池大小 3 -> 128ms / 124ms

池只有 1 个线程时，并行完全退化成串行；2 个线程时，最短的那一路要等前面的任务让出线程，总耗时接近「90+60」而不是最慢的 120ms。扇出的池容量至少要等于同时要跑的分支数，否则 allOf 只是写法上的并行。反过来说，在这个扇出场景里，池开得比分支数大也没有额外收益，只会多出闲置线程。

allOf 不是 fail-fast

很多资料说「一路失败，allOf 立刻失败」，实测不是。三路里 push 在 20ms 抛异常，sms 300ms，mail 1500ms（输出里 a=push、b=sms、c=mail）：

3 路 allOf 失败于 +1511ms（最慢 1500ms）
  b 完成? true，c 完成? true
anyOf 遇失败于 +11ms，cause=IllegalStateException

两个结论：

• allOf 会等所有任务结束才报错。它等齐了 1500ms 的 mail，才把 push 的异常抛出来。所以「一路慢下游会把整条链拖到最慢那一路，哪怕另一路早就失败了」。
• 失败不会取消其余任务。sms、mail 都正常跑完了。allOf 只负责汇合，没有取消传播——取消传播是 StructuredTaskScope 的职责（JDK 25 preview，15.2 展开）。

anyOf 相反：第一个完成的任务就决定结果，失败也算完成。实测 11ms 就失败，cause 是 IllegalStateException。所以用 anyOf 做「多副本取最快」时，必须在每个分支上先 handle 掉失败，否则最快的那个失败会直接终结整条链。

在 libraryd-conc 里落地

AsyncNotifyService 负责预约成功后的三路通知。两个要点：每个分支自己兜住异常，扇出用独立的池。

public final class AsyncNotifyService {
    private final ExecutorService pool;
    private final NotificationGateway gateway;

    public AsyncNotifyService(ExecutorService pool, NotificationGateway gateway) {
        this.pool = pool;
        this.gateway = gateway;
    }

    public NotifyReport notify(String memberId, String message) {
        Map<String, CompletableFuture<ChannelResult>> futures = new LinkedHashMap<>();
        for (String channel : List.of("mail", "sms", "push")) {
            futures.put(channel, CompletableFuture
                    .supplyAsync(() -> gateway.send(channel, memberId, message), pool)
                    .handle((receipt, error) -> error == null
                            ? ChannelResult.ok(channel, receipt)
                            : ChannelResult.failed(channel, error)));
        }
        CompletableFuture.allOf(futures.values().toArray(new CompletableFuture[0])).join();
        List<ChannelResult> results = futures.values().stream()
                .map(CompletableFuture::join)
                .toList();
        return new NotifyReport(memberId, results);
    }
}

handle 把「成功值 + 异常」压成同一种 ChannelResult，于是 allOf 永远不会异常完成，join() 也不需要 try/catch。这是扇出聚合的通用写法：汇合点之前，把每个分支的失败都变成值。三个钩子的差别在 13.2 细讲。

这些方法放在 AsyncLoanService 里，底层仍然是第 10 章的 PooledLoanService。预约本身是「先并行查两个前置条件，再基于两者写一次」：

public CompletableFuture<Reservation> reserveAsync(String isbn, String memberId) {
    CompletableFuture<Book> bookF = CompletableFuture.supplyAsync(
            () -> books.findByIsbn(isbn).orElseThrow(
                    () -> new LoanException(ErrorCode.BOOK_NOT_FOUND, "isbn=" + isbn)), pool);
    CompletableFuture<Member> memberF = CompletableFuture.supplyAsync(
            () -> members.find(memberId).orElseThrow(
                    () -> new LoanException(ErrorCode.QUOTA_EXCEEDED, "no quota for member=" + memberId)), pool);

    return bookF.thenCombine(memberF, Availability::new)
            .thenCompose(a -> CompletableFuture.supplyAsync(() -> reserveNow(a), pool));
}

private Reservation reserveNow(Availability a) {
    if (a.book().stock() <= 0) {
        throw new LoanException(ErrorCode.OUT_OF_STOCK, "isbn=" + a.book().isbn());
    }
    if (a.member().quota() <= 0) {
        throw new LoanException(ErrorCode.QUOTA_EXCEEDED, "member=" + a.member().memberId());
    }
    Instant dueAt = Instant.now().plus(30, ChronoUnit.DAYS);
    Reservation r = new Reservation(UUID.randomUUID().toString(),
            a.book().isbn(), a.member().memberId(), dueAt);
    loans.save(new Loan(r.loanId(), r.isbn(), r.memberId(), r.dueAt()));
    books.save(new Book(a.book().isbn(), a.book().title(), a.book().stock() - 1));
    return r;
}

thenCombine 让两次查询并行（总耗时 = 较慢那次），thenCompose 让写操作在前置条件齐了之后才提交。扣减本身的线程安全不在这里解决——沿用第 9 章的 LockedLoanService，这里只负责编排；把编排和互斥混在一层，是后面很多重复扣减事故的起点。

ErrorCode 里没有「会员不存在」这一项，所以会员查询为空时按「无可用配额」处理；这是枚举定义的缺口，真实项目应该补一个 MEMBER_NOT_FOUND，而不是让 QUOTA_EXCEEDED 承担两种含义。

NotificationGateway、ChannelResult、NotifyReport 是端口与返回值类型，members 是会员查询端口，都不在 COMMON 的领域契约内；对领域类型的读写仍然只有 BookRepository、LoanRepository 和 Reservation/Loan 的既有字段。

常见错误与边界

什么时候不要用 CompletableFuture。 只有一两个任务、下一步又必须等结果时，ExecutorService.submit 加 Future.get 更直白，调试栈也短；CompletableFuture 的栈在异常时经常只剩一个 CompletionException，排查成本明显更高。需要重试、熔断、背压、限流这类编排语义时，也不该用几十行 CompletableFuture 拼一个框架——那是 Reactor 这类库的职责（卷三第 7 章只作对照，不引入依赖）。还有一种情况必须停下来：编排跨越了事务边界。异步任务不继承调用线程的事务上下文，如果写库在 supplyAsync 里，事务注解和连接绑定都可能失效；这种场景应该把事务收在一个同步方法内部，异步只负责调用它。

小结

• supplyAsync 不传执行器时用 ForkJoinPool.commonPool()；本机 10 核，parallelism 实测为 9，线程是 daemon。并行度小于 2 时会退化成每任务一个新线程。
• thenApply 是注册回调而不是提交任务：已完成时在调用线程跑，未完成时由完成者线程跑，thenApplyAsync 才进池。回调里不能阻塞。
• thenApply 返回 future 会得到嵌套类型，接续异步操作要用 thenCompose；两路独立结果合并用 thenCombine，实测 80ms + 60ms 的任务合并耗时 91ms。
• 三路扇出用 allOf，总耗时从串行 290ms 降到 120ms。但 allOf 不 fail-fast：实测等了最慢的 1500ms 才抛异常，且不取消其余任务。anyOf 反之，首个完成即定局，失败也算。
• 扇出聚合的通用写法是「每个分支先 handle 成值，再 allOf」，这样汇合点永远不需要 try/catch。

创建和组合解决了「怎么把结果拼起来」，但没有回答「某一路永远不返回怎么办」。下一节处理异常、超时与取消——那也是异步代码真正开始变难的地方。

阅读导航：上一节 12.3 分治拆分、阈值与异常 ｜ 下一节 13.2 异常、超时与取消