本节目标：在「等齐」「限量」「对齐」三种协调需求里，选对同步器，并知道它什么时候会失效。
适用版本：JDK 25（Temurin 25.0.5）；只使用 JDK 标准库。
承接的问题
5.2 的流水线解决了「一件事怎么异步做」，但 libraryd 里还有三类协调问题，它们的形状完全不同：
- 等齐：启动时要加载 ISBN 索引、罚金规则、馆藏分布，三件事都完成才能开放流量。主线程要等三个并发任务全部结束。
- 限量：数据库连接池只有 10 个，后台任务可以有 400 个，但同时进数据库的不能超过 10 个。
- 对齐：分批导出 12 万条借阅记录，4 个线程各处理一段，每一轮结束后要一起进入下一轮，避免某一轮里有人已经跑到很前面。
这三件事看起来都是「线程之间互相等」，但等的方式不一样。JDK 提供了三个同步器，正好对应：CountDownLatch、Semaphore、CyclicBarrier。它们都建在 AQS 上（4.3 讲过），语义却差得很远。
CountDownLatch：一次性等齐
CountDownLatch 有一个计数，countDown() 减一，await() 等到计数归零。三个任务分别睡 100ms、200ms、300ms：
CountDownLatch ready = new CountDownLatch(3);
long start = System.nanoTime();
long[] delays = {100, 200, 300};
for (int i = 0; i < 3; i++) {
final long delay = delays[i];
new Thread(() -> {
sleep(delay);
ready.countDown();
}).start();
}
ready.await();
System.out.printf("all warmups done after %dms, count=%d%n",
(System.nanoTime() - start) / 1_000_000, ready.getCount());
本机输出：
-- CountDownLatch
warmup-100ms done
warmup-200ms done
warmup-300ms done
all warmups done after 314ms, count=0
second await returned true immediately
两个关键语义：
- 等最慢的那个。总耗时 314ms，接近最慢任务（300ms）加上线程启动开销，而不是三个任务之和（600ms）。三个任务是并发的。
- 计数归零后永久开放。第二次
await(1s)立刻返回true，getCount()永远是 0。CountDownLatch不能重置，这是一次性开关。
这个「一次性」既是优点也是坑。用来做启动门禁很合适：进程启动只有一次，门开了就不再关。用来做「每轮比赛开始」就会错：第二轮 await 会直接穿过去，因为计数早就是 0 了。需要重复使用的场景应该用 CyclicBarrier。
CountDownLatch 还有一个常见用法是反向的：一个线程 await，多个线程 countDown，表示「等 N 个信号」。反过来用（N 个线程 await，一个线程 countDown N 次）表示「一个开关同时放行 N 个线程」，这时它更像一个一次性的闸门。
Semaphore：限量而不是等齐
Semaphore 维护一组许可，acquire() 拿一个（没有就等），release() 还一个。它不关心「谁先来谁后到」，只关心「同时最多几个」。
10 个许可、40 个任务、每个任务睡 20ms：
Semaphore permits = new Semaphore(10, true);
// 40 个线程：acquire -> 记录并发数 -> sleep(20) -> release
本机输出：
-- Semaphore(10) over 40 tasks
elapsedMs=99 maxConcurrent=10 availablePermits=10
40 个任务分成 4 批，每批 10 个，每批 20ms，理论 80ms，实测 99ms（线程启动和调度开销）。maxConcurrent 严格等于 10，availablePermits 最后回到 10，说明没有泄漏许可。
如果去掉信号量，40 个任务会同时进入，maxConcurrent 就是 40。对数据库来说，这意味着 40 个连接请求打向一个只有 10 个连接的池，多出来的 30 个在 getConnection() 上排队——等待没有消失，只是换了个地方。
Semaphore 的三个细节：
- 许可数应该不大于下游容量。libraryd 的
DbConcurrencyGate用 Hikari 的maximumPoolSize作为许可数。设成 20 而连接池只有 10，多出来的 10 个许可毫无意义。 release()不校验是谁拿的。任何线程都能release()，release 多了许可会膨胀，信号量就失去限流作用。写法固定为acquire(); try { ... } finally { release(); }。- 公平性可选。
new Semaphore(n, true)按排队顺序发放，吞吐略低但不会有线程饿死；默认非公平，允许插队。
Semaphore 和 CountDownLatch 的区别可以用一句话概括：latch 等的是「计数到 0」，semaphore 等的是「有许可」。latch 的计数只减不加，semaphore 的许可可以反复借还。
CyclicBarrier：可重用的对齐点
CyclicBarrier 让 N 个线程互相等待，全部到齐后一起继续，并且可以指定一个「到齐时执行」的动作。和 latch 最大的差别是它可以重用：一轮结束后计数自动复位，下一轮继续用。
4 个导出线程、3 轮、每轮各自睡 10/20/30/40ms：
CyclicBarrier barrier = new CyclicBarrier(4, () ->
System.out.println(" [action] round complete on " + Thread.currentThread().getName()));
本机输出：
-- CyclicBarrier(4) over 3 rounds
[action] round complete on exporter-3
worker-2 arrived index=1
worker-0 arrived index=3
worker-1 arrived index=2
worker-3 arrived index=0
[action] round complete on exporter-3
[action] round complete on exporter-3
worker-3 finished all rounds
worker-0 finished all rounds
worker-1 finished all rounds
worker-2 finished all rounds
barrier broken=false
四个观察：
barrier action在最后一个到达的线程上执行。这里永远是exporter-3，因为它每轮睡 40ms，最慢。所以 action 里不能做耗时操作，否则会拖住整轮。await()的返回值是到达序号，0 到 N-1，可以用来让某一个线程做本轮的主持工作。- action 在放行之前执行，所以输出顺序是 action 先于各线程的
arrived打印。 - 3 轮共用了同一个 barrier，
isBroken()为 false，说明没有超时或中断。
CyclicBarrier 的典型用途是分阶段计算：每一轮所有线程处理完自己的分片，然后在 barrier 处对齐，下一轮才能看到完整结果。如果只是「等所有人跑完」，不需要分轮，CountDownLatch 或 invokeAll 更简单。
三者怎么选
| 需求 | 同步器 | 可重用 | 关键方法 |
|---|---|---|---|
| 等 N 个任务全部完成 | CountDownLatch | 否 | countDown() / await() |
| 限制同时进入的数量 | Semaphore | 是 | acquire() / release() |
| N 个线程每轮互相对齐 | CyclicBarrier | 是 | await() |
| 分阶段且阶段数不定 | Phaser | 是 | arriveAndAwaitAdvance() |
Phaser 是三者之外的另一选项，支持动态注册参与者和多个阶段，但 API 更复杂。libraryd 的导出任务阶段数固定，用 CyclicBarrier 就够；如果以后要做「边导出边追加分片」，再考虑 Phaser。
在 libraryd 里落地
启动门禁用 CountDownLatch，包一层是为了带名字和超时信息：
package com.plume.library.app;
import java.time.Duration;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
public final class StartupGate {
private final CountDownLatch latch;
private final String name;
public StartupGate(String name, int tasks) {
if (tasks < 1) {
throw new IllegalArgumentException("tasks 必须大于 0");
}
this.name = name;
this.latch = new CountDownLatch(tasks);
}
public void taskDone() {
latch.countDown();
}
public boolean await(Duration timeout) throws InterruptedException {
return latch.await(timeout.toMillis(), TimeUnit.MILLISECONDS);
}
public long remaining() {
return latch.getCount();
}
public String name() {
return name;
}
}
remaining() 是给失败信息用的：超时的时候能打出「warmup 还剩 1 个任务」，比一句「启动失败」有用得多。三个预热任务里任何一个抛异常，都要在 finally 里 taskDone()，否则 await 会等到超时才失败。
数据库限流用 Semaphore，并带上指标：
package com.plume.library.app;
import java.time.Duration;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
public final class DbConcurrencyGate {
private final Semaphore permits;
private final AtomicInteger inFlight = new AtomicInteger();
private final AtomicInteger maxInFlight = new AtomicInteger();
private final AtomicLong maxWaitNanos = new AtomicLong();
private final AtomicLong timeouts = new AtomicLong();
public DbConcurrencyGate(int maxConcurrent) {
if (maxConcurrent < 1) {
throw new IllegalArgumentException("maxConcurrent 必须大于 0: " + maxConcurrent);
}
this.permits = new Semaphore(maxConcurrent, true);
}
public boolean tryEnter(Duration timeout) throws InterruptedException {
long start = System.nanoTime();
boolean acquired = permits.tryAcquire(timeout.toNanos(), TimeUnit.NANOSECONDS);
long waited = System.nanoTime() - start;
maxWaitNanos.accumulateAndGet(waited, Math::max);
if (!acquired) {
timeouts.incrementAndGet();
return false;
}
int now = inFlight.incrementAndGet();
maxInFlight.accumulateAndGet(now, Math::max);
return true;
}
public void exit() {
inFlight.decrementAndGet();
permits.release();
}
public int inFlight() {
return inFlight.get();
}
public int maxInFlight() {
return maxInFlight.get();
}
public long maxWaitMillis() {
return maxWaitNanos.get() / 1_000_000;
}
public long timeouts() {
return timeouts.get();
}
public int availablePermits() {
return permits.availablePermits();
}
}
tryEnter 带超时，是因为请求不能无限等连接：等 200ms 还拿不到许可，就返回 503 而不是继续挂着。maxWaitMillis 和 timeouts 两个指标能回答「限流是不是设小了」：如果 maxWaitMillis 长期接近超时值，说明许可数不够或者 SQL 太慢；如果 timeouts 一直是 0，说明限流没起作用。
分批导出用 CyclicBarrier：
package com.plume.library.app;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
public final class ExportBarrier {
private final CyclicBarrier barrier;
private final AtomicInteger round = new AtomicInteger();
public ExportBarrier(int parties, Runnable onRoundComplete) {
if (parties < 2) {
throw new IllegalArgumentException("parties 至少为 2");
}
this.barrier = new CyclicBarrier(parties, () -> {
round.incrementAndGet();
onRoundComplete.run();
});
}
public int awaitRound(long timeoutMillis)
throws InterruptedException, BrokenBarrierException, TimeoutException {
return barrier.await(timeoutMillis, TimeUnit.MILLISECONDS);
}
public int completedRounds() {
return round.get();
}
public boolean isBroken() {
return barrier.isBroken();
}
public void reset() {
barrier.reset();
}
}
awaitRound 强制带超时：只要有一个线程卡住，其他线程不应该陪它等到天荒地老。超时会抛 TimeoutException，并且把 barrier 标记为 broken，其他等待者收到 BrokenBarrierException，整批导出可以快速失败重来。这是 barrier 比 latch 更需要注意的地方：一个线程出问题会让所有线程一起失败，所以 action 和每个分片的任务都要能快速失败。
导出的调用侧：
ExportBarrier barrier = new ExportBarrier(4, () ->
metrics.recordExportRound());
for (int shard = 0; shard < 4; shard++) {
final int index = shard;
executor.submit(() -> {
for (int round = 1; round <= 3; round++) {
exportShard(index, round);
barrier.awaitRound(30_000);
}
});
}
常见错误与边界
错误一：把 CountDownLatch 当可重用的栅栏。 计数归零后 await 永远立即返回，第二轮完全对不齐。需要重复对齐就用 CyclicBarrier，或者每轮新建一个 latch。
错误二：release() 次数多于 acquire()。 许可会越还越多，availablePermits() 超过初始值，限流形同虚设。必须放在 finally 里，且确保每条路径只 release 一次。
错误三：在 CyclicBarrier 的 action 里做长任务。 action 由最后一个到达的线程执行，其他线程全部在等它。上面实验里 action 只打印一行；如果里面去查数据库，整轮时间会被 action 拉长 N 倍。
错误四：barrier 超时后继续用。 一旦有线程超时，barrier 进入 broken 状态，后续 await 直接抛 BrokenBarrierException。要么 reset() 重来，要么放弃这一批。reset() 也有代价：正在等待的线程会收到 BrokenBarrierException，必须处理。
边界：Semaphore 不能替代连接池。 它只控制「同时有多少个任务进入」，不管理连接的创建、回收、健康检查。libraryd 仍然是 Hikari 管连接，信号量只是让等待发生在应用层、可观测、可超时。
边界：同步器都是进程内的。 多实例部署时每个 JVM 各有一套 latch 和 semaphore，跨实例的协调要靠数据库、Redis 或消息队列。
小结
CountDownLatch等计数归零，一次性；本机三个任务 100/200/300ms 并发，总耗时 314ms。Semaphore管许可，可反复借还；10 个许可、40 个任务、每个 20ms，实测并发峰值 10、总耗时 99ms。CyclicBarrier可重用，action 在最后一个到达的线程上执行；4 个线程 3 轮共用同一个 barrier。- 选型看语义：等齐用 latch，限量用 semaphore，对齐用 barrier，阶段数不定用 phaser。
- 所有
acquire/countDown都要放在finally或保证执行的路径上，否则等待方会挂到超时。 - libraryd 用
StartupGate、DbConcurrencyGate、ExportBarrier分别封装三者，并暴露剩余任务数、最大等待、超时次数等指标。
并发原语到这里告一段落。下一节换一个维度：不改代码结构，只把平台线程换成虚拟线程，看等待的代价如何变化，以及 JDK 25 里 pinning 的边界到底在哪。
阅读导航：上一节：5.2 阻塞队列：借阅事件流水线 · 下一节：6.1 虚拟线程原理：载体线程、挂起与 pinning 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。