🏷️个人主页：牵着猫散步的鼠鼠 

🏷️系列专栏：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功能很相像，都是让线程等待其他线程，但他们之间有如下区别：

1. 计数器是否可重置:

   • CyclicBarrier 的计数器可以重置，因此它可以被多次使用。一旦所有的线程都到达了屏障点，计数器会自动重置，CyclicBarrier 可以再次使用。
   • CountDownLatch 的计数器只能被使用一次，一旦计数器到达零，就不能再重置或重新使用。

2. 使用场景:

   • 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减1int index = --count;// 4.当 count 数量减为 0 之后说明最后一个线程已经到达栅栏了，// 这时执行我们预先定义好的 Runnable 任务，唤醒全部的等待线程，重置count值为partiesif (index == 0) {  // trippedboolean 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;}@Overridepublic 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时,所有线程被唤醒,重置计数,使其可以被循环利用。

肝文章真的很累，特别是阅读源码，如果有帮助的话就多多点赞支持我吧~