1. 简介
简单描述CyclicBarrier的功能，那就是
它允许一组线程互相等待，直到到达某个公共屏障点 (Common Barrier Point)。在涉及一组固定大小的线程的程序中，这些线程必须不时地互相等待，此时 CyclicBarrier 很有用。因为该 Barrier 在释放等待线程后可以重用，所以称它为循环( Cyclic ) 的 屏障( Barrier )
2. 实现原理
- 借助ReentrantLock和Condition作为线程间通信机制
- 等到所有parties参与线程都到达阻塞屏障，会唤醒所有parties参与线程（会优先执行barrierAction线程），到达数不足parties则所有线程需要阻塞等待
3. 源码结构
- parties 变量，表示拦截线程的总数量。
- count 变量，表示拦截线程的剩余需要数量。
- barrierAction 变量，为 CyclicBarrier 接收的 Runnable 命令，用于在线程到达屏障时，优先执行
barrierAction ，用于处理更加复杂的业务场景 - generation 变量，表示 CyclicBarrier 的更新换代
4. 源码剖析
4.1 await方法
- 每个线程调用 await() 方法，告诉 CyclicBarrier 我已经到达了屏障，然后当前线程被阻塞。当所有线程都到达了屏障，结束阻塞，所有线程可继续执行后续逻辑
- 内部调用 dowait(boolean timed, long nanos) 方法，执行阻塞等待( timed=true )
4.2 解除阻塞情况
如果该线程不是到达的最后一个线程，则他会一直处于等待状态，除非发生以下情况：
- 最后一个线程到达，即 index == 0 。
- 超出了指定时间（超时等待）。
- 其他的某个线程中断当前线程。
- 其他的某个线程中断另一个等待的线程。
- 其他的某个线程在等待 barrier 超时。
- 其他的某个线程在此 barrier 调用 reset() 方法。reset() 方法，用于将屏障重置为初始状态
4.3 BrokenBarrierException异常情况
- 如果一个线程处于等待状态时，如果其他线程调用 reset() 方法
- 调用的 barrier 原本就是被损坏的，则抛出BrokenBarrierException异常
- 任何线程在等待时被中断了，则其他所有线程都将抛出BrokenBarrierException异常，并将 barrier 置于损坏状态
4.4 Generation
Generation 是 CyclicBarrier 内部静态类，描述了 CyclicBarrier 的更新换代。在CyclicBarrier中，同一批线程属于同一代。当有 parties 个线程全部到达 barrier 时，generation 就会被更新换代。其中 broken 属性，标识该当前 CyclicBarrier 是否已经处于中断状态。
4.5 breakBarrier
private void breakBarrier() {
generation.broken = true;//设置为中断状态
count = parties;//重置已到达屏障数量
trip.signalAll();//唤醒全部等待线程
}
在 breakBarrier() 方法中，中除了将 broken设置为 true ，还会调用 #signalAll() 方法，将在 CyclicBarrier 处于等待状态的线程全部唤醒
4.6 nextGeneration
private void nextGeneration() {
trip.signalAll();
count = parties;
generation = new Generation();
}
当所有线程都已经到达 barrier 处（index == 0），则会通过 nextGeneration() 方法，进行更新换代操作。在这个步骤中，做了三件事：
- 唤醒所有线程。
- 重置 count
- 重置 generation
4.7 reset
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
breakBarrier(); // 屏障中断
nextGeneration(); // 重置年代
} finally {
lock.unlock();
}
}
重置 barrier 到初始化状态，通过组合 breakBarrier() 和 nextGeneration() 方法来实现
4.8 getNumberWaiting
public int getNumberWaiting() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
return parties - count;
} finally {
lock.unlock();
}
}
5. 实战用例
汽车准乘人数有限，模拟两个旅行团乘客上车
/**
* created by guanjian on 2020/12/28 17:57
*/
public class CyclicBarrierTest {
private final static int nums = 5;
private final static CyclicBarrier cyclicBarrier = new CyclicBarrier(nums, () -> {
System.out.println("执行");
});
public static void main(String[] args) {
IntStream.range(0, nums).forEach(x -> {
new Thread(()->{
try {
System.out.println("阻塞等待");
cyclicBarrier.await(6000, TimeUnit.MILLISECONDS);
System.out.println("满足条件执行");
} catch (Exception e) {
e.printStackTrace();
}
}).start();
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
}
【控制台输出】
阻塞等待
阻塞等待
阻塞等待
阻塞等待
阻塞等待
执行 //===屏障全部到达后，BarrierAction先执行===
满足条件执行
满足条件执行
满足条件执行
满足条件执行
满足条件执行
6. 总结
- 实现本质是通过ReentrantLock + Condition进行线程间通信
- 本地模拟并发可以选择CyclicBarrier触发，比Jmeter更容易编码操控
