Java并發(fā)系列之CountDownLatch源碼分析

更新時間：2018年03月26日 11:34:02 作者：勞夫子

這篇文章主要為大家詳細介紹了Java并發(fā)系列之CountDownLatch源碼，具有一定的參考價值，感興趣的小伙伴們可以參考一下

CountDownLatch(閉鎖)是一個很有用的工具類，利用它我們可以攔截一個或多個線程使其在某個條件成熟后再執(zhí)行。它的內(nèi)部提供了一個計數(shù)器，在構(gòu)造閉鎖時必須指定計數(shù)器的初始值，且計數(shù)器的初始值必須大于0。另外它還提供了一個countDown方法來操作計數(shù)器的值，每調(diào)用一次countDown方法計數(shù)器都會減1，直到計數(shù)器的值減為0時就代表條件已成熟，所有因調(diào)用await方法而阻塞的線程都會被喚醒。這就是CountDownLatch的內(nèi)部機制，看起來很簡單，無非就是阻塞一部分線程讓其在達到某個條件之后再執(zhí)行。但是CountDownLatch的應用場景卻比較廣泛，只要你腦洞夠大利用它就可以玩出各種花樣。最常見的一個應用場景是開啟多個線程同時執(zhí)行某個任務，等到所有任務都執(zhí)行完再統(tǒng)計匯總結(jié)果。下圖動態(tài)演示了閉鎖阻塞線程的整個過程。

上圖演示了有5個線程因調(diào)用await方法而被阻塞，它們需要等待計數(shù)器的值減為0才能繼續(xù)執(zhí)行。計數(shù)器的初始值在構(gòu)造閉鎖時被指定，后面隨著每次countDown方法的調(diào)用而減1。下面代碼貼出了CountDownLatch的構(gòu)造方法。

//構(gòu)造器
public CountDownLatch(int count) {
   if (count < 0) throw new IllegalArgumentException("count < 0");
   this.sync = new Sync(count);
 }

CountDownLatch只有一個帶參構(gòu)造器，必須傳入一個大于0的值作為計數(shù)器初始值，否則會報錯?？梢钥吹皆跇?gòu)造方法中只是去new了一個Sync對象并賦值給成員變量sync。和其他同步工具類一樣，CountDownLatch的實現(xiàn)依賴于AQS，它是AQS共享模式下的一個應用。CountDownLatch實現(xiàn)了一個內(nèi)部類Sync并用它去繼承AQS，這樣就能使用AQS提供的大部分方法了。下面我們就來看一下Sync內(nèi)部類的代碼。

//同步器
private static final class Sync extends AbstractQueuedSynchronizer {

  //構(gòu)造器
  Sync(int count) {
    setState(count);
  }

  //獲取當前同步狀態(tài)
  int getCount() {
    return getState();
  }

  //嘗試獲取鎖
  //返回負數(shù)：表示當前線程獲取失敗
  //返回零值：表示當前線程獲取成功, 但是后繼線程不能再獲取了
  //返回正數(shù)：表示當前線程獲取成功, 并且后繼線程同樣可以獲取成功
  protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
  }

  //嘗試釋放鎖
  protected boolean tryReleaseShared(int releases) {
    for (;;) {
      //獲取同步狀態(tài)
      int c = getState();
      //如果同步狀態(tài)為0, 則不能再釋放了
      if (c == 0) {
        return false;
      }
      //否則的話就將同步狀態(tài)減1
      int nextc = c-1;
      //使用CAS方式更新同步狀態(tài)
      if (compareAndSetState(c, nextc)) {
        return nextc == 0;
      }
    }
  }
}

可以看到Sync的構(gòu)造方法會將同步狀態(tài)的值設置為傳入的參數(shù)值。之后每次調(diào)用countDown方法都會將同步狀態(tài)的值減1，這也就是計數(shù)器的實現(xiàn)原理。在平時使用CountDownLatch工具類時最常用的兩個方法就是await方法和countDown方法。調(diào)用await方法會阻塞當前線程直到計數(shù)器為0，調(diào)用countDown方法會將計數(shù)器的值減1直到減為0。下面我們來看一下await方法是怎樣調(diào)用的。

//導致當前線程等待, 直到門閂減少到0, 或者線程被打斷
public void await() throws InterruptedException {
  //以響應線程中斷方式獲取
  sync.acquireSharedInterruptibly(1);
}

//以可中斷模式獲取鎖(共享模式)
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
  //首先判斷線程是否中斷, 如果是則拋出異常
  if (Thread.interrupted()) {
    throw new InterruptedException();
  }
  //1.嘗試去獲取鎖
  if (tryAcquireShared(arg) < 0) {
    //2. 如果獲取失敗則進人該方法
    doAcquireSharedInterruptibly(arg);
  }
}

當線程調(diào)用await方法時其實是調(diào)用到了AQS的acquireSharedInterruptibly方法，該方法是以響應線程中斷的方式來獲取鎖的，上面同樣貼出了該方法的代碼。我們可以看到在acquireSharedInterruptibly方法首先會去調(diào)用tryAcquireShared方法嘗試獲取鎖。我們看到Sync里面重寫的tryAcquireShared方法的邏輯，方法的實現(xiàn)邏輯很簡單，就是判斷當前同步狀態(tài)是否為0，如果為0則返回1表明可以獲取鎖，否則返回-1表示不能獲取鎖。如果tryAcquireShared方法返回1則線程能夠不必等待而繼續(xù)執(zhí)行，如果返回-1那么后續(xù)就會去調(diào)用doAcquireSharedInterruptibly方法讓線程進入到同步隊列里面等待。這就是調(diào)用await方法會阻塞當前線程的原理，下面看看countDown方法是怎樣將阻塞的線程喚醒的。

//減少門閂的方法
public void countDown() {
  sync.releaseShared(1);
}

//釋放鎖的操作(共享模式)
public final boolean releaseShared(int arg) {
  //1.嘗試去釋放鎖
  if (tryReleaseShared(arg)) {
    //2.如果釋放成功就喚醒其他線程
    doReleaseShared();
    return true;
  }
  return false;
}

可以看到countDown方法里面調(diào)用了releaseShared方法，該方法同樣是AQS里面的方法，我們在上面也貼出了它的代碼。releaseShared方法里面首先是調(diào)用tryReleaseShared方法嘗試釋放鎖，tryReleaseShared方法在AQS里面是一個抽象方法，它的具體實現(xiàn)邏輯在子類Sync類里面，我們在上面貼出的Sync類代碼里可以找到該方法。tryReleaseShared方法如果返回true表示釋放成功，返回false表示釋放失敗，只有當將同步狀態(tài)減1后該同步狀態(tài)恰好為0時才會返回true，其他情況都是返回false。那么當tryReleaseShared返回true之后就會馬上調(diào)用doReleaseShared方法去喚醒同步隊列的所有線程。這樣就解釋了為什么最后一次調(diào)用countDown方法將計數(shù)器減為0后就會喚醒所有被阻塞的線程。CountDownLatch基本的原理大致就是這些，下面我們看一個它的使用示例。

應用場景：在玩歡樂斗地主時必須等待三個玩家都到齊才可以進行發(fā)牌。

public class Player extends Thread {
  
  private static int count = 1;
  private final int id = count++;
  private CountDownLatch latch;
  
  public Player(CountDownLatch latch) {
    this.latch = latch;
  }

  @Override
  public void run() {
    System.out.println("【玩家" + id + "】已入場");
    latch.countDown();
  }
  
  public static void main(String[] args) throws InterruptedException {
    CountDownLatch latch = new CountDownLatch(3);
    System.out.println("牌局開始, 等待玩家入場...");
    new Player(latch).start();
    new Player(latch).start();
    new Player(latch).start();
    latch.await();
    System.out.println("玩家已到齊, 開始發(fā)牌...");
  }
  
}