CyclicBarrier
5.3.1 概述以及基本使用
CyclicBarrier的字面意思是可循环使用（Cyclic）的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫同步点）时被阻塞，直到最后一个线程到达屏障时，屏障
才会开门，所有被屏障拦截的线程才会继续运行。
例如：公司召集5名员工开会，等5名员工都到了，会议开始。我们创建5个员工线程，1个开会线程，几乎同时启动，使用CyclicBarrier保证5名员工线程全部执行后，再执行开会线程。
CyclicBarrier的相关方法
public CyclicBarrier(int parties, Runnable barrierAction) // 用于在线程到达屏障时，优先执行barrierAction，方便处理更复杂的业务场景
public int await() // 每个线程调用await方法告诉CyclicBarrier我已经到达了屏障，然后当前线程被阻塞
案例演示：模拟员工开会
实现步骤：
- 创建一个员工线程类(EmployeeThread),该线程类中需要定义一个CyclicBarrier类型的形式参数
- 创建一个开会线程类(MettingThread)
- 测试类
- 创建CyclicBarrier对象
- 创建5个EmployeeThread线程对象，把第一步创建的CyclicBarrier对象作为构造方法参数传递过来
- 启动5个员工线程
员工线程类
public class EmployeeThread extends Thread {// CyclicBarrier类型的成员变量private CyclicBarrier cyclicBarrier ;public EmployeeThread(CyclicBarrier cyclicBarrier) { // 使用构造方法对CyclicBarrier进行初始化this.cyclicBarrier = cyclicBarrier ;}@Overridepublic void run() {try {// 模拟开会人员的随机到场Thread.sleep((int) (Math.random() * 1000));System.out.println(Thread.currentThread().getName() + " 到了! ");cyclicBarrier.await();} catch (Exception e) {e.printStackTrace();}}}
开会线程类
public class MettingThread extends Thread {@Overridepublic void run() {System.out.println("好了，人都到了，开始开会......");}}
测试类
public class CyclicBarrierDemo01 {public static void main(String[] args) {// 创建CyclicBarrier对象CyclicBarrier cyclicBarrier = new CyclicBarrier(5 , new MettingThread()) ;// 创建5个EmployeeThread线程对象，把第一步创建的CyclicBarrier对象作为构造方法参数传递过来EmployeeThread thread1 = new EmployeeThread(cyclicBarrier) ;EmployeeThread thread2 = new EmployeeThread(cyclicBarrier) ;EmployeeThread thread3 = new EmployeeThread(cyclicBarrier) ;EmployeeThread thread4 = new EmployeeThread(cyclicBarrier) ;EmployeeThread thread5 = new EmployeeThread(cyclicBarrier) ;// 启动5个员工线程thread1.start();thread2.start();thread3.start();thread4.start();thread5.start();}}
5.3.2 使用场景
使用场景：CyclicBarrier可以用于多线程计算数据，最后合并计算结果的场景。
比如：现在存在两个文件，这个两个文件中存储的是某一个员工两年的工资信息(一年一个文件)，现需要对这两个文件中的数据进行汇总；使用两个线程读取2个文件中的数据，当两个文
件中的数据都读取完毕以后，进行数据的汇总操作。
分析：要想在两个线程读取数据完毕以后进行数据的汇总，那么我们就需要定义一个任务类(该类需要实现Runnable接口)；两个线程读取完数据以后再进行数据的汇总，那么我们可以将
两个线程读取到的数据先存储到一个集合中，而多线程环境下最常见的线程集合类就是ConcurrentHashMap，而这个集合需要被两个线程都可以进行使用，那么我们可以将这个集
合作为我们任务类的成员变量，然后我们在这个任务类中去定义一个CyclicBarrier对象，然后在定义一个方法(count)，当调用这个count方法的时候需要去开启两个线程对象，
使用这两个线程对象读取数据，把读取到的数据存储到ConcurrentHashMap对象，当一个线程读取数据完毕以后，调用CyclicBarrier的awit方法(告诉CyclicBarrier我已经
到达了屏障)，然后在任务类的run方法对ConcurrentHashMap的数据进行汇总操作；
实现步骤:
- 定义一个任务类CyclicBarrierThreadUse(实现了Runnable接口)
- 定义成员变量：CyclicBarrier ，ConcurrentHashMap
private CyclicBarrier cyclicBarrier = new CyclicBarrier(2 , this) ;
private ConcurrentHashMap<Integer , String> concurrentHashMap = new ConcurrentHashMap<Integer , String>() ;
- 定义一个方法count方法，在count方法中开启两个线程对象(可以使用匿名内部类的方式实现)
- 在run方法中对ConcurrentHashMap中的数据进行汇总
- 编写测试类CyclicBarrierThreadUseDemo
- 创建CyclicBarrierThreadUse对象，调用count方法
任务类代代码：
public class CyclicBarrierThreadUse implements Runnable {// 当前我们两个线程到达了屏障点以后，我们需要立即对数据进行汇总, 因此我们需要使用第二个构造方法// 并且我们当前这个类就是一个任务类，因此我们可以直接传递参数thisprivate CyclicBarrier cyclicBarrier = new CyclicBarrier(2 , this) ;private ConcurrentHashMap<Integer , String> concurrentHashMap = new ConcurrentHashMap<Integer , String>() ; // 存储两个线程所读取的数据public void count() {// 定义一个方法count方法，在count方法中开启两个线程对象(可以使用匿名内部类的方式实现)// 线程1new Thread(new Runnable() {@Overridepublic void run() {// 读取数据BufferedReader bufferedReader = null ;try {bufferedReader = new BufferedReader(new FileReader("D:\\salary\\2017-salary.txt")) ;String line = null ;while((line = bufferedReader.readLine()) != null) {concurrentHashMap.put(Integer.parseInt(line) , "") ; // 小的问题，工资信息不能重复}} catch (Exception e) {e.printStackTrace();} finally {if(bufferedReader != null) {try {bufferedReader.close();} catch (IOException e) {e.printStackTrace();}}}// 模拟任务的执行时间try {TimeUnit.SECONDS.sleep(5) ;System.out.println(Thread.currentThread().getName() + "---------------------线程读取数据完毕....");cyclicBarrier.await() ; //通知cyclicBarrier当前线程已经到达了屏障点} catch (Exception e) {e.printStackTrace();}}}).start();// 线程2new Thread(new Runnable() {@Overridepublic void run() {// 读取数据BufferedReader bufferedReader = null ;try {bufferedReader = new BufferedReader(new FileReader("D:\\salary\\2019-salary.txt")) ;String line = null ;while((line = bufferedReader.readLine()) != null) {concurrentHashMap.put(Integer.parseInt(line) , "") ; // 小的问题，工资信息不能重复}} catch (Exception e) {e.printStackTrace();} finally {if(bufferedReader != null) {try {bufferedReader.close();} catch (IOException e) {e.printStackTrace();}}}// 模拟任务的执行时间try {TimeUnit.SECONDS.sleep(10) ;System.out.println(Thread.currentThread().getName() + "---------------------线程读取数据完毕....");cyclicBarrier.await() ; //通知cyclicBarrier当前线程已经到达了屏障点} catch (Exception e) {e.printStackTrace();}}}).start();}@Overridepublic void run() {// 获取concurrentHashMap中的数据进行汇总Enumeration<Integer> enumeration = concurrentHashMap.keys(); // 获取concurrentHashMap中所有的键/*** 这个Enumeration的使用和我们之前所学习过的迭代器类似* boolean hasMoreElements(); 判断集合中是否存在下一个元素* E nextElement(); 获取元素*/int result = 0 ;while(enumeration.hasMoreElements()) {Integer integer = enumeration.nextElement();result += integer ;}// 输出System.out.println(result);}}
测试类代码：
public class CyclicBarrierThreadUseDemo01 {public static void main(String[] args) {// 创建任务类的对象CyclicBarrierThreadUse cyclicBarrierThreadUse = new CyclicBarrierThreadUse();// 调用count方法进行数据汇总cyclicBarrierThreadUse.count();}}
)
