【并发编程】 --- CyclicBarrier原理简介 + 使用方法

源码地址：https://github.com/nieandsun/concurrent-study.git

---

1 原理简介

CyclicBarrier，中文翻译过来是循环屏障的意思，从其命名我们至少可以知道两点：

• 它可以循环
• 它是一个屏障

1.1 CyclicBarrier屏障原理

1.1.1 await为空参时

其原理如下

• （1）比如说我们让CyclicBarrier拦截三个线程（使用默认构造方法CyclicBarrier(int parties)），被拦截的三个线程可以同时开启，也可以非同时开启，如下图所示：

• （2）假如线程1和线程3相继到达了各自设置的await点（或者说屏障点），而线程2没到，他们会等着线程2，如下图所示：

• （3/1）直到线程2也到达了await点，然后三个线程再同时抢占运行

• （3/2）其实 CyclicBarrier还有一点比较牛的是，它不仅可以选择线程的拦截数量，还可以加一个barrierAction（调用构造方法CyclicBarrier（int parties，Runnable barrierAction））。而这个barrierAction的执行时机正是线程1、2、3都达到await点之后，那我们在这里就可以做很多有意义的事了。如果使用了barrierAction，上面的图可以改成下面的：

---

1.1.2 await传入时间参数时

这里就不画图了，其实CyclicBarrier的await方法也可以传入一个时间，比如说线程2的await传入了2秒的等待时间，则如果两秒之后其他线程还没到达await点，它自己就先跑了，并抛出异常。。

---

1.2 CyclicBarrier中的循环是什么意思

1.1中其实只是讲了CyclicBarrier的屏障原理，但是它为什么叫循环屏障呢？？？
其实是这样的，相信看过底层源码的都知道无论是CountDownLatch还是CyclicBarrier他们对线程数量的控制其实都依赖于一个state变量。

• 在CountDownLatch里比如说我指定了state的初始数量为5，则进行一次countDown，我的state就减1，直到state减为0，await就会自动放行 —》 但是在CountDownLatch里这个state一旦初始化了，就只能减少，不能再重新复用了。
• 但是在CyclicBarrier里假如我也指定了state的初始数量为5，比如说我有10个线程，它会先对前五个线程进行阻塞 —>等他们都达到await点 —> 抢占运行； 然后你无需再手动重置state的数量，CyclicBarrier会自动对后五个线程进行同样的操作，这就是所谓的循环的意思。

---

2 CountDownLatch和CyclicBarrier简单比较

看过我上篇文章《【并发编程】 — CountDownLatch原理简介 + 使用方法》2.3小结的再看这里，肯定会觉得CyclicBarrier和 CountDownLatch的实现发令枪的用法好像啊。

其实确实很像，但也有一定的区别：

• 首先，CountDownLatch其实是由一个线程去阻塞其他线程，让其他线程在某一个点达到同步；而CyclicBarrier是
  通过各个工作线程自己调用await从而自行阻塞，直到所有工作线程达到指定屏障，再大家一起往下走。
• 其次，CountDownLatch的计数器(state)只能使用一次，而CyclicBarrier的计数器可以反复使用。
• 还有，CyclicBarrier的barrierAction其实为我们提供了更多的可操作空间
• 最后，从用法上来说
  • CyclicBarrier相当于我指定了state的数量为3，那就是三个线程到达屏障点后，这三个线程一起穿过屏障点继续抢占运行；
  • 而CountDownLatch是我指定了state数量为3，我不管你是几个线程，你只要给我countDown了三次，就可以穿越过我设置的屏障点了。

---

3 具体使用方法

---

3.1 demo1 — awit不传入时间，指定数量的线程到达屏障点后，再一起抢占运行（循环拦截）

• code

package com.nrsc.ch2.juctools;import lombok.extern.slf4j.Slf4j;import java.util.concurrent.*;@Slf4j
public class CyclicBarrierDemo1 {private final static int threadCount = 6;static CyclicBarrier barrier = new CyclicBarrier(2);public static void main(String[] args) throws InterruptedException {ExecutorService exec = Executors.newCachedThreadPool();for (int i = 0; i < threadCount; i++) {final int threadNum = i;//这里故意让后续线程晚一秒开启Thread.sleep(1000);exec.execute(() -> test(threadNum));}exec.shutdown();}private static void test(int threadNum) {try {log.info("线程{}已经准备就绪！", threadNum);barrier.await();} catch (InterruptedException e) {log.error("InterruptedException", e);} catch (BrokenBarrierException e) {log.error("BrokenBarrierException", e);}log.info("线程{}继续运行！", threadNum);}
}

• 测试结果：

---

3.2 demo2 — await传入时间t，前面的线程到达屏障点并在此等待，超过时间t后抛出异常越过屏障继续运行

package com.nrsc.ch2.juctools;import lombok.extern.slf4j.Slf4j;import java.util.concurrent.*;@Slf4j
public class CyclicBarrierDemo2 {private final static int threadCount = 2;static CyclicBarrier barrier = new CyclicBarrier(2);public static void main(String[] args) throws InterruptedException {ExecutorService exec = Executors.newCachedThreadPool();for (int i = 0; i < threadCount; i++) {final int threadNum = i;//这里故意让后续线程晚一秒开启Thread.sleep(2000);exec.execute(() -> test(threadNum));}exec.shutdown();}private static void test(int threadNum) {try {log.info("线程{}已经准备就绪！", threadNum);barrier.await(1, TimeUnit.SECONDS);} catch (InterruptedException e) {log.error("InterruptedException", e);} catch (BrokenBarrierException e) {log.error("BrokenBarrierException", e);} catch (TimeoutException e) {log.error("TimeoutException", e);}log.info("线程{}继续运行！", threadNum);}
}

• 测试结果

---

3.3 demo3 — 利用barrierAction统计到达屏障后前面各个线程运行的某些信息

• code

package com.nrsc.ch2.juctools;import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.MapUtils;import java.util.Map;
import java.util.Random;
import java.util.concurrent.*;@Slf4j
public class CyclicBarrierDemo3 {private final static int threadCount = 5;private static Map<String, Integer> concurrentHashMap = new ConcurrentHashMap();static CyclicBarrier barrier = new CyclicBarrier(5, () -> {MapUtils.verbosePrint(System.out, "前几个线程准备的数据", concurrentHashMap);});public static void main(String[] args) throws InterruptedException {ExecutorService exec = Executors.newCachedThreadPool();for (int i = 0; i < threadCount; i++) {final int threadNum = i;//这里故意让后续线程晚一秒开启Thread.sleep(1000);exec.execute(() -> test(threadNum));}exec.shutdown();}private static void test(int threadNum) {try {//模拟进行部分准备工作,并将准备的结果放入到Map容器int random = new Random().nextInt(10);log.info("线程{}准备的数据-{}",Thread.currentThread().getName(),random);concurrentHashMap.put(Thread.currentThread().getName(), random);log.info("线程{}已经准备就绪！", threadNum);barrier.await();} catch (InterruptedException e) {log.error("InterruptedException", e);} catch (BrokenBarrierException e) {log.error("BrokenBarrierException", e);}log.info("线程{}继续运行！", threadNum);}
}

• 测试结果