同步工具类—— CountDownLatch

本博客系列是学习并发编程过程当中的记录总结。因为文章比较多，写的时间也比较散，因此我整理了个目录贴（传送门），方便查阅。

并发编程系列博客传送门

CountDownLatch简介

CountDownLatch是JDK并发包中提供的一个同步工具类。官方文档对这个同步工具的介绍是：

A synchronization aid that allows one or more threads to wait until a set of operations being performed in other threads completes.

上面的英文介绍大体意思是：CountDownLatch的主要功能是让一个或者多个线程等待直到一组在其余线程中执行的操做完成。

使用列子

观看上面的解释可能并不能直观的说明CountDownLatch的做用，下面咱们经过一个简单的列子看下CountDownLatch的使用。

场景：主人（主线程）请客人（子线程）吃晚饭，须要等待全部客人都到了以后才开饭。咱们用CountDownLatch来模拟下这个场景。

public class CountDownLatchDemo {

    private static final int PERSON_COUNT = 5;

    private static final CountDownLatch c = new CountDownLatch(PERSON_COUNT);

    public static void main(String[] args) throws InterruptedException {
        System.out.println("l am master, waiting guests...");
        for (int i = 0; i < PERSON_COUNT; i++) {
            int finalI = i;
            new Thread(new Runnable() {
                @SneakyThrows
                @Override
                public void run() {
                    System.out.println(Thread.currentThread().getName()+" l am person["+ finalI +"]");
                    TimeUnit.MILLISECONDS.sleep(500);
                    //System.out.println(Thread.currentThread().getName()+" count:"+c.getCount());
                    c.countDown();
                }
            }).start();
        }
        c.await();
        System.out.println("all guests get, begin dinner...");
    }

}

上面的列子中，主人（master线程）请了5个客人吃饭，每一个客人到了以后会将CountDownLatch的值减一，主人（master）会一直等待全部客人到来，最后输出”开饭“。

CountDownLatch的使用方式很简单，下面来看下它的实现原理。

原理剖析

首先咱们先看下CountDownLatch重要的API

- getCount()：获取当前count的值。
- wait()：让当前线程在此CountDownLatch对象上等待，能够中断。与notify()、notifyAll()方法对应。
- await()：让当前线程等待此CountDownLatch对象的count变为0，能够中断。
- await(timeout,TimeUnit)：让当前线程等待此CountDownLatch对象的count变为0，能够超时、能够中断。
- countDown()：使此CountDownLatch对象的count值减1(不管执行多少次，count最小值为0)。

下面咱们看下具体API的源代码

构造函数

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

在构建CountDownLatch对象时须要传入一个int型的初始值，这个值就是计数器的初始值。从上面的代码中能够看出，建立CountDownLatch是new了一个Sync对象。

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

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

    int getCount() {
        return getState();
    }

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

    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;
        }
    }
}

Sync对象是基于AQS机制实现的，本身实现了tryAcquireShared和tryReleaseShared方法。

await方法

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();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

在acquireSharedInterruptibly中先判断了下当前线程有没有被中断，假如线程已经被中断了，直接抛出中断异常。不然进入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);
    }
}

doAcquireSharedInterruptibly的处理逻辑是先判断队列中是否只有当前线程，若是只有当前线程的先尝试获取下资源，若是获取资源成功就直接返回了。获取资源不成功就判断下是否要park当前线程，若是须要park当前线程，
那么当前线程就进入waiting状态。不然在for循环中一直执行上面的逻辑。

countDown方法

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

熟悉AQS机制的会知道上面的代码其实也是调的AQS的releaseShared。releaseShared的方法会调到Sync中的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;
    }
}

上面的代码逻辑很简单：status的值是0的话就返回true，不然返回false。返回true的话，就会唤醒AQS队列中全部阻塞的线程。

使用场景

• 场景一：将任务分割成多个子任务，每一个子任务由单个线程去完成，等全部线程完成后再将结果汇总。（MapReduce）这种场景下，CountDoenLatch做为一个完成信号来使用。
• 场景二：多个线程等到，一直等到某个条件发生。好比多个赛跑运动员都作好了准备，就等待裁判手中的发令枪响。这种场景下，就能够将CountdownLatch的初始值设置成1。

简单总结

• CountDownLatch的初始值不能重置，只能减小不能增长，最多减小到0；
• 当CountDownLatch计数值没减小到0以前，调用await方法可能会让调用线程进组一个阻塞队列，等待计数值减少到0；
• 调用countDown方法会让计数值每次都减少1，可是最多减小到0。当CountDownLatch的计数值减小到0的时候，会唤醒全部在阻塞队列中的线程。