JavaCountDownLatch的源码硬核解析

第JavaCountDownLatch的源码硬核解析目录前言介绍和使用例子概述实现思路源码解析类结构图await()实现原理countDown()实现原理

前言

对于并发执行，Java中的CountDownLatch是一个重要的类，简单理解，CountDownLatch中countdown是倒数的意思，latch则是门闩的含义。在数量倒数到0的时候，打开门闩,一起走，否则都等待在门闩的地方。

为了更好的理解CountDownLatch这个类，本文通过例子和源码带领大家深入解析这个类的原理。

介绍和使用

例子

我们先通过一个例子快速理解下CountDownLatch的妙处。

最近LOLS12赛如火如荼举行，比如我们玩王者荣耀的时候，10个万玩家登入游戏，每个玩家的网速可能不一样，只有每个人进度条走完，才会一起来到王者峡谷，网速快的要等网速慢的。我们通过例子模拟下这个过程。

@Slf4j(topic="a.CountDownLatchTest")

publicclassCountDownLatchTest{

publicstaticvoidmain(String[]args)throwsInterruptedException{

//创建一个倒时器，默认10个数量

CountDownLatchlatch=newCountDownLatch(10);

ExecutorServiceservice=Executors.newFixedThreadPool(10);

//设置进度数据

String[]personProcess=newString[10];

Randomrandom=newRandom();

for(inti=0;ii++){

intfinalJ=i;

service.submit(()-{

//模拟10个人的进度条

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

//模拟网速快慢，随机生成

try{

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

}catch(InterruptedExceptione){

e.printStackTrace();

//设置进度数据

personProcess[finalJ]=j+"%";

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

//运行结束，倒时器-1

latch.countDown();

//打开"阀门"

latch.await();

("王者峡谷到了");

service.shutdown();

运行结果：

概述

CountDownLatch一般用作多线程倒计时计数器，强制它们等待其他一组（CountDownLatch的初始化决定）任务执行完成。

构造器:

publicCountDownLatch(intcount)：设置倒数器需要倒数的数量

常用API:

publicvoidawait()throwsInterruptedException：调用await()方法的线程会被挂起，等待直到count值为0再继续执行。publicbooleanawait(longtimeout,TimeUnitunit)throwsInterruptedException：同await()，若等待timeout时长后，count值还是没有变为0，不再等待，继续执行。时间单位如下常用的毫秒、天、小时、微秒、分钟、纳秒、秒。publicvoidcountDown()：count值递减1publiclonggetCount()：获取当前count值

常见使用场景：

一个程序中有N个任务在执行，我们可以创建值为N的CountDownLatch，当每个任务完成后，调用一下countDown()方法进行递减count值，再在主线程中使用await()方法等待任务执行完成，主线程继续执行。

实现思路

通过前面的例子和介绍我们知道CountDownLatch的大致使用流程：

创建CountDownLatch并设置计数器值。启动多线程并且调用CountDownLatch实例的countDown()方法。主线程调用await()方法，这样主线程的操作就会在这个方法上阻塞，直到其他线程完成各自的任务，count值为0，停止阻塞，主线程继续执行。

不妨我们先思考下，它是怎么实现的呢？我们可以问自己几个问题？

如何做到可以让主线程阻塞等待在那里？是不是可以调用LockSupport.park()方法进行阻塞。那么什么时候该阻塞呢？我们需要有个变量，比如state,如果state大于0，就阻塞主线程。那么什么时候该唤醒呢，又如何唤醒呢？如果任务执行完成后，我们让state减去1，也就是调用countDown()方法，如果发现state是0，那么就调用LockSupport.unpark()唤醒此前阻塞的地方，继续执行。

是不是很熟悉，这就是我们的AQS共享模式的实现原理啊，不了解AQS共享模式的可以参考本篇文章：深入浅出理解Java并发AQS的共享锁模式

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

源码解析

类结构图

以上是CountDownLatch的类结构图，

Sync是CountDownLatch的内部类，被成员变量sync持有。Sync继承了AbstractQueuedSynchronizer，也就是我们大名鼎鼎的AQS。

await()实现原理

1.线程调用await()会阻塞等待其他线程完成任务

//CountDownLatch#await

publicvoidawait()throwsInterruptedException{

//调用AbstractQueuedSynchronizer的acquireSharedInterruptibly方法

sync.acquireSharedInterruptibly(1);

//AbstractQueuedSynchronizer#acquireSharedInterruptibly

publicfinalvoidacquireSharedInterruptibly(intarg)throwsInterruptedException{

//判断线程是否被打断，抛出打断异常

if(Terrupted())

thrownewInterruptedException();

//尝试获取共享锁

//条件成立说明state0，此时线程入队阻塞等待，等待其他线程获取共享资源

//条件不成立说明state=0，此时不需要阻塞线程，直接结束函数调用

if(tryAcquireShared(arg)0)

//阻塞当前线程的逻辑

doAcquireSharedInterruptibly(arg);

//CountDownLatch.Sync#tryAcquireShared

protectedinttryAcquireShared(intacquires){

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

2.doAcquireSharedInterruptibly()方法是实现线程阻塞的核心逻辑

//AbstractQueuedSynchronizer#doAcquireSharedInterruptibly

privatevoiddoAcquireSharedInterruptibly(intarg)throwsInterruptedException{

//将调用latch.await()方法的线程包装成SHARED类型的node加入到AQS的阻塞队列中

finalNodenode=addWaiter(Node.SHARED);

booleanfailed=true;

try{

for(;;){

//获取当前节点的前驱节点

finalNodep=node.predecessor();

//前驱节点时头节点就可以尝试获取锁

if(p==head){

//再次尝试获取锁，获取成功返回1

intr=tryAcquireShared(arg);

if(r=0){

//获取锁成功，设置当前节点为head节点，并且向后传播

setHeadAndPropagate(node,r);

p.next=null;//helpGC

failed=false;

return;

//阻塞在这里

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

thrownewInterruptedException();

}finally{

//阻塞线程被中断后抛出异常，进入取消节点的逻辑

if(failed)

cancelAcquire(node);

3.parkAndCheckInterrupt()方法中会进行阻塞操作

privatefinalbooleanparkAndCheckInterrupt(){

//阻塞线程

LockSupport.park(this);

returnTerrupted();

countDown()实现原理

1.任务结束调用countDown()完成计数器减一（释放锁）的操作

publicvoidcountDown(){

sync.releaseShared(1);

publicfinalbooleanreleaseShared(intarg){

//尝试释放共享锁

if(tryReleaseShared(arg)){

//释放锁成功开始唤醒阻塞节点

doReleaseShared();

returntrue;

returnfalse;

2.调用tryReleaseShared()方法尝试释放锁，true表示state等于0，去唤醒阻塞线程。

protectedbooleantryReleaseShared(intreleases){

for(;;){

intc=getState();

//条件成立说明前面【已经有线程触发唤醒操作】了，这里返回false

if(c==0)

returnfalse;

//计数器减一

intnextc=c-1;

if(compareAndSetState(c,nextc))

//计数器为0时返回true

returnnextc==0;

3.调用doReleaseShared()唤醒阻塞的节点

privatevoiddoReleaseShared(){

for(;;){

Nodeh=head;

//判断队列是否是空队列

if(h!=nullh!=tail){

intws=h.waitStatus;

//头节点的状态为signal，说明后继节点没有被唤醒过

if(ws==Node.SIGNAL){

//cas设置头节点的状态为0，设置失败继续自旋

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

continue;

//唤醒后继节点

unparkSuccessor(h);

//如果有其他线程已经设置了头节点的状态，重新设置为PROPAGATE传播属性

elseif(w