CyclicBarrier

一、简介

CyclicBarrier是在jdk1.5版本引入的，存在于juc包下面。由大佬 Doug Lea（绝对牛逼的大佬，感兴趣的可以自己百度一下）编写。

跟它一同出现的还有CountDownLatch、Semaphore、concurrentHashMap和BlockingQueue。

二、作用

CyclicBarrier用法与CountDownLatch用法基本相似，但是功能却比CountDownLatch更加强大。它包含了CountDownLatch的计数器功能，但是与CountDownLatch不同的是他的初始值在为0时唤醒所有线程以后会重新将初始值复原。

具体用法可以参考CountDownLatch 我就不再赘述了。

跟CountDownLatch一样可以用于分组多线程计算然后合并汇总结果，但是跟CountDownLatch不同的是它可以分阶段汇总，在第四部分会用代码进行示例说明。

三、源码介绍

CyclicBarrier提供了两个构造方法

//该构造方法只传入初始值，默认没有初始值为0的回调方法
public CyclicBarrier(int parties) {this(parties, null);}

//该构造方法只传入初始值和默认的初始值为0的回调方法传入的是一个Runnable
public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;
}

两个比较重要的方法

//默认没有等待时间，如果线程出现异常则会一直等待
public int await() {}


//传入等待时间和时间单位，超出等待时间不再等待
public int await(long timeout, TimeUnit unit){}

四、使用示例

计时器示例


    private static CyclicBarrier cyclicBarrier = new CyclicBarrier(3,() ->{
        //当计数器为0时立即执行
        System.out.println("汇总结果：" + Thread.currentThread().getName() + " 任务合并。");
    });

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(4);
        for (int i = 0; i < 3; i++) {
            final int m = i;
            executorService.submit(() ->{
                try {
                    System.out.println(Thread.currentThread().getName() + "执行任务。====线程"+m);
                    System.out.println("线程"+m+"：到达屏障点");
                    cyclicBarrier.await();
                    System.out.println("线程"+m+"：退出屏障点");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        // 关闭线程池
        executorService.shutdown();
    }

执行结果如下：

循环计时器示例

private static CyclicBarrier cyclicBarrier = new CyclicBarrier(3,() ->{
        //当计数器为0时立即执行
        System.out.println("汇总结果：" + Thread.currentThread().getName() + " 任务合并。");
    });
    
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(4);
        for (int i = 0; i < 3; i++) {
            final int m = i;
            executorService.submit(() ->{
                try {
                    System.out.println(Thread.currentThread().getName() + "执行任务。====线程"+m);
                    System.out.println("线程"+m+"：第一阶段到达屏障点");
                    cyclicBarrier.await();
                    System.out.println("线程"+m+"：第二阶段到达屏障点");
                    cyclicBarrier.await();
                    System.out.println("线程"+m+"：退出屏障点");
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        }
        // 关闭线程池
        executorService.shutdown();
}

执行结果如下