源码分析：CountDownLatch 之倒计时门栓

Posted 2021-04-05 精灵王

简介

CountDownLatch 是JDK1.5 开始提供的一种同步辅助工具，它允许一个或多个线程一直等待，直到其他线程执行的操作完成为止。在初始化的时候给定 CountDownLatch 一个计数，调用await() 方法的线程会一直等待，其他线程执行完操作后调用countDown()，当计数减到0 ，调用await() 方法的线程被唤醒继续执行。

应用场景

1. 多线程并发下载或上传
   主线程初始化一个为5的CountDownLatch ，然后分发给5个线程去完成下载或上传的动作，主线程等待其他线程完成任务后返回成功呢。

2. 首页，一个复杂的查询包含多个子查询，但是子查询结果互相不依赖，也可以使用 CountDownLatch ，等待多个查询完成后再一起返回给首页。

源码分析

CountDownLatch 的源码相对于之前介绍的几个同步类，代码量要少很多很多，在JDK 1.8版本中也就300多行（包含注释），所以分析起来也比较简单。

内部类Sync

同样的，该内部类也继承了AQS，代码展示：

private static final class Sync extends AbstractQueuedSynchronizer {
    private static final long serialVersionUID = 4982264981922014374L;

    Sync(int count) { // 同步器的构造方法，初始化计数
        setState(count);
    }
   ...
}

主要的属性

主要的属性就一个，也就是内部类实例：同步器Sync

private final Sync sync;

构造方法

CountDownLatch 就一个构造方法，必须制定初始化计数

public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count); // 初始化同步器，指定计数
}

CountDownLatch 不算构造方法和toString方法一共也才4个方法，不多，所以我们全部看一下

await() 方法

调用该方法的线程会被阻塞，指定初始化的计数被减为0，或者线程被中断抛出异常。

代码展示：

// CountDownLatch.await()
public void await() throws InterruptedException { // 会抛出中断异常
    sync.acquireSharedInterruptibly(1); //调用的是同步器框架AQS的方法
}
// AQS框架代码
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
    if (Thread.interrupted()) // 检查线程中断状态，抛出异常
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0) // 套路一样，调用Sync里面的方法
        doAcquireSharedInterruptibly(arg); // 阻塞线程，排队，等待被唤醒
}
// 内部类Sync.tryAcquireShared()
protected int tryAcquireShared(int acquires) {
    // 检查计数，如果为0，返回1，如果不为0，返回-1；
    return (getState() == 0) ? 1 : -1;  
}

await() 方法总结：

1. 这应该是最简单的一个tryAcquireShared方法实现了。

2. 仅调用了getState来检查当前计数，如果计数为0，返回1；如果计数不为0，返回-1。

3. 阻塞线程，排队，等待被唤醒，中断抛出异常等逻辑都是在AQS实现的，具体分析请看之前的AQS分析文章

boolean await(timeout, unit)方法

和无参数的await()方法唯一的区别就是该方法指定了等待超时的时间，并且有返回值；
如果计数为0，则返回true；
如果线程被中断，则抛出异常；
如果线程经过了指定的等待时间，则返回false;

代码展示：

public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException {
    if (Thread.interrupted()) // 检查线程中断状态
        throw new InterruptedException();
    // tryAcquireShared 只会返回1或者-1，返回1代表计数已经为0，直接返回true
    // doAcquireSharedNanos 是AQS 框架里面的代码
    return tryAcquireShared(arg) >= 0 || doAcquireSharedNanos(arg, nanosTimeout);
}

// AQS 框架里面的代码
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 (;;) { // 自旋 tryAcquireShared(arg)
            final Node p = node.predecessor();
            if (p == head) { // 轮到当前节点了
                int r = tryAcquireShared(arg);
                if (r >= 0) { // 这里返回的大于等于0，说明计数为0，返回true
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
            }
            nanosTimeout = deadline - System.nanoTime();
            if (nanosTimeout <= 0L)
                return false; // 超时了，直接返回false
            if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold)
                LockSupport.parkNanos(this, nanosTimeout); // 阻塞当前线程
            if (Thread.interrupted()) // 中断抛出异常
                throw new InterruptedException();
        }
    } finally {
        if (failed) // 节点被取消
            cancelAcquire(node);
    }
}

countDown() 方法

如果当前计数大于零，则将其递减，如果计数达到零，则唤醒所有等待的线程（调用了await方法的线程）。如果当前计数等于零，那么什么也不会发生。源码展示：

public void countDown() {
    sync.releaseShared(1); // 调用AQS递减计数
}

// AQS同步框架的代码
public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) { // 调用自己实现的方法tryReleaseShared
        doReleaseShared(); //计数为0，唤醒所有等待的线程，返回true
        return true;
    }
    return false;
}
// CDL 自己实现的递减计数方法
protected boolean tryReleaseShared(int releases) {
    // Decrement count; signal when transition to zero
    for (;;) { // 自旋，保证递减操作成功
        int c = getState(); // 当前的技术
        if (c == 0) // 计数已经是0了，返回false，之后啥也不会发生
            return false;
        int nextc = c-1; // 递减
        if (compareAndSetState(c, nextc)) // cas 更新计数
            return nextc == 0; 计数为0才返回true
    }
}
// 唤醒等待的线程
private void doReleaseShared() {
    for (;;) { //自旋操作
        Node h = head;
        if (h != null && h != tail) { // 等待的线程队列不为空
            int ws = h.waitStatus;
            if (ws == Node.SIGNAL) {//  检查状态是否要唤醒下一个节点的线程
                if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) // CAS 失败了才会继续continue
                    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;
    }
}

countDown() 方法总结：

1. 主要逻辑就是把计数减1

2. 如果计数减到了0，则唤醒所有队列中等待的线程

3. 如果减之前计数已经是0了，则什么也不干

getCount() 方法

public long getCount() { // CDL 的API
    return sync.getCount();
}
// 内部类 Sync
int getCount() {
    return getState();
}
// AQS 框架api
protected final int getState() {
    return state;
}

返回当前的计数。

CountDownLatch 总结

1. 主要功能维护计数，当计数减为零后才放开所有等待的线程

2. CountDownLatch 没有加计数的API，所以一个CountDownLatch不可以重复使用，如果要用可以重置计数的，可以使用CyclicBarrier。

3. CountDownLatch 也会有“死锁”的现象，要避免计数永远减不到0的情况

4. 如果初始化计数为0，那么 CountDownLatch 则毫无作用，不如不用

5. 如果初始化计数为1，调用await时阻塞自己，别人countDown解锁后，再唤醒自己（类似于在等一个资源，拿到资源在继续进行）

和Semaphore的区别

Semaphore 可以用来限流，比如限制一个景区最多允许10000人同时在园内，只有当有人出园后，才允许其他人入园。

CountDownLatch 可以用来计数，比如导游在出发点等待10名游客一起出发，来一名游客就画个叉，直到10名游客到齐后，才一起出发去旅游。

---