CountDownLatch原理详解

介绍

当你看到这篇文章的时候需要先了解AQS的原理，因为本文不涉及到AQS内部原理的讲解。

CountDownLatch是一种同步辅助，让我们多个线程执行任务时，需要等待线程执行完成后，才能执行下面的语句，之前线程操作时是使用Thread.join方法进行等待，CountDownLatch内部使用了AQS锁，前面已经讲述过AQS的内部结构，其实内部有一个state字段，通过该字段来控制锁的操作，CountDownLatch是如何控制多个线程执行都执行结束？其实CountDownLatch内部是将state作为计数器来使用，比如我们初始化时，state计数器为3，同时开启三个线程当有一个线程执行成功，每当有一个线程执行完成后就将state值减少1，直到减少到为0时，说明所有线程已经执行完毕。

源码解析

以一个例子来开始进行源码解析，后面的内容会针对例子来进行源码的分解过程，我们开启三个线程，主线程需要等待三个线程都完成后才能进行后续任务处理，源码如下所示：

public class CountDownLatchDemo {public static void main(String[] args) throws InterruptedException {// 计数器3个。CountDownLatch countDownLatch = new CountDownLatch(3);for (int i = 0; i < 3; ++i) {new Thread(new Worker(countDownLatch, i)).start();}// 等待三个线程都完成countDownLatch.await();System.out.println("3个线程全部执行完成");}// 搬运工人工作线程工作类。static class Worker implements Runnable {private final CountDownLatch countDown;private final Integer id;Worker(CountDownLatch countDown, Integer id) {this.countDown = countDown;this.id = id;}@Overridepublic void run() {try {Thread.sleep(500);doWork();} catch (InterruptedException e) {e.printStackTrace();}countDown.countDown();System.out.println("第" + id + "个线程执行完成工作");}void doWork() {System.out.println("第" + id + "个线程开始工作");}}
}

通过一个例子来说明一下CountDownLatch的工作原理，上面例子我们开启了三个线程，每个线程分别执行自己的任务，主线程等待三个线程执行完成，看一下输出的结果：

等待三个线程完成
第1个线程开始工作
第0个线程开始工作
第0个线程执行完成工作
第1个线程执行完成工作
第2个线程开始工作
第2个线程执行完成工作
3个线程全部执行完成

这里我们将三个线程想象成搬运工人，将货物搬运到车上，三个人必须将自己手头分配的任务都搬运完成后才能触发，也即是货车司机需要等待三个人都完成才能发车，货车司机此时手里有个小本本，记录本次搬运的总人数，线程未启动时如下所示

当搬运工人开始工作时，每个搬运工人各自忙碌自己的任务，假如当工人1完成后，需要跟司机报备一下，说我已经完成任务了，这时候司机会在自己的小本本上记录，工人1已经完成任务，此时还剩下两个工人没有完成任务。

每当工人完成自己手头任务时，都会向司机报备，当所有工人都完成之后，此时工人的小本本记录的完成人数都已完成，司机这时候就可以发车了，因为三个人已经完成了搬运工作。

通过上面的例子，大致了解了CountDownLatch的简单原理，如何保证司机（state）记录谁完成了谁没完成呢？CountDownLatch内部通过AQS的state来完成计数器的功能，接下来通过源码来进行详细分析：

public class CountDownLatch {/*** 同步控制，* 使用 AQS的state来表示计数。*/private static final class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID = 4982264981922014374L;// 初始化state值（也就是需要等待几个线程完成任务）Sync(int count) {setState(count);}// 获取state值。int getCount() {return getState();}// 获得锁。protected int tryAcquireShared(int acquires) {// 这里判断如果state=0的时候才能获得锁，反之获取不到将当前线程放入到队列中阻塞。// 这里是关键点。return (getState() == 0) ? 1 : -1;}protected boolean tryReleaseShared(int releases) {// state进行减少，当state减少为0时，阻塞线程才能进行处理。for (;;) {int c = getState();if (c == 0)return false;int nextc = c-1;if (compareAndSetState(c, nextc))return nextc == 0;}}}// 锁对象。private final Sync sync;/*** 初始化同步锁对象。*/public CountDownLatch(int count) { if (count < 0) throw new IllegalArgumentException("count < 0");this.sync = new Sync(count);}/*** 导致当前线程等待直到闩锁倒计时到零，除非线程是被中断。如果当前计数为零，则此方法立即返回。如果当前计数大于零，* 则当前线程将被禁用以进行线程调度并处于休眠状态，直到发生以下两种情况：* 1.计数达到零。* 2.如果当前线程被中断。*/public void await() throws InterruptedException {sync.acquireSharedInterruptibly(1);}/*** 等待计数器清零或被中断，等待一段时间后如果还是没有*/public boolean await(long timeout, TimeUnit unit)throws InterruptedException {return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));}/*** 使当前线程等待直到闩锁倒计时到零，除非线程被中断或指定的等待时间已过。*/public void countDown() {sync.releaseShared(1);}/*** 返回state值。*/public long getCount() {return sync.getCount();}
}

CountDownLatch源码看上去很少，通过CountDownLatch源码看到内部是基于AQS来进行实现的，内部类Sync类继承自AbstractQueuedSynchronizer并且实现了tryAcquireShared和tryReleaseShared，通过构造函数看到，会创建一个AQS同步对象，并且将state值进行初始化，如果初始化count小于0则抛出异常。

public CountDownLatch(int count) {if (count < 0) throw new IllegalArgumentException("count < 0");// 初始化AQS的state值。this.sync = new Sync(count);
}

根据上面的例子我们来看一下初始化情况下的AQS内部情况：

awit方法

当调用awit方法时，其实内部调用的AQS的acquireSharedInterruptibly方法，这个方法会调用Sync中tryAcquireShared的方法，通过上面例子，我们初始化时将state值初始化2，但是Sync中判断(getState() == 0) ? 1 : -1;此时state值为2，判定为false，则返回-1，当返回负数时，内部会将当前线程挂起，并且放入AQS的队列中，直到AQS的state值减少到0会唤醒当前线程，或者是当前线程被中断，线程会抛出InterruptedException异常，然后返回。

/*** 导致当前线程等待直到闩锁倒计时到零，除非线程是被中断。如果当前计数为零，则此方法立即返回。如果当前计数大于零，* 则当前线程将被禁用以进行线程调度并处于休眠状态，直到发生以下两种情况：* 1.计数达到零。* 2.如果当前线程被中断。*/
public void await() throws InterruptedException {sync.acquireSharedInterruptibly(1);
}

当线程调用await方法时，其实内部调用的是AQS的acquireSharedInterruptibly，我们来看一下AQS内部acquireSharedInterruptibly的方法

    public final void acquireSharedInterruptibly(int arg)throws InterruptedException {// 响应中断if (Thread.interrupted())throw new InterruptedException();// 调用tryAcquireShared 方法。if (tryAcquireShared(arg) < 0)// 阻塞线程，将线程加入到阻塞队列等到其他线程恢复线程。doAcquireSharedInterruptibly(arg);}/*** Acquires in shared interruptible mode.* @param arg the acquire argument*/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 GCfailed = false;return;}}if (shouldParkAfterFailedAcquire(p, node) &&parkAndCheckInterrupt())throw new InterruptedException();}} finally {if (failed)cancelAcquire(node);}}

acquireSharedInterruptibly内部调用的是CountDownLatch内部类Sync实现的tryAcquireShared方法，tryAcquireShared判断state是否已经清空，也就是计数器是否已经清零了，清零时才能进行执行，此时并没有进行清空，则会将当前线程挂起，并且将挂起的线程放入到AQS的阻塞队列，等待其他线程唤醒动作。

coutDown方法

当线程执行完成后，会调用CountDownLatch的countDown，countDown方法内部调用的AQS的releaseShared，releaseShared方法实现在Sync类中，该方法主要作用是将state计数器中的值，进行减1操作，先进行判断是否已经将state值修改为0，如果修改为则不进行下面的操作，防止状态已经修改为0时，其他线程还调用了countDown操作导致state值变为负数，当state值减少1时，会通知阻塞队列中的等待线程，假设上面的例子其中一个线程先执行了countDown方法，则此时state=1，并且唤醒阻塞队列中的线程，线程还是会去调用tryAcquireShared方法，发现还是返回-1，则还会将当前线程进行挂起阻塞并且加入到阻塞队列中。此时队列状态如下所示：

当另外一个线程也执行完成，调用countDown时，state减少1则变为state=0，当这时候唤醒等待的线程时，tryAcquireShared返回的结果是1，则会直接返回成功。

总结

CountDownLatch是利用AQS的state来做计数器功能，当初始化CountDownLatch时，会将state值进行初始化，让调用CountDownLatch的awit时，会判断state计数器是否已经变为0，如果没有变为0则挂起当前线程，并加入到AQS的阻塞队列中，如果有线程调用了CountDownLatch的countDown时，这时的操作是将state计数器进行减少1，每当减少操作时都会唤醒阻塞队列中的线程，线程会判断此时state计数器是否已经都执行完了，如果还没有执行完则继续挂起当前线程，直到state计数器清零或线程被中断为止。

喜欢的朋友可以关注我的微信公众号，不定时推送文章