Java併發系列[7]----CountDownLatch源碼分析

CountDownLatch(閉鎖)是一個頗有用的工具類，利用它咱們能夠攔截一個或多個線程使其在某個條件成熟後再執行。它的內部提供了一個計數器，在構造閉鎖時必須指定計數器的初始值，且計數器的初始值必須大於0。另外它還提供了一個countDown方法來操做計數器的值，每調用一次countDown方法計數器都會減1，直到計數器的值減爲0時就表明條件已成熟，全部因調用await方法而阻塞的線程都會被喚醒。這就是CountDownLatch的內部機制，看起來很簡單，無非就是阻塞一部分線程讓其在達到某個條件以後再執行。可是CountDownLatch的應用場景卻比較普遍，只要你腦洞夠大利用它就能夠玩出各類花樣。最多見的一個應用場景是開啓多個線程同時執行某個任務，等到全部任務都執行完再統計彙總結果。下圖動態演示了閉鎖阻塞線程的整個過程。

上圖演示了有5個線程因調用await方法而被阻塞，它們須要等待計數器的值減爲0才能繼續執行。計數器的初始值在構造閉鎖時被指定，後面隨着每次countDown方法的調用而減1。下面代碼貼出了CountDownLatch的構造方法。

1 //構造器
2 public CountDownLatch(int count) {
3     if (count < 0) throw new IllegalArgumentException("count < 0");
4     this.sync = new Sync(count);
5 }

CountDownLatch只有一個帶參構造器，必須傳入一個大於0的值做爲計數器初始值，不然會報錯。能夠看到在構造方法中只是去new了一個Sync對象並賦值給成員變量sync。和其餘同步工具類同樣，CountDownLatch的實現依賴於AQS，它是AQS共享模式下的一個應用。CountDownLatch實現了一個內部類Sync並用它去繼承AQS，這樣就能使用AQS提供的大部分方法了。下面咱們就來看一下Sync內部類的代碼。

 1 //同步器
 2 private static final class Sync extends AbstractQueuedSynchronizer {
 3 
 4     //構造器
 5     Sync(int count) {
 6         setState(count);
 7     }
 8 
 9     //獲取當前同步狀態
10     int getCount() {
11         return getState();
12     }
13 
14     //嘗試獲取鎖
15     //返回負數：表示當前線程獲取失敗
16     //返回零值：表示當前線程獲取成功, 可是後繼線程不能再獲取了
17     //返回正數：表示當前線程獲取成功, 而且後繼線程一樣能夠獲取成功
18     protected int tryAcquireShared(int acquires) {
19         return (getState() == 0) ? 1 : -1;
20     }
21 
22     //嘗試釋放鎖
23     protected boolean tryReleaseShared(int releases) {
24         for (;;) {
25             //獲取同步狀態
26             int c = getState();
27             //若是同步狀態爲0, 則不能再釋放了
28             if (c == 0) {
29                 return false;
30             }
31             //不然的話就將同步狀態減1
32             int nextc = c-1;
33             //使用CAS方式更新同步狀態
34             if (compareAndSetState(c, nextc)) {
35                 return nextc == 0;
36             }
37         }
38     }
39 }

能夠看到Sync的構造方法會將同步狀態的值設置爲傳入的參數值。以後每次調用countDown方法都會將同步狀態的值減1，這也就是計數器的實現原理。在平時使用CountDownLatch工具類時最經常使用的兩個方法就是await方法和countDown方法。調用await方法會阻塞當前線程直到計數器爲0，調用countDown方法會將計數器的值減1直到減爲0。下面咱們來看一下await方法是怎樣調用的。

 1 //致使當前線程等待, 直到門閂減小到0, 或者線程被打斷
 2 public void await() throws InterruptedException {
 3     //以響應線程中斷方式獲取
 4     sync.acquireSharedInterruptibly(1);
 5 }
 6 
 7 //以可中斷模式獲取鎖(共享模式)
 8 public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
 9     //首先判斷線程是否中斷, 若是是則拋出異常
10     if (Thread.interrupted()) {
11         throw new InterruptedException();
12     }
13     //1.嘗試去獲取鎖
14     if (tryAcquireShared(arg) < 0) {
15         //2. 若是獲取失敗則進人該方法
16         doAcquireSharedInterruptibly(arg);
17     }
18 }

當線程調用await方法時實際上是調用到了AQS的acquireSharedInterruptibly方法，該方法是以響應線程中斷的方式來獲取鎖的，上面一樣貼出了該方法的代碼。咱們能夠看到在acquireSharedInterruptibly方法首先會去調用tryAcquireShared方法嘗試獲取鎖。咱們看到Sync裏面重寫的tryAcquireShared方法的邏輯，方法的實現邏輯很簡單，就是判斷當前同步狀態是否爲0，若是爲0則返回1代表能夠獲取鎖，不然返回-1表示不能獲取鎖。若是tryAcquireShared方法返回1則線程可以沒必要等待而繼續執行，若是返回-1那麼後續就會去調用doAcquireSharedInterruptibly方法讓線程進入到同步隊列裏面等待。這就是調用await方法會阻塞當前線程的原理，下面看看countDown方法是怎樣將阻塞的線程喚醒的。

 1 //減小門閂的方法
 2 public void countDown() {
 3     sync.releaseShared(1);
 4 }
 5 
 6 //釋放鎖的操做(共享模式)
 7 public final boolean releaseShared(int arg) {
 8     //1.嘗試去釋放鎖
 9     if (tryReleaseShared(arg)) {
10         //2.若是釋放成功就喚醒其餘線程
11         doReleaseShared();
12         return true;
13     }
14     return false;
15 }

能夠看到countDown方法裏面調用了releaseShared方法，該方法一樣是AQS裏面的方法，咱們在上面也貼出了它的代碼。releaseShared方法裏面首先是調用tryReleaseShared方法嘗試釋放鎖，tryReleaseShared方法在AQS裏面是一個抽象方法，它的具體實現邏輯在子類Sync類裏面，咱們在上面貼出的Sync類代碼裏能夠找到該方法。tryReleaseShared方法若是返回true表示釋放成功，返回false表示釋放失敗，只有當將同步狀態減1後該同步狀態剛好爲0時纔會返回true，其餘狀況都是返回false。那麼當tryReleaseShared返回true以後就會立刻調用doReleaseShared方法去喚醒同步隊列的全部線程。這樣就解釋了爲何最後一次調用countDown方法將計數器減爲0後就會喚醒全部被阻塞的線程。CountDownLatch基本的原理大體就是這些，下面咱們看一個它的使用示例。

應用場景：在玩歡樂鬥地主時必須等待三個玩家都到齊才能夠進行發牌。

 1 public class Player extends Thread {
 2     
 3     private static int count = 1;
 4     private final int id = count++;
 5     private CountDownLatch latch;
 6     
 7     public Player(CountDownLatch latch) {
 8         this.latch = latch;
 9     }
10 
11     @Override
12     public void run() {
13         System.out.println("【玩家" + id + "】已入場");
14         latch.countDown();
15     }
16     
17     public static void main(String[] args) throws InterruptedException {
18         CountDownLatch latch = new CountDownLatch(3);
19         System.out.println("牌局開始, 等待玩家入場...");
20         new Player(latch).start();
21         new Player(latch).start();
22         new Player(latch).start();
23         latch.await();
24         System.out.println("玩家已到齊, 開始發牌...");
25     }
26     
27 }

運行結果顯示發牌操做必定是在全部玩家都入場後才進行。咱們將23行的latch.await()註釋掉，對比下看看結果。

能夠看到在註釋掉latch.await()這行以後，就不能保證在全部玩家入場後纔開始發牌了。