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

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

       

    原文作者:起床oO
    原文地址: https://www.cnblogs.com/fangpengchengbupter/p/9293600.html
    本文转自网络文章,转载此文章仅为分享知识,如有侵权,请联系博主进行删除。
点赞