前言
在开始之前，就开始一个小实验，在这个实验里面，我希望main线程等待CountDownLatch减少它的计数器数值为0则结束，我规定这个数值为4，每隔一秒启动一个线程去CountDownLatch的计数器值。以下就是这个小实验的代码。
public class CountDownLatchDemo {
public static void main(String[] args) {
//为了保持异步，首先启动一个线程
new Thread(new Runnable() {
@Override
public void run() {
for (int i = 0; i < count; i++) {
try {
//每隔1秒启动一个线程减少计数器
Thread.sleep(1000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
new Thread(new Worker()).start();
}
}
}).start();
System.out.println("等待CountDownLatch...");
try {
cdl.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("CountDownLatch结束...");
}
static final int count = 4;
static final CountDownLatch cdl = new CountDownLatch(count);
static class Worker implements Runnable {
@Override
public void run() {
System.out.println(Thread.currentThread()+" count down");
cdl.countDown();
}
}
}
这样，输出的结果也是意料之中的，如下所示：
等待CountDownLatch...
Thread[Thread-1,5,main] count down
Thread[Thread-2,5,main] count down
Thread[Thread-3,5,main] count down
Thread[Thread-4,5,main] count down
CountDownLatch结束...
分析
就像前面几篇分析的一样，其实都是利用了AQS提供的同步机制。现在就来看一看CountDownLatch的await方法。值得一提的是，CountDownLatch用AQS的state属性表示计数器的值。
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
//这个方法是在AQS里面定义的，CountDownLatch使用CLH队列的时候，它放进去的Node类型是Shared类型
//与ReentrantLock不同，ReentrantLock使用的Exclusive类型。
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//如果线程设置了中断位，重新擦除中断标识以后抛出中断异常，这个方法也随之结束
if (Thread.interrupted())
throw new InterruptedException();
// tryAcquireShared方法返回-1表示失败，如果失败的话，就要将线程“阻塞”
// 这里的阻塞有两种方式，一种是通过死循环自旋，另一种是调用LockSupport的park方法将线程阻塞
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
那让调用await方法的线程停下，无法执行await以下的代码，这样的代码细节又是怎样的呢，再来看一看。
protected int tryAcquireShared(int acquires) {
//state属性=0，说明已经有线程把这个CountDownLatch的计数器操作为0
return (getState() == 0) ? 1 : -1;
}
// 这个方法在countDown方法也会用到
protected boolean tryReleaseShared(int releases) {
// 把拿到的state属性减1（如果大于0），如果state为0，返回true
for (;;) {
int c = getState();
if (c == 0)
return false;
int nextc = c-1;
// 死循环的意义就在这里，如果CAS设置新的state失败，也就是说，线程之间的竞争很激烈，
// 它会一而再再而三的去尝试
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
// 只有在CLH队头的线程才有资格去跳出这个方法
int r = tryAcquireShared(arg);
if (r >= 0) {
// 终于，计数器的值被置为0了
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// state还不是0，就会走到这里，第一次的时候，waitStatus是0，那么node的waitStatus就会被置为SIGNAL
// 所以第二次走到这里的时候，parkAndCheckInterrupt方法就会被执行
// LockSupport的park方法就会把当前线程阻塞住。
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
当计数器为0的时候，不管调用await方法的线程是在自选，还是被LockSupport给阻塞住了，我们都得把它唤醒，唤醒这个线程的具体方法就在setHeadAndPropagate这个方法里边。
setHeadAndPropagate
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
// 要唤醒等待的线程了，那么这个线程对应的node节点也就没有用了，我们就把它作为head指针。
setHead(node);
// 传进来的propagate一般都是1 > 0，或者原来的head没有被cancel掉，
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
// 如果没有后继的等待线程（也有可能因为null而不知道），
// 或者都是Shared模式的节点，就全部释放这些线程。
if (s == null || s.isShared())
doReleaseShared();
}
}
doReleasedShared
private void doReleaseShared() {
for (;;) {
Node h = head;
// h = tail说明没有节点在CLH队列中，也就是没有线程在等待
if (h != null && h != tail) {
int ws = h.waitStatus;
// 在前面我们可以看到，head的waitStatus会被设置成SIGNAL然后被阻塞
// 如果是这样的话，就可以释放这个线程了
if (ws == Node.SIGNAL) {
// 在释放这个线程之前，应该先改变waitStatus，如果改变失败，再来一遍
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
unparkSuccessor(h);
}
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}
unparkSuccessor
private void unparkSuccessor(Node node) {
// 把正在运行状态node设置成无用状态，也就是waitStatus设置为0
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
//找到下一个节点
Node s = node.next;
if (s == null || s.waitStatus > 0) {
如果是null或者被cancel掉，就要从后向前找
s = null;
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
//确实有这么一个后继节点，释放这个节点包含的线程
//线程被唤醒之后，又回到上文写到的doAcquireSharedInterruptibly方法的死循环里
if (s != null)
LockSupport.unpark(s.thread);
}
countDown方法
countDown方法也是调用AQS在CountDownLatch里的子类Sync实现的releaseShared方法。
public void countDown() {
sync.releaseShared(1);
}
sync.releaseShared(1);
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
没错，这里的tryReleaseShared方法和doReleaseShared方法就是上面写到的两个方法，有了上面的基础，理解countDown方法就不难了。首先，最后一个调用countDown方法的线程会先唤醒排在Sync队列队头的线程，然后被唤醒的线程就在doAcquireSharedInterruptibly方法里调用setHeadAndPropagate方法依次唤醒排在Sync队列的线程。
结尾
一个CountDownLatch调用了await方法，要么会被自旋，要么会被LockSupport给阻塞掉。如果是自旋的话，它会不断的去判断计数器是否已经被置0，置0了就可以在CLH队列设置新的head指针从而跳出await()方法。如果是被阻塞住了，那么最后一个调用countDown方法的线程就会负责去唤醒它，是通过调用LockSupport的unpark方法做到的。
CountDownLatch是不能复用的，它和CyclicBarrier不一样，CyclicBarrier设置了一个Generation类作为复用的标识，同时还提供复用的机制，但是CountDownLatch并没有这么设计。
