Java 并发编程CountDownLatch的应用与源码解析

应用场景

CountDownLatch是一个多线程控制工具。用来控制线程的等待。

设置需要countDown的数量，然后每一个线程执行完毕后调用countDown()方法，而在主线程中调用await()方法等待，直到num个子线程执行了countDown()方法，则主线程开始继续执行

示例：

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;public class CountDownLatchDemo {private static final int nums = 5;public static CountDownLatch countDownLatch = new CountDownLatch(nums);public static void main(String[] args) throws Throwable {ExecutorService es = Executors.newFixedThreadPool(5);int i = 0;while (i < 5) {es.submit(new CountDownLatchRunning(countDownLatch, i));i++;}countDownLatch.await();System.out.println("任务全部执行完毕！");es.shutdown();}
}class CountDownLatchRunning implements Runnable {private int i = 0;private CountDownLatch countDownLatch;public CountDownLatchRunning(CountDownLatch countDownLatch, int i) {this.i = i;this.countDownLatch = countDownLatch;}@Overridepublic void run() {try {Thread.sleep(1000); // 任务执行了1秒if (i == 3) {throw new RuntimeException();}System.out.println(Thread.currentThread().getName() + ": 任务执行完毕！");} catch (InterruptedException e) {e.printStackTrace();} finally {countDownLatch.countDown();}}
}

main主线程会一致阻塞等待10条CountDownLatchRunning线程全部执行完成，当countDown()被调用10次之后，mian主线程继续执行

源码解析

CountDownLatch主要是两个方法：await()、countDown()，还有一个构造方法 CountDownLatch(int count)

构造方法：CountDownLatch(int count)

    public CountDownLatch(int count) {if (count < 0) throw new IllegalArgumentException("count < 0");// 实例化一个Sync对象this.sync = new Sync(count);}

通过构造方法去设置AQS state的初始值为count，Sync是在CountDownLatch类中实现的AQS实现类

阻塞方法：await()

在例子中也有提到，await()会阻塞主线程，直到所有线程都countDown()数量等于count，也就是state==0

    public void await() throws InterruptedException {// 共享式获取AQS的同步状态sync.acquireSharedInterruptibly(1);}

调用的是AQS的acquireSharedInterruptibly方法：

 public final void acquireSharedInterruptibly(int arg)throws InterruptedException {if (Thread.interrupted()) // 线程中断 说明闭锁对线程中断敏感throw new InterruptedException();if (tryAcquireShared(arg) < 0) // 闭锁未使用完成 线程进入同步队列自旋等待 doAcquireSharedInterruptibly(arg);}

其中tryAcquireShared依赖于Sync的实现：

    /** 获取共享锁 */protected int tryAcquireShared(int acquires) {// AQS的同步状态为0则闭锁结束 可以进行下一步操作return (getState() == 0) ? 1 : -1;}

也就是需要当getState() == 0的时候，才可以进行继续执行，否则线程进入同步队列自旋等待（AQS同步队列的自旋等待）

计数方法：countDown()

调用countDown()方法会将计数器减1，直到计数器为0，代码如下：

public void countDown() {sync.releaseShared(1);
}

同样调用的是AQS的releaseShared方法：

 public final boolean releaseShared(int arg) {if (tryReleaseShared(arg)) { // 减少闭锁的计数器，只有计数器为0的时候才会返回truedoReleaseShared(); // 唤醒被await方法阻塞的所有线程return true;}return false;}

这里的返回这对CountDownLatch没有用，其中tryReleaseShared方法依赖的是Sync的实现：

        /** 释放共享锁 */protected boolean tryReleaseShared(int releases) {// 死循环，如果CAS操作失败就会不断继续尝试for (;;) {int c = getState();if (c == 0) // 正常不会进入此逻辑return false;int nextc = c-1; // 将计数器-1if (compareAndSetState(c, nextc)) // 更新计数器// 如果操作成功，返回计数器是否为0，直接关系到是否执行doReleaseShared方法来唤醒后续线程return nextc == 0;}}

可以看到，只有当计数器等于0的时候才会返回true，才会唤醒后续线程（调用await()自旋等待的线程）

带有超时等待的方法：await(long timeout, TimeUnit unit)

和await()类似，只是加入了超时检测，在自旋等待的过程中会去检查是否超时，超时则结束

总结

CountDownLatch类使用AQS同步器来实现计数。

在await()的时候，调用await()的线程将进入自旋等待，自旋过程会调用AQS的Sync实现的tryAcquireShared方法，只有当闭锁计数器等于0的时候，线程才能够继续执行
在其它线程调用countDown的方法时候，会将闭锁计数器减1（state-1），直至计数器等于0的时候，会唤醒后续线程获取闭锁（doReleaseShared），自旋等待中的线程才可以继续执行