java相关：CountDownLatch源码解析之await()

发布于 2020-6-18|

复制链接

摘记: CountDownLatch 源码解析—— await()，具体内容如下上一篇文章说了一下CountDownLatch的使用方法。这篇文章就从源码层面说一下await() 的原理。我们已经知道awa ..

CountDownLatch 源码解析—— await()，具体内容如下上一篇文章说了一下CountDownLatch的使用方法。这篇文章就从源码层面说一下await() 的原理。我们已经知道await 能够让当前线程处于阻塞状态，直到锁存器计数为零(或者线程中断)。下面是它的源码。

```java

end.await();

↓

public void await() throws InterruptedException {

sync.acquireSharedInterruptibly(1);

}

```

sync 是CountDownLatch的内部类。下面是它的定义。

```java

private static final class Sync extends AbstractQueuedSynchronizer {

...

}

```

它继承了AbstractQueuedSynchronizer。AbstractQueuedSynchronizer 这个类在java线程中属于一个非常重要的类。它提供了一个框架来实现阻塞锁，以及依赖FIFO等待队列的相关同步器(比如信号、事件等)。继续走下去，就跳到 AbstractQueuedSynchronizer 这个类中。

```java

sync.acquireSharedInterruptibly(1);

↓

public final void acquireSharedInterruptibly(int arg) //AbstractQueuedSynchronizer

throws InterruptedException {

if (Thread.interrupted())

throw new InterruptedException();

if (tryAcquireShared(arg) 这里有两个判断，首先判断线程是否中断，然后再进行下一个判断，这里我们主要看看第二个判断。

```java

protected int tryAcquireShared(int acquires) {

return (getState() == 0) ? 1 : -1;

}

```

需要注意的是 tryAcquireShared 这个方法是在Sync 中实现的。AbstractQueuedSynchronizer 中虽然也有对它的实现，但是默认的实现是抛一个异常。tryAcquireShared 这个方法是用来查询当前对象的状态是否能够被允许获取锁。我们可以看到Sync 中是通过判断state 是否为0 来返回对应的 int 值的。那么 state 又代表什么？

```java

/**

* The synchronization state.

*/

private volatile int state;

```

上面代码很清楚的表明 state 是表示同步的状态 。需要注意的是 state 使用 volatile 关键字修饰。volatile 关键字能够保证 state 的修改立即被更新到主存，当有其他线程需要读取时，会去内存中读取新值。也就是保证了state的可见性。是最新的数据。走到这里 state 是多少呢？这里我们就需要看一看CountDownLatch 的 构造函数了。

```java

CountDownLatch end = new CountDownLatch(2);

↓

public CountDownLatch(int count) {

if (count 原来构造函数中的数字就是这个作用啊，用来set state 。所以我们这里state == 2 了。tryAcquireShared 就返回 -1。进入到下面

```java

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)类中的内部类，定义了一种链式结构。如下所示。

```java

+------+ prev +-----+ +-----+

head | | 千万记住这个结构。第一行代码中还有一个方法 addWaiter(Node.SHARED) 。

```java

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) ;

```java

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 。

```java

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 。等待队列的模型变化如下

```java

+------+ prev +----------------+

head(tail) | | ok，到了这里await 方法 就返回了，是一个 thread 等于当前线程的Node。返回到 doAcquireSharedInterruptibly(int arg) 中，进入下面循环。

```java

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

```java

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。经过这轮循环，队列模型变成下面状态

```java

+--------------------------+ prev +------------------+

head | waitStatus = Node.SIGNAL | 因为shouldParkAfterFailedAcquire返回的是false，所以后面这个条件就不再看了。继续 for (;;)  中的循环。如果state仍然大于0，再次进入到 shouldParkAfterFailedAcquire。这次因为head 中的waitStatus 为 Node.SIGNAL ，所以 shouldParkAfterFailedAcquire 返回true。这次就需要看parkAndCheckInterrupt 这个方法了。

```java

private final boolean parkAndCheckInterrupt() {

LockSupport.park(this);

return Thread.interrupted();

}

```