CountDownLatch簡(jiǎn)介

CountDownLatch顧名思義，count + down + latch ＝ 計(jì)數(shù) ＋ 減 ＋ 門(mén)閂（這么拆分也是便于記憶=_=） 可以理解這個(gè)東西就是個(gè)計(jì)數(shù)器，只能減不能加，同時(shí)它還有個(gè)門(mén)閂的作用，當(dāng)計(jì)數(shù)器不為0時(shí)，門(mén)閂是鎖著的；當(dāng)計(jì)數(shù)器減到0時(shí)，門(mén)閂就打開(kāi)了。

如果你感到懵比的話，可以類(lèi)比考生考試交卷，考生交一份試卷，計(jì)數(shù)器就減一。直到考生都交了試卷（計(jì)數(shù)器為0），監(jiān)考老師（一個(gè)或多個(gè)）才能離開(kāi)考場(chǎng)。至于考生是否做完試卷，監(jiān)考老師并不關(guān)注。只要都交了試卷，他就可以做接下來(lái)的工作了。

CountDownLatch實(shí)現(xiàn)原理

下面從構(gòu)造方法開(kāi)始，一步步解釋實(shí)現(xiàn)的原理：構(gòu)造方法下面是實(shí)現(xiàn)的源碼，非常簡(jiǎn)短，主要是創(chuàng)建了一個(gè)Sync對(duì)象。

public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}

Sync對(duì)象

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;
}
}
}

假設(shè)我們是這樣創(chuàng)建的：new CountDownLatch(5)。其實(shí)也就相當(dāng)于new Sync(5)，相當(dāng)于setState(5)。setState其實(shí)就是共享鎖資源總數(shù),我們可以暫時(shí)理解為設(shè)置一個(gè)計(jì)數(shù)器，當(dāng)前計(jì)數(shù)器初始值為5。

tryAcquireShared方法其實(shí)就是判斷一下當(dāng)前計(jì)數(shù)器的值，是否為0了，如果為0的話返回1（返回1的時(shí)候，就表示獲取鎖成功,awit()方法就不再阻塞）。

tryReleaseShared方法就是利用CAS的方式，對(duì)計(jì)數(shù)器進(jìn)行減一的操作，而我們實(shí)際上每次調(diào)用countDownLatch.countDown()方法的時(shí)候，最終都會(huì)調(diào)到這個(gè)方法，對(duì)計(jì)數(shù)器進(jìn)行減一操作，一直減到0為止。

countDownLatch.await()

public void await() throws InterruptedException { 
sync.acquireSharedInterruptibly(1); 
}

代碼很簡(jiǎn)單，就一句話（注意acquireSharedInterruptibly（）方法是抽象類(lèi)：AbstractQueuedSynchronizer的一個(gè)方法，我們上面提到的Sync繼承了它），我們跟蹤源碼，繼續(xù)往下看：

acquireSharedInterruptibly(int arg) 
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}

源碼也是非常簡(jiǎn)單的，首先判斷了一下，當(dāng)前線程是否有被中斷，如果沒(méi)有的話，就調(diào)用tryAcquireShared(int acquires)方法，判斷一下當(dāng)前線程是否還需要“阻塞”。其實(shí)這里調(diào)用的tryAcquireShared方法，就是我們上面提到的java.util.concurrent.CountDownLatch.Sync.tryAcquireShared(int)這個(gè)方法。

當(dāng)然，在一開(kāi)始我們沒(méi)有調(diào)用過(guò)countDownLatch.countDown()方法時(shí)，這里tryAcquireShared方法肯定是會(huì)返回-1的，因?yàn)闀?huì)進(jìn)入到doAcquireSharedInterruptibly方法。 

doAcquireSharedInterruptibly(int arg)

countDown()方法

// 計(jì)數(shù)器減1
public void countDown() {
sync.releaseShared(1); 
}
//調(diào)用AQS的releaseShared方法
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {//計(jì)數(shù)器減一
doReleaseShared();//喚醒后繼結(jié)點(diǎn),這個(gè)時(shí)候隊(duì)列中可能只有調(diào)用過(guò)await()的線程節(jié)點(diǎn),也可能隊(duì)列為空
return true;
}
return false;
}

這個(gè)時(shí)候，我們應(yīng)該對(duì)于countDownLatch.await()方法是怎么“阻塞”當(dāng)前線程的，已經(jīng)非常明白了。其實(shí)說(shuō)白了，就是當(dāng)你調(diào)用了countDownLatch.await()方法后，你當(dāng)前線程就會(huì)進(jìn)入了一個(gè)死循環(huán)當(dāng)中，在這個(gè)死循環(huán)里面，會(huì)不斷的進(jìn)行判斷，通過(guò)調(diào)用tryAcquireShared方法，不斷判斷我們上面說(shuō)的那個(gè)計(jì)數(shù)器，看看它的值是否為0了（為0的時(shí)候，其實(shí)就是我們調(diào)用了足夠多 countDownLatch.countDown()方法的時(shí)候），如果是為0的話，tryAcquireShared就會(huì)返回1，代碼也會(huì)進(jìn)入到圖中的紅框部分，然后跳出了循環(huán)，也就不再“阻塞”當(dāng)前線程了。

需要注意的是，說(shuō)是在不停的循環(huán)，其實(shí)也并非在不停的執(zhí)行for循環(huán)里面的內(nèi)容，因?yàn)樵诤竺嬲{(diào)用parkAndCheckInterrupt（）方法時(shí)，在這個(gè)方法里面是會(huì)調(diào)用 LockSupport.park(this);來(lái)掛起當(dāng)前線程。

CountDownLatch 使用的注意點(diǎn)：

1、只有當(dāng)count為0時(shí)，await之后的程序才夠執(zhí)行。

2、countDown必須寫(xiě)在finally中，防止發(fā)生異程常時(shí)，導(dǎo)致程序死鎖。

使用場(chǎng)景：

比如對(duì)于馬拉松比賽，進(jìn)行排名計(jì)算，參賽者的排名，肯定是跑完比賽之后，進(jìn)行計(jì)算得出的，翻譯成Java識(shí)別的預(yù)發(fā)，就是N個(gè)線程執(zhí)行操作，主線程等到N個(gè)子線程執(zhí)行完畢之后，在繼續(xù)往下執(zhí)行。

public static void testCountDownLatch(){
int threadCount = 10;
final CountDownLatch latch = new CountDownLatch(threadCount);
for(int i=0; i< threadCount; i++){
new Thread(new Runnable() {
@Override
public void run() {
System.out.println("線程" + Thread.currentThread().getId() + "開(kāi)始出發(fā)");
try {
Thread.sleep(1000);
System.out.println("線程" + Thread.currentThread().getId() + "已到達(dá)終點(diǎn)");
} catch (InterruptedException e) {
e.printStackTrace();
} fianlly {
latch.countDown();
}
}
}).start();
}
try {
latch.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("10個(gè)線程已經(jīng)執(zhí)行完畢！開(kāi)始計(jì)算排名");
}

結(jié)果:

線程10開(kāi)始出發(fā)
線程13開(kāi)始出發(fā)
線程12開(kāi)始出發(fā)
線程11開(kāi)始出發(fā)
線程14開(kāi)始出發(fā)
線程15開(kāi)始出發(fā)
線程16開(kāi)始出發(fā)
線程17開(kāi)始出發(fā)
線程18開(kāi)始出發(fā)
線程19開(kāi)始出發(fā)
線程14已到達(dá)終點(diǎn)
線程15已到達(dá)終點(diǎn)
線程13已到達(dá)終點(diǎn)
線程12已到達(dá)終點(diǎn)
線程10已到達(dá)終點(diǎn)
線程11已到達(dá)終點(diǎn)
線程16已到達(dá)終點(diǎn)
線程17已到達(dá)終點(diǎn)
線程18已到達(dá)終點(diǎn)
線程19已到達(dá)終點(diǎn)
10個(gè)線程已經(jīng)執(zhí)行完畢！開(kāi)始計(jì)算排名

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。

最新評(píng)論