Java 多线程：CountDownLatch 与 CyclicBarrier 线程同步
一、CountDownLatch：等待多个任务完成
1. CountDownLatch 是什么？
CountDownLatch 是 JDK 8 提供的线程同步工具，主要用于让一个或多个线程等待其他任务完成。
例如，某个接口需要分别查询：
商品信息。
库存信息。
价格信息。
三个任务可以并行执行，但主线程必须等待三个任务结束后，才能汇总结果。
主线程
|
┌─────────┼─────────┐
| | |
v v v
线程 A 线程 B 线程 C
查询商品 查询库存 查询价格
| | |
v v v
完成 完成 完成
| | |
└─────────┼─────────┘
|
v
主线程继续执行
|
v
汇总数据CountDownLatch 就是为了解决这种线程等待问题。
2. 核心方法
创建 CountDownLatch：
CountDownLatch latch = new CountDownLatch(3);表示需要收到 3 次完成信号，计数器才能归零。
两个最重要的方法：
// 计数器减 1
latch.countDown();
// 等待计数器归零
latch.await();执行过程：
初始 count = 3
线程 A 完成
|
v
countDown()
|
v
count = 2
线程 B 完成
|
v
countDown()
|
v
count = 1
线程 C 完成
|
v
countDown()
|
v
count = 0
|
v
唤醒等待线程
|
v
主线程继续执行重点理解：
countDown()：减少计数，不需要等待其他任务。await()：等待计数归零，可能阻塞当前线程。count = 0：所有等待线程都可以继续执行。CountDownLatch 只负责同步，不负责判断业务是否成功。
3. Java 8 完整示例
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
public class CountDownLatchDemo {
public static void main(String[] args)
throws InterruptedException {
CountDownLatch latch = new CountDownLatch(3);
ExecutorService executor =
Executors.newFixedThreadPool(3);
try {
for (int i = 1; i <= 3; i++) {
final int taskId = i;
executor.execute(() -> {
try {
System.out.println(
"任务 " + taskId + " 开始"
);
Thread.sleep(taskId * 500L);
System.out.println(
"任务 " + taskId + " 完成"
);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
// 无论正常结束还是异常退出，
// 都发送一次任务结束信号
latch.countDown();
}
});
}
// 主线程最多等待 5 秒
boolean completed = latch.await(
5, TimeUnit.SECONDS
);
if (completed) {
System.out.println("所有任务均已结束");
} else {
System.out.println("等待超时");
}
} finally {
executor.shutdownNow();
}
}
}可能输出：
任务 1 开始
任务 2 开始
任务 3 开始
任务 1 完成
任务 2 完成
任务 3 完成
所有任务均已结束这里有三个重要细节。
第一：countDown() 放在 finally 中。
可以避免任务执行异常后，计数器无法归零。
第二：await() 建议设置超时时间。
boolean completed = latch.await(
5, TimeUnit.SECONDS
);防止某个任务无法完成，导致等待线程无限阻塞。
第三：计数归零不代表所有任务成功。
即使某个任务发生异常，只要执行了 countDown()，主线程仍然可能正常结束等待。
实际业务需要另外检查任务结果或异常。
4. CountDownLatch 底层原理
CountDownLatch 基于 AQS 共享模式实现。
AQS 内部使用：
private volatile int state;在 CountDownLatch 中，state 表示剩余计数。
例如：
CountDownLatch latch = new CountDownLatch(3);初始化：
AQS state = 3调用：
latch.await();执行过程：
await()
|
v
检查 AQS state
|
├── state == 0
| |
| v
| 直接返回
|
└── state > 0
|
v
进入 AQS 同步队列
|
v
等待计数归零调用：
latch.countDown();通过 CAS 更新 state：
state = 3
|
v
state = 2
|
v
state = 1
|
v
state = 0
|
v
AQS 共享释放机制
|
v
唤醒等待线程为什么 CountDownLatch 使用共享模式？
因为计数归零后，可以让多个等待线程同时通过 await()。
5. CountDownLatch 能重复使用吗？
不能。
例如：
CountDownLatch latch = new CountDownLatch(1);
latch.countDown();
// count = 0
latch.countDown();
// count 仍然是 0计数只能减少，不能重置。
如果业务需要再次等待一组任务，必须创建新的 CountDownLatch。
CountDownLatch 是一次性的线程同步工具。
二、CyclicBarrier：多个线程互相等待
1. CyclicBarrier 是什么？
CyclicBarrier 是 JDK 8 提供的循环屏障工具。
主要用于：
多个线程执行分阶段任务时，必须等待所有参与线程到达同一个屏障后，才能一起进入下一阶段。
例如三个线程：
线程 A：处理第一部分数据。
线程 B：处理第二部分数据。
线程 C：处理第三部分数据。
第一阶段全部完成后，才能开始第二阶段。
线程 A：第一阶段 ──┐
|
线程 B：第一阶段 ──┼── 屏障 ── 第二阶段
|
线程 C：第一阶段 ──┘这就是 CyclicBarrier 的典型场景。
2. 核心方法
创建：
CyclicBarrier barrier = new CyclicBarrier(3);表示需要三个线程共同到达屏障。
核心方法：
barrier.await();执行过程：
初始：
需要 3 个线程到达
线程 A 调用 await()
|
v
等待
线程 B 调用 await()
|
v
等待
线程 C 调用 await()
|
v
三个线程全部到达
|
v
屏障打开
|
v
A、B、C 继续执行这里和 CountDownLatch 有明显区别：
CyclicBarrier 的参与线程通过 await() 报到并等待；CountDownLatch 的完成方通过 countDown() 发送完成信号。
3. 为什么叫循环屏障？
因为 CyclicBarrier 可以重复使用。
例如：
第一阶段：
A、B、C 全部到达
|
v
屏障打开
|
v
第二阶段：
A、B、C 全部到达
|
v
屏障再次打开
|
v
第三阶段继续同一个 CyclicBarrier 可以用于多个阶段。
4. Java 8 完整示例
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
public class CyclicBarrierDemo {
public static void main(String[] args) {
CyclicBarrier barrier = new CyclicBarrier(
3,
() -> System.out.println("本阶段全部完成")
);
for (int i = 1; i <= 3; i++) {
final int taskId = i;
Thread thread = new Thread(() -> {
try {
// 第一阶段
System.out.println(
"线程 " + taskId + " 执行第一阶段"
);
Thread.sleep(taskId * 300L);
// 等待全部线程完成第一阶段
barrier.await(5, TimeUnit.SECONDS);
// 第二阶段
System.out.println(
"线程 " + taskId + " 执行第二阶段"
);
Thread.sleep(taskId * 200L);
// 第二轮屏障
barrier.await(5, TimeUnit.SECONDS);
System.out.println(
"线程 " + taskId + " 全部完成"
);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} catch (BrokenBarrierException
| TimeoutException e) {
System.out.println(
"线程 " + taskId + " 屏障失败"
);
}
}, "worker-" + taskId);
thread.start();
}
}
}可能输出：
线程 1 执行第一阶段
线程 2 执行第一阶段
线程 3 执行第一阶段
本阶段全部完成
线程 1 执行第二阶段
线程 2 执行第二阶段
线程 3 执行第二阶段
本阶段全部完成
线程 1 全部完成
线程 2 全部完成
线程 3 全部完成具体的线程执行顺序不固定。
5. barrierAction 是什么？
创建 CyclicBarrier 时，可以传入第二个参数：
CyclicBarrier barrier = new CyclicBarrier(
3,
() -> {
System.out.println("全部线程到齐");
}
);第二个参数叫 barrierAction，即屏障任务。
它具有三个特点：
每一轮屏障正常到齐时执行一次。
由最后到达屏障的线程执行。
执行完成后，其他等待线程才能继续下一阶段。
例如：
线程 A 到达：等待
线程 B 到达：等待
线程 C 最后到达
|
v
执行 barrierAction
|
v
唤醒其他线程
|
v
进入下一阶段生产环境中，不建议在 barrierAction 中执行长时间阻塞操作，否则其他线程需要一起等待。
三、CyclicBarrier 底层原理
1. CyclicBarrier 使用 AQS 吗？
CountDownLatch 直接基于 AQS 实现。
CyclicBarrier 则不同。
CyclicBarrier 自身不继承 AQS，而是使用 ReentrantLock + Condition 实现等待和通知。
底层主要结构：
CyclicBarrier
|
├── ReentrantLock
|
├── Condition
|
├── parties：总参与线程数量
|
├── count：当前轮剩余到达数量
|
└── generation：当前轮次状态其中：
ReentrantLock：保护内部状态。
Condition：让先到达的线程等待。
parties：一轮需要多少个参与线程。
count：本轮还差多少个线程到达。
generation：区分不同屏障轮次及其状态。
2. 底层执行流程
例如：
CyclicBarrier barrier = new CyclicBarrier(3);初始化：
parties = 3
count = 3线程 A 到达：
count = 2
线程 A 通过 Condition.await() 等待线程 B 到达：
count = 1
线程 B 通过 Condition.await() 等待线程 C 到达：
count = 0
执行 barrierAction（如果配置）
开启新一轮
count 重置为 3
Condition.signalAll()
唤醒本轮等待线程这里联系之前学习的 Condition：
condition.await();会释放内部 ReentrantLock，并让线程进入条件等待。
condition.signalAll();会让等待线程有机会重新获取内部锁，完成屏障等待。
注意：
CyclicBarrier 使用 Condition 管理等待，但仍然必须遵守锁竞争规则，signalAll() 不代表所有线程立即同时获得锁。
四、CountDownLatch 与 CyclicBarrier 的区别
1. CountDownLatch 的执行模式
主线程 await()
|
v
等待中
|
┌────────────┼────────────┐
| | |
v v v
任务 A 任务 B 任务 C
| | |
v v v
countDown() countDown() countDown()
| | |
└────────────┼────────────┘
|
v
count = 0
|
v
主线程继续特点：
等待者与发送完成信号的线程可以是不同的。
2. CyclicBarrier 的执行模式
线程 A ── 第一阶段 ── await() ──┐
|
线程 B ── 第一阶段 ── await() ──┼── 屏障打开
|
线程 C ── 第一阶段 ── await() ──┘
|
v
第二阶段
|
v
再次等待到齐特点：
参与线程通常既是任务执行者，也是屏障等待者。
3. 怎么选择？
如果是：
主线程需要等待 5 个任务结束，再汇总最终结果。
优先考虑 CountDownLatch。
如果是：
5 个线程分阶段计算，每个阶段必须全部到齐才能进入下一阶段。
优先考虑 CyclicBarrier。
最简单的记忆方式：
CountDownLatch：等别人做完。
CyclicBarrier：大家互相等齐。
五、异常处理与生产环境常见问题
1. CountDownLatch 计数归零不代表业务成功
例如：
try {
executeTask();
} finally {
latch.countDown();
}即使 executeTask() 失败，也会执行 countDown()。
因此：
count = 0只能表示完成信号已经达到要求。
不能证明所有业务任务执行成功。
如果需要确认任务结果，应结合 Future、CompletableFuture 或单独的结果容器。
2. CountDownLatch 应设置等待超时
不推荐在无法保证所有任务一定完成的场景中无限等待：
latch.await();推荐：
boolean completed = latch.await(
5,
TimeUnit.SECONDS
);
if (!completed) {
// 执行超时处理
}需要注意：
await() 超时只表示等待结束，不会自动取消正在执行的子任务。
如果业务需要取消子任务，应配合 Future.cancel() 等机制处理。
3. CyclicBarrier 的线程池大小不足
假设：
ExecutorService executor =
Executors.newFixedThreadPool(2);
CyclicBarrier barrier =
new CyclicBarrier(3);提交三个任务，每个任务都需要执行：
barrier.await();执行情况：
线程池最大工作线程数量：2
任务 A：运行，进入屏障等待
任务 B：运行，进入屏障等待
任务 C：仍在任务队列，无法开始
屏障需要 3 个参与线程到齐结果：
A 等 C
B 等 C
C 没有空闲工作线程执行如果没有超时或取消机制，就可能一直无法推进。
使用 CyclicBarrier 时，必须保证指定数量的参与任务能够运行到屏障。
生产环境还应该考虑任务在到达屏障前发生异常的情况，避免其他线程无限等待。
4. 什么是 BrokenBarrierException？
CyclicBarrier 有一个重要概念：
BrokenBarrier，即屏障破损。
例如：
线程 A：到达屏障，等待
线程 B：到达屏障，等待
线程 C：尚未到达
线程 A：等待超时
|
v
本轮屏障被标记为 broken
|
v
线程 B：抛出 BrokenBarrierException导致屏障破损的常见情况：
等待线程被中断。
等待线程超时。
barrierAction 执行失败。
外部调用 reset() 破坏当前等待轮次。
可以检查：
barrier.isBroken();也可以重置：
barrier.reset();但是不建议在还有线程正常等待时随意 reset()。
因为它可能导致等待线程收到 BrokenBarrierException。
如果要重试，应先协调参与任务的失败处理，再决定是否重置屏障。
5. barrierAction 不要执行耗时操作
例如：
CyclicBarrier barrier = new CyclicBarrier(
3,
() -> {
// 不推荐执行长时间阻塞操作
callRemoteService();
}
);因为 barrierAction 执行期间，其他已经到达屏障的线程需要继续等待。
推荐将 barrierAction 用于简单的汇总、状态切换等操作。
6. 线程池使用注意事项
学习示例中：
Executors.newFixedThreadPool(3);可以方便地演示并发行为。
但在生产环境中，更推荐显式创建 ThreadPoolExecutor，配置：
核心线程数。
最大线程数。
有界任务队列。
线程工厂。
拒绝策略。
因为 Executors.newFixedThreadPool() 默认使用无界 LinkedBlockingQueue，任务长期积压时可能导致内存压力。
六、重点总结
需要重点记住：
CountDownLatch 用于等待多个任务结束。
CyclicBarrier 用于多个线程分阶段协作。
CountDownLatch 通过 countDown() 减少计数。
CyclicBarrier 通过 await() 到达屏障。
CountDownLatch 的等待者与任务执行者可以不同。
CyclicBarrier 的参与线程通常自己执行 await()。
CountDownLatch 不能重复使用。
CyclicBarrier 正常完成一轮后可以重复使用。
CountDownLatch 直接基于 AQS 共享模式实现。
CyclicBarrier 使用 ReentrantLock + Condition 实现。
CountDownLatch 计数归零不代表任务全部成功。
CyclicBarrier 在超时、中断等情况下可能进入 broken 状态。
CyclicBarrier 的参与任务数量与线程池容量必须合理匹配。
生产环境应考虑超时、异常传播、线程取消和任务结果检查。
七、常见问题与答案
1. CountDownLatch 和 CyclicBarrier 最大的区别是什么？
CountDownLatch 是等待完成信号，CyclicBarrier 是多个线程相互等待到达屏障。
2. CountDownLatch 可以重复使用吗？
不可以。计数归零后无法重置，需要重新创建对象。
3. CyclicBarrier 可以重复使用吗？
可以。正常完成一轮后，自动进入下一轮。
4. CountDownLatch 的 countDown() 会阻塞调用线程吗？
不会因为等待其他任务完成而阻塞。它只是减少计数，并在必要时触发唤醒机制。
5. CountDownLatch 的 await() 为什么会阻塞？
因为 AQS state 还没有归零，当前线程需要等待完成信号。
6. CountDownLatch 的计数归零说明所有任务成功了吗？
不一定，只说明收到了足够数量的完成信号。
7. CyclicBarrier 为什么可以重复使用？
因为每轮全部到齐后，会建立新的屏障轮次并重置剩余到达计数。
8. CyclicBarrier 的最后一个线程到达后会发生什么？
如果设置了 barrierAction，先执行屏障任务，然后释放本轮等待线程，并进入下一轮。
9. CyclicBarrier 为什么可能发生死锁？
例如屏障需要三个任务到齐，但线程池只有两个工作线程，前两个任务阻塞等待，第三个任务无法获得线程执行。
10. BrokenBarrierException 是什么？
表示线程等待的屏障轮次已被破坏，例如其他等待线程超时或中断。
11. CountDownLatch 与 CyclicBarrier 哪个直接基于 AQS？
CountDownLatch 直接基于 AQS；CyclicBarrier 本身不继承 AQS，而是通过 ReentrantLock 和 Condition 实现。
12. 为什么不建议在 barrierAction 中执行耗时操作？
因为它由最后一个到达的线程执行，执行完成前其他等待线程不能进入下一阶段。
八、复习测试
CountDownLatch 是什么？解决什么问题？
CountDownLatch(3) 表示什么？
countDown() 和 await() 分别有什么作用？
CountDownLatch 为什么基于 AQS 共享模式？
CountDownLatch 的 state 记录什么？
CountDownLatch 的 count 降为 0 后能否重置？
为什么 countDown() 通常放在 finally 中？
CountDownLatch 计数归零能否保证所有任务成功？
CyclicBarrier 是什么？为什么叫循环屏障？
CyclicBarrier(3) 有两个线程到达时会发生什么？
CyclicBarrier 为什么可以重复使用？
CyclicBarrier 底层为什么需要 ReentrantLock 和 Condition？
barrierAction 由哪个线程执行？什么时候执行？
为什么线程池只有两个线程，却使用 CyclicBarrier(3)，可能导致任务无法推进？
BrokenBarrierException 在什么情况下可能出现？
CountDownLatch 与 CyclicBarrier 应该如何选择？