Java并发系列[7]----CountDownLatch源码分析

2018-06-18 02:55:35来源：未知阅读 ()

CountDownLatch(闭锁)是一个很有用的工具类，利用它我们可以拦截一个或多个线程使其在某个条件成熟后再执行。它的内部提供了一个计数器，在构造闭锁时必须指定计数器的初始值，且计数器的初始值必须大于0。另外它还提供了一个countDown方法来操作计数器的值，每调用一次countDown方法计数器都会减1，直到计数器的值减为0时就代表条件已成熟，所有因调用await方法而阻塞的线程都会被唤醒。这就是CountDownLatch的内部机制，看起来很简单，无非就是阻塞一部分线程让其在达到某个条件之后再执行。但是CountDownLatch的应用场景却比较广泛，只要你脑洞够大利用它就可以玩出各种花样。最常见的一个应用场景是开启多个线程同时执行某个任务，等到所有任务都执行完再统计汇总结果。下图动态演示了闭锁阻塞线程的整个过程。

上图演示了有5个线程因调用await方法而被阻塞，它们需要等待计数器的值减为0才能继续执行。计数器的初始值在构造闭锁时被指定，后面随着每次countDown方法的调用而减1。下面代码贴出了CountDownLatch的构造方法。

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

CountDownLatch只有一个带参构造器，必须传入一个大于0的值作为计数器初始值，否则会报错。可以看到在构造方法中只是去new了一个Sync对象并赋值给成员变量sync。和其他同步工具类一样，CountDownLatch的实现依赖于AQS，它是AQS共享模式下的一个应用。CountDownLatch实现了一个内部类Sync并用它去继承AQS，这样就能使用AQS提供的大部分方法了。下面我们就来看一下Sync内部类的代码。

 1 //同步器
 2 private static final class Sync extends AbstractQueuedSynchronizer {
 3 
 4     //构造器
 5     Sync(int count) {
 6         setState(count);
 7     }
 8 
 9     //获取当前同步状态
10     int getCount() {
11         return getState();
12     }
13 
14     //尝试获取锁
15     //返回负数：表示当前线程获取失败
16     //返回零值：表示当前线程获取成功, 但是后继线程不能再获取了
17     //返回正数：表示当前线程获取成功, 并且后继线程同样可以获取成功
18     protected int tryAcquireShared(int acquires) {
19         return (getState() == 0) ? 1 : -1;
20     }
21 
22     //尝试释放锁
23     protected boolean tryReleaseShared(int releases) {
24         for (;;) {
25             //获取同步状态
26             int c = getState();
27             //如果同步状态为0, 则不能再释放了
28             if (c == 0) {
29                 return false;
30             }
31             //否则的话就将同步状态减1
32             int nextc = c-1;
33             //使用CAS方式更新同步状态
34             if (compareAndSetState(c, nextc)) {
35                 return nextc == 0;
36             }
37         }
38     }
39 }

可以看到Sync的构造方法会将同步状态的值设置为传入的参数值。之后每次调用countDown方法都会将同步状态的值减1，这也就是计数器的实现原理。在平时使用CountDownLatch工具类时最常用的两个方法就是await方法和countDown方法。调用await方法会阻塞当前线程直到计数器为0，调用countDown方法会将计数器的值减1直到减为0。下面我们来看一下await方法是怎样调用的。

 1 //导致当前线程等待, 直到门闩减少到0, 或者线程被打断
 2 public void await() throws InterruptedException {
 3     //以响应线程中断方式获取
 4     sync.acquireSharedInterruptibly(1);
 5 }
 6 
 7 //以可中断模式获取锁(共享模式)
 8 public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
 9     //首先判断线程是否中断, 如果是则抛出异常
10     if (Thread.interrupted()) {
11         throw new InterruptedException();
12     }
13     //1.尝试去获取锁
14     if (tryAcquireShared(arg) < 0) {
15         //2. 如果获取失败则进人该方法
16         doAcquireSharedInterruptibly(arg);
17     }
18 }

当线程调用await方法时其实是调用到了AQS的acquireSharedInterruptibly方法，该方法是以响应线程中断的方式来获取锁的，上面同样贴出了该方法的代码。我们可以看到在acquireSharedInterruptibly方法首先会去调用tryAcquireShared方法尝试获取锁。我们看到Sync里面重写的tryAcquireShared方法的逻辑，方法的实现逻辑很简单，就是判断当前同步状态是否为0，如果为0则返回1表明可以获取锁，否则返回-1表示不能获取锁。如果tryAcquireShared方法返回1则线程能够不必等待而继续执行，如果返回-1那么后续就会去调用doAcquireSharedInterruptibly方法让线程进入到同步队列里面等待。这就是调用await方法会阻塞当前线程的原理，下面看看countDown方法是怎样将阻塞的线程唤醒的。

 1 //减少门闩的方法
 2 public void countDown() {
 3     sync.releaseShared(1);
 4 }
 5 
 6 //释放锁的操作(共享模式)
 7 public final boolean releaseShared(int arg) {
 8     //1.尝试去释放锁
 9     if (tryReleaseShared(arg)) {
10         //2.如果释放成功就唤醒其他线程
11         doReleaseShared();
12         return true;
13     }
14     return false;
15 }

可以看到countDown方法里面调用了releaseShared方法，该方法同样是AQS里面的方法，我们在上面也贴出了它的代码。releaseShared方法里面首先是调用tryReleaseShared方法尝试释放锁，tryReleaseShared方法在AQS里面是一个抽象方法，它的具体实现逻辑在子类Sync类里面，我们在上面贴出的Sync类代码里可以找到该方法。tryReleaseShared方法如果返回true表示释放成功，返回false表示释放失败，只有当将同步状态减1后该同步状态恰好为0时才会返回true，其他情况都是返回false。那么当tryReleaseShared返回true之后就会马上调用doReleaseShared方法去唤醒同步队列的所有线程。这样就解释了为什么最后一次调用countDown方法将计数器减为0后就会唤醒所有被阻塞的线程。CountDownLatch基本的原理大致就是这些，下面我们看一个它的使用示例。

应用场景：在玩欢乐斗地主时必须等待三个玩家都到齐才可以进行发牌。

 1 public class Player extends Thread {
 2     
 3     private static int count = 1;
 4     private final int id = count++;
 5     private CountDownLatch latch;
 6     
 7     public Player(CountDownLatch latch) {
 8         this.latch = latch;
 9     }
10 
11     @Override
12     public void run() {
13         System.out.println("【玩家" + id + "】已入场");
14         latch.countDown();
15     }
16     
17     public static void main(String[] args) throws InterruptedException {
18         CountDownLatch latch = new CountDownLatch(3);
19         System.out.println("牌局开始, 等待玩家入场...");
20         new Player(latch).start();
21         new Player(latch).start();
22         new Player(latch).start();
23         latch.await();
24         System.out.println("玩家已到齐, 开始发牌...");
25     }
26     
27 }