CountDownLatch 源码解析—— await()，具体内容如下
上一篇文章说了一下CountDownLatch的使用方法。这篇文章就从源码层面说一下await() 的原理。
我们已经知道await 能够让当前线程处于阻塞状态，直到锁存器计数为零（或者线程中断）。
下面是它的源码。
end.await();
↓public void await() throws InterruptedException {
sync.acquireSharedInterruptibly(1);
}
sync 是CountDownLatch的内部类。下面是它的定义。
private static final class Sync extends AbstractQueuedSynchronizer {
...}
它继承了AbstractQueuedSynchronizer。AbstractQueuedSynchronizer 这个类在java线程中属于一个非常重要的类。
它提供了一个框架来实现阻塞锁，以及依赖FIFO等待队列的相关同步器（比如信号、事件等）。
继续走下去，就跳到 AbstractQueuedSynchronizer 这个类中。
sync.acquireSharedInterruptibly(1);
↓public final void acquireSharedInterruptibly(int arg) //AbstractQueuedSynchronizer throws InterruptedException {
if (Thread.interrupted()) throw new InterruptedException();
if (tryAcquireShared(arg) < 0) doAcquireSharedInterruptibly(arg);
}
这里有两个判断，首先判断线程是否中断，然后再进行下一个判断，这里我们主要看看第二个判断。
protected int tryAcquireShared(int acquires) {
return (getState() == 0) ? 1 : -1;
}
需要注意的是 tryAcquireShared 这个方法是在Sync 中实现的。
AbstractQueuedSynchronizer 中虽然也有对它的实现，但是默认的实现是抛一个异常。
tryAcquireShared 这个方法是用来查询当前对象的状态是否能够被允许获取锁。
我们可以看到Sync 中是通过判断state 是否为0 来返回对应的 int 值的。
那么 state 又代表什么？
/** * The synchronization state. */ private volatile int state;
上面代码很清楚的表明 state 是表示同步的状态 。
需要注意的是 state 使用 volatile 关键字修饰。
volatile 关键字能够保证 state 的修改立即被更新到主存，当有其他线程需要读取时，会去内存中读取新值。
也就是保证了state的可见性。是最新的数据。
走到这里 state 是多少呢？
这里我们就需要看一看CountDownLatch 的 构造函数了。
CountDownLatch end = new CountDownLatch(2);
↓public CountDownLatch(int count) {
if (count < 0) throw new IllegalArgumentException("count < 0");
this.sync = new Sync(count);
}
↓Sync(int count) {
setState(count);
}
原来构造函数中的数字就是这个作用啊，用来set state 。
所以我们这里state == 2 了。tryAcquireShared 就返回 -1。进入到下面
doAcquireSharedInterruptibly(arg);
↓private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
final Node node = addWaiter(Node.SHARED);
boolean failed = true;
try {
for (;
;
) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null;
// help GC failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException();
}
}
finally {
if (failed) cancelAcquire(node);
}
}
OK，这段代码有点长，里面还调用了几个函数。我们一行一行的看。
第一行 出现了一个新的类 Node。
Node 是AQS（AbstractQueuedSynchronizer）类中的内部类，定义了一种链式结构。如下所示。
+------+ prev +-----+ +-----+head | | <---- | | <---- | | tail +------+ +-----+ +-----+
千万记住这个结构。
第一行代码中还有一个方法 addWaiter(Node.SHARED) 。
addWaiter(Node.SHARED) //Node.SHARED 表示该结点处于共享模式 ↓private Node addWaiter(Node mode) {
Node node = new Node(Thread.currentThread(), mode);
// Try the fast path of enq;
backup to full enq on failure Node pred = tail;
// private transient volatile Node tail;
if (pred != null) {
node.prev = pred;
if (compareAndSetTail(pred, node)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
首先是构造了一个Node，将当前的线程存进去了，模式是共享模式。
tail 表示 这个等待队列的队尾，此刻是null. 所以 pred == null ，进入到enq(node) ;
enq(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;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
同样tail 为 null , 进入到 compareAndSetHead 。
compareAndSetHead(new Node()) ↓/** * CAS head field. Used only by enq. */private final boolean compareAndSetHead(Node update) {
return unsafe.compareAndSwapObject(this, headOffset, null, update);
}
这是一个CAS操作，如果head 是 null 的话，等待队列的 head 就会被设置为 update 的值，也就是一个新的结点。
tail = head; 那么此时 tail 也不再是null了。进入下一次的循环。
这次首先将node 的 prev 指针指向 tail ，然后通过一个CAS 操作将node 设置为尾部，并返回了队列的 tail ，也就是 node 。
等待队列的模型变化如下
+------+ prev +----------------+head(tail) | | <---- node | currentThread | +------+ +----------------+ ↓ +------+ prev +----------------+head | | <---- node(tail) | currentThread | +------+ +----------------+
ok，到了这里await 方法 就返回了，是一个 thread 等于当前线程的Node。
返回到 doAcquireSharedInterruptibly(int arg) 中，进入下面循环。
for (;
;
) {
final Node p = node.predecessor();
if (p == head) {
int r = tryAcquireShared(arg);
if (r >= 0) {
setHeadAndPropagate(node, r);
p.next = null;
// help GC failed = false;
return;
}
}
if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) throw new InterruptedException();
}
这个时候假设state 仍然大于0，那么此时 r < 0，所以进入到 shouldParkAfterFailedAcquire 这个方法 。
shouldParkAfterFailedAcquire(p, node) ↓private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL) //static final int SIGNAL = -1;
/* * This node has already set status asking a release * to signal it, so it can safely park. */ return true;
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. */ compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
}
return false;
}
↓/** * CAS waitStatus field of a node. */private static final boolean compareAndSetWaitStatus(Node node, int expect, int update) {
return unsafe.compareAndSwapInt(node, waitStatusOffset, expect, update);
}
可以看到 shouldParkAfterFailedAcquire 也是一路走，走到 compareAndSetWaitStatus。
compareAndSetWaitStatus 将 prev 的 waitStatus 设置为 Node.SIGNAL 。
Node.SIGNAL 表示后续结点中的线程需要被unparking（类似被唤醒的意思）。该方法返回false。
经过这轮循环，队列模型变成下面状态
+--------------------------+ prev +------------------+head | waitStatus = Node.SIGNAL | <---- node(tail) | currentThread | +--------------------------+ +------------------+
因为shouldParkAfterFailedAcquire返回的是false，所以后面这个条件就不再看了。继续 for (;;) 中的循环。
如果state仍然大于0，再次进入到 shouldParkAfterFailedAcquire。
这次因为head 中的waitStatus 为 Node.SIGNAL ，所以 shouldParkAfterFailedAcquire 返回true。
这次就需要看parkAndCheckInterrupt 这个方法了。
private final boolean parkAndCheckInterrupt() {
LockSupport.park(this);
return Thread.interrupted();
}
ok，线程没有被中断，所以，返回false。继续 for (;;) 中的循环。
如果state 一直大于0，并且线程一直未被中断，那么就一直在这个循环中。也就是我们上篇文章说的裁判一直不愿意宣布比赛结束的情况。
那么什么情况下跳出循环呢？也就是什么情况下state 会 小于0呢？ 下一篇文章 我将说明。
总结一下，await() 方法 其实就是初始化一个队列，将需要等待的线程（state > 0）加入一个队列中，并用waitStatus 标记后继结点的线程状态。
以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持脚本之家。
您可能感兴趣的文章:
- 理解Koa2中的async&await的用法
- 详解ES6之async+await 同步/异步方案
- 深入理解ES7的async/await的用法
- 浅谈Async和Await如何简化异步编程（几个实例让你彻底明白）
- 关于async和await的一些误区实例详解
