概述
CountDownLatch是J.U.C包中提供的一个并发工具类，其主要作用是协调多个线程之间的同步，其可以让一个线程在等待其他线程执行完任务之后再继续执行。
demo1
假设现在有一场考试，考场中有五个人，考试时间是1s，那么监考老师只能等考试时间到了才能收卷。使用CountDownLatch如下：
public static void main(String[] args) {
final CountDownLatch latch = new CountDownLatch(10);
ExecutorService service = new ThreadPoolExecutor(10, 10, 0L, TimeUnit.SECONDS, new ArrayBlockingQueue<>(10));
try {
for (int i = 0; i < 10; i++) {
service.execute(()->{
try {
System.out.println("学生i" + Thread.currentThread().getName() + " 开始答题。。。。");
Thread.sleep(1000);
System.out.println("时间到，学生i"+Thread.currentThread().getName()+"交卷。。。。");
} catch (InterruptedException e) {
e.printStackTrace();
}finally {
latch.countDown();
}
});
}
System.out.println("考试开始。。。");
latch.await();
service.shutdown();
System.out.println("考试结束。。。");
} catch (InterruptedException e) {
e.printStackTrace();
}
}
启动运行，输出如下：
考试开始。。。
学生ipool-1-thread-1 开始答题。。。。
学生ipool-1-thread-3 开始答题。。。。
学生ipool-1-thread-2 开始答题。。。。
学生ipool-1-thread-4 开始答题。。。。
学生ipool-1-thread-5 开始答题。。。。
学生ipool-1-thread-6 开始答题。。。。
学生ipool-1-thread-7 开始答题。。。。
学生ipool-1-thread-8 开始答题。。。。
学生ipool-1-thread-9 开始答题。。。。
学生ipool-1-thread-10 开始答题。。。。
时间到，学生ipool-1-thread-1交卷。。。。
时间到，学生ipool-1-thread-3交卷。。。。
时间到，学生ipool-1-thread-6交卷。。。。
时间到，学生ipool-1-thread-9交卷。。。。
时间到，学生ipool-1-thread-2交卷。。。。
时间到，学生ipool-1-thread-10交卷。。。。
时间到，学生ipool-1-thread-8交卷。。。。
时间到，学生ipool-1-thread-7交卷。。。。
时间到，学生ipool-1-thread-5交卷。。。。
时间到，学生ipool-1-thread-4交卷。。。。
考试结束。。。
源码分析
成员变量
//私有静态内部类
private final Sync sync;
// Sync继承自AQS 从而具有队列同步的功能
private static final class Sync extends AbstractQueuedSynchronizer {
。。。。
}
构造方法
public CountDownLatch(int count) {
//入参校验
if (count < 0) throw new IllegalArgumentException("count < 0");
//初始化 sync
this.sync = new Sync(count);
}
Sync(int count) {
setState(count);
}
//AbstractQueuedSynchronizer类
private volatile int state;
protected final void setState(int newState) {
state = newState;
}
CountDownLatch的构造方法实际上是对Sync进行初始化，而Sync的构造方法底层又是调用AQS框架的setState方法来设置计数器的值。
countdown方法
public void countDown() {
sync.releaseShared(1);
}
countDown方法很简单，就是调用AQS框架里的releaseShared来改变state的值，源码如下：
public final boolean releaseShared(int arg) {
//tryReleaseShared是一个模板方法，由其子类进行实现
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
我们来看一下tryReleaseShared这个方法，在AQS框架中，这是一个模板方法，由继承它的子类来具体实现。在CountDownLatch中其私有静态内部类Sync继承了AQS，所以也会重写该方法，通过自旋和CAS来实现释放锁的目的。如下所示：
protected boolean tryReleaseShared(int releases) {
//自旋
for (;;) {
//获取计数器的值
int c = getState();
// 每次释放的时候，也就是子任务完成的时候计数值减一
if (c == 0)
return false;
//否则的话 将state-1
int nextc = c-1;
//使用 CAS 修改 state的值
if (compareAndSetState(c, nextc))
// 子任务均处理完毕后，返回 true； 也就是真正的释放
// 将唤醒阻塞在同步队列的线程
// 否则继续自旋
return nextc == 0;
}
}
当tryReleaseShared的返回结果true时，继续执行AQS中的doReleaseShared方法，源码如下：
private void doReleaseShared() {
//自旋
for (;;) {
//获取CLH队列的头节点
Node h = head;
//如果不为null 且CLH队列不只一个节点
if (h != null && h != tail) {
//获取节点的waitStatus
int ws = h.waitStatus;
//如果是 ws是SINGAL状态
if (ws == Node.SIGNAL) {
//设置节点状态由 SINGAL变为0失败
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
//继续自旋
continue;
//唤醒头节点
unparkSuccessor(h);
}
// 如果ws==0 且 设置状态PROPAGATE失败
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
//继续自旋
continue;
}
//判断头节点是否已经反正变化
if (h == head)
//结束循环
break;
}
}
首先获得head节点。如果head节点不等于空且head节点不等于tail节点，说明CLH队列中此时不止一个节点在排队，获得head节点的waitStatus。判断当前head节点状态是否是SINGAL。处于SINGAL状态的节点，说明当前节点的后继节点处于被唤醒的状态。如果CAS操作将head节点的waitStatus重置为0失败，那么跳出当前循环，继续执行下一次循环(重新检查)。如果重置成功，那么调用unparkSuccessor方法唤醒后继节点。 如果当前head节点状态等于0，通过CAS操作将waitStatus设置为PROPAGATE(传播)状态，确保可以向后一个节点传播下去。如果CAS操作失败，那么当前循环，继续执行下一次循环。最后的h == head，是判断head节点是否发生变化。如果没有发生变化，结束循环。如果发生变化，必须再次循环。
private void unparkSuccessor(Node node) {
//获取节点的waitStatus
int ws = node.waitStatus;
//小于0的话 使用CAS 设置为0
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
//当前节点的下一个节点
Node s = node.next;
//如果为null 或者waitStatus>0,即被取消
if (s == null || s.waitStatus > 0) {
s = null;
//从尾开始向前遍历 找到第一个waitStatus小于等于0的节点
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
//不为空的话，就唤醒这个节点
if (s != null)
LockSupport.unpark(s.thread);
}
获取head节点的waitStatus，如果小于0，进行CAS操作重置为0。获取head节点的后继节点，如果后继节点等于null或者后继节点的waitStaus大于0(说明后继节点处于CANCELLED状态),那么从队列从尾部往前进行遍历寻找waitStatus小于等于0的节点。如果这个遍历出来的节点不等于null的话，那么通过LockSupport.unpark()唤醒这个节点中的线程。
从实现可以看出，每次子任务在调用 countDown 时，会将同步状态值减一，当所有子任务均完成时 (state = 0) 此时会唤醒阻塞在同步队列的节点。
await方法
await方法会使当前线程在计数器变为0之前，一直处于等待状态，除非被打断。
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
await方法底层调用的还是AQS框架中的acquireSharedInterruptibly方法，源码如下：
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//判断线程是否中断
if (Thread.interrupted())
//如果被中断 则直接抛出异常
throw new InterruptedException();
//tryAcquireShared依然是一个模板方法 由子类实现
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
AQS中的acquireSharedInterruptibly方法，会判断线程是否中断。如果中断， 抛出InterruptedException异常。值得注意的是Thread.interrupted()方法，是测试当前线程是否中断。该方法会清除线程的中断状态。换句话说，如果调用这个方法2次，那么第二次会直接返回false，除非当前线程在第一次调用之后再次被中断。如果tryAcquireShared()小于0(说明该计数器值大于0)，继续执行doAcquireSharedInterruptibly。
Sync中的tryAcquireShared方法很简单，源码如下：
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
这里只是简单的判断state变量。如果state等于0(说明计数值为0)，返回1，否则返回-1(说明计数器值大于0)。
最后是AQS里的doAcquireSharedInterruptibly方法：
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) {
int r = tryAcquireShared(arg);
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);
}
}
很明显，是通过轮询的方式去获取共享锁。首先将当前线程包装成类型为SHARED的节点，标志为共享类型的节点。获取当前节点的前驱节点。如果当前节点的前驱节点为head节点的话，说明该节点是在AQS队列中等待获取锁的第一个节点。调用CountDownLatch中的tryAcquireShared()尝试去获取锁。返回的值大于0的话，说明获取锁成功。如果获取共享锁成功，那么把当前节点设置为AQS同步队列中的head节点，同时将p.next置为null(方便GC)。回到头看，如果当前节点的前驱节点不是head节点或者获取锁失败，我们需要调用shouldParkAfterFailedAcquire()方法判断当前线程是否需要挂起，如果需要挂起调用 parkAndCheckInterrupt()。
await方法还有一个重载方法，加入超时机制。
// 返回false，代表超时
public boolean await(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}
// 返回false，代表超时。返回true，代表获得共享锁成功
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
return tryAcquireShared(arg) >= 0 ||
doAcquireSharedNanos(arg, nanosTimeout);
}
private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
final long deadline = System.nanoTime() + nanosTimeout;
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;;) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return true;
}
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L)
return false;
if (shouldParkAfterFailedAcquire(p, node) &&
//static final long spinForTimeoutThreshold = 1000L;
nanosTimeout > spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
if (Thread.interrupted())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}
在AQS中的doAcquireSharedNanos方法中，如果在nanosTimeout时间范围内，还没有获取共享锁成功的话，直接返回false。spinForTimeoutThreadshold的值为1000nanoseconds。如果shouldParkAfterFailedAcquire(p, node)返回true且超时时间大于阀值spinForTimeoutThreadshold的话，会通过LockSupport.parkNanos(this, nanosTimeout);让线程挂起nanosTimeout时间。这样的策略体现是：如果超时时间很短的话，就不把当前线程挂起，而是通过自旋，这样线程获取锁很快就释放的情况下，可以减少cpu资源和线程挂起和恢复的性能损耗。