【并发编程】CyclicBarrier 源码分析

本是古典何须时尚 2022-04-23 11:10 229阅读 0赞

# 前言 #

Github：[https://github.com/yihonglei/jdk-source-code-reading][https_github.com_yihonglei_jdk-source-code-reading]（java-concurrent）

# 一 原理 #

CyclicBarrier（循环屏障），也即可以循环利用的屏障。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70][]

CyclicBarrier内部维护了整型变量count（拦截线程计数器），每个线程调用await方法时，

count就会减1，如果count不为0，调用await的线程等待执行，如果count为0，所有调用await方法

的线程被唤醒执行。

CyclicBarrier底层基于ReentrantLock和Condition实现，如果count不为0，

则调用Condition的await方法让线程等待执行，当count为0时，调用Condition的singleAll唤醒

全部等待的线程执行。

关于ReentrantLock和Condition：[https://blog.csdn.net/yhl\_jxy/article/details/87088314][https_blog.csdn.net_yhl_jxy_article_details_87088314]。

CyclicBarrier原理可以简单理解为等待/通知模型，count不为0，调用await的线程等待，

直到count为0，才唤醒等待的线程执行。

# 二 实战 #

**实例场景：**

操场5条跑道，起跑线相当于屏障，5个运动员先后就绪后，鸣枪，开跑！

**AthleteThread**

package com.jpeony.concurrent.cyclicbarrier;
    
    import java.util.concurrent.*;
    
    /**
     * 运动员跑步全部就绪-->鸣枪-->开跑！
     *
     * @author yihonglei
     */
    public class AthleteThread extends Thread {
        /**
         * 操场上有5条跑道，当运动员全部就绪-->鸣枪-->开跑！
         * 这里new Runnable作为barrierAction，也就是屏障打破后触发的动作，比如运行员就绪后，鸣枪！
         */
        private static final CyclicBarrier playGround = new CyclicBarrier(5, new Runnable() {
            @Override
            public void run() {
                System.out.println("鸣枪!!!!!!!!!!!!!!!!");
            }
        });
    
        // 运动员编号
        private Integer num;
    
        /**
         * 构造器
         *
         * @param num 运动员编号
         */
        public AthleteThread(Integer num) {
            this.num = num;
        }
    
        @Override
        public void run() {
            try {
                System.out.println("运动员编号：" + num + "就绪...................");
                /*
                 * playGround 每次调用await时，count减1，直到count为0时，打破屏障，所有调用await的线程同时开始执行。
                 */
                playGround.await();
                System.out.println("运动员编号：" + num + "开跑>>>>>>>>>>>>>>>>>>>");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    
        public static void main(String[] args) {
            ExecutorService executorService = Executors.newFixedThreadPool(5);
    
            // 5个运动员参加运动会
            for (int i = 1; i <= 5; i++) {
                executorService.execute(new AthleteThread(i));
            }
    
            executorService.shutdown();
        }
    }

**运行结果**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70 1][]

**程序分析**

1）AthleteThread中通过CyclicBarrier构造器创建拦截数为5，以及打破屏障时执行什么线程的CyclicBarrier；

2）每一个线程在开跑前都要等待其它线程就绪，所有运行员就绪后，才打破屏障，唤醒所有调用await的线程，

一起开跑！

下面通过源码分析CyclicBarrier的实现原理，看源码可能会理解更深刻些！

# 三 源码分析 #

## 1、构造器 ##

![70][]

1）传入线程拦截数，调用2，创建CyclicBarrier；

2）或基于parties，barrierAction创建CyclicBarrier；

![70 1][]

1）parties为CyclicBarrier拦截线程数量，变量被定义为final，表示不可以改变，主要用于CyclicBarrier重置时

创建CyclicBarrier对象；

2）barrierCommand为打破屏障时执行的线程，这个线程可选，并不是必须定义的；

3）count为线程拦截计数器，每一次await都会减少1，初始化时count数量等于parties大小；

## 2、await逻辑 ##

await中包含线程等待、唤醒以及重置CyclicBarrier逻辑，await有能够指定超时等的重载方法，

这里只分析默认方法。

![70 2][]

具体实现在dowait中，dowait方法内容比较多，源码如下：

![70 3][]

![70 4][]

在进行源码分析前，需要先说明CyclicBarrier中的锁和Condition变量定义：

![70 5][]

ReentrantLock为重入锁，Condition用于实现一组线程的等待/唤醒机制，也就CyclicBarrier能实现等待/唤醒的底层实现。

1）每一个线程进入dowait的时候，需要上锁，避免并发；

2）这里有个Generation，下一代的意思，源码：

private static class Generation {
        boolean broken = false;
    }

Generation维护了一个broken变量，表示是否打破屏障。

如果屏障已经打破或线程已经停止则都会抛出异常。

3）count减少1，如果index为0，表示已经达到拦截线程数已经用完。

4）如果barrierCommand不为空，则执行run方法。

5）重点在nextGeneration()里面，源码：

private void nextGeneration() {
        // signal completion of last generation
        trip.signalAll();
        // set up next generation
        count = parties;
        generation = new Generation();
    }

打破屏障的时候，会调trip.signalAll()唤醒所有等待的线程，count重置为parties，generation也重置。

相当于把CyclicBarrier的count重置，也就CyclicBarrier重置了，因为parties，barrierCommand都没有变。

6）这个一般情况下不会触发。

7）如果index不为0，程序会执行到for循环，默认timed为false，调用trip.await()线程会进入等待状态。

8）这个是显示指定超时时间，线程等待超时时间自动唤醒执行，在使用CyclicBarrier的非默认await

方法，指定超时时间时使用，看一眼CyclicBarrier的await重载方法就可以理解。

9）如果已经打破屏障，直接抛异常。

10）如果generation没有被重置，说明还没有打破，返回剩余线程拦截数。

11）如果调用的是超时的await，传入参数时间错误，屏障恢复，同时会抛超时异常，

并且唤醒其他等待线程。

# 四 总结 #

1）CyclicBarrier底层基于ReentrantLock和Condition实现，如果count不为0，则调用Condition的await方法

让线程等待执行，当count为0时，调用Condition的singleAll唤醒全部等待的线程执行。

2）使用场景就是等待/唤醒场景。

[https_github.com_yihonglei_jdk-source-code-reading]: https://github.com/yihonglei/jdk-source-code-reading
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70]: /images/20220320/6a7488baaf174355a7821ce60a0e84cb.png
[https_blog.csdn.net_yhl_jxy_article_details_87088314]: https://blog.csdn.net/yhl_jxy/article/details/87088314
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70 1]: /images/20220320/51a3496bc129462cbf6bac005fe5fa07.png
[70]: /images/20220320/f5f8a5d9efa547db9c043a5056c13f4c.png
[70 1]: /images/20220320/5a4db82a476c4ecb8eac26b7e768b050.png
[70 2]: /images/20220320/21d7166740f049bfbd17538732209247.png
[70 3]: /images/20220320/543bf1fa31134667b71d7ba4848ebc77.png
[70 4]: /images/20220320/c1fc8f13f1d34e399665fb249fae8389.png
[70 5]: /images/20220320/ba61504cf841470da1a0d3041830bf0d.png