前言 CountDownLatch 一个同步辅助工具,同样是基于 AQS 实现,本篇文件主要是介绍 CountDownLatch 的使用,以及源码。

介绍

一个同步辅助工具,允许一个或多个线程等待,直到在其他线程中执行的一组操作完成为止
一个 CountDownLatch 初始化为给定计数。 在 await 方法阻塞,调用 countDown 方法会减少计数直到达到零,此后所有等待的线程被释放,任何后续调用 await 都会立即返回。 这是一次性的现象 - 计数不能复位。 如果你需要一个版本重置计数,请考虑使用CyclicBarrier 。
CountDownLatch 是一种通用的同步工具,可用于多种用途。

  1. 用作一个简单的开/关锁存器,或者门:所有线程调用await在门口等待,直到被调用 countDown 的线程打开。
  2. 初始化计数为 N ,用一个线程等待,直到 N 个线程完成某项操作,或某些动作已经完成 N 次。

CountDownLatch 一个有用的属性是,它不要求调用 countDown 线程等待计数到达零之前继续,它只是阻止任何线程通过await ,直到所有线程可以通过。

基本使用

在我之前 CAS 那篇文章《从JUC源码看CAS,我做了个笔记 ……》中介绍 CAS 举例时使用了 CountDownLatch,其代码如下:

  1. public class CasTest {
  2. private static final CountDownLatch LATCH = new CountDownLatch(10);
  3. private static int NUM_I = 0;
  4. private static volatile int NUM_J = 0;
  5. private static final AtomicInteger NUM_K = new AtomicInteger(0);
  6. public static void main(String[] args) throws InterruptedException {
  7. ExecutorService threadPool = Executors.newFixedThreadPool(10);
  8. for (int i = 0; i < 10; i++) {
  9. threadPool.execute(new Runnable() {
  10. public void run() {
  11. for (int j = 0; j < 10000; j++) {
  12. NUM_I++;
  13. NUM_J++;
  14. NUM_K.incrementAndGet();
  15. }
  16. LATCH.countDown();
  17. }
  18. });
  19. }
  20. LATCH.await();
  21. System.out.println("NUM_I = " + NUM_I);
  22. System.out.println("NUM_J = " + NUM_J);
  23. System.out.println("NUM_K = " + NUM_K.get());
  24. threadPool.shutdown();
  25. }
  26. }

简单介绍下这段代码的主要逻辑及功能:

  1. CountDownLatch 初始化计数为 10 。
  2. 开 10 个线程去处理业务逻辑,业务逻辑结束会调用 LATCH.countDown() 对计数进行 -1 操作。
  3. 在 LATCH.await() 处会阻塞等待,直到 LATCH 的值为 0 ,即 10 个线程业务都处理结束。
  4. 然后主线程继续执行。

    问题疑问

  5. CountDownLatch 和 AQS 有什么关系?

  6. CountDownLatch 的实现原理是什么?

    源码分析

    基本结构

    CountDownLatch - 图1
    通过类图可以看出,CountDownLatch 内部存在一个静态类 Sync,而 Sync 继承了 AbstractQueuedSynchronizer。具体内部是如何实现的,则下面通过源码和画图一步一步的进行介绍。

    初始化

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

    通过初始化构造器可以看出,在 new 创建对象时必须传递一个 int 类型的非负数。实现逻辑可以看出,是创建了一个 Sync 对象。

    1. private static final class Sync extends AbstractQueuedSynchronizer {
    2. private static final long serialVersionUID = 4982264981922014374L;
    3. Sync(int count) {
    4. setState(count);
    5. }
    6. int getCount() {
    7. return getState();
    8. }
    9. }

    之前在介绍 AQS 源码中已经介绍了 state 的含义,state 在不同子类中代表不同的含义。

  7. 在 ReentrantLock 中 state 代表加锁状态,0 没有线程获得锁,大于等于 1 已经有线程获得锁,大于 1 说明该获得锁的线程多次重入。

  8. 在 ReentrantReadWriteLock 中 state 代表锁的状态。state 为 0 ,没有线程持有锁,state 的高 16 为代表读锁状态,低 16 为代表写锁状态。通过位运算可以获取读写锁的实际值。
  9. 而在这里 (CountDownLatch)则代表门闩或者说计数的值。

    countDown

    1. public void countDown() {
    2. sync.releaseShared(1);
    3. }
    递减锁存器的计数:
  • 如果当前计数大于零,则递减。
  • 如果计数到达零,则释放所有等待的线程。
  • 如果那么当前计数等于零没有任何反应。

此处调用的是 AQS 的 releaseShard() 方法,释放共享资源。

  1. // AQS 代码
  2. public final boolean releaseShared(int arg) {
  3. if (tryReleaseShared(arg)) {
  4. doReleaseShared();
  5. return true;
  6. }
  7. return false;
  8. }

在 AQS 释放共享资源方法中 tryReleaseShared(arg) 部分是在 CountDownLatch 的内部类 Sync 中实现的,代码部分如下:

  1. protected boolean tryReleaseShared(int releases) {
  2. // Decrement count; signal when transition to zero
  3. for (;;) {
  4. int c = getState();
  5. if (c == 0)
  6. return false;
  7. int nextc = c-1;
  8. if (compareAndSetState(c, nextc))
  9. return nextc == 0;
  10. }
  11. }

递减更新 state ,如果 state 为 0 则返回 false,否则返回 true 。
此时再对照上面 AQS 代码,发现:如果 tryReleaseShared 返回 true ,则会唤醒后续节点开始执行操作。所以也就是说,如果 state 不为 0,则不会唤醒后续节点,直到 state 为 0 。

await

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

导致当前线程等待,直到锁存器倒计数至零,除非线程被中断。

  • 如果当前计数为零,则此方法立即返回。
  • 如果当前计数大于零,则当前线程用于线程调度目的,禁用并一直处于休眠状态的两件事情之一发生:
    • 因调用countDown方法使计数达到0;
    • 其他某些线程中断当前线程。
      1. public final void acquireSharedInterruptibly(int arg)
      2. throws InterruptedException {
      3. if (Thread.interrupted())
      4. throw new InterruptedException();
      5. if (tryAcquireShared(arg) < 0)
      6. doAcquireSharedInterruptibly(arg);
      7. }
      AQS 定义了 tryAcquireShared 返回值分为 3 种:
  1. 小于 0: 表示失败;
  2. 等于 0: 表示共享模式获取资源成功,但后续的节点不能以共享模式获取成功;
  3. 大于 0: 表示共享模式获取资源成功,后续节点在共享模式获取也可能会成功,在这种情况下,后续等待线程必须检查可用性。

其中 tryAcquireShared 同样由 CountDownLatch 的内部类 Sync 中实现,内部逻辑主要是判断 state 的值,进行返回。
在内部实现中返回的值只有 1 和 -1 ,说明在 state == 0 时,返回 1 ,即唤醒后续节点。不等于 0 时,会阻塞。

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

总结

Q: CountDownLatch 和 AQS 有什么关系?
A: CountDownLatch 是基于 AQS 的共享模式实现的。
Q: CountDownLatch 的实现原理是什么?
A: 可以参考上面的源码解析,进行总结介绍。 CountDownLatch 是基于 AQS 共享模式实现的,在初始化时必须传入计数,该计数实际上是 AQS 的 state 值。在 countDown 时对 state 进行递减,在 当 state 为 0 时 会唤醒 AQS 队列中的所有等待的节点 (因为是共享模式)。而 await 方法是判断 state 的值,如果不是 0 ,则所有线程在队列中阻塞,等待唤醒。
Q: state 在代表的含义是什么?
A:

  1. 在 ReentrantLock 中 state 代表加锁状态,0 没有线程获得锁,大于等于 1 已经有线程获得锁,大于 1 说明该获得锁的线程多次重入。
  2. 在 ReentrantReadWriteLock 中 state 代表锁的状态。state 为 0 ,没有线程持有锁,state 的高 16 为代表读锁状态,低 16 为代表写锁状态。通过位运算可以获取读写锁的实际值。
  3. 而在这里 (CountDownLatch)则代表门闩或者说计数的值。