一、導(dǎo)語

最近在學(xué)習(xí)并發(fā)編程原理，所以準(zhǔn)備整理一下自己學(xué)到的知識(shí)，先寫一篇CountDownLatch的源碼分析，之后希望可以慢慢寫完整個(gè)并發(fā)編程。

二、什么是CountDownLatch

CountDownLatch是java的JUC并發(fā)包里的一個(gè)工具類，可以理解為一個(gè)倒計(jì)時(shí)器，主要是用來控制多個(gè)線程之間的通信。
比如有一個(gè)主線程A，它要等待其他4個(gè)子線程執(zhí)行完畢之后才能執(zhí)行，此時(shí)就可以利用CountDownLatch來實(shí)現(xiàn)這種功能了。

三、簡單使用

public static void main(String[] args){System.out.println("主線程和他的兩個(gè)小兄弟約好去吃火鍋");System.out.println("主線程進(jìn)入了飯店");System.out.println("主線程想要開始動(dòng)筷子吃飯");//new一個(gè)計(jì)數(shù)器，初始值為2，當(dāng)計(jì)數(shù)器為0時(shí)，主線程開始執(zhí)行CountDownLatch latch = new CountDownLatch(2);new Thread(){public void run() {try {System.out.println("子線程1——小兄弟A 正在到飯店的路上");Thread.sleep(3000);System.out.println("子線程1——小兄弟A 到飯店了");//一個(gè)小兄弟到了，計(jì)數(shù)器-1latch.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 到飯店了");//另一個(gè)小兄弟到了，計(jì)數(shù)器-1latch.countDown();} catch (InterruptedException e) {e.printStackTrace();}};}.start();//主線程等待，直到其他兩個(gè)小兄弟也進(jìn)入飯店（計(jì)數(shù)器==0），主線程才能吃飯latch.await();System.out.println("主線程終于可以開始吃飯了～"); } 復(fù)制代碼

四、源碼分析

核心代碼：

CountDownLatch latch = new CountDownLatch(1);latch.await();latch.countDown(); 復(fù)制代碼

其中構(gòu)造函數(shù)的參數(shù)是計(jì)數(shù)器的值；
await()方法是用來阻塞線程，直到計(jì)數(shù)器的值為0
countDown()方法是執(zhí)行計(jì)數(shù)器-1操作

1、首先來看構(gòu)造函數(shù)的代碼

public CountDownLatch(int count) {if (count < 0) throw new IllegalArgumentException("count < 0");this.sync = new Sync(count);} 復(fù)制代碼

這段代碼很簡單，首先if判斷傳入的count是否<0，如果小于0直接拋異常。
然后new一個(gè)類Sync，這個(gè)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 zerofor (;;) {int c = getState();if (c == 0)return false;int nextc = c-1;if (compareAndSetState(c, nextc))return nextc == 0;}}} 復(fù)制代碼

可以看到Sync是一個(gè)內(nèi)部類，繼承了AQS，AQS是一個(gè)同步器，之后我們會(huì)詳細(xì)講。
其中有幾個(gè)核心點(diǎn)：

變量 state是父類AQS里面的變量，在這里的語義是計(jì)數(shù)器的值

getState()方法也是父類AQS里的方法，很簡單，就是獲取state的值

tryAcquireShared和tryReleaseShared也是父類AQS里面的方法，在這里CountDownLatch對(duì)他們進(jìn)行了重寫，先有個(gè)印象，之后詳講。

2、了解了CountDownLatch的構(gòu)造函數(shù)之后，我們?cè)賮砜此暮诵拇a，首先是await()。

public void await() throws InterruptedException {sync.acquireSharedInterruptibly(1);} 復(fù)制代碼

可以看到，其實(shí)是通過內(nèi)部類Sync調(diào)用了父類AQS的acquireSharedInterruptibly()方法。

public final void acquireSharedInterruptibly(int arg)throws InterruptedException {//判斷線程是否是中斷狀態(tài)if (Thread.interrupted())throw new InterruptedException();//嘗試獲取state的值if (tryAcquireShared(arg) < 0)//step1doAcquireSharedInterruptibly(arg);//step2} 復(fù)制代碼

tryAcquireShared(arg)這個(gè)方法就是我們剛才在Sync內(nèi)看到的重寫父類AQS的方法，意思就是判斷是否getState() == 0，如果state為0,返回1，則step1處不進(jìn)入if體內(nèi)acquireSharedInterruptibly(int arg)方法執(zhí)行完畢。若state!=0，則返回-1，進(jìn)入if體內(nèi)step2處。

下面我們來看acquireSharedInterruptibly(int arg)方法：

private void doAcquireSharedInterruptibly(int arg)throws InterruptedException {//step1、把當(dāng)前線程封裝為共享類型的Node，加入隊(duì)列尾部final Node node = addWaiter(Node.SHARED);boolean failed = true;try {for (;;) {//step2、獲取當(dāng)前node的前一個(gè)元素final Node p = node.predecessor();//step3、如果前一個(gè)元素是隊(duì)首if (p == head) {//step4、再次調(diào)用tryAcquireShared()方法，判斷state的值是否為0int r = tryAcquireShared(arg);//step5、如果state的值==0if (r >= 0) {//step6、設(shè)置當(dāng)前node為隊(duì)首，并嘗試釋放共享鎖setHeadAndPropagate(node, r);p.next = null; // help GCfailed = false;return;}}//step7、是否可以安心掛起當(dāng)前線程，是就掛起；并且判斷當(dāng)前線程是否中斷if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())throw new InterruptedException();}} finally {//step8、如果出現(xiàn)異常,failed沒有更新為false，則把當(dāng)前node從隊(duì)列中取消if (failed)cancelAcquire(node);}} 復(fù)制代碼

按照代碼中的注釋，我們可以大概了解該方法的內(nèi)容，下面我們來仔細(xì)看下其中調(diào)用的一些方法是干什么的。
1、首先看addWaiter()

//step1 private Node addWaiter(Node mode) {//把當(dāng)前線程封裝為nodeNode node = new Node(Thread.currentThread(), mode);// Try the fast path of enq; backup to full enq on failure//獲取當(dāng)前隊(duì)列的隊(duì)尾tail，并賦值給predNode pred = tail;//如果pred!=null，即當(dāng)前隊(duì)尾不為nullif (pred != null) {//把當(dāng)前隊(duì)尾tail，變成當(dāng)前node的前繼節(jié)點(diǎn)node.prev = pred;//cas更新當(dāng)前node為新的隊(duì)尾if (compareAndSetTail(pred, node)) {pred.next = node;return node;}}//如果隊(duì)尾為空，走enq方法enq(node);//step1.1return node;}----------------------------------------------------------------- //step1.1 private Node enq(final Node node) {for (;;) {Node t = tail;//如果隊(duì)尾tail為null，初始化隊(duì)列if (t == null) { // Must initialize//cas設(shè)置一個(gè)新的空node為隊(duì)首if (compareAndSetHead(new Node()))tail = head;} else {//cas把當(dāng)前node設(shè)置為新隊(duì)尾，把前隊(duì)尾設(shè)置成當(dāng)前node的前繼節(jié)點(diǎn)node.prev = t;if (compareAndSetTail(t, node)) {t.next = node;return t;}}}} 復(fù)制代碼

2、接下來我們?cè)趤砜磗etHeadAndPropagate()方法，看其內(nèi)部實(shí)現(xiàn)

//step6 private void setHeadAndPropagate(Node node, int propagate) {//獲取隊(duì)首headNode h = head; // Record old head for check below//設(shè)置當(dāng)前node為隊(duì)首，并取消node所關(guān)聯(lián)的線程setHead(node);//if (propagate > 0 || h == null || h.waitStatus < 0 ||(h = head) == null || h.waitStatus < 0) {Node s = node.next;//如果當(dāng)前node的后繼節(jié)點(diǎn)為null或者是shared類型的if (s == null || s.isShared())//釋放鎖，喚醒下一個(gè)線程doReleaseShared();//step6.1}} -------------------------------------------------------------------- //step6.1 private void doReleaseShared() {for (;;) {//找到頭節(jié)點(diǎn)Node h = head;if (h != null && h != tail) {//獲取頭節(jié)點(diǎn)狀態(tài)int ws = h.waitStatus;if (ws == Node.SIGNAL) {if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))continue; // loop to recheck cases//喚醒head節(jié)點(diǎn)的next節(jié)點(diǎn)unparkSuccessor(h);}else if (ws == 0 &&!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))continue; // loop on failed CAS}if (h == head) // loop if head changedbreak;}} 復(fù)制代碼

3、接下來我們來看countDown()方法。

public void countDown() {sync.releaseShared(1);} 復(fù)制代碼

可以看到調(diào)用的是父類AQS的releaseShared 方法

public final boolean releaseShared(int arg) {//state-1if (tryReleaseShared(arg)) {//step1//喚醒等待線程，內(nèi)部調(diào)用的是LockSupport.unpark方法doReleaseShared();//step2return true;}return false;} ------------------------------------------------------------------ //step1 protected boolean tryReleaseShared(int releases) {// Decrement count; signal when transition to zerofor (;;) {//獲取當(dāng)前state的值int c = getState();if (c == 0)return false;int nextc = c-1;//cas操作來進(jìn)行原子減1if (compareAndSetState(c, nextc))return nextc == 0;}} 復(fù)制代碼

五、總結(jié)

CountDownLatch主要是通過計(jì)數(shù)器state來控制是否可以執(zhí)行其他操作，如果不能就通過LockSupport.park()方法掛起線程，直到其他線程執(zhí)行完畢后喚醒它。 下面我們通過一個(gè)簡單的圖來幫助我們理解一下：

juc
PS：本人也是還在學(xué)習(xí)的路上，理解的也不是特別透徹，如有錯(cuò)誤，愿傾聽教誨。^_^

轉(zhuǎn)載于:https://juejin.im/post/5cd5339151882535786bc644

總結(jié)

以上是生活随笔為你收集整理的Java并发编程之CountDownLatch源码解析的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。