导读：这篇文章介绍的是java并发组件aqs之CyclicBarrier
Cyclicbarrier概念：
-
Cyclicbarrier是一个同步辅助类，它允许一组线程相互等待直到到达某个公共的屏障点，通过它可以完成多个线程之间相互等待，直有当每个线程都准备就绪后才能各自继续往下执行后面的操作，和countdownlatch有相似的地方，都是通过计数器来实现的，当某个线程调用了await()方法之后该线程就进入了等待状态而且计数器执行的是加一的操作，当计数器的值达到我们设置的初始值的时候，因为调用了await()方法进入等待状态的线程会被唤醒继续执行它们后续的操作。由于CyclicBarrier释放等待线程后可以重用所以我们又称为循环屏障。
cyclicbarrier
Cyclicbarrier使用场景：
-
Cyclicbarrier的使用场景跟Countdownlatch很相似，可以用于多线程计算数据最后合并计算结果的应用场景。比如我们需要统计多个Excel中的数据，然后等到一个总结果。我们可以通过多线程处理每一个Excel，执行完成后得到相应的结果，最后通过Barrieraction来计算这些线程的计算结果，得到所有Excel的总和。
CyclicBarrier中的方法
下面通过实例来展示下Cyclicbarrier的使用
- 1、示列一
@Slf4j
public class CyclicBarrierExample1 {
private static CyclicBarrier barrier = new CyclicBarrier(5); //定义5个线程同步等待
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int threadNum = i;
Thread.sleep(1000);
executor.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
log.error("exception", e);
}
});
}
executor.shutdown();
}
private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
barrier.await(); //线程调用await()告诉当前有个线程ok了
log.info("{} continue", threadNum); //到达定义的数目时，await()方法之后的操作就可以执行
}
}
返回结果：可以看出通过下面的返回结果。代码中定义了5个线程同步等待，当rece()中每个线程调用await()告诉当前线程ok了，当累计到我们之前定义的线程数时，await()方法后的操作开始执行。
- 2、示列二
@Slf4j
public class CyclicBarrierExample2 {
private static CyclicBarrier barrier = new CyclicBarrier(5);
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
final int threadNum = i;
Thread.sleep(1000);
executor.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
log.error("exception", e);
}
});
}
executor.shutdown();
}
private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
try {
// 由于状态可能会改变，所以会抛出BarrierException异常，如果想继续往下执行，需要加上try-catch
barrier.await(2000, TimeUnit.MILLISECONDS); //用于做超时处理
} catch (Exception e) {
log.warn("BarrierException", e);
}
log.info("{} continue", threadNum);
}
}
返回结果：注意查看下返回结果的时间，示例二和示例一的代码差不多，我们加了个await的超时处理，
这个过程中如果超时状态可能会改变，所以会抛出BarrierException异常，这时候如果要让后续的操作继续执行，得捕捉异常。
- 3、示列三
@Slf4j
public class CyclicBarrierExample3 {
private static CyclicBarrier barrier = new CyclicBarrier(5, () -> {
// 当线程全部到达屏障时，优先执行这里的runnable
log.info("callback is running");
});
public static void main(String[] args) throws Exception {
ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < 10; i++) {
final int threadNum = i;
Thread.sleep(1000);
executor.execute(() -> {
try {
race(threadNum);
} catch (Exception e) {
log.error("exception", e);
}
});
}
executor.shutdown();
}
private static void race(int threadNum) throws Exception {
Thread.sleep(1000);
log.info("{} is ready", threadNum);
barrier.await();
log.info("{} continue", threadNum);
}
}
返回结果：这里我们定义CyclicBarrier(int parties, Runnable barrierAction)，指定了一个runnable就是在线程到达屏障的时候优先执行runnable。
总结：CyclicBarrier的内部是使用重入锁ReentrantLock和Condition，在CyclicBarrier中最重要的方法莫过于await()方法，表示在此的所有线程都已经在此barrier上调用了await()方法之前，将一直等待。同时await()方法内部调用了dowait(boolean timed, long nanos)方法，更多详细的可以看CyclicBarrier的类源码(目前先不记录了)，await()的处理逻辑还是比较简单的：如果该线程不是到达的最后一个线程，则他会一直处于等待状态，除非发生以下情况:
1、最后一个线程到达，即index == 0
2、超出了指定时间（超时等待）
3、其他的某个线程中断当前线程
4、其他的某个线程中断另一个等待的线程
5、其他的某个线程在等待barrier超时
6、其他的某个线程在此barrier调用reset()方法。reset()方法用于将屏障重置为初始状态。
