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