AbstractQueuedSynchronizer（AQS） 源码细致分析 - CyclicBarrier 源码分析

1、简介

CyclicBarrier，回环栅栏，它会阻塞一组线程直到这些线程同时达到某一个条件才继续执行。它与 CountDownLatch 很类似，但是又不同，CountDownLatch 需要调用 countDown() 方法触发事件，而 CyclicBarrier 不需要，它就像一个栅栏一样，当一组线程都到达了栅栏处才继续往下走。

工作原理图

CyclicBarrier 与 CountDownLatch 的异同？

• 两者都能实现阻塞一组线程被唤醒；
• 前者是最后一个线程到达的时候自动唤醒；
• 后者是通过显示地调用 countDown() 实现的；
• 前者是通过重入锁及其条件队列实现的，后者是通过直接基于 AQS 实现的
• 前者具有 “代” 的概念，可以重复使用，而后者只能使用一次
• 前者只能实现多个线程到达栅栏处一起运行
• 后者不仅可以实现多个线程等待一个线程条件成立，还能实现一个线程等待多个线程条件成立（详见 CountDownLatch 那章使用案例）

2、入门案例

在分析源码之前，先看一个入门案例：

• 使用一个 CyclicBarrier 使得 5 个玩家保持同步，当 5 个线程同时到达 cyclicBarrier.await() 处，大家在一起往下运行。

/*** @author wcc* @date 2022/2/17 11:12*/
public class CycliBarrierTest01 {/*** 案例：* 模拟过气游戏 "王者荣耀" 有戏开始逻辑*/public static void main(String[] args) {// 第一步：定义玩家，定义 5 个String[] heros = {"安琪拉", "亚瑟", "张飞", "刘备", "成吉思汗"};// 第二步：创建固定线程数量的线程池，线程数量为5ExecutorService service = Executors.newFixedThreadPool(5);// 第三步：创建barrier， parties 设置为5CyclicBarrier barrier = new CyclicBarrier(5);// 第四步：通过for循环开启 5个任务，模拟游戏开始，传递给每个任务，英雄名名称 和 barrierfor (int i = 0; i < 5; i++) {service.execute(new Player(heros[i], barrier));}service.shutdown();}static class Player implements Runnable{private String hero;private CyclicBarrier barrier;public Player(String hero, CyclicBarrier barrier) {this.hero = hero;this.barrier = barrier;}@Overridepublic void run() {try {// 每个玩家加载进度不一样，这里使用随机数来模拟TimeUnit.SECONDS.sleep(new Random().nextInt(10));System.out.println(hero + "：加载进度 100%，等待其他玩家加载完成中");barrier.await();System.out.println(hero + "：发现所有英雄加载完成，开始战斗吧");}catch (Exception e){e.printStackTrace();}}}
}

运行结果如下：

3、源码分析

成员属性

// 因为 barrier 的实现是依赖于 Condition 条件队列的，Condition 条件队列必须依赖 lock 才能使用
private final ReentrantLock lock = new ReentrantLock();
// 线程挂起实现使用的 Condition 队列 条件：当前代所有线程到位，这个条件队列内的线程才会被唤醒
private final Condition trip = lock.newCondition();
// barrier 需要参与进来的线程数量
private final int parties;
// 当前代最后一个到位的线程的需要执行的事件
private final Runnable barrierCommand;
// 表示 barrier 对象当前 "代"
private Generation generation = new Generation();
// 表示当前 "代"还有多少个线程未到位，初始值为 parties
private int count;

构造方法

/*** 有参构造方法* @param parties barrier 需要参与的线程数量，每次屏障需要参与的线程数* @param barrierAction 当前"代"最后一个到位的线程需要执行的事件，可以为null*/
public CyclicBarrier(int parties, Runnable barrierAction) {// 因为小于等于0的barrier没有任何意义if (parties <= 0) throw new IllegalArgumentException();this.parties = parties;// count 的初始值就是 parties，后面当前代每到位一个线程，count--this.count = parties;// 初始化到栅栏所要执行的任务this.barrierCommand = barrierAction;
}public CyclicBarrier(int parties) {this(parties, null);
}

成员方法

3.1、nextGeneration() 方法

// 开启下一代，当所有线程到位后（假设barrierComment不为空，还需要最后一个线程执行完事件），会调用 nextGeneration()开启下一代
private void nextGeneration() {// 将在trip条件队列内挂起的线程 全部唤醒trip.signalAll();// 重置 count 为 partiescount = parties;// 开启新的一代，使用一个新的 generation 对象，表示新的一代，新的一代和上一代没有任何关系generation = new Generation();
}

3.2、breakBarrier() 方法

// 打破当前 barrier 屏障，在屏障内的线程都会抛出异常...
private void breakBarrier() {// 将代中 broken 设置为 true，表示这一代是被打破了，再来到这一代的线程，直接抛出异常generation.broken = true;// 重置 count 为 partiescount = parties;// 将再 trip 条件队列内挂起的线程全部唤醒，唤醒后的线程，会检查当前这个代是否是打破的，// 如果是打破的话，接下来的逻辑和开启下一代唤醒的逻辑不一样trip.signalAll();
}

3、await() 方法

• 每个需要在栅栏处等待的线程都需要显式的调用 await() 方法等待其他线程的到来

public int await() throws InterruptedException, BrokenBarrierException {try {// 调用 dowait() 方法，不需要超时return dowait(false, 0L);} catch (TimeoutException toe) {throw new Error(toe); // cannot happen}
}

3.4、dowait() 方法（重点）

• dowait() 方法里面的整个逻辑分成两个部分
• 最后一个线程走上面的逻辑，当 count 减为 0 的时候，打破栅栏，它调用 nextGeneration() 方法通知条件队列中的等待线程转移到 AQS 队列中等待被唤醒，并进入下一代。
• 非最后一个线程走下面的 for 循环逻辑，这些线程会阻塞在 condition 的 await() 方法处，它们会加入到条件队列中，等待被通知，当它们唤醒的时候已经更新换代了，这时候返回。

/*** @param timed：表示当前调用 await() 方法的线程是否指定了超时时长，如果是true，表示线程是响应超时的* @param nanos：线程等待超时时长 纳秒，如果timed 为 false的话，那么 nanos == 0* @return*/
private int dowait(boolean timed, long nanos)throws InterruptedException, BrokenBarrierException,TimeoutException {// 获取 barrier 全局锁对象final ReentrantLock lock = this.lock;// 加锁// 为什么要加锁呢？// 因为 barrier 挂起和唤醒依赖的组件是 Conditionlock.lock();try {// 获取 barrier 当前的代final Generation g = generation;// 如果当前代已经是被打破状态，则当前调用 await方法的线程，直接抛出 BrokenBarrierException 异常if (g.broken)throw new BrokenBarrierException();// 如果当前线程的中断标记位为 true，则打断当前 代 generation 整个 barrier 屏障，然后当前线程抛出中断异常if (Thread.interrupted()) {// 1.设置当前代状态 broken 为 true// 2.唤醒在 trip 条件内的线程，breakBarrier();throw new InterruptedException();}// 执行到这里，说明当前线程的中断状态是正常的，false，当前代的broken为false（未打破的状态）// 正常逻辑// 假设 parties 给的是5，那么index对应的值为4，3，2，1，0int index = --count;// 条件成立：说明当前线程是最后一个到达 barrier 的线程if (index == 0) {  // tripped// 标记：false：表示最后一个线程执行  的时候抛出异常，true：表示最后一个线程执行 barrierCommand 的时候未抛出异常// barrierCommand 就是创建 barrier 对象的时候指定的第二个 Runnable 接口实现对象，这个可以为nullboolean ranAction = false;try {final Runnable command = barrierCommand;// 条件成立：说明创建 barrier 对象的时候指定 Runnable 接口了，这个时候最后一个到达的线程就需要执行这个任务if (command != null)command.run();// command.run 未抛出异常的话，那么线程会执行到这里ranAction = true;// 开启新的一代// 1.唤醒 trip 条件队列内挂起的线程，被唤醒的线程会依次获取到 lock，然后依次退出await方法// 2.重置 count 为 parties// 3.创建一个新的 Generation对象，表示新的一代nextGeneration();// 因为当前线程是此代最后一个到达的线程，所以 index == 0return 0;} finally {if (!ranAction)// 如果 command.run() 执行抛出异常的话，会进入到这里，打破当前代breakBarrier();}}// 执行到这里：说明当前线程不是最后一个到达 barrier 的线程，此时需要进入一个自旋中// loop until tripped, broken, interrupted, or timed out// 自旋一直到条件满足当前代被打破、线程被中断、等待超时for (;;) {try {// 条件成立：说明当前线程是不指定超时时间的if (!timed)// 当前线程会释放掉lock，然后进入到 trip 条件队列的尾部，然后挂起自己，等待被唤醒trip.await();else if (nanos > 0L)// 执行到这里，说明当前线程调用 await() 方法的时候，是指定了超时时间的nanos = trip.awaitNanos(nanos);} catch (InterruptedException ie) {// 抛出中断异常，会进入这里// 什么时候会抛出 InterruptedException 中断异常呢？// node 节点在条件队列内的时候收到中断信号的时候会抛出中断异常// 条件一成立：g == generation 说明当前代没有变化// 条件二：前置条件：当前代没有被打破 ! g.broken：成立：当前代如果没有被打破，那么当前线程就会去打破，然后抛出中断异常if (g == generation && ! g.broken) {breakBarrier();throw ie;} else {// 执行到 else有几种情况？// 1.代发生变化了，这个时候就不需要抛出中断异常了，因为代已经更新了，// 这里唤醒后就走正常逻辑了，只不过设置下中断标记（表示是被中断唤醒的）// 2.代没有发生变化，但是代被打破了，此时也不用返回中断异常，// 执行到下面的时候，会抛出 BrokenBarrierException异常,只不过设置下中断标记（表示是被中断唤醒的）Thread.currentThread().interrupt();}}// 唤醒后，执行到这里，有几种情况？// 1.正常情况，当前 barrier 开启了新的一代，nextGeneration中（trip.signalAll）// 2.第二种情况，当前 generation 被打破了，此时也会唤醒所有在条件队列中挂起的线程// 3.第三种情况：当前线程在条件队列trip中 等待超时，主动转移到阻塞队列，获取到锁，唤醒// 条件成立：当前代已经被打破if (g.broken)// 线程唤醒后依次抛出 BrokenBarrierException 异常throw new BrokenBarrierException();// 条件成立：说明当前线程挂起期间，最后一个线程到位了，然后触发了开启新的一代的逻辑，此时唤醒trip条件队列内的线程// 正常情况，当前 barrier 开启了新的一代，nextGeneration中（trip.signalAll）if (g != generation)// 返回各自对应的 index：表示还有多少代未到位return index;// 第三种情况：当前线程在条件队列trip中 等待超时，主动转移到阻塞队列，获取到锁，唤醒if (timed && nanos <= 0L) {// 打破 barrierbreakBarrier();// 抛出超时异常throw new TimeoutException();}}} finally {lock.unlock();}
}

总结：

• CyclicBarrier 会使一组线程阻塞在 await() 处，当最后一个线程到达的时候唤醒（只是从条件队列中转移到 AQS 队列中）卡前面的线程大家再继续走下去
• CyclicBarrier 不是直接使用 AQS 实现的一个同步器
• CyclicBarrier 基于 ReentrantLock 及其 Condition 实现整个同步逻辑。