本节目标：用
CompletableFuture把 libraryd 的独立查询并行起来，并把超时、异常和取消写成可检查的代码。
适用版本：JDK 25（Temurin 25.0.5）；Javalin 7.2.4；HikariCP 7.1.0。
承接的问题
9.3 的 BookSearchService 用 Future.get(2, TimeUnit.SECONDS) 把查询丢进工作池，然后请求线程原地等结果。它解决了拒绝处理，但没有解决等待本身：搜索要查一页书目，还要查总数，两个 SQL 是独立的，却串行执行；一个 20 条的页面可能要等 400ms，而不是 200ms。
Future 的问题不只是「会阻塞」，而是「不能组合」。拿到一个 Future 之后，你只能 get() 或轮询 isDone()，没法写「等这两个都完成再合并」「如果超时就返回缓存」「失败时走降级」。这些逻辑只能手写在调用方，越写越像回调地狱。
CompletableFuture 把「完成」「转换」「合并」「异常」「超时」做成阶段（stage），可以链式表达。但它的 API 很像，语义差别很大：thenApply 和 thenCompose 一字之差，类型完全不同；allOf 看起来像「任一失败就失败」，实际上会等所有任务结束；orTimeout 看起来像「取消任务」，实际上只是让这个 CompletableFuture 以超时结束，上游任务还在跑。
这一节先把这几个语义用真实输出钉死，再落到 libraryd 的搜索服务。
组合：map、flatMap 与 zip
先看三个最常用的组合方法。它们对应函数式编程里的 map、flatMap 和 zip：
CompletableFuture<BookDto> dto = CompletableFuture
.supplyAsync(() -> repository.findByIsbn(isbn), io)
.thenApply(optional -> optional.orElseThrow(() ->
new LibraryException(ErrorCode.BOOK_NOT_FOUND, "图书不存在: " + isbn)))
.thenApply(BookDto::from);
thenApply 接收「上一个阶段的值」，返回「一个新值」。上面的链条里，第一个 thenApply 把 Optional<Book> 变成 Book，第二个把 Book 变成 BookDto。它不改变异步边界：如果前一个阶段已经完成，thenApply 可能在当前线程直接执行；如果还没完成，就在完成它的线程上执行。
如果函数本身返回 CompletableFuture，用 thenApply 会得到嵌套类型：
CompletableFuture<CompletableFuture<LoanDto>> nested =
findBook(isbn).thenApply(book -> findActiveLoan(book.isbn()));
这不是编译错误，但类型是 CompletableFuture<CompletableFuture<LoanDto>>，后面还要再解一层。正确的方法是 thenCompose，它会把两层压平：
CompletableFuture<LoanDto> flat =
findBook(isbn).thenCompose(book -> findActiveLoan(book.isbn()));
thenCombine 用于两个互不依赖的阶段，等它们都完成后合并：
CompletableFuture<List<BookDto>> items = CompletableFuture.supplyAsync(
() -> repository.search(q, offset, size).stream().map(BookDto::from).toList(), io);
CompletableFuture<Long> total = CompletableFuture.supplyAsync(
() -> repository.count(q), io);
CompletableFuture<PageResponse<BookDto>> page = items.thenCombine(total,
(list, count) -> new PageResponse<>(list, count, pageNumber, size));
三个方法的区别可以压成一张表：
| 方法 | 输入 | 返回 | 用途 |
|---|---|---|---|
thenApply | T -> U | CF<U> | 同步转换值 |
thenCompose | T -> CF<U> | CF<U> | 串联依赖的异步调用 |
thenCombine | CF<U> + (T,U) -> V | CF<V> | 合并两个独立结果 |
supplyAsync 不传 executor 时使用 ForkJoinPool.commonPool()。本机输出：
--- common pool default ---
ForkJoinPool.commonPool-worker-1
common pool 的线程数是 CPU 核数减一，适合短小的 CPU 计算，不适合阻塞 JDBC。libraryd 的所有数据库调用都显式传 libraryd-worker-* 池，避免把 common pool 占满后影响其他使用 common pool 的库。
异常与超时
先看 orTimeout。一个 500ms 的任务，配 200ms 超时：
long t0 = System.nanoTime();
try {
slow("isbn-1", 500)
.orTimeout(200, TimeUnit.MILLISECONDS)
.join();
} catch (CompletionException e) {
System.out.printf("elapsed=%dms cause=%s%n",
(System.nanoTime() - t0) / 1_000_000,
e.getCause().getClass().getName());
}
本机输出：
elapsed=208ms cause=java.util.concurrent.TimeoutException
join() 抛的是 CompletionException，真正的原因在 getCause() 里。get() 则抛 ExecutionException，原因同样在 getCause()。这两个包装很容易漏掉，日志里只打 e.getMessage() 往往只能看到 java.util.concurrent.CompletionException，看不到真正的原因。
orTimeout 让 CompletableFuture 以 TimeoutException 结束，但不会取消上游任务。500ms 的 Thread.sleep 还在跑，如果它持有数据库连接，连接会一直被占到任务自己结束。orTimeout 只是让调用方提前拿到失败，不是资源回收。
如果超时后要返回降级值，用 completeOnTimeout：
String value = slow("isbn-2", 500)
.completeOnTimeout("cached-book", 200, TimeUnit.MILLISECONDS)
.join();
本机输出：
value=cached-book elapsed=201ms
它不会抛异常，而是用给定的值完成这个 CompletableFuture。适合「有缓存、可以接受旧值」的读路径；写路径不能用它，否则调用方会以为写成功了。
异常恢复用 exceptionally 或 handle：
CompletableFuture<String> recovered = slow("isbn-1", 500)
.orTimeout(200, TimeUnit.MILLISECONDS)
.exceptionally(error -> "fallback-book");
本机输出：
recovered=fallback-book elapsed=200ms
exceptionally 只处理异常，正常值直接透传；handle 同时接收值和异常，总是执行，适合「无论成功失败都要记一笔」的场景。whenComplete 也总是执行，但它不改变结果，异常会继续往下传。
allOf 的语义最容易误解。三个任务：a 100ms 成功，b 立刻失败，c 800ms 成功。直觉上 allOf 应该在 b 失败时立刻失败，实际输出是：
--- allOf with failure ---
allOf failed after=804ms cause=java.lang.IllegalStateException: db down
a done=true c done=true
它等了 804ms，直到所有任务都结束，才把异常抛给 join()。allOf 的契约是「所有任务完成时完成」，不是「任一失败就失败」。如果要快速失败，需要自己用 anyOf 或者对每个任务单独处理，并在失败时取消其余任务；而取消能不能真正停下任务，又取决于任务是否响应中断。
取消也不等于停止：
CompletableFuture<String> cancelled = slow("never", 5000);
System.out.println("cancel result=" + cancelled.cancel(true));
System.out.println("isCancelled=" + cancelled.isCancelled()
+ " isCompletedExceptionally=" + cancelled.isCompletedExceptionally());
本机输出：
cancel result=true
isCancelled=true isCompletedExceptionally=true
cancel(true) 会尝试中断执行任务的线程，但 CompletableFuture 本身不保证任务停下来。如果任务里的 JDBC 驱动不响应中断，查询还会继续跑。真正的超时要在 JDBC 层设置 Statement.setQueryTimeout，在 HTTP 层设置请求超时，在异步层用 orTimeout 做兜底，三层各管一段。
在 libraryd 里落地
把搜索改成两个并行查询，再用 thenCombine 合并：
package com.plume.library.app;
import com.plume.library.domain.ErrorCode;
import com.plume.library.domain.LibraryException;
import com.plume.library.web.BookDto;
import com.plume.library.web.PageResponse;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.Executor;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public final class BookQueryService {
private final BookSearchRepository repository;
private final Executor io;
public BookQueryService(BookSearchRepository repository, Executor io) {
this.repository = repository;
this.io = io;
}
public PageResponse<BookDto> search(String q, int page, int size) {
int offset = page * size;
CompletableFuture<List<BookDto>> items = CompletableFuture.supplyAsync(
() -> repository.search(q, offset, size).stream()
.map(BookDto::from)
.toList(),
io);
CompletableFuture<Long> total = CompletableFuture.supplyAsync(
() -> repository.count(q),
io);
try {
return items.thenCombine(total,
(list, count) -> new PageResponse<>(list, count, page, size))
.orTimeout(2, TimeUnit.SECONDS)
.join();
} catch (CompletionException e) {
throw translate(e.getCause());
}
}
private LibraryException translate(Throwable cause) {
if (cause instanceof TimeoutException) {
return new LibraryException(ErrorCode.DB_UNAVAILABLE,
"book search timed out", cause);
}
if (cause instanceof RejectedExecutionException) {
return new LibraryException(ErrorCode.SERVICE_OVERLOADED,
"worker queue is full", cause);
}
if (cause instanceof LibraryException libraryException) {
return libraryException;
}
return new LibraryException(ErrorCode.INTERNAL,
"book search failed", cause);
}
}
两个查询都提交到同一个 io 池。池大小是 8 时，一个搜索请求占 2 个线程；4 个并发搜索就能把池占满，第 5 个请求开始在队列里等。这个账要和 9.1 的队列容量一起算，不能只看单请求的延迟。
orTimeout(2, TimeUnit.SECONDS) 是异步层的兜底，但 repository.search 里的 JDBC 查询不会因为 orTimeout 而停止。Repository 里要设置查询超时：
try (PreparedStatement ps = connection.prepareStatement(sql)) {
ps.setQueryTimeout(2);
ps.setString(1, "%" + q + "%");
try (ResultSet rs = ps.executeQuery()) {
// 映射结果
}
}
setQueryTimeout(2) 让 PostgreSQL 在服务端取消查询，连接可以更快回到池里；orTimeout 让调用方在 2 秒时拿到失败，不必等 JDBC 驱动返回。两者不是重复，前者管资源，后者管响应时间。
BookDto::from 是一个静态工厂，把领域对象转成 Web DTO。第 6 章已经定过 DTO 边界，这里不把 Book 直接暴露给 Jackson，也不在异步链里做序列化。
常见错误与边界
「thenApply 里返回 CompletableFuture。」 类型会变成 CompletableFuture<CompletableFuture<T>>，后续的 thenApply 拿到的是内层 future，不是值。需要展平时用 thenCompose。这个错误在编译期不一定报错，因为嵌套类型本身合法，直到你调用 join() 才发现拿到的是 future 而不是结果。
「以为 allOf 会 fail-fast。」 本机实测：一个 100ms 失败、一个 800ms 成功，allOf 仍然等到 804ms 才失败。它的语义是「全部完成」，不是「任一失败」。要在失败时尽快返回，需要对每个 future 单独 handle，或者用 anyOf 加显式取消。
「以为 orTimeout 会取消上游。」 它只让当前的 CompletableFuture 以 TimeoutException 完成，上游任务继续跑。对 JDBC 来说，这意味着连接可能一直被占着。必须同时设置 Statement.setQueryTimeout，并在必要时用 cancel(true) 尝试中断。
「在 common pool 里跑 JDBC。」 supplyAsync 不传 executor 时用 ForkJoinPool.commonPool()，线程数是 CPU 核数减一。JDBC 是阻塞调用，占住 common pool 的线程会影响其他依赖 common pool 的库。所有阻塞 I/O 都显式传自己的 executor。
「在请求线程上 join()，以为它是异步。」 join() 仍然会阻塞调用线程。它的价值是把阻塞点收在一个边界上，而不是让整条链路都阻塞。更好的做法是在 Web 层组合好 future 再返回，或者用 Javalin 的异步能力；但对 libraryd 这种同步 JDBC 服务，至少在服务层把并行查询合并完再返回，不要让每个 repository 调用都 join() 一次。
「exceptionally 吞掉异常，日志里只剩一个降级值。」 降级是产品行为，不是可观测性。exceptionally 里要记录原始异常和请求 ID，否则用户看到旧数据，运维不知道为什么。第 10.3 节会把上下文和异常日志接起来。
小结
thenApply做同步转换，thenCompose展平返回CompletableFuture的函数，thenCombine合并两个独立结果。supplyAsync默认用 common pool；阻塞 JDBC 必须传自己的 executor。orTimeout以TimeoutException完成当前 future，但不会取消上游任务；资源回收要靠 JDBC 查询超时和cancel(true)。completeOnTimeout返回降级值，适合读路径；写路径不能用它冒充成功。allOf等所有任务结束才失败，本机一个 100ms 失败加一个 800ms 成功，allOf在 804ms 才抛异常。join包CompletionException，get包ExecutionException，原因都在getCause()里。- libraryd 的搜索用
thenCombine并行查页和总数，再用orTimeout做响应兜底。
搜索的两个查询已经能并行，但一次会员借阅详情不止两个查询：借阅、会员、书目之间有依赖，还有 N+1 的风险。下一节把扇出、依赖和结果合并画清楚。
阅读导航：上一节：9.3 背压与拒绝策略：让过载可见 · 下一节：10.2 异步编排：并行查询与结果合并 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。