Java並發編程之CountDownLatch源碼解析
最近在學習並發編程原理，所以準備整理一下自己學到的知識，先寫一篇CountDownLatch的源碼分析，之後希望可以慢慢寫完整個並發編程。
二、什麽是CountDownLatch
CountDownLatch是java的JUC並發包裏的一個工具類，可以理解為一個倒計時器，主要是用來控制多個線程之間的通信。
比如有一個主線程A，它要等待其他4個子線程執行完畢之後才能執行，此時就可以利用CountDownLatch來實現這種功能了。
三、簡單使用
public static void main(String[] args){ System.out.println("主線程和他的兩個小兄弟約好去吃火鍋"); System.out.println("主線程進入了飯店"); System.out.println("主線程想要開始動筷子吃飯"); //new一個計數器，初始值為2，當計數器為0時，主線程開始執行 CountDownLatch latch = new CountDownLatch(2); new Thread(){ public void run() { try { System.out.println("子線程1——小兄弟A 正在到飯店的路上"); Thread.sleep(3000); System.out.println("子線程1——小兄弟A 到飯店了"); //一個小兄弟到了，計數器-1 latch.countDown(); } catch (InterruptedException e) { e.printStackTrace(); } }; }.start(); new Thread(){ public void run() { try { System.out.println("子線程2——小兄弟B 正在到飯店的路上"); Thread.sleep(3000); System.out.println("子線程2——小兄弟B 到飯店了"); //另一個小兄弟到了，計數器-1 latch.countDown(); } catch (InterruptedException e) { e.printStackTrace(); } }; }.start(); //主線程等待，直到其他兩個小兄弟也進入飯店（計數器==0），主線程才能吃飯 latch.await(); System.out.println("主線程終於可以開始吃飯了～"); }
四、源碼分析
核心代碼：
CountDownLatch latch = new CountDownLatch(1);
latch.await();
latch.countDown();
其中構造函數的參數是計數器的值；
await()方法是用來阻塞線程，直到計數器的值為0
countDown()方法是執行計數器-1操作
1、首先來看構造函數的代碼
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
這段代碼很簡單，首先if判斷傳入的count是否<0，如果小於0直接拋異常。
然後new一個類Sync，這個Sync是什麽呢？我們一起來看下
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
Sync(int count) { setState(count); } int getCount() { return getState(); } //嘗試獲取共享鎖 protected int tryAcquireShared(int acquires) { return (getState() == 0) ? 1 : -1; } //嘗試釋放共享鎖 protected boolean tryReleaseShared(int releases) { // Decrement count; signal when transition to zero for (;;) { int c = getState(); if (c == 0) return false; int nextc = c-1; if (compareAndSetState(c, nextc)) return nextc == 0; } } }
可以看到Sync是一個內部類，繼承了AQS，AQS是一個同步器，之後我們會詳細講。
其中有幾個核心點：
變量 state是父類AQS裏面的變量，在這裏的語義是計數器的值
getState()方法也是父類AQS裏的方法，很簡單，就是獲取state的值
tryAcquireShared和tryReleaseShared也是父類AQS裏面的方法，在這裏CountDownLatch對他們進行了重寫，先有個印象，之後詳講。
2、了解了CountDownLatch的構造函數之後，我們再來看它的核心代碼，首先是await()。
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
可以看到，其實是通過內部類Sync調用了父類AQS的acquireSharedInterruptibly()方法。
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//判斷線程是否是中斷狀態
if (Thread.interrupted())
throw new InterruptedException();
//嘗試獲取state的值
if (tryAcquireShared(arg) < 0)//step1
doAcquireSharedInterruptibly(arg);//step2
}
tryAcquireShared(arg)這個方法就是我們剛才在Sync內看到的重寫父類AQS的方法，意思就是判斷是否getState() == 0，如果state為0,返回1，則step1處不進入if體內acquireSharedInterruptibly(int arg)方法執行完畢。若state!=0，則返回-1，進入if體內step2處。
下面我們來看acquireSharedInterruptibly(int arg)方法：
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
//step1、把當前線程封裝為共享類型的Node，加入隊列尾部
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
//step2、獲取當前node的前一個元素
final Node p = node.predecessor();
//step3、如果前一個元素是隊首
if (p == head) {
//step4、再次調用tryAcquireShared()方法，判斷state的值是否為0
int r = tryAcquireShared(arg);
//step5、如果state的值==0
if (r >= 0) {
//step6、設置當前node為隊首，並嘗試釋放共享鎖
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
//step7、是否可以安心掛起當前線程，是就掛起；並且判斷當前線程是否中斷
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
//step8、如果出現異常,failed沒有更新為false，則把當前node從隊列中取消
if (failed)
cancelAcquire(node);
}
}按照代碼中的註釋，我們可以大概了解該方法的內容，下面我們來仔細看下其中調用的一些方法是幹什麽的。
1、首先看addWaiter()
//step1
private Node addWaiter(Node mode) {
//把當前線程封裝為node
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq; backup to full enq on failure
//獲取當前隊列的隊尾tail，並賦值給pred
Node pred = tail;
//如果pred!=null，即當前隊尾不為null
if (pred != null) {
//把當前隊尾tail，變成當前node的前繼節點
node.prev = pred;
//cas更新當前node為新的隊尾
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
//如果隊尾為空，走enq方法
enq(node);//step1.1
return node;
}//step1.1
private Node enq(final Node node) {
for (;;) {
Node t = tail;
//如果隊尾tail為null，初始化隊列
if (t == null) { // Must initialize
//cas設置一個新的空node為隊首
if (compareAndSetHead(new Node()))
tail = head;
} else {
//cas把當前node設置為新隊尾，把前隊尾設置成當前node的前繼節點
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}2、接下來我們在來看setHeadAndPropagate()方法，看其內部實現
//step6
private void setHeadAndPropagate(Node node, int propagate) {
//獲取隊首head
Node h = head; // Record old head for check below
//設置當前node為隊首，並取消node所關聯的線程
setHead(node);
//
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
//如果當前node的後繼節點為null或者是shared類型的
if (s == null || s.isShared())
//釋放鎖，喚醒下一個線程
doReleaseShared();//step6.1
}
}//step6.1
private void doReleaseShared() {
for (;;) {
//找到頭節點
Node h = head;
if (h != null && h != tail) {
//獲取頭節點狀態
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
//喚醒head節點的next節點
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}3、接下來我們來看countDown()方法。
public void countDown() {
sync.releaseShared(1);
}
可以看到調用的是父類AQS的releaseShared 方法
public final boolean releaseShared(int arg) {
//state-1
if (tryReleaseShared(arg)) {//step1
//喚醒等待線程，內部調用的是LockSupport.unpark方法
doReleaseShared();//step2
return true;
}
return false;
}//step1
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
for (;;) {
//獲取當前state的值
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
//cas操作來進行原子減1
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}五、總結
CountDownLatch主要是通過計數器state來控制是否可以執行其他操作，如果不能就通過LockSupport.park()方法掛起線程，直到其他線程執行完畢後喚醒它。 下面我們通過一個簡單的圖來幫助我們理解一下：
Java並發編程之CountDownLatch源碼解析