Java并发包中CyclicBarrier的源码分析和使用

Posted 2020-12-03 起床oO

CyclicBarrier的介绍和源码分析

CyclicBarrier的字母意思是可循环（Cyclic）使用的屏障（Barrier）。它要做的事情是，让一组线程到达一个屏障（也可以叫做同步点）时被阻塞，直到最后一个线程到达屏障，屏障才会开门，所有被屏障拦截的线程才会继续干活。线程进入屏障通过CyclicBarrier的await()方法。

 

CyclicBarrier默认的构造方法是CyclicBarrier(int parties)。其参数表示屏障拦截的线程数量，每个线程调用await方法告诉CyclicBarrier我已经到达屏障，然后当前线程被阻塞。

 

CyclicBarrier还提供一个更高级的构造函数CyclicBarrier(int parties，Runnable barrier Action)，用于在线程到达屏障时，优先执行barrier Action这个Runnable对象，方便处理更复杂的业务场景。

 

public CyclicBarrier(int parties) {         this(parties, null);     }    public CyclicBarrier(int parties, Runnable barrierAction) {         if (parties <= 0) throw new IllegalArgumentException();         this.parties = parties;         this.count = parties;         this.barrierCommand = barrierAction;     }

 

实现原理，在CyclicBarrier的内部定义了一个Lock对象（ReentrantLock），每当一个线程调用CyclicBarrier的await()方法时，将剩余拦截的线程数减一，然后判断剩余拦截数是否为0，如果不是，进入Lock对象的条件队列等待。如果是，执行barrierAction对象的Runnable方法，然后将所的条件队列中的所有线程放入锁等待队列中，这些线程会依次获取锁，释放锁，接着先从await()方法返回，在从CyclicBarrier的await()方法返回。

 

/** The lock for guarding barrier entry */     private final ReentrantLock lock = new ReentrantLock();

 

await()源码：

 

public int await() throws InterruptedException, BrokenBarrierException {         try {             return dowait(false, 0L);         } catch (TimeoutException toe) {             throw new Error(toe); // cannot happen         }     }

dowait源码：

 

/**      * Main barrier code, covering the various policies.      */     private int dowait(boolean timed, long nanos)         throws InterruptedException, BrokenBarrierException,                TimeoutException {         final ReentrantLock lock = this.lock;         lock.lock();         try {             final Generation g = generation;             if (g.broken)                 throw new BrokenBarrierException();             if (Thread.interrupted()) {                 breakBarrier();                 throw new InterruptedException();             }             int index = --count;             if (index == 0) {  // tripped                 boolean ranAction = false;                 try {                     final Runnable command = barrierCommand;                     if (command != null)                         command.run();                     ranAction = true;                     nextGeneration();                     return 0;                 } finally {                     if (!ranAction)                         breakBarrier();                 }             }             // loop until tripped, broken, interrupted, or timed out             for (;;) {                 try {                     if (!timed)                         trip.await();                     else if (nanos > 0L)                         nanos = trip.awaitNanos(nanos);                 } catch (InterruptedException ie) {                     if (g == generation && ! g.broken) {                         breakBarrier();                         throw ie;                     } else {                         // We‘re about to finish waiting even if we had not                         // been interrupted, so this interrupt is deemed to                         // "belong" to subsequent execution.                         Thread.currentThread().interrupt();                     }                 }                 if (g.broken)                     throw new BrokenBarrierException();                 if (g != generation)                     return index;                 if (timed && nanos <= 0L) {                     breakBarrier();                     throw new TimeoutException();                 }             }         } finally {             lock.unlock();         }     }

 

 

 

private void nextGeneration() {         // signal completion of last generation         trip.signalAll();         // set up next generation         count = parties;         generation = new Generation();     }     /**      * Sets current barrier generation as broken and wakes up everyone.      * Called only while holding lock.      */     private void breakBarrier() {         generation.broken = true;         count = parties;         trip.signalAll();     }   nextGeneration和breakBarrier方法都可以停止阻塞。

CyclicBarrier主要用于一组线程之间的相互等待，而CountDownLatch一般用于一组线程等待另一组线程。实际上可以通过CountDownLatch的countDown()和await()来实现CyclicBarrier的功能。

即CountDownLatch中的countDown()和await() = CyclicBarrier中的await()。注意在一个线程中先调用countDown()再调用await()

 

CyclicBarrier对象可以重复使用，重用之前应当调用CyclicBarrier的reset方法：

 

public void reset() {         final ReentrantLock lock = this.lock;         lock.lock();         try {             breakBarrier();   // break the current generation             nextGeneration(); // start a new generation         } finally {             lock.unlock();         }     }

CyclicBarrier使用：

 

package com.fpc.Test; import java.util.concurrent.BrokenBarrierException; import java.util.concurrent.CyclicBarrier; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.Random; public class CyclicBarrierTest {      private CyclicBarrier cyclicBarrier = new CyclicBarrier(4);      private Random rnd = new Random();            class taskDemo implements Runnable{            private String taskId;                        public taskDemo( String taskId ) {                 this.taskId = taskId;            }                        @Override            public void run() {                 try {                      int time = rnd.nextInt(1000);                      Thread.sleep(time);                      System.out.println(" Thread : " + taskId + " sleep : " + time + "ms");                      try {                            cyclicBarrier.await();                            System.out.println(" Thread : " + taskId + " sleep is over");                      } catch (BrokenBarrierException e) {                            // TODO Auto-generated catch block                            e.printStackTrace();                      }                 } catch (InterruptedException e) {                      // TODO Auto-generated catch block                      e.printStackTrace();                 }            }      } //   CyclicBarrier cyclicBarrier = new CyclicBarrier();      public static void main( String[] args ) {            CyclicBarrierTest c = new CyclicBarrierTest();                        ExecutorService pool = Executors.newCachedThreadPool();            pool.submit(c.new taskDemo("1"));            pool.submit(c.new taskDemo("2"));            pool.submit(c.new taskDemo("3"));            pool.submit(c.new taskDemo("4"));      } }

运行结果：

 

Thread : 1 sleep : 102ms  Thread : 3 sleep : 254ms  Thread : 4 sleep : 394ms  Thread : 2 sleep : 943ms  Thread : 2 sleep is over  Thread : 1 sleep is over  Thread : 4 sleep is over  Thread : 3 sleep is over

 

如果构造CyclicBarrier时，给传的大小是5，但是你进入屏障的线程数只有4个，那么会发生什么现象？

这些4个线程都无法结束，因为CyclicBarrier还在等待第5个线程结束，但此时根本没有第5个线程，所以之前的4个线程根本无法结束。

Thread : 3 sleep : 166ms  Thread : 4 sleep : 281ms  Thread : 2 sleep : 444ms  Thread : 1 sleep : 776ms

那么如果构造CyclicBarrier时，给传递的参数是3呢，但此时有4个线程进入屏障呢？

结果是其中三个线程会执行然后结束，第四个线程永远无法结束，这是因为CyclicBarrier是可以循环利用的。

执行结果：

Thread : 3 sleep : 500ms  Thread : 1 sleep : 567ms  Thread : 2 sleep : 912ms  Thread : 2 sleep is over  Thread : 3 sleep is over  Thread : 1 sleep is over  Thread : 4 sleep : 988ms