并发工具 - 并发工具类 CountDownLatch 和 CyclicBarrier - 《Java 并发专题笔记》

1. 倒计时器：CountDownLatch
2. 循环栅栏：CyclicBarrier
- CountDownLatch 与 CyclicBarrier 比较

1. 倒计时器：CountDownLatch

在多线程协作完成业务功能时，有时候需要等待其他多个线程完成任务之后，主线程才能继续往下执行业务功能，在这种的业务场景下，通常可以使用 Thread 类的 join 方法，让主线程等待被 join 的线程执行完之后，主线程才能继续往下执行。当然，使用线程间消息通信机制也可以完成。其实，java并发工具类中为我们提供了类似「倒计时」这样的工具类，可以十分方便的完成所说的这种业务场景。

为了能够理解 CountDownLatch，举一个很通俗的例子，运动员进行跑步比赛时，假设有6个运动员参与比赛，裁判员在终点会为这6个运动员分别计时，可以想象没当一个运动员到达终点的时候，对于裁判员来说就少了一个计时任务。直到所有运动员都到达终点了，裁判员的任务也才完成。这6个运动员可以类比成6个线程，当线程调用CountDownLatch.countDown()方法时就会对计数器的值减 1，直到计数器的值为 0 的时候，裁判员（调用await()方法的线程）才能继续往下执行。

countDownLatch 的计数器不能重置，如果业务上需要一个可以重置次数的计数器，可以用 CyclicBarrier

CountDownLatch 的构造方法：**

public CountDownLatch(int count)

构造方法会传入一个整型数N，之后调用 CountDownLatch 的countDown方法会对N减1，知道N减到0的时候，当前调用await方法的线程继续执行。

countDownLatch 的方法

await() throws InterruptedException：调用该方法的线程等到构造方法传入的N减到0的时候，才能继续往下执行；
await(long timeout, TimeUnit unit)：与上面的 await 方法功能一致，只不过这里有了时间限制，调用该方法的线程等到指定的 timeout 时间后，不管N是否减至为 0，都会继续往下执行；
countDown()：使 CountDownLatch 初始值 N 减 1；
long getCount()：获取当前 CountDownLatch 维护的值；

Demo1

public class CountDownLatchDemo {
    private static CountDownLatch startSignal = new CountDownLatch(1);
    //用来表示裁判员需要维护的是6个运动员
    private static CountDownLatch endSignal = new CountDownLatch(6);
    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(6);
        for (int i = 0; i < 6; i++) {
            executorService.execute(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + " 运动员等待裁判员响哨！！！");
                    startSignal.await();
                    System.out.println(Thread.currentThread().getName() + "正在全力冲刺");
                    endSignal.countDown();
                    System.out.println(Thread.currentThread().getName() + "  到达终点");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            });
        }
        System.out.println("裁判员发号施令啦！！！");
        startSignal.countDown();
        endSignal.await();
        System.out.println("所有运动员到达终点，比赛结束！");
        executorService.shutdown();
    }
}

输出结果：

pool-1-thread-2 运动员等待裁判员响哨！！！
pool-1-thread-3 运动员等待裁判员响哨！！！
pool-1-thread-1 运动员等待裁判员响哨！！！
pool-1-thread-4 运动员等待裁判员响哨！！！
pool-1-thread-5 运动员等待裁判员响哨！！！
pool-1-thread-6 运动员等待裁判员响哨！！！
裁判员发号施令啦！！！
pool-1-thread-2正在全力冲刺
pool-1-thread-2  到达终点
pool-1-thread-3正在全力冲刺
pool-1-thread-3  到达终点
pool-1-thread-1正在全力冲刺
pool-1-thread-1  到达终点
pool-1-thread-4正在全力冲刺
pool-1-thread-4  到达终点
pool-1-thread-5正在全力冲刺
pool-1-thread-5  到达终点
pool-1-thread-6正在全力冲刺
pool-1-thread-6  到达终点
所有运动员到达终点，比赛结束！

Demo2

@Slf4j
public class CountDownLatchExample2 {
    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();
                }
            });
        }
        // 10 毫秒之后，虽然计数器没有减为0，也会执行 shutdown 
        countDownLatch.await(10, TimeUnit.MILLISECONDS);
        log.info("finish");
        exec.shutdown();
    }
    private static void test(int threadNum) throws Exception {
        Thread.sleep(100);
        log.info("{}", threadNum);
    }
}

使用场景：程序执行需要等待某个条件完成后才能继续执行后续操作，典型的应用：并行计算。当运算量很大时，可以将运算拆分成多个子任务，等多个任务都完成之后，父任务再拿到多个子任务的计算结果进行汇总。

注意：

如果是 countDownLatch.await(10, TimeUnit.MILLISECONDS); 则 10 毫秒（10 毫秒是 test 方法执行的时间）之后，虽然计数器没有减为0，也会执行 shutdown （执行完 shutdown 方法后，线程程序不会立刻销毁所有线程，而是让当前已有的线程全部执行完，再把当前线程池销毁）。
countDownLatch.countDown(); 为了因程序异常而无法执行，最好放到 finally 里执行。

2. 循环栅栏：CyclicBarrier

一个同步输助类，允许一组 thread 互相等待，直到某个公共屏障点（common barriar point：countdown=0等）。通过它可以完成过个线程之间的相互等待，只有当每个线程都准备就绪后，才能各自继续往下执行。当某个线程执行了 await 方法后，该线程进入等待状态，而且计数器执行 +1 操作，当计数器的值达到设置的初始值时，之前调用 await 方法的线程会被唤醒，继续执行后续的操作。因为 CyclicBarriar 释放后可以重用，因此被称为「循环屏障」。

使用场景：用于多线程计算数据，最后合并计算结果的应用场景。

主要方法：

//等到所有的线程都到达指定的临界点
await() throws InterruptedException, BrokenBarrierException 
//与上面的await方法功能基本一致，只不过这里有超时限制，阻塞等待直至到达超时时间为止
await(long timeout, TimeUnit unit) throws InterruptedException, 
BrokenBarrierException, TimeoutException 
//获取当前有多少个线程阻塞等待在临界点上
int getNumberWaiting()
//用于查询阻塞等待的线程是否被中断
boolean isBroken()
//将屏障重置为初始状态。如果当前有线程正在临界点等待的话，将抛出BrokenBarrierException。
void reset()

另外需要注意的是，CyclicBarrier提供了这样的构造方法：

public CyclicBarrier(int parties, Runnable barrierAction)

示例：

private static CyclicBarrier barrier = new CyclicBarrier(5);
public static void main(String[] args) throws Exception {
    ExecutorService executor = Executors.newCachedThreadPool();
    for (int i = 0; i < 10; i++) {
        final int threadNum = i;
        Thread.sleep(1000);
        executor.execute(() -> {
            try {
                race(threadNum);
            } catch (Exception e) {
                log.error("exception", e);
            }
        });
    }
    executor.shutdown();
}
private static void race(int threadNum) throws Exception {
    Thread.sleep(1000);
    log.info("{} is ready", threadNum);
    barrier.await();
    log.info("{} continue", threadNum);
}

结果是先每隔 1s 执行一个线程，当5个线程都执行完，再同时执行 log.info("{} continue", threadNum); ，然后再依次这样执行下面5个。

int await(long timeout, TimeUnit unit)也可以控制等待时间，但是注意为了不影响其他线程执行，需要捕获异常。

private static void race(int threadNum) throws Exception {
    Thread.sleep(1000);
    log.info("{} is ready", threadNum);
    try {
        barrier.await(2000, TimeUnit.MILLISECONDS);
    } catch (Exception e) {
        log.warn("BarrierException", e);
    }
    log.info("{} continue", threadNum);
}

输出：

16:53:34.934 [pool-1-thread-1] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample2 - 0 is ready
16:53:35.930 [pool-1-thread-2] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample2 - 1 is ready
16:53:36.933 [pool-1-thread-3] INFO com.mmall.concurrency.example.aqs.CyclicBarrierExample2 - 2 is ready
16:53:36.948 [pool-1-thread-1] WARN com.mmall.concurrency.example.aqs.CyclicBarrierExample2 - BarrierException
java.util.concurrent.TimeoutException: null
    at java.util.concurrent.CyclicBarrier.dowait(CyclicBarrier.java:257)
    at java.util.concurrent.CyclicBarrier.await(CyclicBarrier.java:435)
    at com.mmall.concurrency.example.aqs.CyclicBarrierExample2.race(CyclicBarrierExample2.java:37)
    at com.mmall.concurrency.example.aqs.CyclicBarrierExample2.lambda$main$0(CyclicBarrierExample2.java:24)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
    at java.lang.Thread.run(Thread.java:748)
16:53:36.948 [pool-1-thread-3] WARN com.mmall.concurrency.example.aqs.CyclicBarrierExample2 - BarrierException
...

在声明 CyclicBarriar 的时候，可以指定一个 runnable，此时在线程到达屏障的时候，优先执行 runnable 。

private static CyclicBarrier barrier = new CyclicBarrier(5, () -> {
        log.info("callback is running");
    });
public static void main(String[] args) throws Exception {
    ExecutorService executor = Executors.newCachedThreadPool();
    for (int i = 0; i < 10; i++) {
        final int threadNum = i;
        Thread.sleep(1000);
        executor.execute(() -> {
            try {
                race(threadNum);
            } catch (Exception e) {
                log.error("exception", e);
            }
        });
    }
    executor.shutdown();
}
private static void race(int threadNum) throws Exception {
    Thread.sleep(1000);
    log.info("{} is ready", threadNum);
    barrier.await();
    log.info("{} continue", threadNum);
}

输出：

为了理解CyclicBarrier，这里举一个通俗的例子。开运动会时，会有跑步这一项运动，我们来模拟下运动员入场时的情况，假设有6条跑道，在比赛开始时，就需要6个运动员在比赛开始的时候都站在起点了，裁判员吹哨后才能开始跑步。跑道起点就相当于“barrier”，是临界点，而这6个运动员就类比成线程的话，就是这6个线程都必须到达指定点了，意味着凑齐了一波，然后才能继续执行，否则每个线程都得阻塞等待，直至凑齐一波即可。cyclic是循环的意思，也就是说CyclicBarrier当多个线程凑齐了一波之后，仍然有效，可以继续凑齐下一波。CyclicBarrier 的执行示意图如下：
并发工具类 CountDownLatch 和 CyclicBarrier - 图5

示例

public class CyclicBarrierDemo {
    //指定必须有6个运动员到达才行
    private static CyclicBarrier barrier = new CyclicBarrier(6, () -> {
        System.out.println("所有运动员入场，裁判员一声令下！！！！！");
    });
    public static void main(String[] args) {
        System.out.println("运动员准备进场，全场欢呼............");
        ExecutorService service = Executors.newFixedThreadPool(6);
        for (int i = 0; i < 6; i++) {
            service.execute(() -> {
                try {
                    System.out.println(Thread.currentThread().getName() + " 运动员，进场");
                    barrier.await();
                    System.out.println(Thread.currentThread().getName() + "  运动员出发");
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (BrokenBarrierException e) {
                    e.printStackTrace();
                }
            });
        }
    }
}

输出结果：

运动员准备进场，全场欢呼............
pool-1-thread-2 运动员，进场
pool-1-thread-1 运动员，进场
pool-1-thread-3 运动员，进场
pool-1-thread-4 运动员，进场
pool-1-thread-5 运动员，进场
pool-1-thread-6 运动员，进场
所有运动员入场，裁判员一声令下！！！！！
pool-1-thread-6  运动员出发
pool-1-thread-1  运动员出发
pool-1-thread-5  运动员出发
pool-1-thread-4  运动员出发
pool-1-thread-3  运动员出发
pool-1-thread-2  运动员出发

CountDownLatch 与 CyclicBarrier 比较

相同点
- 用于控制并发的工具类
- 都是通过计数器实现
不同点
- CountDownLatch 的计数器只能使用一次，而 CyclicBarriar 的计数器可以使用reset 方法重置，循环使用。
- CountDownLatch 是一个或多个线需要等待其他线程完成某项操作之后才能继续往下执行，描述的是一个或多个线程等他其他线程的关系。CyclicBarriar 是多个线程之间相互等待，直到所有线程都满足之后才能继续执行，描述的是各个线程内部相互等待的关系。
- CountDownLatch方法比较少，操作比较简单，而CyclicBarrier提供的方法更多，比如能够通过getNumberWaiting()，isBroken()这些方法获取当前多个线程的状态，并且CyclicBarrier的构造方法可以传入barrierAction，指定当所有线程都到达时执行的业务功能；
- CountDownLatch是不能复用的，而CyclicLatch是可以复用的。