本节目标：把「查借阅、查会员、查书目」画成依赖图，用最少的阶段并行执行，并避免 N+1 和连接池过载。
适用版本：JDK 25（Temurin 25.0.5）；PostgreSQL 17；HikariCP 7.1.0。
承接的问题
GET /api/members/{id}/loans 需要返回该会员当前在借的图书列表。每个 LoanDto 至少要有借阅信息，页面上还要显示书名。数据分散在三张表：
loans：按member_id查活动借阅。members：按 id 查会员，不存在时返回MEMBER_NOT_FOUND。books：按借阅记录里的 ISBN 查书名、作者。
串行执行是三次数据库往返。用一个每次 200ms 的模拟查询做实验，本机输出：
sequential=606ms values=loans,member,books
parallel=207ms values=loans,member,books
前两个查询互不依赖，可以并行；第三个查询依赖前两个的结果，特别是依赖 loans 里的 ISBN 集合。所以它不是一个「三个都并行」的问题，而是两波：
| 波次 | 查询 | 依赖 |
|---|---|---|
| 第一波 | loans、members | 无 |
| 第二波 | books | 第一波的 ISBN 集合 |
第二个实验把依赖关系写进 thenCompose，本机输出：
thenCompose at 1ms ids=[b1, b2, b3]
inner query took 205ms
twoWave=214ms value=books-for-[b1, b2, b3]
两波总耗时约 214ms，接近单次查询的 200ms 加上调度开销。
真正的风险不在延迟，而在扇出。如果每本书都发一个异步查询，20 条借阅就是 20 个任务，10 个并发请求就是 200 个任务。工作池只有 8 个线程，队列很快满，拒绝策略开始工作，用户看到 503。异步编排的第一原则是先把依赖图和数据量算清楚，再决定哪些任务可以并行。
先画依赖图，再写 thenCompose
第一波的两个查询没有依赖，用 thenCombine 合并：
CompletableFuture<List<Loan>> loansFuture = CompletableFuture.supplyAsync(
() -> loanRepository.findActiveByMember(memberId), io);
CompletableFuture<Member> memberFuture = CompletableFuture.supplyAsync(
() -> memberRepository.findById(memberId)
.orElseThrow(() -> new LibraryException(
ErrorCode.MEMBER_NOT_FOUND, "会员不存在: " + memberId)),
io);
CompletableFuture<MemberLoans> base = loansFuture.thenCombine(memberFuture,
(loans, member) -> new MemberLoans(member, loans));
thenCombine 等两个 future 都完成，再把两个值交给合并函数。如果 memberFuture 先失败，thenCombine 不会立刻失败，它会等 loansFuture 也结束；这一点和 10.1 的 allOf 一样，是「全部完成」语义。对两个查询来说，最多多等一个查询的时间，通常可以接受；如果要 fail-fast，需要在合并前对每个 future 单独 handle。
第二波依赖第一波的 ISBN 集合，用 thenCompose：
CompletableFuture<MemberLoansView> view = base.thenCompose(loaded -> {
Set<String> isbns = loaded.loans().stream()
.map(Loan::isbn)
.collect(Collectors.toUnmodifiableSet());
if (isbns.isEmpty()) {
return CompletableFuture.completedFuture(
new MemberLoansView(loaded.member(), List.of()));
}
return CompletableFuture.supplyAsync(
() -> bookRepository.findByIsbns(isbns), io)
.thenApply(booksByIsbn -> toView(loaded, booksByIsbn));
});
这里用 thenCompose 而不是 thenApply，因为 lambda 返回的是 CompletableFuture<Map<String, Book>>。thenCompose 会把它展平，最终类型是 CompletableFuture<MemberLoansView>，不是嵌套的 future。
合并结果：allOf 与逐项处理
如果第一波不止两个查询，比如还要查罚金汇总、预约数量，thenCombine 写起来会嵌套。可以用 allOf 把 N 个 future 收在一起：
CompletableFuture<List<Loan>> loans = CompletableFuture.supplyAsync(
() -> loanRepository.findActiveByMember(memberId), io);
CompletableFuture<Member> member = CompletableFuture.supplyAsync(
() -> memberRepository.findById(memberId).orElseThrow(), io);
CompletableFuture<Long> fineTotal = CompletableFuture.supplyAsync(
() -> loanRepository.sumOutstandingFineCents(memberId), io);
CompletableFuture<MemberDashboard> dashboard = CompletableFuture
.allOf(loans, member, fineTotal)
.thenApply(ignored -> new MemberDashboard(
member.join(), loans.join(), fineTotal.join()));
allOf 返回 CompletableFuture<Void>，本身不携带结果；结果要从各个 future 里 join() 出来。因为 allOf 完成时所有 future 都已完成，join() 不会阻塞。这个写法比嵌套 thenCombine 清楚，但要注意两点：
allOf不会 fail-fast。任何一个任务失败，它仍然等所有任务结束才把异常抛给thenApply；在thenApply里第一次join()才会看到异常。allOf接收的是CompletableFuture<?>...，没有类型安全；结果类型靠调用方自己保证。
对「部分失败可以降级」的场景，不要用 allOf 一把梭，而是逐项 handle：
CompletableFuture<Map<String, Book>> books = CompletableFuture
.supplyAsync(() -> bookRepository.findByIsbns(isbns), io)
.handle((result, error) -> {
if (error == null) {
return result;
}
logBookLookupFailure(isbns, error);
return Map.of();
});
这样书目查询失败时，借阅列表仍然可以返回，只是书名退化成 ISBN。降级策略要写进 DTO，让前端知道数据不完整，而不是悄悄返回一个空标题。
避免 N+1：一次批量查询
最糟的写法是给每条借阅发一个异步任务：
// 反例：20 条借阅会产生 20 个任务
List<CompletableFuture<Book>> futures = loans.stream()
.map(loan -> CompletableFuture.supplyAsync(
() -> bookRepository.findByIsbn(loan.isbn()).orElseThrow(), io))
.toList();
这段代码看起来「并行了」，实际上把数据库查询次数从 1 次变成 N 次，还把 N 个任务塞进容量有限的工作池。20 条借阅、8 个线程、队列 200，单请求还能撑住；10 个并发请求就是 200 个任务，队列满，开始拒绝。异步没有减少工作量，只是把工作量分散到更多线程上。
正确做法是一次批量查询。Repository 接口接收 ISBN 集合：
public interface BookRepository {
Optional<Book> findByIsbn(String isbn);
Map<String, Book> findByIsbns(Set<String> isbns);
}
JDBC 实现用 PostgreSQL 的数组参数，避免手工拼 in (?, ?, ...)：
public Map<String, Book> findByIsbns(Set<String> isbns) {
if (isbns.isEmpty()) {
return Map.of();
}
String sql = """
select isbn, title, author, published_year
from books
where isbn = any (?)
""";
try (Connection connection = dataSource.getConnection();
PreparedStatement ps = connection.prepareStatement(sql)) {
ps.setArray(1, connection.createArrayOf("varchar", isbns.toArray()));
try (ResultSet rs = ps.executeQuery()) {
Map<String, Book> result = new HashMap<>();
while (rs.next()) {
Book book = new Book(
rs.getString("isbn"),
rs.getString("title"),
rs.getString("author"),
rs.getInt("published_year"));
result.put(book.isbn(), book);
}
return Map.copyOf(result);
}
} catch (SQLException e) {
throw new LibraryException(ErrorCode.DB_UNAVAILABLE,
"批量查询图书失败", e);
}
}
isbn = any (?) 等价于 isbn in (...)，但参数是一个数组，PreparedStatement 只编译一次。ISBN 集合来自数据库里的借阅记录，大小受会员借阅上限约束；即使没有上限，也应该在应用层截断或分页，不能把上万个 ISBN 塞进一个数组参数。
限流：一次请求最多占几个连接
即使没有 N+1，一个请求也可能在第一波占 2 个连接、第二波占 1 个连接。工作池 8 个线程时，4 个并发请求就能把池占满；Hikari 只有 10 个连接，第 5 个请求开始等连接。
可以在服务层加一个信号量，限制「同时进入数据库的异步任务数」：
package com.plume.library.app;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.concurrent.Semaphore;
public final class BoundedDbExecutor {
private final Executor delegate;
private final Semaphore permits;
public BoundedDbExecutor(Executor delegate, int maxConcurrent) {
this.delegate = delegate;
this.permits = new Semaphore(maxConcurrent, true);
}
public <T> CompletableFuture<T> supplyAsync(Callable<T> task) {
return CompletableFuture.supplyAsync(() -> {
try {
permits.acquire();
try {
return task.call();
} finally {
permits.release();
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new CompletionException(e);
} catch (Exception e) {
throw new CompletionException(e);
}
}, delegate);
}
}
maxConcurrent 应该不大于 Hikari 的 maximumPoolSize，并且要留出连接给健康检查和其他路径。比如 Hikari 10、限流 8，剩下 2 个连接给 /health 和后台任务。信号量用公平模式，避免某个大请求一直抢不到许可。
工作池要比 maxConcurrent 稍大，否则任务会占着线程等许可。第 9.2 节的 DbConcurrencyGate 是同一个思路的同步版本。
在 libraryd 里落地
把上面的片段合成 MemberLoansService：
package com.plume.library.app;
import com.plume.library.domain.Book;
import com.plume.library.domain.ErrorCode;
import com.plume.library.domain.LibraryException;
import com.plume.library.domain.Loan;
import com.plume.library.domain.Member;
import java.time.LocalDate;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
public final class MemberLoansService {
public record LoanLine(long id, String isbn, String title,
LocalDate dueOn, LocalDate returnedOn, long fineCents) {}
public record MemberLoansView(Member member, List<LoanLine> loans) {}
private final LoanRepository loans;
private final MemberRepository members;
private final BookRepository books;
private final BoundedDbExecutor db;
public MemberLoansService(LoanRepository loans, MemberRepository members,
BookRepository books, BoundedDbExecutor db) {
this.loans = loans;
this.members = members;
this.books = books;
this.db = db;
}
public MemberLoansView load(String memberId) {
CompletableFuture<List<Loan>> loansFuture = db.supplyAsync(
() -> loans.findActiveByMember(memberId));
CompletableFuture<Member> memberFuture = db.supplyAsync(
() -> members.findById(memberId).orElseThrow(() ->
new LibraryException(ErrorCode.MEMBER_NOT_FOUND,
"会员不存在: " + memberId)));
CompletableFuture<MemberLoansView> result = loansFuture
.thenCombine(memberFuture, Loaded::new)
.thenCompose(loaded -> {
Set<String> isbns = loaded.loans().stream()
.map(Loan::isbn)
.collect(Collectors.toUnmodifiableSet());
if (isbns.isEmpty()) {
return CompletableFuture.completedFuture(
new MemberLoansView(loaded.member(), List.of()));
}
return db.supplyAsync(() -> books.findByIsbns(isbns))
.thenApply(byIsbn -> toView(loaded, byIsbn));
});
try {
return result.orTimeout(2, TimeUnit.SECONDS).join();
} catch (CompletionException e) {
throw translate(e.getCause());
}
}
private record Loaded(Member member, List<Loan> loans) {}
private MemberLoansView toView(Loaded loaded, Map<String, Book> byIsbn) {
List<LoanLine> lines = loaded.loans().stream()
.map(loan -> {
Book book = byIsbn.get(loan.isbn());
String title = book == null ? loan.isbn() : book.title();
return new LoanLine(loan.id(), loan.isbn(), title,
loan.dueOn(), loan.returnedOn(), 0L);
})
.toList();
return new MemberLoansView(loaded.member(), lines);
}
private LibraryException translate(Throwable cause) {
if (cause instanceof TimeoutException) {
return new LibraryException(ErrorCode.DB_UNAVAILABLE,
"member loans timed out", cause);
}
if (cause instanceof LibraryException libraryException) {
return libraryException;
}
return new LibraryException(ErrorCode.INTERNAL,
"member loans failed", cause);
}
}
常见错误与边界
「用 thenCompose 给每条记录发一个查询。」 这是 N+1 的异步版本。任务数随数据量增长，工作池和连接池都会被拖垮。正确做法是先收集业务键，再批量查询，把任务数从 N 降到 1。
「allOf 里放 100 个任务。」 一页 100 条借阅，每条一个任务，再加上其他查询，单请求就能打出 100 多个任务。工作池队列 200，两个并发请求就满了。allOf 不会限制并发，它只是等所有任务结束；限制并发要靠信号量、批量查询或分页。
「在异步任务里再 join() 另一个任务。」 如果两个任务提交到同一个有界池，池满时会出现互相等待：任务 A 占着线程等任务 B，任务 B 在队列里等线程。这是异步版的死锁。避免在任务内部阻塞等待同一池的任务；需要依赖时用 thenCompose 组合，而不是在任务体里 join。
「合并结果时共享一个可变 List。」 多个线程往同一个 ArrayList 里 add，结果可能丢数据或抛 ConcurrentModificationException。用 stream().map(...).toList() 或 Collectors.toUnmodifiableList() 生成不可变结果，再交给合并函数。
「只算查询时间，不算连接等待。」 三个查询各 200ms，不代表请求 200ms 完成；如果连接池被占满，任务会先等连接，再执行查询。压测时要看 Hikari 的等待时间和工作池的队列深度，不能只看 SQL 本身的耗时。
小结
- 先画依赖图：借阅和会员可以并行，书目依赖借阅结果，是两波而不是一波。
thenCombine合并两个独立结果，thenCompose串联有依赖的异步调用，allOf收拢 N 个独立任务。allOf不 fail-fast，等所有任务结束才失败；部分失败可降级时用逐项handle。- 异步 N+1 和同步 N+1 一样危险：20 条借阅产生 20 个任务，必须改成一次
findByIsbns批量查询。 - 信号量把同时进入数据库的任务数压到连接池容量以内，避免异步任务把连接池打满。
- 不要在任务体里
join()同一池的任务，否则可能死锁；依赖关系用组合表达。
并行查询和结果合并已经能跑通，但异步线程里的日志会丢掉 requestId，异常也可能在 CompletionException 里迷路。下一节专门处理异步边界上的上下文、阻塞和可观测性。
阅读导航：上一节：10.1 CompletableFuture：组合、异常与超时 · 下一节：10.3 异步边界：阻塞、上下文与可观测性 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。