CountDownLatch源码解析
读一些无用的书，做一些无用的事，花一些无用的时间，都是为了在一切已知之外，保留一个超越自己的机会，人生中一些很了不起的变化，就是来自这种时刻。
使用CountDownLatch可以让一个或者多个线程等待其他线程操作执行完成后再执行，这种操作的应用场景也很多，例如某个线程需要用到其他线程的执行结果，这时候可以通过CountDownLatch来做控制。这里首先来看个示例，熟悉下CountDownLatch的用法。
这里我们模拟了一个使用场景，我们有一个主线程和10个辅助线程，辅助线程会等待主线程准备就绪了，然后这10个辅助线程开始运行，等待10个辅助线程运行完毕了，主线程继续往下运行。
package test.java.util.juc.CountDownLatch;
import jdk.management.resource.internal.inst.FileOutputStreamRMHooks;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
/**
* @ClassName CountDownLatchDemo
* @Description: TODO
* @Author lirui
* @Date 2020/4/26
* @Version V1.0
**/
public class CountDownLatchDemo {
static class Work implements Runnable {
private int taskId;
private CountDownLatch countDownLatch1;
private CountDownLatch countDownLatch2;
public Work(int t, CountDownLatch start, CountDownLatch end) {
this.taskId = t;
this.countDownLatch1 = start;
this.countDownLatch2 = end;
}
@Override
public void run() {
System.out.println("taskId" + taskId);
try {
countDownLatch1.await();
System.out.println("taskId" + taskId + "开始时间来了" + System.currentTimeMillis());
} catch (InterruptedException e) {
e.printStackTrace();
}
countDownLatch2.countDown();
}
}
public static void main(String[] args) {
CountDownLatch countDownLatch1 = new CountDownLatch(1);
CountDownLatch countDownLatch2 = new CountDownLatch(10);
ExecutorService executorService = Executors.newFixedThreadPool(10);
for (int i = 1; i <= 10; i++) {
executorService.execute(new Work(i, countDownLatch1, countDownLatch2));
}
try {
Thread.sleep(1000);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("子线程等着我释放");
countDownLatch1.countDown();
try {
countDownLatch2.await();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("子线程跑完主线程开始跑起来");
}
}
// result-----------------------
taskId1
taskId2
taskId4
taskId3
taskId7
taskId8
taskId5
taskId6
taskId9
taskId10
子线程等着我释放
taskId8开始时间来了1588583281154
taskId3开始时间来了1588583281154
taskId7开始时间来了1588583281154
taskId9开始时间来了1588583281154
taskId2开始时间来了1588583281154
taskId10开始时间来了1588583281154
taskId1开始时间来了1588583281154
taskId6开始时间来了1588583281154
taskId5开始时间来了1588583281154
taskId4开始时间来了1588583281154
子线程跑完主线程开始跑起来复制代码这段代码分成两段：
第一段，10个辅助线程等待开始的信号，信号由主线程发出，所以10个辅助线程调用 countDownLatch1.await()方法等待开始信号，当主线程的事儿干完了，调用countDownLatch1.countDown()通知辅助线程开始干活。
第二段，主线程等待10个辅助线程完成的信号，信号由10个辅助线程发出，所以主线程调用countDownLatch2.await()方法等待完成信号，10个辅助线程干完自己的活儿的时候调用countDownLatch2.countDown()方法发出自己的完成的信号，当完成信号达到10个的时候，唤醒主线程继续执行后续的逻辑。
CountDownLatch的内部结构也很简单，存在一个Sync内部类，其所有的功能都是通过Sync来完成，Sync也是继承了AQS,它不像ReentrantLock,是不存在公平与非公平的概念，所以相对也简单许多。
Sync源码分析
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
//传入初始次数
Sync(int count) {
setState(count);
}
// 获取还剩的次数
int getCount() {
return getState();
}
// 尝试获取共享锁
protected int tryAcquireShared(int acquires) {
// state==0 返回1,state!=0 返回-1
return (getState() == 0) ? 1 : -1;
}
// 尝试释放锁
protected boolean tryReleaseShared(int releases) {
// 自旋操作
for (;;) {
// state值赋予c
int c = getState();
// 等于0代表无法再释放
if (c == 0)
return false;
// count值减-1
int nextc = c-1;
// 将值更新到state
if (compareAndSetState(c, nextc))
// 减为0返回true,这时会唤醒后面排队的线程
return nextc == 0;
}
}
}复制代码它的构造方法就是传入一个count值，将count值赋值给state,这个count值就代表初始次数。
它重写了AQS的两个方法，一个是tryAcquireShared，用来判断是否可以获取共享锁，只有state值等于0，代表其他线程已经执行完成，才可以获取锁。
tryReleaseShared尝试释放锁，每个线程调用countDown方法都会调用这个方法将state值减1，直到state==0,则可以唤醒阻塞的线程，所以只会有一个线程会达到state减1后等于0，去唤醒阻塞的线程。
await（）方法源码分析
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
// AQS.acquireSharedInterruptibly()
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
//当前线程是中断状态，抛出异常
if (Thread.interrupted())
throw new InterruptedException();
// 尝试获取锁,返回值大于等于0代表获取到锁
// 返回值小于0代表没有获取锁，加入到等待队列
if (tryAcquireShared(arg) < 0)
// 1.加入阻塞队列
// 2. 自旋不断尝试获取锁
// 3. 被挂起
// 4. 获取到锁之后唤醒下一个线程
doAcquireSharedInterruptibly(arg);
}
// AQS.doAcquireSharedInterruptibly
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
// 将调用latch.await()方法的线程 包装成node加入到阻塞线程队列当中
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
// 自旋，获取当前节点的前驱节点
for (; ; ) {
// 获取当前节点的前驱节点
final Node p = node.predecessor();
// 当前线程对应的结点是head.next节点
if (p == head) {
// head.next节点key尝试获取共享锁
int r = tryAcquireShared(arg);
// r>=0说明被唤醒
if (r >= 0) {
//将当前节点设置为头节点，并调用doReleaseShared唤醒下一个节点
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
// shouldParkAfterFailedAcquire:给当前节点找一个符合要求的前置节点，将结点的状态设置为signal,最终返回true
//parkAndCheckInterrupt挂起当前线程
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
if (failed)
cancelAcquire(node);
}
}复制代码await方法是等待其他线程执行完成的方法，它首先会尝试获取锁，只有state值等于0才会获取到锁所以只要state值不等于0，所有调用await方法的线程都不会拿到锁，都会加入阻塞队列。
通过addWaiter方法将线程加入阻塞队列，然后自旋尝试获取锁，获取不到，就会被挂起，直到被唤醒后，接着自旋获取锁。
如果获取到锁，会调用setHeadAndPropagate方法将自己设置为头结点，并唤醒下一个被阻塞的的节点，具体来看下这个方法。
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
// 将当前节点设置为头节点
setHead(node);
// 调用setHeadAndPropagete时 propagate==1
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
// s：当前节点的后继节点
Node s = node.next;
// s==null:说明当前节点是队尾了
// S!=NULL 但 s是共享模式
if (s == null || s.isShared())
// 基本所有情况都会执行到doReleaseShared方法
doReleaseShared();
}
}复制代码步骤也很简单，首先通过setHead方法将自己设置为头节点，然后调用doReleaseShared方法去唤醒下一个节点。
countDown（）方法源码分析
public void countDown() {
sync.releaseShared(1);
}
// AQS.releaseShared
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
// 调用countDown方法的线程只会有一个线程进入里面
doReleaseShared();
return true;
}
return false;
}
/**
* 都有哪几种路径会调用到doReleaseShared方法呢？
* 1.latch.countDown() -> AQS.state == 0 -> doReleaseShared() 唤醒当前阻塞队列内的 head.next 对应的线程。
* 2.被唤醒的线程 -> doAcquireSharedInterruptibly parkAndCheckInterrupt() 唤醒 -> setHeadAndPropagate() -> doReleaseShared()
*/
private void doReleaseShared() {
for (; ; ) {
// 获取头节点
Node h = head;
//条件一：h != null 成立，说明阻塞队列不为空..
//不成立：h == null 什么时候会是这样呢？
//latch创建出来后，没有任何线程调用过 await() 方法之前，有线程调用latch.countDown()操作 且触发了 唤醒阻塞节点的逻辑..
//条件二：h != tail 成立，说明当前阻塞队列内，除了head节点以外 还有其他节点。
//h == tail -> head 和 tail 指向的是同一个node对象。 什么时候会有这种情况呢？
//1. 正常唤醒情况下，依次获取到 共享锁，当前线程执行到这里时 （这个线程就是 tail 节点。）
//2. 第一个调用await()方法的线程 与 调用countDown()且触发唤醒阻塞节点的线程 出现并发了..
// 因为await()线程是第一个调用 latch.await()的线程，此时队列内什么也没有，它需要补充创建一个Head节点，然后再次自旋时入队
// 在await()线程入队完成之前，假设当前队列内 只有 刚刚补充创建的空元素 head 。
// 同期，外部有一个调用countDown()的线程，将state 值从1，修改为0了，那么这个线程需要做 唤醒 阻塞队列内元素的逻辑..
// 注意：调用await()的线程 因为完全入队完成之后，再次回到上层方法 doAcquireSharedInterruptibly 会进入到自旋中，
// 获取当前元素的前驱，判断自己是head.next， 所以接下来该线程又会将自己设置为 head，然后该线程就从await()方法返回了
if (h != null && h != tail) {
int ws = h.waitStatus;
if (ws == Node.SIGNAL) {
//唤醒后继节点前 将head节点的状态改为 0
//这里为什么，使用CAS呢？
//当doReleaseShared方法 存在多个线程 唤醒 head.next 逻辑时，
//CAS 可能会失败...
//案例：
//t3 线程在if(h == head) 返回false时，t3 会继续自旋. 参与到 唤醒下一个head.next的逻辑..
//t3 此时执行到 CAS WaitStatus(h,Node.SIGNAL, 0) 成功.. t4 在t3修改成功之前，也进入到 if (ws == Node.SIGNAL) 里面了，
//但是t4 修改 CAS WaitStatus(h,Node.SIGNAL, 0) 会失败，因为 t3 改过了...
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue;
// 唤醒后继节点
// 如果头节点的next节点不符合唤醒条件
// 就从队列尾节点向前遍历找到符合的节点进行唤醒
unparkSuccessor(h);
} else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
//条件成立：
//1.说明刚刚唤醒的 后继节点，还没执行到 setHeadAndPropagate方法里面的 设置当前唤醒节点为head的逻辑。
//这个时候，当前线程 直接跳出去...结束了..
//此时用不用担心，唤醒逻辑 在这里断掉呢？、
//不需要担心，因为被唤醒的线程 早晚会执行到doReleaseShared方法。
//2.h == null latch创建出来后，没有任何线程调用过 await() 方法之前，
//有线程调用latch.countDown()操作 且触发了 唤醒阻塞节点的逻辑..
//3.h == tail -> head 和 tail 指向的是同一个node对象
//条件不成立：
//被唤醒的节点 非常积极，直接将自己设置为了新的head，此时 唤醒它的节点（前驱），执行h == head 条件会不成立..
//此时 head节点的前驱，不会跳出 doReleaseShared 方法，会继续唤醒 新head 节点的后继...
if (h == head) // loop if head changed
break;
}
}
/**
* AQS.unparkSuccessor
* 唤醒该节点的下一个节点，如果下一个节点状态为Cancel，则从队尾向前遍历找到不是CANCEL的节点
*/
private void unparkSuccessor(Node node) {
/*
* If status is negative (i.e., possibly needing signal) try
* to clear in anticipation of signalling. It is OK if this
* fails or if status is changed by waiting thread.
*/
int ws = node.waitStatus;
// 小于0说明当前节点是signal状态，可以唤醒它的下一个结点
if (ws < 0)
// cas 将 waitstatus更新为0
compareAndSetWaitStatus(node, ws, 0);
// 获取当前节点的下一个节点
Node s = node.next;
//s 什么时候等于null？
//1.当前节点就是tail节点时 s == null。
//2.当新节点入队未完成时（1.设置新节点的prev 指向pred 2.cas设置新节点为tail 3.（未完成）pred.next -> 新节点 ）
//需要找到可以被唤醒的节点..
// 如果当前节点的下一个节点等于空或者状态为cancel状态
if (s == null || s.waitStatus > 0) {
s = null;
// for循环遍历找到离当前节点最近的一个结点并且不是非取消状态
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
if (s != null)
// 唤醒当前节点线程
LockSupport.unpark(s.thread);
}复制代码countDown方法是用来唤醒等待的线程，但在之前我们分析tryReleaseShared方法可知，每当调用这个方法state值会减1，只有当state值为0的时候才会唤醒等待的线程。
唤醒等待的线程调用的是AQS的doReleaseShared方法，这个方法最终会调用unparkSuccessor方法来实现对等待的线程的唤醒操作，如果头结点的next结点不符合唤醒条件，就会从队尾开始遍历，找到符合条件的结点来唤醒。
总结
CountDownLatch表示允许一个或多个线程等待其它线程的操作执行完毕后再执行后续的操作；
CountDownLatch使用AQS的共享锁机制实现；
CountDownLatch初始化的时候需要传入次数count；
每次调用countDown()方法count的次数减1；
每次调用await()方法的时候会尝试获取锁，这里的获取锁其实是检查AQS的state变量的值是否为0；
当count的值（也就是state的值）减为0的时候会唤醒排队着的线程（这些线程调用await()进入队列）；
高频问题解答
CountDownLatch的初始次数是否可以调整？
答案是不能的，它没有提供修改（增加或减少）次数的方法。
CountDownLatch为什么使用共享锁？
因为CountDownLatch的await()多个线程可以调用多次，当调用多次的时候这些线程都要进入AQS队列中排队，当count次数减为0的时候，它们都需要被唤醒，继续执行任务，如果使用互斥锁则不行，互斥锁在多个线程之间是互斥的，一次只能唤醒一个，不能保证当count减为0的时候这些调用了await()方法等待的线程都被唤醒，而共享锁当唤醒一个线程后，它会接着唤醒下一个线程，一个接着一个，保证阻塞的线程会一个接一个被全部唤醒。
CountDownLatch与Thread.join()有何不同？
Thread.join()是在主线程中调用的，它只能等待被调用的线程结束了才会通知主线程，而CountDownLatch则不同，它的countDown()方法可以在线程执行的任意时刻调用，灵活性更大。
如果你觉得文章还不错，就请给我点个赞，您的支持和鼓励是我最大的动力。喜欢就请关注我吧~
作者：卧龙小蛋
链接：https://juejin.cn/post/7015228370459820046