同步器选用:CountDownLatch、CyclicBarrier、Semaphore、Phaser 与 Exchanger
版本基线:API 契约以 Java SE 17 为准,实现分析固定到 OpenJDK 17u
jdk-17.0.12+7。本机实际编译运行版本为 Oracle JDK 17.0.12+8(x86_64)。排队骨架、hasQueuedPredecessors公平门与 Condition 的源码链在AQS 正文;锁的选型在锁正文;线程池等待与 Future 在线程池与异步正文。本页按"等待谁、谁推进、失败了传染谁"三个问题讲五个同步器的协议、失败分支与选型。
先问协议,再选工具
五个同步器都建立在 AQS 共享获取或独立状态机上(骨架见AQS 正文),但回答的问题不同:
| 工具 | 等待者与推进者 | 状态可逆吗 | 失败传染 |
|---|---|---|---|
| CountDownLatch | N 方 countDown,任意多方 await;推进者是别人 | 否:归零不可重置 | 无:countDown 不阻塞不抛受检异常 |
| CyclicBarrier | N 方互等,最后一方推进全员 | 是:代际重置 | 强:一方超时/中断 → 全员 BrokenBarrierException |
| Semaphore | 获取者自等;许可是资源不是身份 | 许可可增减 | 无:acquire 失败只影响自己 |
| Phaser | 注册方按阶段互等,最后一方触发 onAdvance 推进 | 是:逐代推进,可终止 | 可选:forceTermination 释放全员,返回负值 |
| Exchanger | 恰好两方成对,互相推进 | 否:每次 exchange 新配对 | 超时/中断只影响自己与已配对者 |
选型错误通常是把"等待"和"推进"的角色搞混:用 latch 等一个会重复发生的事件(不可重置)、用 barrier 做资源池(它不是资源模型)、把 semaphore 当"可以重入的锁"(许可没有所有权)。逐个展开。
CountDownLatch:一次性的计数门闩
初始化 N,每次 countDown 减一,await 在计数到零后放行。语义全部浓缩在这段共享模式实现里:
原码连续摘录(仅统一公共缩进,截取内部类局部):CountDownLatch.java,jdk-17.0.12+7,L171–L185。
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c - 1;
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}读出四条协议:获取不消耗状态(tryAcquireShared 只看 state==0,多个 await 都能过);释放是递减(tryReleaseShared 减一),只有恰好减到零的那次返回 true 触发通知——AQS 共享模式"返回正数可获取"的骨架约定(见AQS 正文);已归零后再 countDown 是无操作(c == 0 返回 false,计数不会变负);没有加回去的路径——归零不可重置,"复用"只能新建实例。
使用边界:countDown 不阻塞也不抛受检异常,放在 finally 里保证任务异常也放行等待者;await() 可被中断,await(timeout, unit) 超时返回 false(不抛 TimeoutException——与后面 barrier 不同),返回值就是超时探针。getCount() 返回剩余计数,仅监控用。一方等待多方完成用 latch;多方互等用 barrier;要"开始信号+完成信号"就是两个 latch(官方 javadoc 的 startSignal/doneSignal 模式)。等待线程池任务完成时注意 shutdown+awaitTermination 与 Future 的边界归线程池正文。
实验第 8 节断言:N=2 时 await(200ms) 返回 false;两次 countDown 后 await(2s) 返回 true;第三次 countDown 后 getCount() 仍为 0。
CyclicBarrier:代际与 broken 的传染
CyclicBarrier(N) 让 N 方到齐后同时放行,并可循环使用。它的可重用不是"计数清零"而是换一代(Generation):
原码连续摘录(仅统一公共缩进,截取方法局部):CyclicBarrier.java,jdk-17.0.12+7,L180–L196。
private void nextGeneration() {
// signal completion of last generation
trip.signalAll();
// set up next generation
count = parties;
generation = new Generation();
}
/**
* Sets current barrier generation as broken and wakes up everyone.
* Called only while holding lock.
*/
private void breakBarrier() {
generation.broken = true;
count = parties;
trip.signalAll();
}正常 trip 走 nextGeneration:signalAll 叫醒上一代全部等待者,然后新建 Generation、计数重置——这就是"cyclic"的全部实现。任何一方出问题走 breakBarrier:把当前代标 broken、同样唤醒全员。等待循环里每个被唤醒者都会检查自己进入时记住的那一代:
原码连续摘录(仅统一公共缩进,截取方法局部):CyclicBarrier.java,jdk-17.0.12+7,L232–L260。
// loop until tripped, broken, interrupted, or timed out
for (;;) {
try {
if (!timed)
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
// We're about to finish waiting even if we had not
// been interrupted, so this interrupt is deemed to
// "belong" to subsequent execution.
Thread.currentThread().interrupt();
}
}
if (g.broken)
throw new BrokenBarrierException();
if (g != generation)
return index;
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}这是本页最重要的失败分支,四条传染规则:等待者的中断(若代未换、未破)会 breakBarrier 并把 BrokenBarrierException 传染给同代所有人;带时限等待超时同样 breakBarrier——超时不只影响自己,整个屏障作废(对比 latch 的 await 超时只返回 false);醒来后查 g.broken 抛 BrokenBarrierException、查 g != generation 判定"新一代已开"正常返回 await 的序号(最后一方得 0,先到者得 N-1…1,总和恒为 N(N-1)/2);若屏障在 trip 边缘被中断,中断"归属后续执行"(恢复标记不抛)。
重置与命令:reset() 等价于 break 当前代再开新代,等待中的各方收 BrokenBarrierException;barrierCommand(构造参数)在最后一方 trip 前持锁执行,抛异常即 breakBarrier——所以 barrierAction 里不能再碰这把锁(自死锁),也不要做长 I/O(最后一方持锁等待,全员卡在 signalAll 之前)。
实验第 9 节断言:两方 await 序号 {1,0};trip 后同实例可直接进入第二代(复用);一方 await(150ms) 超时后,未超时的另一方收到 BrokenBarrierException(传染实测)且 isBroken() 为 true;reset() 后屏障重新可用。
Semaphore:许可是非所有权资源
Semaphore 用 AQS 共享 state 表示许可数。它与锁的根本区别:许可没有所有权——任何线程都能 release,包括从未 acquire 过的。非公平获取就是"有余量就 CAS 扣减":
原码连续摘录(仅统一公共缩进,截取方法局部):Semaphore.java,jdk-17.0.12+7,L183–L191。
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}公平版在循环首装上AQS 正文的公共公平门:
原码连续摘录(仅统一公共缩进,截取嵌套类局部):Semaphore.java,jdk-17.0.12+7,L242–L260。
static final class FairSync extends Sync {
private static final long serialVersionUID = 2014338818796000944L;
FairSync(int permits) {
super(permits);
}
protected int tryAcquireShared(int acquires) {
for (;;) {
if (hasQueuedPredecessors())
return -1;
int available = getState();
int remaining = available - acquires;
if (remaining < 0 ||
compareAndSetState(available, remaining))
return remaining;
}
}
}hasQueuedPredecessors() 为真就返回 -1 排队,与公平 ReentrantLock 同一门。释放的非所有权契约写在 javadoc 里:
原码连续摘录(仅统一公共缩进,截取文档与方法):Semaphore.java,jdk-17.0.12+7,L421–L433。
* <p>Releases a permit, increasing the number of available permits by
* one. If any threads are trying to acquire a permit, then one is
* selected and given the permit that was just released. That thread
* is (re)enabled for thread scheduling purposes.
*
* <p>There is no requirement that a thread that releases a permit must
* have acquired that permit by calling {@link #acquire}.
* Correct usage of a semaphore is established by programming convention
* in the application.
*/
public void release() {
sync.releaseShared(1);
}"释放使可用许可加一"且"释放者不必是获取者"——正确性靠应用层约定。两条工程含义:额外 release 会扩容许可(availablePermits 超过初始值),泄漏的 release 把"限 3 并发"悄悄变成无限并发;acquire/release 跨线程配对是合法且常用的模式(池的借出/归还、生产者扣减/消费者归还),这是锁做不到的。本票实验第 10 节把第一条做成真实红绿:按"release 后回到初始值"的误解写断言,实测许可数从 1 涨到 2(扩容而非复原),修正断言后通过。
获取族谱与锁一致:acquire() 可中断、acquireUninterruptibly() 不响应中断但记录、tryAcquire() 立即、tryAcquire(t,u) 限时、drainPermits() 一次性取走全部。选型:保护固定容量资源池(连接、缓冲区)用 semaphore;互斥保护临界区用锁——把 semaphore(1) 当锁用没有重入、没有 Condition、且任何线程都能"替你解锁"。
Phaser:把 barrier 的代际变成一等状态
Phaser 是 barrier/latch 的合体升级:阶段(phase)自动推进、参与方可动态注册/注销、终止可编程。全部状态压在一个 64 位字段里:
原码连续摘录(仅统一公共缩进,截取字段声明):Phaser.java,jdk-17.0.12+7,L299–L306。
private static final int MAX_PARTIES = 0xffff;
private static final int MAX_PHASE = Integer.MAX_VALUE;
private static final int PARTIES_SHIFT = 16;
private static final int PHASE_SHIFT = 32;
private static final int UNARRIVED_MASK = 0xffff; // to mask ints
private static final long PARTIES_MASK = 0xffff0000L; // to mask longs
private static final long COUNTS_MASK = 0xffffffffL;
private static final long TERMINATION_BIT = 1L << 63;一个 long 同时编码:阶段号(31 位,回绕于 MAX_PHASE)、注册方数与未到方数(各 16 位,上限 65535)、最高位终止标志。对比 CyclicBarrier 的 count + Generation 两个字段:Phaser 用位域 CAS 同时改"到齐人数与阶段",支持分层(tiered,父子 phaser)以降低大参与方下的争用。
arrive() 到达不等待;arriveAndAwaitAdvance() 到达并等本阶段全员到达;最后一方到达时触发推进并回调终止谓词:
原码连续摘录(仅统一公共缩进,截取文档与方法):Phaser.java,jdk-17.0.12+7,L925–L937。
* <pre> {@code
* Phaser phaser = new Phaser() {
* protected boolean onAdvance(int phase, int parties) { return false; }
* };}</pre>
*
* @param phase the current phase number on entry to this method,
* before this phaser is advanced
* @param registeredParties the current number of registered parties
* @return {@code true} if this phaser should terminate
*/
protected boolean onAdvance(int phase, int registeredParties) {
return registeredParties == 0;
}默认 onAdvance 只在"注销到零方"时终止;固定轮次任务覆写成 phase >= iterations - 1,永不终止则恒返回 false。onAdvance 相当于 barrier 的 barrierCommand + 终止判定合一,同样由最后一方在推进前调用(不要在里面等待)。终止后所有同步方法立刻返回负值(阶段号变负),注册无效——用返回值符号判终止,而不是猜。forceTermination() 是 Phaser 版的 breakBarrier:释放全员、置终止位,等待者拿到负阶段号返回。与 CyclicBarrier 的差别:broken 是"这一代作废、可重来";terminated 是"整个 phaser 永久关闭"。
实验第 11 节断言:3 方 arriveAndAwaitAdvance() 后 phase 0→1(注册方动态 register() 计入);覆写 onAdvance 到 phase>=1 终止后 isTerminated() 为 true、arriveAndAwaitAdvance() 返回负值;arriveAndDeregister() 使注册方减一(默认终止谓词在减到零时触发)。
Exchanger:恰好两方的会合
Exchanger<V> 只做一件事:两个线程在会合点交换各自持有的对象,exchange(v) 阻塞直到另一方到达,各自拿到对方给的值。它的内存语义是官方写明的配对 happens-before:
原码连续摘录(仅统一公共缩进,截取类文档):Exchanger.java,jdk-17.0.12+7,L95–L100。
* <p>Memory consistency effects: For each pair of threads that
* successfully exchange objects via an {@code Exchanger}, actions
* prior to the {@code exchange()} in each thread
* <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>
* those subsequent to a return from the corresponding {@code exchange()}
* in the other thread.配对成功时,双方 exchange 之前的动作互相 happens-before——缓冲区交接不需要再加锁。失败分支:exchange(v, t, u) 超时抛 TimeoutException(与 latch 的返回 false、barrier 的传染不同,只影响自己);被中断抛 InterruptedException;第三个线程到达时会与等待者之一配对(两两配对,多余者继续等)——它不是三方向步工具。典型用法是官方示例的生产/消费双缓冲:填充线程把满缓冲交出去换空缓冲,两方永远不需要锁。实现上高竞争时用 arena 槽位数组分散,低竞争单槽 CAS——这是实现细节,选型时只需要"恰好两方、交替节奏接近"这个前提。
实验第 12 节断言:两线程 exchange 互取对方值;无伙伴时 exchange(v, 200ms) 抛 TimeoutException。
设计案例:一批任务的"起跑—分段—收口"(算例,非实测)
假设数据处理作业:1 个协调线程 + 8 个工作线程,要求同时起跑(避免计时偏置)、分 3 个阶段推进(每阶段全员完成才进下一阶段)、个别工作线程可能中途失败、收口后释放资源。以下为假设算例。
四个方案:
- 两个 CountDownLatch(start + done)+ 每阶段一个新 latch:起跑门
startSignal(1),阶段门每阶段new CountDownLatch(8)。失败处理麻烦:某工作线程异常后必须countDown(放finally)否则阶段门永远凑不齐;阶段数固定时可行,"每阶段新建"的分配在高频小阶段下成为负担。收口再一个 latch。三把不同生命周期的闩,状态分散。 - CyclicBarrier(9)(含协调线程)+ barrierAction:起跑、每阶段、收口全是同一个 barrier 的代际,
await返回序号可让"得 0 的一方"做阶段汇总(barrierAction 亦可)。代价是强传染:任何一个工作线程在任一阶段超时/中断,全队BrokenBarrierException——对"个别失败即全批重跑"的作业这是正确语义,对"个别失败只需降级"的作业是灾难;且 broken 后要明确 reset 重开还是放弃。 - Phaser(1+8):
register()动态增减参与方(工作线程可退出:arriveAndDeregister()),阶段自动推进,onAdvance(phase)里做阶段汇总与固定轮次终止,个别失败可forceTermination()释放全员或自行注销后其余继续。它是唯一"参与方会变"的方案,也是这个案例的最优解:3 阶段固定 + 8+1 方 + 失败要传染全队(数据不完整就停批)。 - Semaphore/Exchanger:不解决多方分段同步——semaphore 是资源闸门,Exchanger 只有两方。列在这里是为了说明"不该选"也要能论证。
不变量与恢复责任:阶段推进的前置是"本阶段全部到达";失败策略要么传染停批(barrier/phaser 的 broken/terminate),要么注销降级(phaser 独有),要么外部监控兜底超时——恢复责任必须写在作业协议里,不能依赖"线程一般不会失败"。验证方式:故障注入(杀一个工作线程)看全队是停是降级、收口计数是否守恒,而不是只测正常路径。协调线程自身的等待归线程池与异步正文的 cancel/orTimeout 链。
竞争时间线:barrier 超时传染与 latch 归零通知
| 时间 | A(await 无时限) | B(await 200ms) | barrier 状态 |
|---|---|---|---|
| t1 | await,计数 2→1 | — | parties=2,等待 |
| t2 | — | awaitNanos 计时中 | 双方等待 |
| t3 | — | 超时:breakBarrier(),抛 TimeoutException | broken=true,signalAll |
| t4 | 醒来:g.broken → 抛 BrokenBarrierException | 已退出 | 全员失败:传染完成 |
| 时间 | T1 | T2(等待者) | latch 状态 |
|---|---|---|---|
| t1 | — | await:state=2≠0,park | state=2 |
| t2 | countDown:2→1,CAS 成功返回 false | 仍睡 | state=1 |
| t3 | countDown:1→0,返回 true 触发 signalNext | 被唤醒重查 state==0 | state=0 |
| t4 | — | tryAcquireShared 得 1,await 返回 | 放行且不再消耗状态 |
左表是 barrier 的强传染(一方超时全员 broken);右表是 latch 的"只有归零那次才通知"——中间的 countDown 不唤醒任何人,归零后任意多次 await 都直接通过。
实验:五件套的公开入口
公开入口(仓库根):
JAVA_HOME=/Library/Java/JavaVirtualMachines/jdk-17.jdk/Contents/Home \
bash labs/java-senior/run.sh sync探针为 labs/java-senior/LocksProbe.java。本节相关输出(2026-10-25,Oracle JDK 17.0.12+8,macOS 26.7.1 x86_64):
== 8. CountDownLatch: one-shot gate ==
await(200ms) before zero -> false(超时返回值探针,不抛 TimeoutException)
after 2x countDown: await(2s)=true getCount=0; extra countDown -> no-op, still 0(归零不可重置、不会变负)
repeated awaits on opened latch pass without consuming state(门开了就不消耗状态)
== 9. CyclicBarrier: generations and broken propagation ==
two parties awaited, indices sum=1(序号 0/1,最后一方得 0)
same barrier tripped again for generation 2(复用:代际自动重置)
one party timed out -> other got BrokenBarrierException(超时传染,断言)
isBroken=true; after reset() barrier reusable(reset 重开新代)
== 10. Semaphore: permits are not ownership ==
main acquired permit; worker tryAcquire(200ms) -> false
worker (never acquired) released; availablePermits=1(跨线程归还是合法模式)
extra release -> availablePermits=2(release 是加一不是复原:扩容,红绿修正点)
== 11. Phaser: phases, dynamic parties, termination ==
3 parties arrived; phase 0 -> 1(动态 register 计入)
onAdvance>=1 terminated: isTerminated=true, sync returns -2147483646(终止后同步方法返回负值)
arriveAndDeregister -> registered=1; remaining party tripped solo (last arriver returns next phase=1); at 0 parties default onAdvance terminated(注销到零即默认终止)
== 12. Exchanger: pairwise rendezvous ==
two threads exchanged values mutually(配对互取)
exchange(200ms) without partner -> TimeoutException(只影响自己)断言口径:全部为确定性断言(固定参与方与超时上限);超时上限取 150–500ms,断言不依赖精确耗时。Semaphore 第 3 行曾按"release 后回到初始值"的误解写成红断言(实测 2,扩容),红绿过程见实测证据。
边界:不覆盖 Phaser 分层(tiered phasers)与 bulkRegister 的并发注册竞争;Exchanger 的 arena 实现细节未实验(正文以 API 契约为据);五个同步器与线程池的组合等待(如 awaitTermination 门)归线程池正文。
追问链
第一轮:为什么 await(timeout) 在 latch 返回 false,在 barrier 抛 TimeoutException,语义差别是什么?
latch 的等待者各自独立:超时只是"我这一票不等了",状态(计数)没变;barrier 的参与方互为推进条件:一方超时意味着这一代永远凑不齐,继续让其他人等就是丢失唤醒的放大版,所以 breakBarrier 传染全员。评分点:等待者独立性 vs 互等、传染的必要性。
第二轮:工作线程可能中途退出的多阶段批处理,五个工具怎么选?
参与方固定+失败即全停:CyclicBarrier;参与方会变/要编程终止/失败可选降级:Phaser(注销或 forceTermination);单次起跑+单次收口:两个 latch;资源并发上限:semaphore;两方交替交接:Exchanger。评分点:按"等待谁、谁推进、失败传染谁"三问排除。
第三轮:semaphore(1) 能当互斥锁用吗?
不能等价替换:许可无所有权(任何线程可 release,包括未获取者——"替你解锁"破坏互斥责任)、无重入(再 acquire 自阻塞)、无 Condition。它能做锁做不到的跨线程借还配对。评分点:所有权差异、重入缺失、各自不可替代的场景。
第四轮:latch 的 countDown 放在 finally 里防什么?不放会怎样?
防任务异常路径漏 countDown:等待者永远凑不齐,await() 无限阻塞(无时限版无兜底)。放 finally 保证异常也放行;若要求"失败也停批",等待侧要用带时限 await + 计数核对,不能只靠 finally。评分点:异常路径、无限等待风险、超时兜底。
闭卷验收
本页 13 道闭卷题收进了下方自测面板:先回忆,再展开参考答案与评分点,最后给三档自评。
来源与版本
| 来源 | 版本 | 用途 |
|---|---|---|
| Java SE 17 CountDownLatch API | Java SE 17 | 一次性、countDown 不阻塞、await 超时返回 false |
| Java SE 17 CyclicBarrier API | Java SE 17 | 代际重用、broken 传染、reset、barrierAction |
| Java SE 17 Semaphore API | Java SE 17 | 许可非所有权、获取族谱、公平性 |
| Java SE 17 Phaser API | Java SE 17 | 动态注册、阶段推进、onAdvance、终止语义 |
| Java SE 17 Exchanger API | Java SE 17 | 两方会合、配对 happens-before、超时 |
| OpenJDK 17u CountDownLatch | jdk-17.0.12+7 | tryAcquireShared/tryReleaseShared(L171–L185) |
| OpenJDK 17u CyclicBarrier | jdk-17.0.12+7 | nextGeneration/breakBarrier(L180–L196)、dowait 等待循环(L232–L260) |
| OpenJDK 17u Semaphore | jdk-17.0.12+7 | nonfairTryAcquireShared(L183–L191)、FairSync(L242–L260)、release 非所有权文档(L421–L433) |
| OpenJDK 17u Phaser | jdk-17.0.12+7 | state 位域编码(L299–L306)、onAdvance 默认与覆写(L925–L937) |
| OpenJDK 17u Exchanger | jdk-17.0.12+7 | 配对 happens-before 文档(L95–L100) |
本地 LocksProbe 实验 | Oracle JDK 17.0.12+8 | 五件套协议、失败分支与红绿实测输出 |
访问日期:2026-10-25。源码行号以 jdk-17.0.12+7 标签为准;运行环境为 Oracle JDK 17.0.12+8。Exchanger 的 arena 与 Phaser 的分层实现属机制细节,正文引用处以 API 契约与固定源码为界,不外推到其他版本。