CountDownLatch 源码解析—— await()

上一篇文章说了一下CountDownLatch的使用方法。这篇文章就从源码层面说一下await() 的原理。

我们已经知道await 能够让当前线程处于阻塞状态，直到锁存器计数为零（或者线程中断）。

下面是它的源码。

1end.await();  
2    ↓
3public void await() throws InterruptedException {
4    sync.acquireSharedInterruptibly(1);
5}

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

1private static final class Sync extends AbstractQueuedSynchronizer {
2　　...
3}

它继承了AbstractQueuedSynchronizer。AbstractQueuedSynchronizer 这个类在java线程中属于一个非常重要的类。

它提供了一个框架来实现阻塞锁，以及依赖FIFO等待队列的相关同步器（比如信号、事件等）。

继续走下去，就跳到 AbstractQueuedSynchronizer 这个类中。

1sync.acquireSharedInterruptibly(1);  
2    ↓
3public final void acquireSharedInterruptibly(int arg)  //AbstractQueuedSynchronizer
4            throws InterruptedException {
5    if (Thread.interrupted())
6        throw new InterruptedException();
7    if (tryAcquireShared(arg) < 0)
8        doAcquireSharedInterruptibly(arg);
9}

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

1protected int tryAcquireShared(int acquires) {
2    return (getState() == 0) ? 1 : -1;
3}

需要注意的是 tryAcquireShared 这个方法是在Sync 中实现的。

AbstractQueuedSynchronizer 中虽然也有对它的实现，但是默认的实现是抛一个异常。

tryAcquireShared 这个方法是用来查询当前对象的状态是否能够被允许获取锁。

我们可以看到Sync 中是通过判断state 是否为0 来返回对应的 int 值的。

那么 state 又代表什么？

1/**
2 * The synchronization state.
3 */
4   private volatile int state;

上面代码很清楚的表明 state 是表示同步的状态。

需要注意的是 state 使用 volatile 关键字修饰。

volatile 关键字能够保证 state 的修改立即被更新到主存，当有其他线程需要读取时，会去内存中读取新值。

也就是保证了state的可见性。是最新的数据。

走到这里 state 是多少呢？

这里我们就需要看一看CountDownLatch 的构造函数了。

1CountDownLatch end = new CountDownLatch(2);
2    ↓
3public CountDownLatch(int count) {
4    if (count < 0) throw new IllegalArgumentException("count < 0");
5    this.sync = new Sync(count);
6}
7    ↓
8Sync(int count) {
9    setState(count);
10}

原来构造函数中的数字就是这个作用啊，用来set state 。

所以我们这里state == 2 了。tryAcquireShared 就返回 -1。进入到下面

1doAcquireSharedInterruptibly(arg);
2    ↓
3private void doAcquireSharedInterruptibly(int arg)
4        throws InterruptedException {
5        final Node node = addWaiter(Node.SHARED);
6        boolean failed = true;
7        try {
8            for (;;) {
9                final Node p = node.predecessor();
10                if (p == head) {
11                    int r = tryAcquireShared(arg);
12                    if (r >= 0) {
13                        setHeadAndPropagate(node, r);
14                        p.next = null; // help GC
15                        failed = false;
16                        return;
17                    }
18                }
19                if (shouldParkAfterFailedAcquire(p, node) &&
20                    parkAndCheckInterrupt())
21                    throw new InterruptedException();
22            }
23        } finally {
24            if (failed)
25                cancelAcquire(node);
26        }
27    }

OK，这段代码有点长，里面还调用了几个函数。我们一行一行的看。

第一行出现了一个新的类 Node。

Node 是AQS（AbstractQueuedSynchronizer）类中的内部类，定义了一种链式结构。如下所示。

1+------+  prev +-----+       +-----+
2head |      | <---- |     | <---- |     |  tail
3     +------+       +-----+       +-----+

千万记住这个结构。

第一行代码中还有一个方法 addWaiter(Node.SHARED) 。

1addWaiter(Node.SHARED)  //Node.SHARED  表示该结点处于共享模式
2    ↓
3private Node addWaiter(Node mode) {
4    Node node = new Node(Thread.currentThread(), mode);
5    // Try the fast path of enq; backup to full enq on failure
6    Node pred = tail; // private transient volatile Node tail;
7    if (pred != null) {
8        node.prev = pred;
9        if (compareAndSetTail(pred, node)) {
10            pred.next = node;
11            return node;
12        }
13    }
14    enq(node);
15    return node;
16}

首先是构造了一个Node，将当前的线程存进去了，模式是共享模式。

tail 表示这个等待队列的队尾，此刻是null. 所以 pred == null ，进入到enq(node) ;

1enq(node)
2    ↓
3private Node enq(final Node node) {
4    for (;;) {
5        Node t = tail;
6        if (t == null) { // Must initialize
7            if (compareAndSetHead(new Node()))
8                tail = head;
9        } else {
10            node.prev = t;
11            if (compareAndSetTail(t, node)) {
12                t.next = node;
13                return t;
14            }
15        }
16    }
17}

同样tail 为 null , 进入到 compareAndSetHead 。

1compareAndSetHead(new Node())
2    ↓
3/**
4 * CAS head field. Used only by enq.
5 */
6private final boolean compareAndSetHead(Node update) {
7    return unsafe.compareAndSwapObject(this, headOffset, null, update);
8}

这是一个CAS操作，如果head 是 null 的话，等待队列的 head 就会被设置为 update 的值，也就是一个新的结点。

tail = head; 那么此时 tail 也不再是null了。进入下一次的循环。

这次首先将node 的 prev 指针指向 tail ，然后通过一个CAS 操作将node 设置为尾部，并返回了队列的 tail ，也就是 node 。

等待队列的模型变化如下

1+------+  prev      +----------------+
2head(tail) |      | <---- node | currentThread  |
3           +------+            +----------------+
4           
5                    ↓
6                    
7       +------+  prev            +----------------+
8head   |      | <---- node(tail) | currentThread  |
9       +------+                  +----------------+

ok，到了这里await 方法就返回了，是一个 thread 等于当前线程的Node。

返回到 doAcquireSharedInterruptibly(int arg) 中，进入下面循环。

1for (;;) {
2    final Node p = node.predecessor();
3    if (p == head) {
4        int r = tryAcquireShared(arg);
5        if (r >= 0) {
6            setHeadAndPropagate(node, r);
7            p.next = null; // help GC
8            failed = false;
9            return;
10        }
11    }
12    if (shouldParkAfterFailedAcquire(p, node) &&
13        parkAndCheckInterrupt())
14        throw new InterruptedException();
15}

这个时候假设state 仍然大于0，那么此时 r < 0，所以进入到 shouldParkAfterFailedAcquire 这个方法。

1shouldParkAfterFailedAcquire(p, node)
2    ↓
3private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
4    int ws = pred.waitStatus;
5    if (ws == Node.SIGNAL)  //static final int SIGNAL    = -1;
6        /*
7         * This node has already set status asking a release
8         * to signal it, so it can safely park.
9         */
10        return true;
11    if (ws > 0) {
12        /*
13         * Predecessor was cancelled. Skip over predecessors and
14         * indicate retry.
15         */
16        do {
17            node.prev = pred = pred.prev;
18        } while (pred.waitStatus > 0);
19        pred.next = node;
20    } else {
21        /*
22         * waitStatus must be 0 or PROPAGATE.  Indicate that we
23         * need a signal, but don't park yet.  Caller will need to
24         * retry to make sure it cannot acquire before parking.
25         */
26        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
27    }
28    return false;
29}
30    ↓
31/**
32 * CAS waitStatus field of a node.
33 */
34private static final boolean compareAndSetWaitStatus(Node node,
35                                                     int expect,
36                                                     int update) {
37    return unsafe.compareAndSwapInt(node, waitStatusOffset,
38                                    expect, update);
39}
40    
41

可以看到 shouldParkAfterFailedAcquire 也是一路走，走到 compareAndSetWaitStatus。

compareAndSetWaitStatus 将 prev 的 waitStatus 设置为 Node.SIGNAL 。

Node.SIGNAL 表示后续结点中的线程需要被unparking（类似被唤醒的意思）。该方法返回false。

经过这轮循环，队列模型变成下面状态

1+--------------------------+   prev           +------------------+
2head   | waitStatus = Node.SIGNAL | <---- node(tail) | currentThread    |
3       +--------------------------+                  +------------------+

因为shouldParkAfterFailedAcquire返回的是false，所以后面这个条件就不再看了。继续 for (;;) 中的循环。

如果state仍然大于0，再次进入到 shouldParkAfterFailedAcquire。

这次因为head 中的waitStatus 为 Node.SIGNAL ，所以 shouldParkAfterFailedAcquire 返回true。

这次就需要看parkAndCheckInterrupt 这个方法了。

1private final boolean parkAndCheckInterrupt() {
2        LockSupport.park(this);
3        return Thread.interrupted();
4    }

ok，线程没有被中断，所以，返回false。继续 for (;;) 中的循环。

如果state 一直大于0，并且线程一直未被中断，那么就一直在这个循环中。也就是我们上篇文章说的裁判一直不愿意宣布比赛结束的情况。

那么什么情况下跳出循环呢？也就是什么情况下state 会小于0呢？下一篇文章我将说明。

总结一下，await() 方法其实就是初始化一个队列，将需要等待的线程（state > 0）加入一个队列中，并用waitStatus 标记后继结点的线程状态。

CountDownLatch 源码解析—— await()

评论区

相关推荐