【同步工具类:CyclicBarrier】

介绍

官方介绍:
一种同步辅助工具，允许一组线程都等待对方到达共同的障碍点。CyclicBarrier在涉及固定大小的线程组的程序中非常有用，这些线程组偶尔必须彼此等待。该屏障被称为循环屏障，因为它可以在释放等待线程后重新使用。
CyclicBarrier支持可选的Runnable命令，该命令在参与方中的最后一个线程到达后，但在释放任何线程之前，在每个障碍点运行一次。此屏障动作对于在任何一方继续之前更新共享状态都很有用。
通俗理解：
它可以协同多个线程，让多个线程在这个栅栏前等待，直到所有线程都达到了这个栅栏时，再一起继续执行后面的动作.
举个例子，你和朋友约定在公交站汇合，去公园玩。这个公交站相当于栅栏。只有你们都到了公交站，才一起去公园。

源码分析

CyclicBarrier 基于ReetrantLock + Condition实现。

    /** The lock for guarding barrier entry *///用于线程之间互相唤醒private final ReentrantLock lock = new ReentrantLock();/** Condition to wait on until tripped */private final Condition trip = lock.newCondition();//总线程数private final int parties;

构造函数

可以看到，不仅可以传入 参与方的总数量(即 parties)。还可以传入一个回调函数，当所有的线程被唤醒时，barrierAction 被执行，该参数可以为空。

    /*** 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;this.count = parties;this.barrierCommand = barrierAction;}

await() 函数

1.CyclicBarrier 是可以被重用的。
2.CyclicBarrier 会响应中断，N 个线程还没有到齐，如果有线程收到了中断信号，所有阻塞的线程也会被唤醒。也就是 breakBarrier函数。然后count 被重置为初始值(parties),重新开始
3.构造函数传入的回调函数，barrierAction 只会被最后一个线程执行一次。

 public int await() throws InterruptedException, BrokenBarrierException {try {return dowait(false, 0L);} catch (TimeoutException toe) {throw new Error(toe); // cannot happen}}

    /*** Main barrier code, covering the various policies.*/private int dowait(boolean timed, long nanos)throws InterruptedException, BrokenBarrierException,TimeoutException {final ReentrantLock lock = this.lock;lock.lock();try {final Generation g = generation;if (g.broken)throw new BrokenBarrierException();if (Thread.interrupted()) {  //响应中断breakBarrier();  //唤醒所有阻塞的线程throw new InterruptedException();}int index = --count;  //每个线程调用一次await(). count 减一，当count==0时，则唤醒其他的所有线程if (index == 0) {  // trippedboolean ranAction = false;try {final Runnable command = barrierCommand;if (command != null)// 一起唤醒之和，如果回调函数不为空，还需要执行回调函数command.run();ranAction = true;nextGeneration();//唤醒其他所有线程，并将count值复原。//用于下一次的CyclicBarrier.这是可以复用的原因return 0;} finally {if (!ranAction)breakBarrier();}}// loop until tripped, broken, interrupted, or timed out//当count>0,说明 人没有到齐，需要阻塞自己for (;;) {try {if (!timed)trip.await();//当阻塞自己的时候，await方法会释放锁，这样其他线程调用await方法时会执行--countelse 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.//如果不是响应的中断，说明是被 sigalAll唤醒。则自己唤醒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();}}

     private void nextGeneration() {// signal completion of last generation// 唤醒所有阻塞的线程trip.signalAll();// set up next generation// 设置初始值，开始下一个轮回count = parties;generation = new Generation();}

业务场景

10 个求职者一起来公司应聘，招聘方式为笔试和面试。首先，需要等10个人到期后，开始笔试，笔试结束之后，再一起参加面试。把10个人看作10个线程。如图所示:

方案一:

采用一个CyclicBarrier.重复实现两次等待

代码实现

class Solver {public static void main(String[] args) {CyclicBarrier barrier=new CyclicBarrier(10);for (int i=0;i<10;i++){//开启10个线程模拟10个求职者new Thread(new JobHunt(barrier)).start();}}
}class JobHunt implements Runnable {private CyclicBarrier cyclicBarrier;public JobHunt(CyclicBarrier cyclicBarrier) {this.cyclicBarrier = cyclicBarrier;}@Overridepublic void run() {//赶来公司路上doOnTheWay();//到公司后，看人是否到齐，如果没有到齐，就阻塞，// 到齐了就开始笔试try {System.out.println(Thread.currentThread().getName()+" 已经来公司了...");cyclicBarrier.await();doWriteExam();System.out.println(Thread.currentThread().getName()+" 笔试做完了....");cyclicBarrier.await();doInterview();System.out.println(Thread.currentThread().getName()+"  面试完啦.....");} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}}/*** 模拟在路上方法*/public void doOnTheWay(){doCostTime(2000);}/*** 模拟笔试过程*/public void doWriteExam(){doCostTime(3000);}/*** 模拟面试过程*/public void doInterview(){doCostTime(5000);}private void doCostTime(int time){Random random=new Random();try {//随机休眠时间int count=random.nextInt(time);// System.out.println(count);Thread.sleep(count);} catch (InterruptedException e) {e.printStackTrace();}}}

测试截图

从截图中我们可以看出，CyclicBarrier 实现了大家一起等待，直至人到齐了再去一起做笔试或者面试。

方案二

由于两次等待结束后，打印的消息不一样。所以我们采用两个 CyclicBarrier。分别传入不同的 barrierAction，来实现自定义的 等待结束后的打印事件。

代码实现

class Solver {public static void main(String[] args) {//将笔试等待的回调函数传入CyclicBarrier barrierOnWriteExam=new CyclicBarrier(10,new BarrierActionOnWriteExam());//将面试等待的回调函数传入CyclicBarrier barrierOnInterview=new CyclicBarrier(10,new BarrierActionOnInterview());for (int i=0;i<10;i++){//开启10个线程模拟10个求职者new Thread(new JobHunt(barrierOnWriteExam,barrierOnInterview)).start();}}
}class JobHunt implements Runnable {private CyclicBarrier cyclicBarrierOnWriteExam;private CyclicBarrier cyclicBarrierOnInterview;public JobHunt(CyclicBarrier cyclicBarrierOnWriteExam,CyclicBarrier cyclicBarrierOnInterview) {this.cyclicBarrierOnWriteExam = cyclicBarrierOnWriteExam;this.cyclicBarrierOnInterview=  cyclicBarrierOnInterview;}@Overridepublic void run() {//赶来公司路上doOnTheWay();//到公司后，看人是否到齐，如果没有到齐，就阻塞，// 到齐了就开始笔试try {System.out.println(Thread.currentThread().getName()+" 已经来公司了...");cyclicBarrierOnWriteExam.await();doWriteExam();System.out.println(Thread.currentThread().getName()+" 笔试做完了....");cyclicBarrierOnInterview.await();doInterview();System.out.println(Thread.currentThread().getName()+"  面试完啦.....");} catch (InterruptedException e) {e.printStackTrace();} catch (BrokenBarrierException e) {e.printStackTrace();}}/*** 模拟在路上方法*/public void doOnTheWay(){doCostTime(2000);}/*** 模拟笔试过程*/public void doWriteExam(){doCostTime(3000);}/*** 模拟面试过程*/public void doInterview(){doCostTime(5000);}private void doCostTime(int time){Random random=new Random();try {//随机休眠时间int count=random.nextInt(time);// System.out.println(count);Thread.sleep(count);} catch (InterruptedException e) {e.printStackTrace();}}}class BarrierActionOnWriteExam implements Runnable{@Overridepublic void run() {//自定义等待完成后的回调函数System.out.println("大家人到齐了,开始笔试吧");}}class BarrierActionOnInterview implements Runnable{@Overridepublic void run() {//自定义等待完成后的回调函数System.out.println("大家人到齐了,开始面试吧");}
}

测试打印

通过打印结果可以看到，首先是能正确实现效果。其次 是通过传入 回调事件参数给 CyclicBarrier，可以很方便实现 自己的业务逻辑。

总结

虽然 CountDownLatch 和CyclicBarrier 都能实现多个线程一起等待然后一起做某些事情。
CountDownLatch 更多的是 一个主线程等待 分支线程完成。然后主线程去做其他事情。
CyclicBarrier 是 大家分别做某些事情，等每个人都做完后，大家再一起去做另外一件事情。
并且两者实现的 原理完全不同。
希望通过本文大家能对 CyclicBarrier 有个更加理性的认识。多敲敲小demo。看能否有优化的地方。这样才能更好的理解。
CountDownLatch 学习的地址:
https://blog.csdn.net/echohuangshihuxue/article/details/129280219