CountDownlatch实现原理

类图及概要

CountDownLatch 内部有个计数器，并且这个计数器是递减的 。 下面就通过源码看看 JDK 开发组在何时初始化计数器，在何时递减计数器，当计数器变为 0 时做了什么操作， 多个线程是如何通过计时器值实现同步的 。

从类图可以看出， CountDownLatch 是使用 AQS 实现的 。 通过下面的构造函数，你会发现，实际上是把计数器的值赋给了 AQS 的状态变量 state，也就是这里使用 AQS 的状态值来表示计数器值。

``

public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException(“count < 0”);
this.sync = new Sync(count);
}

Sync(int count) {
setState(count);
}

``

核心方法

await() 方法

当线程调 用 CountDownLatch 对象的 await 方法后 ， 当前线程会被 阻塞 ， 直到下面的情况之一发生才会返回 ： 当所有线程都调用了 CountDownLatch 对象的 countDown 方法后，也就是计数器的值为 0 时；其他线程调用了当前线程的 interrupt （）方法中断了当前线程 ，当前线程就会抛出InterruptedException 异常 ， 然后返回。

``

public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}

public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
// 如果线程被中断则抛出异常
if (Thread.interrupted())
throw new InterruptedException();
// 查看当前计数器千直是否为 0 ， 为 0 则直接返回 ， 否则进入AQS的队列等待
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}

// sync类实现的 AQS 的接口
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}

``
await（） 方法委托 sync 调用 了 AQS 的 acquiresharedInterruptibIy方法 。

该方法 的特点是线程获取资源时可 以被中 断， 并且获取 的资源是共享资源。 acquireSharedlnterruptibly 首先判断当前线程是否己被中断 ， 若是则抛出异常，否则调用 sync 实现 的 tryAcquireShared 方法查看当前状态值（ 计数器值）是否为 0 ， 是则当前线程 的 await（） 方法直接返回 ， 否 则调用 AQS 的doAcquireSharedlnterruptibly 方法让当前线程阻塞。另外可 以看到，这里tryAcquireShared 传递的 arg 参数没有被用 到， 调用tryAcquireShared 的方法仅仅是为了检查当前状态值是不是为 0 ， 并没有调用 CAS 让 当 前状态值减 1 。

await(long timeout, TimeUnit unit) 方法

当线程调用了 CountDownLatch 对象 的该方法后 ， 当前线程会被阻塞 ， 直到下面的情况之一发生才会返回：当 所有线程都调用了 CountDownLatch 对象 的 countDown 方法后 ，也就是计数器值为 0 时 ，这时候会返 回 true ；设置的 timeout 时间到了，因为超时而返回false ； 其他线程调用了当前线程的 interrupt ( ）方法中断了当前线程 ， 当前线程会抛出InterruptedException 异常，然后返回。

countDown() 方法

线程调用该方法后 ，计数器的值递减 ， 递减后如果计数器值为 0 则唤醒所有因调用await 方 法而被阻塞的线程，否则什么都不做。下面看下 countDown() 方法是 如何调用AQS 的方法的。

``

public void countDown() {
// 委托sync调 用 AQS的方法
sync.releaseShared(1);
}

public final boolean releaseShared(int arg) {
// 调用 sync 实现的 tryReleaseShared
if (tryReleaseShared(arg)) {
// AQS 的释放资源方法
doReleaseShared();
return true;
}
return false;
}

``

CountDownLatch 的 countDown ( ）方法委 托 sync 调用了 AQS 的releaseShared 方法。

re leaseShared 首先调用了 sync 实现的 AQS 的 tryReleaseShared 方法。

``

protected boolean tryReleaseShared(int releases) {

        // 循环进行CAS ，直到当前线程成功完成CAS使计数器值（状态值state ）减 1并更新到state
        for (;;) {
            int c = getState();
            // 如果当前状态值为0则直接返回（ 1 )
            if (c == 0)
                return false;
                // 使用 CAS让计数器佳减 1 ( 2)
            int nextc = c-1;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }

}

``

如上代码首先获取当前状态值（计数器值） 。 代码 (1) 判断如果当前状态值为 0 则直接返回 false，从而 countDown ( ）方法直接返回：否则执行码 (2) 使用 CAS 将计数器值减 1, CAS 失败则循环重试，否则如果当前计数器值为 0 则返回 true，返回 true 说明是最后一个线程调用的 countdown 方法，那么该线程除了让计数器值减 1 外，还需要唤醒因调用 CountDownLatch 的 await 方法而被阻塞的线程，具体是调用 AQS 的 doReleaseShared方法来激活阻塞的线程。这里代码 (1) 貌似是多余的，其实不然，之所以添加代码 (1) 是为了防止当计数器值为 0 后，其他线程又调用了 countDown 方法，如果没有代码 (1) 状态值就可能会变成负数。

getCount() 方法

获取当前计数器的值，也就是 AQS 的 state 的值，在其内部还是调用了 AQS 的 getState 方法来获取 state 的值（计数器当前值）。

``

public long getCount() {
return sync.getCount();
}

int getCount() {
return getState();
}

``

总结

使用 AQS 的状态变量来存放计数器 的值。首先在 初始化CountDownLatch 时设置状态值（计数器值），当多个线程调用 countdown 方法时实际是原子性递减 AQS 的状态值。当线程调用 await 方法后当前线程会被放入 AQS 的阻塞队列等待计数器为 0 再返 回 。其他线程调用 countdown 方法让计数器值递减 1 ，当计数器值变为0 时， 当 前线程还要调用 AQS 的 doReleaseShared 方法来激活由于调用 await() 方法而被阻塞的线程 。