J.U.C之AQS—CountDownLatch

概述

它是一个同步辅助类，通过它可以在一个线程（线程间会轮换）执行countdown() -> count值减至0的期间，保证其他线程会调用await()一直阻塞等待，最后等待的线程执行resume()，所有线程再一起执行另一个实务操作。其中有一个原子性的且不会被重置的计数器以保证上述的实现。

原理图如下：

使用场景

当一个程序需在另一个条件完成后才可以继续执行后续操作。
如：并行计算中最后的汇总操作场

例子演示

情景一：指定计数次数

@Slf4j
public class CountDownLatchExample1 {

    private final static int threadCount = 200;

    public static void main(String[] args) throws Exception {

        ExecutorService exec = Executors.newCachedThreadPool();

        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);

        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (Exception e) {
                    log.error("exception", e);
                } finally {
                    //保证了方法被调用即计数执行减一
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
        log.info("finish");
        exec.shutdown();
    }

    private static void test(int threadNum) throws Exception {
        Thread.sleep(100);
        log.info("{}", threadNum);
        Thread.sleep(100);
    }
}

运行结果：（截取部分）

根据结果，多线程并发期间，核心方法以乱序执行，但总数仍一定，且最后执行到测试语句“finish”。其中，countDownLatch.await()语句循环检查计数是否已经减为0，即保证了此时全部线程执行结束。

情景二：指定计数时间

@Slf4j
public class CountDownLatchExample2 {

    private final static int threadCount = 2000;

    public static void main(String[] args) throws Exception {

        ExecutorService exec = Executors.newCachedThreadPool();

        final CountDownLatch countDownLatch = new CountDownLatch(threadCount);

        for (int i = 0; i < threadCount; i++) {
            final int threadNum = i;
            exec.execute(() -> {
                try {
                    test(threadNum);
                } catch (Exception e) {
                    log.error("exception", e);
                } finally {
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await(3, TimeUnit.MILLISECONDS);
        log.info("finish");
        exec.shutdown();
    }

    private static void test(int threadNum) throws Exception {
        Thread.sleep(100);
        log.info("{}", threadNum);
    }
}

测试结果：

先来看一下await()的源码：

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

/**
 * Causes the current thread to wait until the latch has counted down to
 * zero, unless the thread is {@linkplain Thread#interrupt interrupted},
 * or the specified waiting time elapses.
 *
 * <p>If the current count is zero then this method returns immediately
 * with the value {@code true}.
 *
 * <p>If the current count is greater than zero then the current
 * thread becomes disabled for thread scheduling purposes and lies
 * dormant until one of three things happen:
 * <ul>
 * <li>The count reaches zero due to invocations of the
 * {@link #countDown} method; or
 * <li>Some other thread {@linkplain Thread#interrupt interrupts}
 * the current thread; or
 * <li>The specified waiting time elapses.
 * </ul>
 */
public boolean await(long timeout, TimeUnit unit)
    throws InterruptedException {
    return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

通过countDownLatch.await(3, TimeUnit.MILLISECONDS)方法，指定timeout参数实现。其中，该参数指的是在test方法执行时开始计数，延时timeout的时间后或计数已经减为0后，继续执行后续方法，此例为log.info("finish")，即输出finish。
而无参的await()方法只有计数到0时才会继续执行后续方法。

结果分析：
例子中为了测试timeout，将timeout设置为1ms，而线程执行核心方法时设定休眠100ms，那么与结果是正好对应的，即finish总是第一个被输出的日志。

但是，finish输出后就直接执行了exec.shutdown()即关闭线程池的操作了啊，怎么还会有线程日志输出？

其实，exec.shutdown()操作：不会再接受新的线程任务，只会等待当前已经分配的线程执行完操作后再关闭，而不是在第一时间销毁所有的线程并强制关闭线程池。但是线程池还有一个立即关闭的线程池的方法 -> 在第一时间销毁所有的线程并强制关闭线程池，即shutdownNow()。

现在修改shutdown()为shutdownNow()，再进行测试，运行结果为：（部分截图）

结果分析：
finish日志字段输出和其后200个exception java.lang.InterruptedException: sleep interrupted异常说明了测试的正确性。也是对应了上面await()源代码中的说明：unless the thread is {@linkplain Thread#interrupt interrupted}。

小总结

使用CountDownLatch中，最好是计数指定配合指定超时时间使用，避免计数因为意外的情况难以到达使得系统资源空耗或业务逻辑无法继续执行情况，以提高程序的高效性，实用性。