🏷️个人主页：牵着猫散步的鼠鼠
🏷️系列专栏：Java全栈-专栏
🏷️个人学习笔记，若有缺误，欢迎评论区指正
目录
1. 前言
2. 什么是CyclicBarrier？
3. CyclicBarrier与CountDownLatch的区别
4. CyclicBarrier底层原理实现
5. CyclicBarrier的使用
6. 总结
1. 前言
我们在前面的文章讲解了AQS的底层原理，以及CountDownLatch(倒计时器)，Semaphore(信号量)等同步工具类的底层原理与使用，甚至基于AQS手搓了一个简易的线程同步器，今天我们就继续来学习一个新的线程同步器CyclicBarrier(循环屏障)，他与前两者相比其内部没有使用AQS。
2. 什么是CyclicBarrier？
CyclicBarrier 是 Java 中的一个线程同步工具类，它允许一组线程互相等待，直到所有线程都达到某个屏障点（同步点）后才继续执行。这个屏障被称为循环屏障，因为它可以在释放等待线程后重新使用。
当线程调用 CyclicBarrier 的 await() 方法时，该线程会被阻塞，直到所有线程都调用了 await() 方法。一旦最后一个线程调用了 await() 方法，所有等待的线程将被释放，并且可以选择性地执行一个预先定义好的 Runnable 任务。这个机制特别适合用于并行计算中，例如，当多个线程需要处理任务的多个部分，并且必须等待所有部分都完成才能进行下一步操作时。
CyclicBarrier 的一个特点是它可以循环使用。一旦所有线程都通过了屏障，CyclicBarrier 可以重置并再次使用。
3. CyclicBarrier与CountDownLatch的区别
有同学可能会感觉CyclicBarrier和CountDownLatch功能很相像，都是让线程等待其他线程，但他们之间有如下区别：
-
计数器是否可重置:
- CyclicBarrier 的计数器可以重置，因此它可以被多次使用。一旦所有的线程都到达了屏障点，计数器会自动重置，CyclicBarrier 可以再次使用。
- CountDownLatch 的计数器只能被使用一次，一旦计数器到达零，就不能再重置或重新使用。
-
使用场景:
- CyclicBarrier 通常用于一组线程必须相互等待，直到所有线程都达到某个屏障点的情况。它适合用于模拟并发问题的场景，例如多线程计算，每个线程处理一部分数据，然后等待所有线程完成再进行汇总。
- CountDownLatch 通常用于一个或多个线程等待其他线程完成某些操作的情况。它适用主线程等待一系列的子任务线程完成这种场景。
以及底层实现不同，CountDownLatch内部是通过AQS和CAS操作实现的，而CyclicBarrier的实现原理我们接下来会讲解。
4. CyclicBarrier底层原理实现
在CyclicBarrier有两个成员变量分别为parties，count，前者代表每次拦截的线程数量，后者是计数器，每有一个线程执行到同步点时，count减1，当count值变为0时说明所有线程都走到了同步点，这时就可以尝试执行我们在构造方法中设计的任务啦。
//每次拦截的线程数
private final int parties;
//计数器
private int count;
//一个参数的构造
public CyclicBarrier(int parties) {
this(parties, null);
}
//多参构造，parties为拦截线程数，barrierAction这个 Runnable会在
// CyclicBarrier 的计数器为 0 的时候执行，也就是所有线程都完成任务后执行
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
this.parties = parties;
this.count = parties;
this.barrierCommand = barrierAction;
}
每个线程通过调用await方法告诉CyclicBarrier已经到达屏障，然后进行阻塞等待，知道count等于0，所有线程都到达了屏障，因此，我们跟入await方法的源码中去看一下，在dowait(boolean timed, long nanos)，可以通过时间参数来设置阻塞的时间，默认为false。
public int await() throws InterruptedException, BrokenBarrierException {
try {
// false参数代表不设置阻塞超时时间
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
//await方法内部，继续调用dowait方法实现功能
private int dowait(boolean timed, long nanos)
throws InterruptedException, BrokenBarrierException,
TimeoutException {
final ReentrantLock lock = this.lock;
// 1.使用重入锁对操作上锁
lock.lock();
try {
final Generation g = generation;
if (g.broken)
throw new BrokenBarrierException();
// 2.如果线程中断了，抛出异常，唤醒其他线程
if (Thread.interrupted()) {
//打破屏障，唤醒全部等待的线程
breakBarrier();
throw new InterruptedException();
}
// 3.cout减1
int index = --count;
// 4.当 count 数量减为 0 之后说明最后一个线程已经到达栅栏了，
// 这时执行我们预先定义好的 Runnable 任务，唤醒全部的等待线程，重置count值为parties
if (index == 0) { // tripped
boolean ranAction = false;
try {
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
// 将 count 重置为 parties 属性的初始化值
// 唤醒之前等待的线程
// 下一波执行开始
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
// 超时时间判断，如果没有配置超时时间就会之间等待
for (;;) {
try {
if (!timed)
// 使用lock接口提供的Condition信号量让当前线程阻塞等待
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();
}
}
} finally {
lock.unlock();
}
}
这么长的源码，其实原理很简单，我给大家总结下总体逻辑(可结合代码注解阅读)
内部通过使用重入锁上锁来保证对count修改的原子性，每次线程调用await后，都会进行--count操作，直到 count 为0时，会去执行我们预先定义好的 Runnable 任务command，然后唤醒线程继续向下执行，并且重置当前计数器，所以它能处理循环使用的场景。
简言之：CyclicBarrier内部通过维护一个int类型的count变量代表还需要等待到达的线程数，通过ReentrantLock保证了内部修改操作的原子性，通过ReentrantLock提供的Condition实现了线程的等待与唤醒，每次调用await都会让count-1，直到count变为0，就唤醒所有等待中的线程，并执行预设的Runnable任务，并重置同步器中count等数据，让该同步器可以被循环利用
5. CyclicBarrier的使用
大致的了解了CyclicBarrier的原理之后，我们写个小demo测试一下它如何使用
public class Test {
public static void main(String[] args) {
int numberOfThreads = 3; // 线程数量
CyclicBarrier barrier = new CyclicBarrier(numberOfThreads, () -> {
// 当所有线程都到达障碍点时执行的操作
System.out.println("所有线程都已到达屏障，进入下一阶段");
});
for (int i = 0; i < numberOfThreads; i++) {
new Thread(new Task(barrier), "Thread " + (i + 1)).start();
}
}
static class Task implements Runnable {
private final CyclicBarrier barrier;
public Task(CyclicBarrier barrier) {
this.barrier = barrier;
}
@Override
public void run() {
try {
System.out.println(Thread.currentThread().getName() + " 正在屏障处等待");
barrier.await(); // 等待所有线程到达障碍点
System.out.println(Thread.currentThread().getName() + " 已越过屏障.");
} catch (Exception e) {
e.printStackTrace();
}
}
}
}
执行结果：
Thread 2 正在屏障处等待
Thread 1 正在屏障处等待
Thread 3 正在屏障处等待
所有线程都已到达屏障，进入下一阶段
Thread 3 已越过屏障.
Thread 1 已越过屏障.
Thread 2 已越过屏障.
6. 总结
CyclicBarrier可以让一组线程互相等待,直到最后一个线程也准备就绪后,它们才能继续运行。就好比几个朋友约好一起吃晚餐,必须等到所有人到齐后才能入座就餐。CyclicBarrier实现了这种"集体出发"的功能,每次所有线程就位后,它们可以执行一个预先指定的任务,然后继续向前推进。
有趣的是,CyclicBarrier与CountDownLatch不同,它可以重复使用。这也从侧面体现了两者在设计理念上的差异。
借鉴了AQS的精髓,CyclicBarrier内部通过ReentrantLock和Condition实现了高效、安全的线程同步和通信。count变量作为"计数器",当减至0时,所有线程被唤醒,重置计数,使其可以被循环利用。
肝文章真的很累，特别是阅读源码，如果有帮助的话就多多点赞支持我吧~
