java并发包之CountDownLatch

一、前言

CountDownLatch是一个同步辅助器，用于一个线程等待多个线程执行完操作后再执行。

二、主要特点

1. 拥有一个初使值，定义等待并发线程执行的数目
2. await方法，表示将调用该方法的线程休眠，直到其他操作完成，初使值变为0，才唤醒该线程
3. countDown方法，每个被等到的线程执行完后调用该方法，初使值会减1
4. CountDownLatch是一致性的，一旦初始化变为0，就不能再次使用
5. 会利用AQS中的Condition队列和AQS同步队列，并且使用的是AQS的共享模式

三、使用场景

1. 实现最大的并发性：可以模拟同时启动多个线程的情况
2. 开始执行前等待多个线程完成各自的任务：比如下文会说到的运动员赛跑问题

四、使用实例

赛跑问题，子线程是运动员，主线程是裁判。运动员全员等待裁判的令枪指示，令枪响起，运动员全部开跑，代码如下：

public class RunnerDemo {

    private static final int RUNNER_NUMBER = 5;

    //5位运动员准备
    private static CountDownLatch ready = new CountDownLatch(RUNNER_NUMBER);

    private static CountDownLatch begin = new CountDownLatch(1);

    public static class Runner implements Runnable{

        public void run() {
            System.out.println(Thread.currentThread().getName()+"选手已经准备好");
            ready.countDown();
            try {
                begin.await();
                System.out.println("选手"+Thread.currentThread().getName()+"开跑");
            } catch (InterruptedException e) {
                e.printStackTrace();
            }

        }
    }

    public static void main(String[] args) {
        for (int i = 0 ; i < RUNNER_NUMBER ;i ++){
            new Thread(new Runner()).start();
        }

        try {
            ready.await();
            System.out.println("运动员们已经准备好,开跑!");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        
        begin.countDown();
    }

}

运行结果：

Thread-1选手已经准备好
Thread-2选手已经准备好
Thread-0选手已经准备好
Thread-3选手已经准备好
Thread-4选手已经准备好
运动员们已经准备好,开跑!
选手Thread-1开跑
选手Thread-3开跑
选手Thread-4开跑
选手Thread-0开跑
选手Thread-2开跑

五、源码分析

5.1 await()

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

    public final void acquireSharedInterruptibly(int arg)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)
            doAcquireSharedInterruptibly(arg);
    }

调用了CountDownLatch内部的Sync的tryAcquireShared和AQS的doAcquireSharedInterruptibly函数。tryAcquireShared函数的源码如下

 protected int tryAcquireShared(int acquires) {
            return (getState() == 0) ? 1 : -1;
        }

如果state的资源数值如上文中已经提到，在Sync的构造函数中已经初始化好。如果state＝0则返回，否则调用doAcquireSharedInterruptibly进入AQS队列中等待 代码如下：

  private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        //将节点标记为共享模式，并存入到AQS队列中
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head) {
                    int r = tryAcquireShared(arg);
                    //由于实例化后state基本不为0，因此r基本为-1。
                    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);
        }
    }

处理的方法基本上和reentrantLock中的类似。当执行parkAndCheckInterrupt方法后线程会被阻塞住(park())。直到其他线程唤醒此线程。但是针对r值有两种情况：

1. r=-1，表示还没有线程调用countDown方法，state没有变化导致state一直大于0或者调用了countDown方法，但state不等于0，仍然等待state＝0
2. r=1，表示countDown已经将state修改为0，那么就需要唤醒AQS队列中的线程。

5.2 countDown()

  public void countDown() {
        sync.releaseShared(1);
    }

调用 Sync的releaseShared方法

public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }

然后是tryReleaseShared

protected boolean tryReleaseShared(int releases) {
            // Decrement count; signal when transition to zero
            for (;;) {
                int c = getState();
                if (c == 0)
                    return false;
                int nextc = c-1;
                if (compareAndSetState(c, nextc))
                    return nextc == 0;
            }
        }

利用cas减少state的值，如果state=0则尝试唤醒await阻塞的线程，执行doReleaseShared来处理

private void doReleaseShared() {
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                //如果头节点的状态为SIGNAL
                if (ws == Node.SIGNAL) {
                    //将头节点状态通过CAS设置为0，
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;
                    //唤醒后续的节点(详细看前文对于unparkSuccessor介绍) 
                    unparkSuccessor(h);
                ｝
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;             
            }
            if (h == head)                
                break;
        }
    }