Java并发编程笔记之 CountDownLatch闭锁的源码分析

JUC 中倒数计数器 CountDownLatch 的使用与原理分析，当须要等待多个线程执行完毕后在作一件事情时候 CountDownLatch 是比调用线程的 join 方法更好的选择，CountDownLatch 与 线程的 join 方法区别是什么？

平常开发中常常会遇到须要在主线程中开启多线程去并行执行任务，而且主线程须要等待全部子线程执行完毕后再进行汇总的场景，它的内部提供了一个计数器，在构造闭锁时必须指定计数器的初始值，且计数器的初始值必须大于0。另外它还提供了一个countDown方法来操做计数器的值，每调用一次countDown方法计数器都会减1，直到计数器的值减为0时就表明条件已成熟，全部因调用await方法而阻塞的线程都会被唤醒。这就是CountDownLatch的内部机制，看起来很简单，无非就是阻塞一部分线程让其在达到某个条件以后再执行。可是CountDownLatch的应用场景却比较普遍，只要你脑洞够大利用它就能够玩出各类花样。最多见的一个应用场景是开启多个线程同时执行某个任务，等到全部任务都执行完再统计汇总结果。下图动态演示了闭锁阻塞线程的整个过程。

 

 

在CountDownLatch出现以前通常都是使用线程的join（）方法来实现，可是join不够灵活，不可以知足不一样场景的需求。接下来咱们看看CountDownLatch的原理实现。

 

一.CountDownLatch原理探究

　　从CountDownLatch的名字能够猜想内部应该有个计数器，而且这个计数器是递减的，下面就经过源码看看JDK开发组是什么时候初始化计数器，什么时候递减的，计数器变为 0 的时候作了什么操做，多个线程是如何经过计时器值实现同步的，首先咱们先看看CountDownLatch内部结构，类图以下：

从类图能够知道CountDownLatch内部仍是使用AQS实现的，经过下面构造函数初始化计数器的值，可知其实是把计数器的值赋值给了AQS的state，也就是这里AQS的状态值来表示计数器值。

构造函数源码以下：

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

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

接下来主要看一下CountDownLatch中几个重要的方法内部是如何调用AQS来实现功能的。

　　1.void await()方法，当前线程调用了CountDownLatch对象的await方法后，当前线程会被阻塞，直到下面的状况之一才会返回：（1）当全部线程都调用了CountDownLatch对象的countDown方法后，

也就是说计时器值为 0 的时候。（2）其余线程调用了当前线程的interrupt（）方法中断了当前线程，当前线程会抛出InterruptedException异常后返回。接下来让咱们看看await（）方法内部是如何调用

AQS的方法的，源码以下：

//CountDownLatch的await（）方法
public void await() throws InterruptedException {
   sync.acquireSharedInterruptibly(1);
}
    //AQS的获取共享资源时候可被中断的方法
public final void acquireSharedInterruptibly(int arg)throws InterruptedException {
    //若是线程被中断则抛异常
    if (Thread.interrupted())
         throw new InterruptedException();
        //尝试看当前是否计数值为0，为0则直接返回，否者进入AQS的队列等待
    if (tryAcquireShared(arg) < 0)
         doAcquireSharedInterruptibly(arg);
}

 //sync类实现的AQS的接口
 protected int tryAcquireShared(int acquires) {
       return (getState() == 0) ? 1 : -1;
 }

　　从上面代码能够看到await（）方法委托sync调用了AQS的acquireSharedInterruptibly方法，该方法的特色是线程获取资源的时候能够被中断，而且获取到的资源是共享资源，这里为何要调用AQS的这个方法，而不是调用独占锁的accquireInterruptibly方法呢？这是由于这里状态值须要的并非非 0 即 1 的效果，而是和初始化时候指定的计数器值有关系，好比你初始化的时候计数器值为 8 ，那么state的值应该就有 0 到 8 的状态，而不是只有  0  和  1 的独占效果。

　　这里await（）方法调用acquireSharedInterruptibly的时候传递的是 1 ，就是说明要获取一个资源，而这里计数器值是资源总数，也就是意味着是让总的资源数减 1 ，acquireSharedInterruptibly内部首先判断若是当前线程被中断了则抛出异常，不然调用sync实现的tryAcquireShared方法看当前状态值(计数器值)是否为 0  ，是则当前线程的await（）方法直接返回，不然调用AQS的doAcquireSharedInterruptibly让当前线程阻塞。另外调用tryAcquireShared的方法仅仅是检查当前状态值是否是为 0 ，并无调用CAS让当前状态值减去 1 。

 

　　2.boolean await(long timeout, TimeUnit unit)，当线程调用了 CountDownLatch 对象的该方法后，当前线程会被阻塞，直到下面的状况之一发生才会返回： （1）当全部线程都调用了 CountDownLatch 对象的 countDown 方法后，也就是计时器值为 0 的时候，这时候返回 true； (2) 设置的 timeout 时间到了，由于超时而返回 false； （3）其它线程调用了当前线程的 interrupt（）方法中断了当前线程，当前线程会抛出 InterruptedException 异常后返回。源码以下：

public boolean await(long timeout, TimeUnit unit)
        throws InterruptedException {
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
}

 

　　3.void countDown() 当前线程调用了该方法后，会递减计数器的值，递减后若是计数器为 0 则会唤醒全部调用await 方法而被阻塞的线程，不然什么都不作，接下来看一下countDown（）方法内部是如何调用AQS的方法的，源码以下：

   //CountDownLatch的countDown（）方法
    public void countDown() {
       //委托sync调用AQS的方法
        sync.releaseShared(1);
    }
   //AQS的方法
    public final boolean releaseShared(int arg) {
        //调用sync实现的tryReleaseShared
        if (tryReleaseShared(arg)) {
            //AQS的释放资源方法
            doReleaseShared();
            return true;
        }
        return false;
    }

如上面代码能够知道CountDownLatch的countDown（）方法是委托sync调用了AQS的releaseShared方法，后者调用了sync 实现的AQS的tryReleaseShared，源码以下：

//syn的方法
protected boolean tryReleaseShared(int releases) {
  //循环进行cas，直到当前线程成功完成cas使计数值（状态值state）减一并更新到state
  for (;;) {
      int c = getState();

      //若是当前状态值为0则直接返回（1）
      if (c == 0)
          return false;

      //CAS设置计数值减一（2）
      int nextc = c-1;
      if (compareAndSetState(c, nextc))
          return nextc == 0;
  }
}

如上代码能够看到首先获取当前状态值(计数器值)，代码（1）若是当前状态值为 0 则直接返回 false ，则countDown（）方法直接返回；不然执行代码（2）使用CAS设置计数器减一，CAS失败则循环重试，不然若是当前计数器为 0 则返回 true 。返回 true 后，说明当前线程是最后一个调用countDown()方法的线程，那么该线程除了让计数器减一外，还须要唤醒调用CountDownLatch的await 方法而被阻塞的线程。这里的代码（1）貌似是多余的，其实否则，之因此添加代码 (1) 是为了防止计数器值为 0 后，其余线程又调用了countDown方法，若是没有代码（1），状态值就会变成负数。

 

　　4.long getCount() 获取当前计数器的值，也就是 AQS 的 state 的值，通常在 debug 测试时候使用，源码以下：

public long getCount() {
     return sync.getCount();
}

int getCount() {
     return getState();
}

如上代码可知内部仍是调用了 AQS 的 getState 方法来获取 state 的值（计数器当前值）。

 

到目前为止原理理解的差很少了，接下来用一个例子进行讲解CountDownLatch的用法，例子以下：

package com.hjc;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Created by cong on 2018/7/6.
 */
public class CountDownLatchTest {

    private static AtomicInteger id = new AtomicInteger();

    // 建立一个CountDownLatch实例，管理计数为ThreadNum
    private static volatile CountDownLatch countDownLatch = new CountDownLatch(3);

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

        Thread threadOne = new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    Thread.sleep(3000);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }

                System.out.println("【玩家" + id.getAndIncrement() + "】已入场");
                countDownLatch.countDown();
            }
        });

        Thread threadTwo = new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }

                System.out.println("【玩家" + id.getAndIncrement() + "】已入场");
                countDownLatch.countDown();

            }
        });

        Thread threadThree = new Thread(new Runnable() {

            @Override
            public void run() {
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }

                System.out.println("【玩家" + id.getAndIncrement() + "】已入场");
                countDownLatch.countDown();

            }
        });

        // 启动子线程
        threadOne.start();
        threadTwo.start();
        threadThree.start();
        System.out.println("等待斗地主玩家进场");

        // 等待子线程执行完毕，返回
        countDownLatch.await();

        System.out.println("斗地主玩家已经满人，开始发牌.....");

    }
}

运行结果以下：

 

 

如上代码，建立了一个 CountDownLatch 实例，由于有两个子线程因此构造函数参数传递为 3，主线程调用 countDownLatch.await（）方法后会被阻塞。子线程执行完毕后调用 countDownLatch.countDown() 方法让 countDownLatch 内部的计数器减一，等全部子线程执行完毕调用 countDown（）后计数器会变为 0，这时候主线程的 await（）才会返回。

 

若是把上面的代码中Thread.sleep和countDownLatch.await()的代码注释掉，运行几遍，运行结果就可能会出现以下结果，以下图：

 能够看到在注释掉latch.await()这行以后，就不能保证在全部玩家入场后才开始发牌了。

 

总结：CountDownLatch 与 join 方法的区别，一个区别是调用一个子线程的 join（）方法后，该线程会一直被阻塞直到该线程运行完毕，而 CountDownLatch 则使用计数器容许子线程运行完毕或者运行中时候递减计数，也就是 CountDownLatch 能够在子线程运行任什么时候候让 await 方法返回而不必定必须等到线程结束；另外使用线程池来管理线程时候通常都是直接添加 Runable 到线程池这时候就没有办法在调用线程的 join 方法了，countDownLatch 相比 Join 方法让咱们对线程同步有更灵活的控制。