CyclicBarrier 线程同步器

雨点打透心脏的1/2处 2023-03-12 12:23 68阅读 0赞

CountDownLatch，Semaphore，CyclicBarrier 都是在 JUC 下包含了一些常用的同步工具类，其实将其放在一起介绍印象会更加深刻，由于对其了解使用的先后顺序，造成并没有一起来介绍。

**CountDownLatch：**  
当一个线程调用await方法时，就会阻塞当前线程。每当有线程调用一次 countDown 方法时，计数就会减 1。当 count 的值等于 0 的时候，被阻塞的线程才会继续运行。  
[CountDownLatch 的同步使用方法][CountDownLatch]

**Semaphore：**  
Semaphore 信号量，用来控制同一时间，资源可被访问的线程数量，一般可用于流量的控制。  
[Semaphore 控制线程并发量][Semaphore]

**CyclicBarrier:**  
CyclicBarrier 是多个线程互相等待,一直到指定线程数到达指定位置。  
1、CyclicBarrier 有两个构造函数，

/**
         * Creates a new {@code CyclicBarrier} that will trip when the
         * given number of parties (threads) are waiting upon it, and which
         * will execute the given barrier action when the barrier is tripped,
         * performed by the last thread entering the barrier.
         *
         * @param parties the number of threads that must invoke {@link #await}
         *        before the barrier is tripped
         * @param barrierAction the command to execute when the barrier is
         *        tripped, or {@code null} if there is no action
         * @throws IllegalArgumentException if {@code parties} is less than 1
         */
        public CyclicBarrier(int parties, Runnable barrierAction) {
            if (parties <= 0) throw new IllegalArgumentException();
            this.parties = parties;
            this.count = parties;
            this.barrierCommand = barrierAction;
        }
    
     /**
         * Creates a new {@code CyclicBarrier} that will trip when the
         * given number of parties (threads) are waiting upon it, and
         * does not perform a predefined action when the barrier is tripped.
         *
         * @param parties the number of threads that must invoke {@link #await}
         *        before the barrier is tripped
         * @throws IllegalArgumentException if {@code parties} is less than 1
         */
        public CyclicBarrier(int parties) {
            this(parties, null);
        }

根据源码上面注释，public CyclicBarrier(int parties) 当指定线程数到达等待的地方，所有线程往下走。 public CyclicBarrier(int parties, Runnable barrierAction) 这个方法当指定线程数到达等待的地方 ，先执行构造函数里面的Runnable 的run方法，所有线程再继续往下走。

1、public CyclicBarrier(int parties) 构造函数

public class CyclicBarrierTest {
    
        public static void main(String[] args) throws InterruptedException {
            CyclicBarrier barrier = new CyclicBarrier(3);
            SynchronizeRunner runner1 = new SynchronizeRunner("barrierq1", barrier);
            SynchronizeRunner runner2 = new SynchronizeRunner("barrierq2", barrier);
            SynchronizeRunner runner3 = new SynchronizeRunner("barrierq3", barrier);
            new Thread(runner1).start();
            new Thread(runner2).start();
            new Thread(runner3).start();
    
            Thread.currentThread().join(1000);
        }
    
    }
    
        class SynchronizeRunner implements Runnable{
    
            private String name;
            private CyclicBarrier cyclicBarrier;
    
            public SynchronizeRunner(String name, CyclicBarrier cyclicBarrier) {
                this.name = name;
                this.cyclicBarrier = cyclicBarrier;
            }
    
            @Override
            public void run() {
                try {
                    System.out.println(name+"：开始进入run方法");
                    cyclicBarrier.await();
                    System.out.println(name+"：跳过await方法");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        }

执行结果：

barrierq3：开始进入run方法
    barrierq1：开始进入run方法
    barrierq2：开始进入run方法
    barrierq2：跳过await方法
    barrierq1：跳过await方法
    barrierq3：跳过await方法

2、 public CyclicBarrier(int parties, Runnable barrierAction) 构造函数

public class CyclicBarrierTest {
    
    
        public static void main(String[] args) throws InterruptedException {
            CyclicBarrier barrier = new CyclicBarrier(3, new Runnable() {
                @Override
                public void run() {
                    try {
                        System.out.println("CyclicBarrier 构造器中的run 休眠之前");
                        TimeUnit.SECONDS.sleep(2);
                        System.out.println("CyclicBarrier 构造器中的run 休眠之后");
    
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            });
            SynchronizeRunner runner1 = new SynchronizeRunner("barrierq1", barrier);
            SynchronizeRunner2 runner2 = new SynchronizeRunner2("barrierq2", barrier);
            SynchronizeRunner runner3 = new SynchronizeRunner("barrierq3", barrier);
            new Thread(runner1).start();
            new Thread(runner2).start();
            new Thread(runner3).start();
    
            Thread.currentThread().join(1000);
        }
    }
    
        class SynchronizeRunner implements Runnable{
    
            private String name;
            private CyclicBarrier cyclicBarrier;
    
            public SynchronizeRunner(String name, CyclicBarrier cyclicBarrier) {
                this.name = name;
                this.cyclicBarrier = cyclicBarrier;
            }
    
            @Override
            public void run() {
                try {
                    System.out.println(name+"：开始进入run方法");
                    cyclicBarrier.await();
                    System.out.println(name+"：跳过await方法");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            }
        }
    
    class SynchronizeRunner2 implements Runnable {
    
        private String name;
        private CyclicBarrier cyclicBarrier;
    
        public SynchronizeRunner2(String name, CyclicBarrier cyclicBarrier) {
            this.name = name;
            this.cyclicBarrier = cyclicBarrier;
        }
    
        @Override
        public void run() {
            try {
                System.out.println(name + "：开始进入SynchronizeRunner2类中的run方法");
                cyclicBarrier.await();
                System.out.println(name + "：跳过开始进入SynchronizeRunner2类中的await方法");
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (BrokenBarrierException e) {
                e.printStackTrace();
            }
        }
    
    }

执行结果：

barrierq3：开始进入run方法
    barrierq1：开始进入run方法
    barrierq2：开始进入SynchronizeRunner2类中的run方法
    CyclicBarrier 构造器中的run 休眠之前
    CyclicBarrier 构造器中的run 休眠之后
    barrierq3：跳过await方法
    barrierq2：跳过开始进入SynchronizeRunner2类中的await方法
    barrierq1：跳过await方法

由2结果可知，await不一定必须是同一个类，只要指定线程到达对应地方就可以了。

如果看了可能会发现CyclicBarrier 和 CountDownLatch 非常相似，CyclicBarrier 是多个线程相互等待。CountDownLatch 是一个线程等待多个线程。

[CountDownLatch]: https://blog.csdn.net/fajing_feiyue/article/details/101878289
[Semaphore]: https://blog.csdn.net/fajing_feiyue/article/details/103932011