CountdownLatch 源码

构造方法

publicCountDownLatch(intcount){if(count<0)thrownewIllegalArgumentException("count < 0");this.sync=newSync(count);}

Sync

privatestaticfinalclassSyncextendsAbstractQueuedSynchronizer{privatestaticfinallongserialVersionUID=4982264938641297070L;Sync(intcount){setState(count);}intgetCount(){returngetState();}// await调用：共享模式尝试获取锁protectedinttryAcquireShared(intacquires){// state == 0 返回1（获取成功），否则-1（阻塞）return(getState()==0)?1:-1;}// countDown调用：共享模式释放锁，state自减1protectedbooleantryReleaseShared(intreleases){// CAS循环自旋减statefor(;;){intc=getState();if(c==0)returnfalse;intnextc=c-1;if(compareAndSetState(c,nextc))returnnextc==0;}}}

countdown

publicvoidcountDown(){sync.releaseShared(1);}

releaseShared

publicfinalbooleanreleaseShared(intarg){if(tryReleaseShared(arg)){doReleaseShared();returntrue;}returnfalse;}

tryReleaseShared

protectedbooleantryReleaseShared(intreleases){for(;;){intc=getState();if(c==0)returnfalse;intnextc=c-1;if(compareAndSetState(c,nextc))returnnextc==0;}}

doReleaseShared

• 遍历 AQS 阻塞队列，唤醒头部等待节点；
• 等待线程被 unpark 唤醒，再次执行tryAcquireShared；
• 如果 state=0，返回 1，获取共享锁成功，退出阻塞逻辑，主线程继续执行后续代码

privatevoiddoReleaseShared(){for(;;){Nodeh=head;if(h!=null&&h!=tail){intws=h.waitStatus;// 头节点标记SIGNAL，说明后继需要唤醒if(ws==Node.SIGNAL){// CAS重置头节点状态为0，防止重复唤醒if(!compareAndSetWaitStatus(h,Node.SIGNAL,0))continue;// 唤醒头节点的后继等待线程（主线程）unparkSuccessor(h);}// 无等待线程，设置PROPAGATE传播标记elseif(ws==0&&!compareAndSetWaitStatus(h,0,Node.PROPAGATE))continue;}// 循环终止条件：中途head发生变化，说明有新节点被唤醒，继续循环传播if(h==head)break;}}

await

publicvoidawait()throwsInterruptedException{sync.acquireSharedInterruptibly(1);}

acquireSharedInterruptibly

publicfinalvoidacquireSharedInterruptibly(intarg)throwsInterruptedException{if(Thread.interrupted())thrownewInterruptedException();if(tryAcquireShared(arg)<0)doAcquireSharedInterruptibly(arg);}

tryAcquireShared

protectedinttryAcquireShared(intacquires){return(getState()==0)?1:-1;}

doAcquireSharedInterruptibly

• 把当前主线程封装成 共享 SHARED 类型 Node
• CAS 把节点加入 AQS 阻塞双向链表尾部
• 循环自旋：
  1. 如果当前节点是队列头节点，再次调用 tryAcquireShared
  2. 执行 parkAndCheckInterrupt()，调用 LockSupport.park()
  3. 当前线程挂起，卡住，不再往下执行，等待其他线程 countDown 把 state 降到 0 唤醒

privatevoiddoAcquireSharedInterruptibly(intarg)throwsInterruptedException{// 1. 创建共享SHARED类型节点，加入队尾finalNodenode=addWaiter(Node.SHARED);booleanfailed=true;try{for(;;){// 获取当前节点前驱finalNodep=node.prev;// 如果前驱是头结点，有资格尝试获取共享锁if(p==head){intr=tryAcquireShared(arg);if(r>=0){// 获取共享锁成功，设置新head，传播唤醒setHeadAndPropagate(node,r);p.next=null;// help GCfailed=false;return;}}// 调整节点waitStatus，无资格则park阻塞if(shouldParkAfterFailedAcquire(p,node)&&parkAndCheckInterrupt())thrownewInterruptedException();}}finally{if(failed)cancelAcquire(node);}}

setHeadAndPropagate

privatevoidsetHeadAndPropagate(Nodenode,intpropagate){Nodeh=head;// 保存旧头setHead(node);// 当前等待节点晋升为新head，清空thread、prev// propagate>0 说明还有剩余共享许可，需要继续唤醒后继if(propagate>0||h==null||h.waitStatus<0||(h=head)==null||h.waitStatus<0){Nodes=node.next;// 如果下一个节点也是共享节点，继续唤醒传播if(s==null||s.isShared())doReleaseShared();}}

getCount

publiclonggetCount(){returnsync.getCount();}

总结

await

latch.await() → sync.acquireSharedInterruptibly(1) → tryAcquireShared 返回 -1 → doAcquireSharedInterruptibly(1) addWaiter(SHARED) 入队 for 自旋 tryAcquireShared 失败 shouldParkAfterFailedAcquire 设置前驱 SIGNAL parkAndCheckInterrupt() 阻塞主线程

countdown

latch.countDown() → sync.releaseShared(1) → tryReleaseShared CAS state到0，返回true → doReleaseShared() CAS清空head SIGNAL标记 unparkSuccessor 唤醒主线程 等待线程被唤醒，回到 doAcquireSharedInterruptibly 自旋 tryAcquireShared 返回 1 setHeadAndPropagate 晋升新 head、传播唤醒 退出阻塞，await执行完毕