贴贴乐咕卡
93.13M · 2026-04-15
在 Java 并发编程中，CyclicBarrier、CountDownLatch 和 Semaphore 是三个最常用的同步辅助类。它们各自解决不同场景下的线程协调问题，但初看有些相似。本文将从源码层面深入剖析它们的实现原理、性能差异和适用场景，帮助你在实际开发中做出正确选型。
| 工具 | 核心作用 | 典型应用场景 | 一句话总结 |
|---|---|---|---|
| CyclicBarrier | 让一组线程到达一个屏障（同步点）时被阻塞，直到最后一个线程到达屏障，所有被阻塞的线程才能继续执行。 | 多线程分阶段计算（如并行处理后再聚合）、多线程数据加载后再统一处理。 | 可重用的多线程等待点 |
| CountDownLatch | 一个或多个线程等待其他线程完成一组操作后才继续执行。 | 主线程等待所有子任务完成、并行任务启动前的同步。 | 一次性倒计时门闩 |
| Semaphore | 控制同时访问某个特定资源的线程数量，相当于一个计数器（许可集）。 | 数据库连接池限流、限流器、有界缓冲区的互斥访问。 | 信号量限流 |
| 方法签名 | 参数 | 返回值 | 异常 |
|---|---|---|---|
CyclicBarrier(int parties) | parties：屏障拦截的线程数量 | - | - |
CyclicBarrier(int parties, Runnable barrierAction) | parties：屏障拦截的线程数量；barrierAction：当所有线程到达屏障后优先执行的命令 | - | - |
int await() | 无 | 当前线程到达屏障的索引（0 表示最后一个到达） | InterruptedException，BrokenBarrierException |
int await(long timeout, TimeUnit unit) | timeout：超时时间；unit：时间单位 | 同 await() | InterruptedException，BrokenBarrierException，TimeoutException |
void reset() | 无 | - | - |
int getNumberWaiting() | 无 | 当前正在屏障处等待的线程数 | - |
boolean isBroken() | 无 | 屏障是否被破坏 | - |
| 方法签名 | 参数 | 返回值 | 异常 |
|---|---|---|---|
CountDownLatch(int count) | count：需要倒计数的次数（必须大于 0） | - | - |
void await() | 无 | - | InterruptedException |
boolean await(long timeout, TimeUnit unit) | timeout：超时时间；unit：时间单位 | 如果计数到达 0 返回 true，超时返回 false | InterruptedException |
void countDown() | 无 | - | - |
long getCount() | 无 | 当前计数 | - |
| 方法签名 | 参数 | 返回值 | 异常 |
|---|---|---|---|
Semaphore(int permits) | permits：初始许可数量 | - | - |
Semaphore(int permits, boolean fair) | permits：初始许可数量；fair：是否公平模式 | - | - |
void acquire() | 无 | - | InterruptedException |
void acquire(int permits) | permits：需要获取的许可数 | - | InterruptedException |
void release() | 无 | - | - |
void release(int permits) | permits：释放的许可数 | - | - |
boolean tryAcquire() | 无 | 获取成功返回 true，否则 false | - |
int availablePermits() | 无 | 当前可用的许可数 | - |
CyclicBarrier：可重用。当所有线程都到达屏障后，屏障会自动重置，可以再次使用。通过 reset() 方法也可以手动重置。
源码体现：CyclicBarrier 内部维护 int count（剩余未到达线程数）和 int parties（总线程数）。每次 await() 减少 count，当 count == 0 时触发 barrierAction 并执行 nextGeneration()：重置 count = parties，唤醒所有等待线程。
CountDownLatch：一次性。计数器的值只能递减，当计数变为 0 后，所有等待线程被唤醒，之后 await() 将立即返回，但计数器无法重置。
源码体现：CountDownLatch.Sync 继承 AQS，state 表示计数。tryReleaseShared 每次将 state 减 1，直到 0；一旦为 0，后续 countDown() 无任何效果。
Semaphore：可重用。许可的获取和释放是动态的，可以反复增减，不存在“一次性”限制。
parties 递减到 0。每调用一次 await()，剩余等待线程数减 1。当减到 0 时触发屏障动作并重置。count 递减到 0。每次 countDown() 减 1，减到 0 后唤醒等待线程。acquire() 减少许可，release() 增加许可。没有固定的归零点，许可数可以为 0 甚至负数（当释放多于获取时）。release() 增加许可。即使一个线程从未 acquire()，也可以 release() 增加许可，这可能导致许可数超出初始值。| 工具 | AQS 模式 | 实现方式 |
|---|---|---|
| CyclicBarrier | 不使用 AQS | 内部组合 ReentrantLock + Condition（trip），使用独占锁和条件队列实现阻塞/唤醒。 |
| CountDownLatch | 共享模式 | 内部类 Sync 继承 AQS，重写 tryAcquireShared / tryReleaseShared。 |
| Semaphore | 共享模式 | 内部类 Sync 继承 AQS，公平/非公平版本分别重写 tryAcquireShared。 |
源码依据：
CountDownLatch.Sync.tryAcquireShared：返回 (getState() == 0) ? 1 : -1。只有当 state == 0 时获取成功，否则阻塞。Semaphore.NonfairSync.tryAcquireShared：直接尝试 CAS 减少 state；FairSync 则会先检查队列中是否有更早的等待线程。Runnable barrierAction，当最后一个线程到达屏障时，该动作会在所有被唤醒线程执行前由最后一个到达的线程执行。| 工具 | 典型场景 |
|---|---|
| CyclicBarrier | 分阶段计算（如 MapReduce 中的 barrier）、多线程同时开始执行（模拟并发请求）、多线程数据加载完成后统一处理。 |
| CountDownLatch | 主线程等待多个子任务完成后再继续（如批量初始化）、并行任务启动信号（所有 worker 等待 startLatch）。 |
| Semaphore | 控制数据库连接池的最大连接数、限流器（限制 QPS）、有界阻塞队列中的互斥写入。 |
CountDownLatch 内部类 Sync 继承 AQS，state 表示还需要倒计数的次数。
关键方法：
// CountDownLatch.Sync
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c - 1;
if (compareAndSetState(c, nextc))
return nextc == 0; // 只有当计数变为 0 时才唤醒等待线程
}
}
流程时序图（Mermaid）：
sequenceDiagram
participant Main as 主线程
participant Worker as 工作线程
participant CDL as CountDownLatch(state=N)
Main->>CDL: await()
CDL->>CDL: tryAcquireShared => (state==0?) 返回-1
CDL->>AQS: doAcquireSharedInterruptibly
AQS->>Main: 线程阻塞，进入同步队列
Worker->>CDL: countDown()
CDL->>CDL: tryReleaseShared CAS减1
alt state变为0
CDL-->>AQS: 返回true，唤醒后继节点
AQS-->>Main: 唤醒主线程
Main->>CDL: 再次尝试获取，成功返回
else state未到0
CDL-->>Worker: 继续，不唤醒
end
Semaphore.Sync 继承 AQS，state 表示当前可用许可数。
非公平模式 tryAcquireShared：
// NonfairSync
protected int tryAcquireShared(int acquires) {
return nonfairTryAcquireShared(acquires);
}
final int nonfairTryAcquireShared(int acquires) {
for (;;) {
int available = getState();
int remaining = available - acquires;
if (remaining < 0 || compareAndSetState(available, remaining))
return remaining;
}
}
公平模式：
// FairSync
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;
}
}
release 逻辑（共享模式通用）：
protected final boolean tryReleaseShared(int releases) {
for (;;) {
int current = getState();
int next = current + releases;
if (next < current) // overflow
throw new Error("Maximum permit count exceeded");
if (compareAndSetState(current, next))
return true;
}
}
acquire / release 流程时序图：
sequenceDiagram
participant Thread as 线程
participant Sem as Semaphore(state=permits)
participant AQS as AQS同步队列
Thread->>Sem: acquire()
Sem->>Sem: tryAcquireShared (非公平)
alt 剩余许可>=请求数
Sem->>Sem: CAS减少state
Sem-->>Thread: 成功，继续执行
else 许可不足
Sem->>AQS: doAcquireSharedInterruptibly
AQS->>Thread: 线程阻塞入队
end
Note over Thread,Sem: 其他线程释放许可
Thread->>Sem: release()
Sem->>Sem: tryReleaseShared CAS增加state
Sem->>AQS: 唤醒队列中的等待线程
AQS-->>Thread: 被唤醒的线程尝试重新获取
CyclicBarrier 不使用 AQS，而是组合 ReentrantLock 和 Condition。核心字段：
private final ReentrantLock lock = new ReentrantLock();
private final Condition trip = lock.newCondition();
private int count; // 剩余未到达线程数
private int parties;
private Runnable barrierCommand;
await 核心逻辑：
private int dowait(boolean timed, long nanos) {
final ReentrantLock lock = this.lock;
lock.lock();
try {
int index = --count;
if (index == 0) { // 最后一个线程到达
// 执行 barrierAction
if (barrierCommand != null)
barrierCommand.run();
nextGeneration(); // 重置 count，唤醒所有等待线程
return 0;
}
// 循环等待，直到被唤醒、超时或屏障破坏
for (;;) {
if (!timed)
trip.await();
else if (nanos > 0)
nanos = trip.awaitNanos(nanos);
// ...
}
} finally {
lock.unlock();
}
}
nextGeneration() 实现重置和唤醒：
private void nextGeneration() {
trip.signalAll(); // 唤醒所有等待线程
count = parties; // 重置计数
}
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class CountDownLatchExample {
public static void main(String[] args) throws InterruptedException {
int taskCount = 5;
CountDownLatch latch = new CountDownLatch(taskCount);
ExecutorService executor = Executors.newFixedThreadPool(taskCount);
for (int i = 1; i <= taskCount; i++) {
final int taskId = i;
executor.submit(() -> {
try {
Thread.sleep((long) (Math.random() * 1000));
System.out.println("任务 " + taskId + " 完成");
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
latch.countDown();
}
});
}
System.out.println("主线程等待所有子任务完成...");
latch.await(); // 阻塞直到计数为0
System.out.println("所有任务完成，主线程继续执行");
executor.shutdown();
}
}
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
public class CyclicBarrierExample {
private static final int THREAD_COUNT = 4;
private static final CyclicBarrier barrier = new CyclicBarrier(THREAD_COUNT,
() -> System.out.println("=== 本轮所有线程已到达屏障，开始下一阶段 ==="));
public static void main(String[] args) {
for (int i = 0; i < THREAD_COUNT; i++) {
final int threadId = i;
new Thread(() -> {
try {
// 第一阶段：计算部分数据
System.out.println("线程 " + threadId + " 完成第一阶段计算");
barrier.await(); // 等待其他线程
// 第二阶段：基于第一阶段结果继续计算
System.out.println("线程 " + threadId + " 完成第二阶段计算");
barrier.await(); // 再次等待
System.out.println("线程 " + threadId + " 所有阶段完成");
} catch (InterruptedException | BrokenBarrierException e) {
e.printStackTrace();
}
}).start();
}
}
}
import java.util.concurrent.Semaphore;
class LimitedConnectionPool {
private final Semaphore semaphore;
private final int maxConnections;
public LimitedConnectionPool(int maxConnections) {
this.maxConnections = maxConnections;
this.semaphore = new Semaphore(maxConnections, true); // 公平模式
}
public void getConnection() throws InterruptedException {
semaphore.acquire();
System.out.println(Thread.currentThread().getName() + " 获取连接，剩余许可: " + semaphore.availablePermits());
// 模拟使用连接
Thread.sleep(500);
}
public void releaseConnection() {
semaphore.release();
System.out.println(Thread.currentThread().getName() + " 释放连接，剩余许可: " + semaphore.availablePermits());
}
}
public class SemaphoreExample {
public static void main(String[] args) {
LimitedConnectionPool pool = new LimitedConnectionPool(3);
for (int i = 1; i <= 10; i++) {
final int id = i;
new Thread(() -> {
try {
pool.getConnection();
pool.releaseConnection();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}, "Thread-" + id).start();
}
}
}
| 工具 | 锁竞争程度 | 上下文切换 | 内存开销 | 性能特点与原因 |
|---|---|---|---|---|
| CyclicBarrier | 中等（独占锁 + Condition） | 高（每次屏障点所有线程阻塞再唤醒） | 较低（一个锁 + 一个条件队列） | 屏障点会造成所有线程停止，然后全部唤醒，适合批处理场景，但高并发下唤醒风暴可能影响性能。 |
| CountDownLatch | 低（共享模式，CAS） | 中等（等待线程只被唤醒一次） | 低（AQS 节点较少） | countDown() 仅 CAS 操作，极少阻塞；await() 在计数未到 0 时阻塞，唤醒后不再重新阻塞。性能很好，尤其适合一次性同步。 |
| Semaphore | 公平模式高，非公平低 | 取决于许可争用程度 | 低 | 非公平模式下 CAS 快速尝试，减少上下文切换；公平模式下会增加队列检查和唤醒开销。非公平模式性能更高，适合高并发限流。 |
为什么会有这些差异？
ReentrantLock，每次 await 都会加锁，并且所有线程在屏障点同时唤醒会引发 惊群效应（尽管 Condition 的 signalAll 会逐个唤醒，但大量线程同时竞争锁仍会有开销）。countDown() 只在计数变为 0 时触发一次 releaseShared，然后唤醒所有等待线程。大部分时间只有 CAS 操作，没有锁竞争。因此性能损耗小。state，失败后才进入队列，减少了无效唤醒。公平模式需要检查 hasQueuedPredecessors()，增加了内存屏障开销。await() 会立即返回，countDown() 无效。如果需要重用，请使用 CyclicBarrier。countDown() 调用次数少于初始值，等待线程将永远阻塞。countDown() 会导致主线程永久阻塞。建议使用 finally 块确保 countDown() 被执行。countUp() 方法。isBroken() == true），其他等待线程会抛出 BrokenBarrierException。此时需要调用 reset() 恢复。reset() 方法会中断当前正在等待的线程（抛出 BrokenBarrierException），小心使用。parties，屏障永远不会触发。acquire() 后忘记 release() 会导致许可逐渐耗尽，最终所有线程阻塞。务必使用 try-finally 或 try-with-resources（自定义 AutoCloseable）。release() 可以在没有 acquire 的情况下调用，导致许可数超出初始值。这可能是 Bug 或故意设计（例如动态补充许可）。acquire() 响应中断，被中断时会抛出 InterruptedException，不会获取许可。graph TD
A[需要线程同步] --> B{是否需要屏障动作?}
B -->|是| C[CyclicBarrier]
B -->|否| D{是否需要重复使用?}
D -->|是| E{是否需要动态增减计数?}
E -->|是| F[Semaphore]
E -->|否| G[CyclicBarrier<br/>或重置CountDownLatch?]
D -->|否| H{是等待事件发生一次<br/>还是控制并发数?}
H -->|等待事件发生一次| I[CountDownLatch]
H -->|控制同时访问线程数| J[Semaphore]
| 需求 | 推荐工具 | 理由 |
|---|---|---|
| 等待 N 个任务完成，不重用 | CountDownLatch | 简单、轻量、一次性 |
| 多个线程分阶段执行，每阶段都同步 | CyclicBarrier | 可重用，支持屏障动作 |
| 限制同时访问资源的线程数 | Semaphore | 许可动态增减，天然限流 |
| 模拟高并发，让所有线程同时开始 | CyclicBarrier 或 CountDownLatch | 两者都可以，CyclicBarrier 更直观 |
| 需要公平地获取许可（避免饥饿） | Semaphore(true) | 公平模式保证先到先得 |
| 工作线程完成后通知主线程 | CountDownLatch | 主线程 await，工作线程 countDown |
| 特性 | CyclicBarrier | CountDownLatch | Semaphore |
|---|---|---|---|
| 可重用 | |||
| 计数器方向 | 递减至0 | 递减至0 | 可增可减 |
| 支持释放/增加 | |||
| AQS 模式 | 无（Lock+Condition） | 共享 | 共享 |
| 屏障动作 | |||
| 典型场景 | 分阶段并行 | 等待完成信号 | 限流/池化 |
| 初始化参数 | 线程数 | 计数值 | 许可数 |
| 主要方法 | await() | await(), countDown() | acquire(), release() |
CountDownLatch 和 Semaphore 都是 AQS 共享模式的经典应用，掌握 tryAcquireShared / tryReleaseShared 的语义有助于快速理解其他同步工具（如 ReentrantReadWriteLock）。CyclicBarrier 是一个等待点（所有线程必须到达），Semaphore 是一个资源计数器（线程可以随时进出）。通过以上对比和源码分析，相信你已经能够根据业务需求在三者之间做出正确的选择，并避免常见陷阱。在实际开发中，优先考虑语义最匹配的工具，不要为了“炫技”而使用复杂的同步机制。
