一，前言
CyclicBarrier 是另一种多线程并发控制的实用工具，和 CountDownLatch 相似，可以实现线程间的计数等待，但它的功能比 CountDownLatch 更强大。
CyclicBarrier 可以接收一个参数作为 待执行的动作 barrierAction，当计数器一次计数完成后，系统就会执行此动作。Cyclic 意为循环，也就是说这个计数器可以反复使用。
循环栅栏：要求线程在栅栏处等待，达到要求的N个线程数后，则放行，同时计数器清零。然后接着凑齐下一批N个线程。
底层实现
CyclicBarrier 的关键在于利用 ReentrantLock 和 Condition，每个调用 await() 方法的线程都被加入到 condition 队列中进行等待，所有参与线程都调用了await() 之后，执行设置的后续动作 barrierCommand，再唤醒 condition 队列中的所有等待线程，重置 count，并"更新换代"。
二，构造方法
/**
* Creates a new {@code CyclicBarrier} that will trip when the
* given number of parties (threads) are waiting upon it, and which
* will execute the given barrier action when the barrier is tripped,
* performed by the last thread entering the barrier.
*
* @param parties the number of threads that must invoke {@link #await}
* before the barrier is tripped
* @param barrierAction the command to execute when the barrier is
* tripped, or {@code null} if there is no action
* @throws IllegalArgumentException if {@code parties} is less than 1
*/
public CyclicBarrier(int parties, Runnable barrierAction) {
if (parties <= 0) throw new IllegalArgumentException();
// 需要等待的线程数
this.parties = parties;
// 剩余等待的线程数，reset() 时重置count = parties
this.count = parties;
this.barrierCommand = barrierAction;
}
第一个参数：计数总数，也就是参与的线程总数。
第二个参数：当计数器一次计数完成后，系统会执行的动作。
public CyclicBarrier(int parties) {
this(parties, null);
}
三，维护锁状态逻辑
其底层使用 ReentrantLock+Condition 进行锁状态的维护
1、维护锁状态
private final ReentrantLock lock = new ReentrantLock();
private final Condition trip = lock.newCondition();
2、参与线程的数量
private final int parties;
3、所有参与线程到达屏障点后执行的动作
private final Runnable barrierCommand;
4、还在等待的线程个数(注意：这里的等待不是condition队列中的等待线程，而是还没执行await()方法的线程，得明白每个等待的含义)
private int count;
6、当前的generation,同一批线程属于同一个generation,每跨越一次屏障点就换一个新的generation
private static class Generation {
boolean broken = false;
}
四，CyclicBarrier await()
具体看看其是如何实现等待逻辑的，线程等待需要调用 await 方法
public int await() throws InterruptedException, BrokenBarrierException {
try {
return dowait(false, 0L);
} catch (TimeoutException toe) {
throw new Error(toe); // cannot happen
}
}
最终调用的是 dowait() 方法
private int dowait(boolean timed, long nanos){
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、等待线程数减1
int index = --count;
// 4、所有线程均调用了await()，即到达公共屏障点
if (index == 0) {
boolean ranAction = false;
try {
// 5、执行所有线程都到达等待点之后的Runnable
final Runnable command = barrierCommand;
if (command != null)
command.run();
ranAction = true;
// 6、**唤醒所有的 condition 队列中的线程并生成下一代 **
nextGeneration();
return 0;
} finally {
if (!ranAction)
breakBarrier();
}
}
// 7、如果等待线程数不为0
for (;;) { //自旋
try {
// 8、根据传入的参数来决定是定时等待还是非定时等待
if (!timed)
// **将当前线程加入到 condition 队列，释放lock锁资源,使得下一个线程能获得锁资源进入lock块**
trip.await();
else if (nanos > 0L)
nanos = trip.awaitNanos(nanos);
} catch (InterruptedException ie) {
// 9、线程中断之后唤醒所有线程并进入下一代
if (g == generation && ! g.broken) {
breakBarrier();
throw ie;
} else {
Thread.currentThread().interrupt();
}
}
// 10、如果线程因为打翻屏障操作而被唤醒则抛出异常
if (g.broken)
throw new BrokenBarrierException();
// 11、如果线程因为换代操作而被唤醒则返回计数器的值
if (g != generation)
return index;
// 12、如果线程因为时间到了而被唤醒则打翻栅栏并抛出异常
if (timed && nanos <= 0L) {
breakBarrier();
throw new TimeoutException();
}
}
} finally {
lock.unlock();
}
}
可以看到，是通过index字段控制线程等待的，当index不为0的时候，线程统一会进行阻塞，直到index为0的时候，才会唤醒所有线程，这时候所有线程才会继续往下执行。
再看上面代码提到的breakBarrier()和nextGeneration()方法:
private void breakBarrier() {
generation.broken = true; //设置当前generation被破坏
count = parties; //重置count
trip.signalAll(); //唤醒condition队列中的所有等待线程
}
private void nextGeneration() {
trip.signalAll(); //唤醒condition队列中的所有等待线程
count = parties; //重置count
generation = new Generation(); //”更新换代”
}
五，重复使用
这个跟 CountdownLatch 不一样的是，CountdownLatch 是一次性的，而 CycliBarrier 是可以重复使用的，只需调用一下 reset() 方法。
public void reset() {
final ReentrantLock lock = this.lock;
lock.lock();
try {
// 1、破坏当前的屏障点并唤醒所有线程
breakBarrier();
// 2、生成下一代
nextGeneration();
} finally {
lock.unlock();
}
}
private void breakBarrier() {
generation.broken = true;
// 将等待线程数量重置
count = parties;
// 唤醒所有线程
trip.signalAll();
}
private void nextGeneration() {
// 唤醒所有线程
trip.signalAll();
// 将等待线程数量重置
count = parties;
generation = new Generation();
}
六，简单使用
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* 循环栅栏：要求线程在栅栏处等待，达到要求的N个线程数后，则放行，同时计数器清零。然后接着凑齐下一批N个线程。
*/
public class CyclicBarrierDemo {
public static void main(String[] args) {
final int N = 10;
ExecutorService exec = Executors.newFixedThreadPool(10);
boolean flag = false;
//设置屏障
CyclicBarrier cyclic = new CyclicBarrier(N, new BarrierRun(flag, N));
System.out.println("集合队伍！" + ":线程：" + Thread.currentThread().getName());
for (int i = 0; i < N; i++) {
exec.submit(new Soldier(cyclic, "士兵" + i));
}
exec.shutdown();
}
public static class Soldier implements Runnable {
private String soldier;
private final CyclicBarrier cyclic;
Soldier(CyclicBarrier cyclic, String soldierName) {
this.cyclic = cyclic;
this.soldier = soldierName;
}
@Override
public void run() {
try {
// 等待所有士兵报道
report();
cyclic.await();
// 等待所有士兵完成工作
doWord();
cyclic.await();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (BrokenBarrierException e) {
//表示当前CyclicBarrier已经破损了
e.printStackTrace();
}
}
void report() {
System.out.println(soldier + "报道！" + ":线程：" + Thread.currentThread().getName());
}
void doWord() {
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(soldier + ":任务完成" + ":线程：" + Thread.currentThread().getName());
}
}
public static class BarrierRun implements Runnable {
boolean flag;
int N;
public BarrierRun(boolean flag, int N) {
this.flag = flag;
this.N = N;
}
@Override
public void run() {
if (flag) {
System.out.println("大佬：士兵" + N + "个，任务完成" + ":线程：" + Thread.currentThread().getName());
}else {
System.out.println("大佬：士兵" + N + "个，集合完毕" + ":线程：" + Thread.currentThread().getName());
flag = true;
}
}
}
}
执行结果：
集合队伍！:线程：main
士兵0报道！:线程：pool-1-thread-1
士兵4报道！:线程：pool-1-thread-5
士兵8报道！:线程：pool-1-thread-9
士兵3报道！:线程：pool-1-thread-4
士兵7报道！:线程：pool-1-thread-8
士兵2报道！:线程：pool-1-thread-3
士兵6报道！:线程：pool-1-thread-7
士兵1报道！:线程：pool-1-thread-2
士兵5报道！:线程：pool-1-thread-6
士兵9报道！:线程：pool-1-thread-10
大佬：士兵10个，集合完毕:线程：pool-1-thread-10
士兵2:任务完成:线程：pool-1-thread-3
士兵6:任务完成:线程：pool-1-thread-7
士兵9:任务完成:线程：pool-1-thread-10
士兵1:任务完成:线程：pool-1-thread-2
士兵5:任务完成:线程：pool-1-thread-6
士兵3:任务完成:线程：pool-1-thread-4
士兵7:任务完成:线程：pool-1-thread-8
士兵0:任务完成:线程：pool-1-thread-1
士兵4:任务完成:线程：pool-1-thread-5
士兵8:任务完成:线程：pool-1-thread-9
大佬：士兵10个，任务完成:线程：pool-1-thread-9
七，总结
CyclicBarrier 的关键在于利用 ReentrantLock 和 Condition，每个调用 await() 方法的线程都被加入到 condition 队列中进行等待，所有参与线程都调用了await() 之后，执行设置的后续动作 barrierCommand，再唤醒 condition 队列中的所有等待线程，重置 count，并"更新换代"。
评论区