一、前言
CountDownLatch是一个同步辅助器，用于一个线程等待多个线程执行完操作后再执行。
二、主要特点
- 拥有一个初使值，定义等待并发线程执行的数目
- await方法，表示将调用该方法的线程休眠，直到其他操作完成，初使值变为0，才唤醒该线程
- countDown方法，每个被等到的线程执行完后调用该方法，初使值会减1
- CountDownLatch是一致性的，一旦初始化变为0，就不能再次使用
- 会利用AQS中的Condition队列和AQS同步队列，并且使用的是AQS的共享模式
三、使用场景
- 实现最大的并发性：可以模拟同时启动多个线程的情况
- 开始执行前等待多个线程完成各自的任务：比如下文会说到的运动员赛跑问题
四、使用实例
赛跑问题，子线程是运动员，主线程是裁判。运动员全员等待裁判的令枪指示，令枪响起，运动员全部开跑，代码如下：
public class RunnerDemo {
private static final int RUNNER_NUMBER = 5;
//5位运动员准备
private static CountDownLatch ready = new CountDownLatch(RUNNER_NUMBER);
private static CountDownLatch begin = new CountDownLatch(1);
public static class Runner implements Runnable{
public void run() {
System.out.println(Thread.currentThread().getName()+"选手已经准备好");
ready.countDown();
try {
begin.await();
System.out.println("选手"+Thread.currentThread().getName()+"开跑");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) {
for (int i = 0 ; i < RUNNER_NUMBER ;i ++){
new Thread(new Runner()).start();
}
try {
ready.await();
System.out.println("运动员们已经准备好,开跑!");
} catch (InterruptedException e) {
e.printStackTrace();
}
begin.countDown();
}
}
运行结果：
Thread-1选手已经准备好
Thread-2选手已经准备好
Thread-0选手已经准备好
Thread-3选手已经准备好
Thread-4选手已经准备好
运动员们已经准备好,开跑!
选手Thread-1开跑
选手Thread-3开跑
选手Thread-4开跑
选手Thread-0开跑
选手Thread-2开跑
五、源码分析
5.1 await()
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
调用了CountDownLatch内部的Sync的tryAcquireShared和AQS的doAcquireSharedInterruptibly函数。tryAcquireShared函数的源码如下
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
如果state的资源数值如上文中已经提到，在Sync的构造函数中已经初始化好。如果state＝0则返回，否则调用doAcquireSharedInterruptibly进入AQS队列中等待 代码如下：
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
//将节点标记为共享模式，并存入到AQS队列中
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
//由于实例化后state基本不为0，因此r基本为-1。
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
//当前节点的前驱节点不是头节点时，线程会进入等待阻塞状态，直到获取到共享锁后被唤醒
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
处理的方法基本上和reentrantLock中的类似。当执行parkAndCheckInterrupt方法后线程会被阻塞住(park())。直到其他线程唤醒此线程。但是针对r值有两种情况：
- r=-1，表示还没有线程调用countDown方法，state没有变化导致state一直大于0或者调用了countDown方法，但state不等于0，仍然等待state＝0
- r=1，表示countDown已经将state修改为0，那么就需要唤醒AQS队列中的线程。
5.2 countDown()
public void countDown() {
sync.releaseShared(1);
}
调用 Sync的releaseShared方法
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
然后是tryReleaseShared
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;
}
}
利用cas减少state的值，如果state=0则尝试唤醒await阻塞的线程，执行doReleaseShared来处理
private void doReleaseShared() {
for (;;) {
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
//如果头节点的状态为SIGNAL
if (ws == Node.SIGNAL) {
//将头节点状态通过CAS设置为0，
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
//唤醒后续的节点(详细看前文对于unparkSuccessor介绍)
unparkSuccessor(h);
｝
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue;
}
if (h == head)
break;
}
}
