CountDownLatch源码探究 (JDK 1.8)
纳兰小依 人气:0CountDownLatch能够实现让线程等待某个计数器倒数到零的功能，之前对它的了解也仅仅是简单的使用，对于其内部如何实现线程等待却不是很了解，最好的办法就是通过看源码来了解底层的实现细节。CountDownLatch的源码并不是很复杂，因为其核心的功能是依赖AbstractQueuedSynchronizer（下文简称AQS）来实现的。CountDownLatch常用的方法很少，但是因为涉及到AQS，逻辑有些绕，要理清中间的逻辑稍微要费一些时间。
1.内部类Sync
CountDownLatch的核心功能是通过内部类Sync实现的，这个类继承了AQS：
private static final class Sync extends AbstractQueuedSynchronizer {
private static final long serialVersionUID = 4982264981922014374L;
//构造器，根据传入的整数初始化状态字段state
Sync(int count) {
setState(count);
}
int getCount() {
return getState();
}
//tryAcquireShared唯一的作用是查看状态字段是不是等于0
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
protected boolean tryReleaseShared(int releases) {
// Decrement count; signal when transition to zero
//自旋，在两种条件下会退出自旋：a)state字段已经为0；b)线程成功地将state字段减1
for (;;) {
int c = getState();
//如果state已经为0，就返回false
if (c == 0)
return false;
int nextc = c-1;
//从下面的语句可以看到，只有当state=0才会返回true
if (compareAndSetState(c, nextc))
return nextc == 0;
}
}
}2.构造器
CountDownLatch只有一个构造器，在构造器中会初始化sync字段，结合Sync类的定义可知，构造器的唯一工作是将state字段初始化为传入的参数：
public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}3.节点状态waitStatus
等待的线程会构造成节点放在等待队列中，节点的状态waitStatus有如下几种：
/** waitStatus value to indicate thread has cancelled */
static final int CANCELLED = 1;
/** waitStatus value to indicate successor's thread needs unparking */
static final int SIGNAL = -1;
/** waitStatus value to indicate thread is waiting on condition */
static final int CONDITION = -2;
/**
* waitStatus value to indicate the next acquireShared should
* unconditionally propagate
*/
static final int PROPAGATE = -3;注意，在CountDownLatch中并没有用到CONDITION状态，因此后文将会直接忽略该状态，当waitStatus > 0时，指的就是CANCELLED状态。
4.核心方法
await()
当计数器没不等于0时，await()方法会让当前线程挂起，该方法调用了AQS类的acquireSharedInterruptibly方法，如下：
public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
//显然，tryAcquireShared方法只有在state=0时才返回1，表示计数器已归零，此时方法直接返回，被阻塞的线程就可以继续执行
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}通常，调用await()的线程在执行到acquireSharedInterruptibly方法时，计数器并不为0，那么当前线程就需要执行doAcquireSharedInterruptibly方法中的阻塞逻辑了。由于该方法内部调用了三个主要方法：addWaiter、shouldParkAfterFailedAcquire和parkAndCheckInterrupt，在解析的过程中难免会穿插对这些方法的介绍，从而引入跳跃性。为了避免跳跃性引发的阅读和理解上的困难，这里准备先介绍addWaiter方法。
addWaiter
private Node addWaiter(Node mode) {
//将当前线程构造成一个Node节点
Node node = new Node(Thread.currentThread(), mode);
//获取尾节点
Node pred = tail;
//尾节点不为空，说明队列已完成初始化
if (pred != null) {
//将node节点放到对尾，这里的做法是先将node的prev指针指向尾节点，然后通过原子操作将新添加的node更新成尾节点，成功的话addWaiter方法结束
node.prev = pred;
if (compareAndSetTail(pred, node)) {
//原子操作成功的话，更新原尾节点的next指针
pred.next = node;
return node;
}
}
//执行到这里有两种情况：1)尾节点为空，即队列还没初始化；2)队列已初始化，但是上文将node节点设置成尾节点失败，此时node节点还没有真正添加进队列
enq(node);
return node;
}
private Node enq(final Node node) {
for (;;) {
Node t = tail;
//如果队列还没初始化，则先初始化，做法是将一个空节点作为头结点，然后让尾节点也指向这个空节点
if (t == null) { // Must initialize
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
//这里会一直自旋，直到成功地将node节点更新成尾节点
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
addWaiter方法的主要作用就是将当前线程添加到等待队列的队尾，如果队列还没初始化，则先初始化，enq方法使用自旋避免入队失败。
doAcquireSharedInterruptibly
接下来正式开始介绍doAcquireSharedInterruptibly方法，源码如下：
private void doAcquireSharedInterruptibly(int arg)
throws InterruptedException {
//将当前线程添加到等待队列，注意参数是Node.SHARED，下文会用到
final Node node = addWaiter(Node.SHARED);
//该字段在state=0时才会被设置为false
boolean failed = true;
try {
//又是自旋，该自旋的终止条件有两种：1)state=0，计数器正常结束，执行return语句返回；2)线程响应中断异常，跳出自旋
for (;;) {
//获取node的前驱节点
final Node p = node.predecessor();
//如果前驱节点是头结点，则执行if代码块的逻辑
if (p == head) {
//获取state字段的状态，如果state=0则返回1，否则返回-1
int r = tryAcquireShared(arg);
//r>=0，说明计数器结束了，需要唤醒阻塞的线程
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
//计数器正常结束时，会将failed设置为false，避免执行finally中的语句
failed = false;
return;
}
}
//执行到这里说明state!=0，真正的阻塞逻辑在parkAndCheckInterrupt方法里
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}
} finally {
//如果线程被中断，那么failed=true，执行cancelAcquire方法
if (failed)
cancelAcquire(node);
}
}doAcquireSharedInterruptibly先通过addWaiter方法将当前线程添加到等待队列尾部，然后开始自旋。如果state字段不为0，那么会执行到末尾的条件语句：
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();先来看看shouldParkAfterFailedAcquire干了些什么：
//注意pred是node的前驱节点
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
//如果已经是SIGNAL状态，则之间返回true
if (ws == Node.SIGNAL)
/*
* This node has already set status asking a release
* to signal it, so it can safely park.
*/
return true;
//ws>0只能是cancelled状态，此时通过修改指针将这些cancelled的节点从队列删除
if (ws > 0) {
/*
* Predecessor was cancelled. Skip over predecessors and
* indicate retry.
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* waitStatus must be 0 or PROPAGATE. Indicate that we
* need a signal, but don't park yet. Caller will need to
* retry to make sure it cannot acquire before parking.
*/
//如果前驱节点的状态既不是SIGNAL，也不是CANCELLED，那么只可能是0或者PROPAGATE，就把前驱节点的状态更新为 Node.SIGNAL。注意：1)CONDITION状态在CountDownLatch中并没有用到；2)节点新建的时候状态都是0，是在这里才被修改成了SIGNAL
compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}之前对节点的SIGNAL状态是怎么来的一直有点迷糊，看了上面的代码才发现是在最后一个else分支中设置的。从shouldParkAfterFailedAcquire源码了解到，该方法只有在前驱节点状态是SIGNAL时才返回true，此时才有机会执行parkAndCheckInterrupt方法。parkAndCheckInterrupt是真正让线程挂起的地方，来看看其源码：
private final boolean parkAndCheckInterrupt() {
//线程最终会阻塞在这里，线程恢复之后也将从这里继续执行
LockSupport.park(this);
return Thread.interrupted();
}parkAndCheckInterrupt方法借助LockSupport实现线程阻塞，被阻塞的线程在被唤醒后会返回当前线程的中断状态（注意Thread.interrupted()会清除线程的中断状态）。好了，到这里整个逻辑就比较清楚了，如果线程是正常被唤醒（即state=0），那么parkAndCheckInterrupt返回false，doAcquireSharedInterruptibly方法会接着自旋一次，这里再次将自旋代码贴出：
for (;;) {
//获取node的前驱节点
final Node p = node.predecessor();
//如果前驱节点是头结点，则执行if代码块的逻辑
if (p == head) {
//获取state字段的状态，如果state=0则返回1，否则返回-1
int r = tryAcquireShared(arg);
//r>=0，说明计数器结束了，需要唤醒阻塞的线程
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null; // help GC
failed = false;
return;
}
}
//执行到这里说明state!=0，真正的阻塞逻辑在parkAndCheckInterrupt方法里
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt())
throw new InterruptedException();
}那么setHeadAndPropagate方法做了些什么事呢，看看它的源码（删掉了源码中的注释）：
//回忆一下，显然propagate=1，node是当前插入到对尾的新节点
private void setHeadAndPropagate(Node node, int propagate) {
Node h = head; // Record old head for check below
//把node设置为头结点
setHead(node);
//此时propagate > 0的条件已经满足，直接执行if代码块的逻辑
if (propagate > 0 || h == null || h.waitStatus < 0 ||
(h = head) == null || h.waitStatus < 0) {
Node s = node.next;
//如果没有下一个节点，或者下一个节点的isShared返回true，就释放。还记得吗，在构造新节点的时候addWaiter的参数是Node.SHARED，这里就是判断这个字段
if (s == null || s.isShared())
doReleaseShared();
}
}
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}接下来看一下doReleaseShared是如何实现的：
private void doReleaseShared() {
/*
* Ensure that a release propagates, even if there are other
* in-progress acquires/releases. This proceeds in the usual
* way of trying to unparkSuccessor of head if it needs
* signal. But if it does not, status is set to PROPAGATE to
* ensure that upon release, propagation continues.
* Additionally, we must loop in case a new node is added
* while we are doing this. Also, unlike other uses of
* unparkSuccessor, we need to know if CAS to reset status
* fails, if so rechecking.
*/
for (;;) {
//获取头结点
Node h = head;
if (h != null && h != tail) {
int ws = h.waitStatus;
//如果头结点的状态是SIGNAL，那么会将其状态修改为0，该步骤一直自旋直到成功为止
if (ws == Node.SIGNAL) {
if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
continue; // loop to recheck cases
//成功修改头结点的状态后，会执行下面这个方法
unparkSuccessor(h);
}
//如果头结点状态已经改成0了，就再次将其状态更新为Node.PROPAGATE，目的是？？？
else if (ws == 0 &&
!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
continue; // loop on failed CAS
}
if (h == head) // loop if head changed
break;
}
}头结点的状态成功更新为0后，会执行unparkSuccessor方法的逻辑，该方法源码如下：
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;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
/*
* Thread to unpark is held in successor, which is normally
* just the next node. But if cancelled or apparently null,
* traverse backwards from tail to find the actual
* non-cancelled successor.
*/
//获取后继节点
Node s = node.next;
//如果没有后继节点，或者后继节点是CANCELLED状态，则执行下面的代码块
if (s == null || s.waitStatus > 0) {
s = null;
//从队列末尾向开头遍历，找到靠近头结点的第一个不为CANCELLED状态的节点
for (Node t = tail; t != null && t != node; t = t.prev)
if (t.waitStatus <= 0)
s = t;
}
//找到这样的非CANCELLED节点，就将其唤醒
if (s != null)
LockSupport.unpark(s.thread);
}unparkSuccessor的主要工作是将头结点后面第一个非CANCELLED状态的节点所对应的线程唤醒。
cancelAcquire
到目前为止，并没有发现CANCELLED状态是在哪里设置，因为还有一个方法没有分析。doAcquireSharedInterruptibly中的finally语句块会处理线程被中断的情况，执行的是cancelAcquire方法的逻辑，其源码如下：
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
if (node == null)
return;
//线程中断后，将其对应的节点中保存的线程清空
node.thread = null;
// Skip cancelled predecessors
//从队列中删除状态为CANCELLED的节点
Node pred = node.prev;
while (pred.waitStatus > 0)
node.prev = pred = pred.prev;
// predNext is the apparent node to unsplice. CASes below will
// fail if not, in which case, we lost race vs another cancel
// or signal, so no further action is necessary.
Node predNext = pred.next;
// Can use unconditional write instead of CAS here.
// After this atomic step, other Nodes can skip past us.
// Before, we are free of interference from other threads.
//CANCELLED状态在这里设置
node.waitStatus = Node.CANCELLED;
// If we are the tail, remove ourselves.
//如果当前是尾节点，其第一个非CANCELLED状态的前驱节点设置为新的尾节点，pred后面的节点将会被GC回收。注意，下面的两个原子操作，不管是否成功，都没有重试
if (node == tail && compareAndSetTail(node, pred)) {
compareAndSetNext(pred, predNext, null);
} else {
// If successor needs signal, try to set pred's next-link
// so it will get one. Otherwise wake it up to propagate.
int ws;
if (pred != head &&
((ws = pred.waitStatus) == Node.SIGNAL ||
(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
pred.thread != null) {
Node next = node.next;
//当前线程对应的节点不是尾节点，其有后继节点并且后继节点不是CANCELLED状态，通过修改指针将当前线程节点从队列删除
if (next != null && next.waitStatus <= 0)
compareAndSetNext(pred, predNext, next);
} else {
//根据前面的if条件，在以下几种情况时会执行到这里，唤醒node节点的后继节点
//1)pred=head，即当前被中断的线程前面的所有线程都是CANCELLED状态
//2)pred!=head，但是pred节点的状态不等于SIGNAL，且将pred节点的状态修改为SIGNAL失败
//3)pred节点记录的线程是null，目前已知头结点的thread字段确实为null，除此之外还有其他情况吗？？？
unparkSuccessor(node);
}
node.next = node; // help GC
}
}分析到这里，才刚把await()的逻辑分析完，但是仅仅分析代码仍然是不够的，因为本人分析到这里的时候，脑袋仍然是蒙的，主要原因是缺少一个全局的认识。代码放在这里都能看懂，但是代码为什么这样写？当计数器结束（即state=0）时，队列中的等待线程是一起全部换新，还是一个一个依次唤醒？线程被唤醒后重新执行doAcquireSharedInterruptibly中的自旋时，和第一次执行到底有哪些地方不一样呢？因此，有必要对以上的逻辑进行整体梳理。
看完这部分源码之后，发现核心的逻辑都包含在doAcquireSharedInterruptibly中，现在是时候回过头来整理一下该方法的逻辑了。
假设有现在有一个线程t1执行了await方法，由于等待队列还没初始化，因此先构造一个空的头节点，并且把t1构造成节点加到队列中，如下图：
接着，在shouldParkAfterFailedAcquire方法中修改头结点的状态：
现在又有新的t2线程执行了await，此时队列的结构将更新为下图：
即每添加一个节点到等待队尾，就将其前驱节点的状态更新为Node.SIGNAL(即-1)，然后所有的线程都阻塞在parkAndCheckInterrupt方法里。现在，计数器已经结束，最后一个执行countDown方法的线程顺带执行了doReleaseShared方法，将头结点的waitStatus更新成了0，如下图：
继续向下执行到unparkSuccessor方法，唤醒线程t1，t1从parkAndCheckInterrupt方法中醒来，继续自旋。t1的前置节点就是头结点head，且state=0，t1开始执行setHeadAndPropagate，将自己设置为头结点，并在setHead方法中将thread和prev字段都设置为空，如下图：
线程t1接着执行doReleaseShared方法，把头节点（此时t1就是头结点）状态更新为0，并唤醒t2，开始执行await之后的逻辑，如下图：
唤醒t2后，t1退出await方法，此时队列如下：
t2开始执行后，同样把自己设置为头结点，如下：
在执行setHeadAndPropagate方法时，t2没有后继节点了，仍然会执行doReleaseShared方法，但是在doReleaseShared方法中，t2即使头结点也是尾节点，那就什么也不做，直接结束并退出await方法，此时队列里就只剩下一个头结点了。
countDown
现在，终于可以开始看看countDown方法的逻辑了：
public void countDown() {
sync.releaseShared(1);
}
public final boolean releaseShared(int arg) {
//之前分析过，该方法会将state的值减1，并且只有在减1后state=0才会返回true，表示计数器结束了
if (tryReleaseShared(arg)) {
//唤醒后继节点中第一个不为CANCELLED状态的节点
doReleaseShared();
return true;
}
return false;
}当一个线程将state修改成0时，顺便还要执行doReleaseShared方法，这个方法会将头结点的后继节点唤醒。
有一个小细节需要注意，doReleaseShared方法在源码中有两个地方调用，一个入口就是刚讲的countDown方法，另一个就是从await方法进入，在setHeadAndPropagate中调用，但是二者是有先后顺序的是，是countDown方法唤醒最前面的线程之后，再由该线程依次唤醒后面的线程。
加载全部内容