CountDownLatch源码解析之await()
CountDownLatch源码解析——await()，具体内容如下
上一篇文章说了一下CountDownLatch的使用方法。这篇文章就从源码层面说一下await()的原理。
我们已经知道await能够让当前线程处于阻塞状态，直到锁存器计数为零（或者线程中断）。
下面是它的源码。
end.await();
↓
publicvoidawait()throwsInterruptedException{
sync.acquireSharedInterruptibly(1);
}
sync是CountDownLatch的内部类。下面是它的定义。
privatestaticfinalclassSyncextendsAbstractQueuedSynchronizer{
...
}
它继承了AbstractQueuedSynchronizer。AbstractQueuedSynchronizer这个类在java线程中属于一个非常重要的类。
它提供了一个框架来实现阻塞锁，以及依赖FIFO等待队列的相关同步器（比如信号、事件等）。
继续走下去，就跳到AbstractQueuedSynchronizer这个类中。
sync.acquireSharedInterruptibly(1);
↓
publicfinalvoidacquireSharedInterruptibly(intarg)//AbstractQueuedSynchronizer
throwsInterruptedException{
if(Thread.interrupted())
thrownewInterruptedException();
if(tryAcquireShared(arg)<0)
doAcquireSharedInterruptibly(arg);
}
这里有两个判断，首先判断线程是否中断，然后再进行下一个判断，这里我们主要看看第二个判断。
protectedinttryAcquireShared(intacquires){
return(getState()==0)?1:-1;
}
需要注意的是tryAcquireShared这个方法是在Sync中实现的。
AbstractQueuedSynchronizer中虽然也有对它的实现，但是默认的实现是抛一个异常。
tryAcquireShared这个方法是用来查询当前对象的状态是否能够被允许获取锁。
我们可以看到Sync中是通过判断state是否为0来返回对应的int值的。
那么state又代表什么？
/** *Thesynchronizationstate. */ privatevolatileintstate;
上面代码很清楚的表明state是表示同步的状态。
需要注意的是state使用volatile关键字修饰。
volatile关键字能够保证state的修改立即被更新到主存，当有其他线程需要读取时，会去内存中读取新值。
也就是保证了state的可见性。是最新的数据。
走到这里state是多少呢？
这里我们就需要看一看CountDownLatch的构造函数了。
CountDownLatchend=newCountDownLatch(2);
↓
publicCountDownLatch(intcount){
if(count<0)thrownewIllegalArgumentException("count<0");
this.sync=newSync(count);
}
↓
Sync(intcount){
setState(count);
}
原来构造函数中的数字就是这个作用啊，用来setstate。
所以我们这里state==2了。tryAcquireShared就返回-1。进入到下面
doAcquireSharedInterruptibly(arg);
↓
privatevoiddoAcquireSharedInterruptibly(intarg)
throwsInterruptedException{
finalNodenode=addWaiter(Node.SHARED);
booleanfailed=true;
try{
for(;;){
finalNodep=node.predecessor();
if(p==head){
intr=tryAcquireShared(arg);
if(r>=0){
setHeadAndPropagate(node,r);
p.next=null;//helpGC
failed=false;
return;
}
}
if(shouldParkAfterFailedAcquire(p,node)&&
parkAndCheckInterrupt())
thrownewInterruptedException();
}
}finally{
if(failed)
cancelAcquire(node);
}
}
OK，这段代码有点长，里面还调用了几个函数。我们一行一行的看。
第一行出现了一个新的类Node。
Node是AQS（AbstractQueuedSynchronizer）类中的内部类，定义了一种链式结构。如下所示。
+------+prev+-----++-----+ head||<----||<----||tail +------++-----++-----+
千万记住这个结构。
第一行代码中还有一个方法addWaiter(Node.SHARED)。
addWaiter(Node.SHARED)//Node.SHARED表示该结点处于共享模式
↓
privateNodeaddWaiter(Nodemode){
Nodenode=newNode(Thread.currentThread(),mode);
//Trythefastpathofenq;backuptofullenqonfailure
Nodepred=tail;//privatetransientvolatileNodetail;
if(pred!=null){
node.prev=pred;
if(compareAndSetTail(pred,node)){
pred.next=node;
returnnode;
}
}
enq(node);
returnnode;
}
首先是构造了一个Node，将当前的线程存进去了，模式是共享模式。
tail表示这个等待队列的队尾，此刻是null.所以pred==null，进入到enq(node);
enq(node)
↓
privateNodeenq(finalNodenode){
for(;;){
Nodet=tail;
if(t==null){//Mustinitialize
if(compareAndSetHead(newNode()))
tail=head;
}else{
node.prev=t;
if(compareAndSetTail(t,node)){
t.next=node;
returnt;
}
}
}
}
同样tail为null,进入到compareAndSetHead。
compareAndSetHead(newNode())
↓
/**
*CASheadfield.Usedonlybyenq.
*/
privatefinalbooleancompareAndSetHead(Nodeupdate){
returnunsafe.compareAndSwapObject(this,headOffset,null,update);
}
这是一个CAS操作，如果head是null的话，等待队列的head就会被设置为update的值，也就是一个新的结点。
tail=head; 那么此时tail也不再是null了。进入下一次的循环。
这次首先将node的prev指针指向tail，然后通过一个CAS操作将node设置为尾部，并返回了队列的tail，也就是node。
等待队列的模型变化如下
+------+prev+----------------+ head(tail)||<----node|currentThread| +------++----------------+ ↓ +------+prev+----------------+ head||<----node(tail)|currentThread| +------++----------------+
ok，到了这里await方法就返回了，是一个thread等于当前线程的Node。
返回到doAcquireSharedInterruptibly(intarg)中，进入下面循环。
for(;;){
finalNodep=node.predecessor();
if(p==head){
intr=tryAcquireShared(arg);
if(r>=0){
setHeadAndPropagate(node,r);
p.next=null;//helpGC
failed=false;
return;
}
}
if(shouldParkAfterFailedAcquire(p,node)&&
parkAndCheckInterrupt())
thrownewInterruptedException();
}
这个时候假设state仍然大于0，那么此时r<0，所以进入到shouldParkAfterFailedAcquire这个方法。
shouldParkAfterFailedAcquire(p,node)
↓
privatestaticbooleanshouldParkAfterFailedAcquire(Nodepred,Nodenode){
intws=pred.waitStatus;
if(ws==Node.SIGNAL)//staticfinalintSIGNAL=-1;
/*
*Thisnodehasalreadysetstatusaskingarelease
*tosignalit,soitcansafelypark.
*/
returntrue;
if(ws>0){
/*
*Predecessorwascancelled.Skipoverpredecessorsand
*indicateretry.
*/
do{
node.prev=pred=pred.prev;
}while(pred.waitStatus>0);
pred.next=node;
}else{
/*
*waitStatusmustbe0orPROPAGATE.Indicatethatwe
*needasignal,butdon'tparkyet.Callerwillneedto
*retrytomakesureitcannotacquirebeforeparking.
*/
compareAndSetWaitStatus(pred,ws,Node.SIGNAL);
}
returnfalse;
}
↓
/**
*CASwaitStatusfieldofanode.
*/
privatestaticfinalbooleancompareAndSetWaitStatus(Nodenode,
intexpect,
intupdate){
returnunsafe.compareAndSwapInt(node,waitStatusOffset,
expect,update);
}
可以看到shouldParkAfterFailedAcquire 也是一路走，走到compareAndSetWaitStatus。
compareAndSetWaitStatus将prev的waitStatus设置为Node.SIGNAL。
Node.SIGNAL表示后续结点中的线程需要被unparking（类似被唤醒的意思）。该方法返回false。
经过这轮循环，队列模型变成下面状态
+--------------------------+prev+------------------+ head|waitStatus=Node.SIGNAL|<----node(tail)|currentThread| +--------------------------++------------------+
因为shouldParkAfterFailedAcquire返回的是false，所以后面这个条件就不再看了。继续for(;;) 中的循环。
如果state仍然大于0，再次进入到shouldParkAfterFailedAcquire。
这次因为head中的waitStatus为Node.SIGNAL，所以shouldParkAfterFailedAcquire返回true。
这次就需要看parkAndCheckInterrupt这个方法了。
privatefinalbooleanparkAndCheckInterrupt(){
LockSupport.park(this);
returnThread.interrupted();
}
ok，线程没有被中断，所以，返回false。继续for(;;) 中的循环。
如果state一直大于0，并且线程一直未被中断，那么就一直在这个循环中。也就是我们上篇文章说的裁判一直不愿意宣布比赛结束的情况。
那么什么情况下跳出循环呢？也就是什么情况下state会小于0呢？下一篇文章我将说明。
总结一下，await() 方法其实就是初始化一个队列，将需要等待的线程（state>0）加入一个队列中，并用waitStatus标记后继结点的线程状态。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持毛票票。