JavaCountDownLatch的源碼硬核解析

第JavaCountDownLatch的源碼硬核解析目錄前言介紹和使用例子概述實(shí)現(xiàn)思路源碼解析類結(jié)構(gòu)圖await()實(shí)現(xiàn)原理countDown()實(shí)現(xiàn)原理

前言

對于并發(fā)執(zhí)行，Java中的CountDownLatch是一個(gè)重要的類，簡單理解，CountDownLatch中countdown是倒數(shù)的意思，latch則是門閂的含義。在數(shù)量倒數(shù)到0的時(shí)候，打開門閂,一起走，否則都等待在門閂的地方。

為了更好的理解CountDownLatch這個(gè)類，本文通過例子和源碼帶領(lǐng)大家深入解析這個(gè)類的原理。

介紹和使用

例子

我們先通過一個(gè)例子快速理解下CountDownLatch的妙處。

最近LOLS12賽如火如荼舉行，比如我們玩王者榮耀的時(shí)候，10個(gè)萬玩家登入游戲，每個(gè)玩家的網(wǎng)速可能不一樣，只有每個(gè)人進(jìn)度條走完，才會一起來到王者峽谷，網(wǎng)速快的要等網(wǎng)速慢的。我們通過例子模擬下這個(gè)過程。

@Slf4j(topic="a.CountDownLatchTest")

publicclassCountDownLatchTest{

publicstaticvoidmain(String[]args)throwsInterruptedException{

//創(chuàng)建一個(gè)倒時(shí)器，默認(rèn)10個(gè)數(shù)量

CountDownLatchlatch=newCountDownLatch(10);

ExecutorServiceservice=Executors.newFixedThreadPool(10);

//設(shè)置進(jìn)度數(shù)據(jù)

String[]personProcess=newString[10];

Randomrandom=newRandom();

for(inti=0;ii++){

intfinalJ=i;

service.submit(()-{

//模擬10個(gè)人的進(jìn)度條

for(intj=0;j=100;j++){

//模擬網(wǎng)速快慢，隨機(jī)生成

try{

Thread.sleep(random.nextInt(100));

}catch(InterruptedExceptione){

e.printStackTrace();

//設(shè)置進(jìn)度數(shù)據(jù)

personProcess[finalJ]=j+"%";

("{}",Arrays.toString(personProcess));

//運(yùn)行結(jié)束，倒時(shí)器-1

latch.countDown();

//打開"閥門"

latch.await();

("王者峽谷到了");

service.shutdown();

運(yùn)行結(jié)果：

概述

CountDownLatch一般用作多線程倒計(jì)時(shí)計(jì)數(shù)器，強(qiáng)制它們等待其他一組（CountDownLatch的初始化決定）任務(wù)執(zhí)行完成。

構(gòu)造器:

publicCountDownLatch(intcount)：設(shè)置倒數(shù)器需要倒數(shù)的數(shù)量

常用API:

publicvoidawait()throwsInterruptedException：調(diào)用await()方法的線程會被掛起，等待直到count值為0再繼續(xù)執(zhí)行。publicbooleanawait(longtimeout,TimeUnitunit)throwsInterruptedException：同await()，若等待timeout時(shí)長后，count值還是沒有變?yōu)?，不再等待，繼續(xù)執(zhí)行。時(shí)間單位如下常用的毫秒、天、小時(shí)、微秒、分鐘、納秒、秒。publicvoidcountDown()：count值遞減1publiclonggetCount()：獲取當(dāng)前count值

常見使用場景：

一個(gè)程序中有N個(gè)任務(wù)在執(zhí)行，我們可以創(chuàng)建值為N的CountDownLatch，當(dāng)每個(gè)任務(wù)完成后，調(diào)用一下countDown()方法進(jìn)行遞減count值，再在主線程中使用await()方法等待任務(wù)執(zhí)行完成，主線程繼續(xù)執(zhí)行。

實(shí)現(xiàn)思路

通過前面的例子和介紹我們知道CountDownLatch的大致使用流程：

創(chuàng)建CountDownLatch并設(shè)置計(jì)數(shù)器值。啟動多線程并且調(diào)用CountDownLatch實(shí)例的countDown()方法。主線程調(diào)用await()方法，這樣主線程的操作就會在這個(gè)方法上阻塞，直到其他線程完成各自的任務(wù)，count值為0，停止阻塞，主線程繼續(xù)執(zhí)行。

不妨我們先思考下，它是怎么實(shí)現(xiàn)的呢？我們可以問自己幾個(gè)問題？

如何做到可以讓主線程阻塞等待在那里？是不是可以調(diào)用LockSupport.park()方法進(jìn)行阻塞。那么什么時(shí)候該阻塞呢？我們需要有個(gè)變量，比如state,如果state大于0，就阻塞主線程。那么什么時(shí)候該喚醒呢，又如何喚醒呢？如果任務(wù)執(zhí)行完成后，我們讓state減去1，也就是調(diào)用countDown()方法，如果發(fā)現(xiàn)state是0，那么就調(diào)用LockSupport.unpark()喚醒此前阻塞的地方，繼續(xù)執(zhí)行。

是不是很熟悉，這就是我們的AQS共享模式的實(shí)現(xiàn)原理啊，不了解AQS共享模式的可以參考本篇文章：深入淺出理解Java并發(fā)AQS的共享鎖模式

我們把思路理清楚后，直接看CountDownLatch的源碼。

源碼解析

類結(jié)構(gòu)圖

以上是CountDownLatch的類結(jié)構(gòu)圖，

Sync是CountDownLatch的內(nèi)部類，被成員變量sync持有。Sync繼承了AbstractQueuedSynchronizer，也就是我們大名鼎鼎的AQS。

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

1.線程調(diào)用await()會阻塞等待其他線程完成任務(wù)

//CountDownLatch#await

publicvoidawait()throwsInterruptedException{

//調(diào)用AbstractQueuedSynchronizer的acquireSharedInterruptibly方法

sync.acquireSharedInterruptibly(1);

//AbstractQueuedSynchronizer#acquireSharedInterruptibly

publicfinalvoidacquireSharedInterruptibly(intarg)throwsInterruptedException{

//判斷線程是否被打斷，拋出打斷異常

if(Terrupted())

thrownewInterruptedException();

//嘗試獲取共享鎖

//條件成立說明state0，此時(shí)線程入隊(duì)阻塞等待，等待其他線程獲取共享資源

//條件不成立說明state=0，此時(shí)不需要阻塞線程，直接結(jié)束函數(shù)調(diào)用

if(tryAcquireShared(arg)0)

//阻塞當(dāng)前線程的邏輯

doAcquireSharedInterruptibly(arg);

//CountDownLatch.Sync#tryAcquireShared

protectedinttryAcquireShared(intacquires){

return(getState()==0)1:-1;

2.doAcquireSharedInterruptibly()方法是實(shí)現(xiàn)線程阻塞的核心邏輯

//AbstractQueuedSynchronizer#doAcquireSharedInterruptibly

privatevoiddoAcquireSharedInterruptibly(intarg)throwsInterruptedException{

//將調(diào)用latch.await()方法的線程包裝成SHARED類型的node加入到AQS的阻塞隊(duì)列中

finalNodenode=addWaiter(Node.SHARED);

booleanfailed=true;

try{

for(;;){

//獲取當(dāng)前節(jié)點(diǎn)的前驅(qū)節(jié)點(diǎn)

finalNodep=node.predecessor();

//前驅(qū)節(jié)點(diǎn)時(shí)頭節(jié)點(diǎn)就可以嘗試獲取鎖

if(p==head){

//再次嘗試獲取鎖，獲取成功返回1

intr=tryAcquireShared(arg);

if(r=0){

//獲取鎖成功，設(shè)置當(dāng)前節(jié)點(diǎn)為head節(jié)點(diǎn)，并且向后傳播

setHeadAndPropagate(node,r);

p.next=null;//helpGC

failed=false;

return;

//阻塞在這里

if(shouldParkAfterFailedAcquire(p,node)parkAndCheckInterrupt())

thrownewInterruptedException();

}finally{

//阻塞線程被中斷后拋出異常，進(jìn)入取消節(jié)點(diǎn)的邏輯

if(failed)

cancelAcquire(node);

3.parkAndCheckInterrupt()方法中會進(jìn)行阻塞操作

privatefinalbooleanparkAndCheckInterrupt(){

//阻塞線程

LockSupport.park(this);

returnTerrupted();

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

1.任務(wù)結(jié)束調(diào)用countDown()完成計(jì)數(shù)器減一（釋放鎖）的操作

publicvoidcountDown(){

sync.releaseShared(1);

publicfinalbooleanreleaseShared(intarg){

//嘗試釋放共享鎖

if(tryReleaseShared(arg)){

//釋放鎖成功開始喚醒阻塞節(jié)點(diǎn)

doReleaseShared();

returntrue;

returnfalse;

2.調(diào)用tryReleaseShared()方法嘗試釋放鎖，true表示state等于0，去喚醒阻塞線程。

protectedbooleantryReleaseShared(intreleases){

for(;;){

intc=getState();

//條件成立說明前面【已經(jīng)有線程觸發(fā)喚醒操作】了，這里返回false

if(c==0)

returnfalse;

//計(jì)數(shù)器減一

intnextc=c-1;

if(compareAndSetState(c,nextc))

//計(jì)數(shù)器為0時(shí)返回true

returnnextc==0;

3.調(diào)用doReleaseShared()喚醒阻塞的節(jié)點(diǎn)

privatevoiddoReleaseShared(){

for(;;){

Nodeh=head;

//判斷隊(duì)列是否是空隊(duì)列

if(h!=nullh!=tail){

intws=h.waitStatus;

//頭節(jié)點(diǎn)的狀態(tài)為signal，說明后繼節(jié)點(diǎn)沒有被喚醒過

if(ws==Node.SIGNAL){

//cas設(shè)置頭節(jié)點(diǎn)的狀態(tài)為0，設(shè)置失敗繼續(xù)自旋

if(!compareAndSetWaitStatus(h,Node.SIGNAL,0))

continue;

//喚醒后繼節(jié)點(diǎn)

unparkSuccessor(h);

//如果有其他線程已經(jīng)設(shè)置了頭節(jié)點(diǎn)的狀態(tài)，重新設(shè)置為PROPAGATE傳播屬性

elseif(w