Java线程：CountDownLatch 原理

朱雀 2022-03-11 11:16 175阅读 0赞

### CountDownLatch的使用 ###

CountDownLatch是同步工具类之一，可以指定一个计数值，在并发环境下由线程进行减1操作，当计数值变为0之后，被await方法阻塞的线程将会唤醒，实现线程间的同步。

1.  `public void startTestCountDownLatch() {`
2.  `int threadNum = 10;`
3.  `final CountDownLatch countDownLatch = new CountDownLatch(threadNum);`
4.   
5.  `for (int i = 0; i < threadNum; i++) {`
6.  `final int finalI = i + 1;`
7.  `new Thread(() -> {`
8.  `System.out.println("thread " + finalI + " start");`
9.  `Random random = new Random();`
10. `try {`
11. `Thread.sleep(random.nextInt(10000) + 1000);`
12. `} catch (InterruptedException e) {`
13. `e.printStackTrace();`
14. `}`
15. `System.out.println("thread " + finalI + " finish");`
16.  
17. `countDownLatch.countDown();`
18. `}).start();`
19. `}`
20.  
21. `try {`
22. `countDownLatch.await();`
23. `} catch (InterruptedException e) {`
24. `e.printStackTrace();`
25. `}`
26. `System.out.println(threadNum + " thread finish");`
27. `}`
28.

主线程启动10个子线程后阻塞在await方法，需要等子线程都执行完毕，主线程才能唤醒继续执行。

### 构造器 ###

CountDownLatch和ReentrantLock一样，内部使用Sync继承AQS。构造函数很简单地传递计数值给Sync，并且设置了state。

1.  `Sync(int count) {`
2.  `setState(count);`
3.  `}`
4.

上文已经介绍过AQS的state，这是一个由子类决定含义的“状态”。对于ReentrantLock来说，state是线程获取锁的次数；对于CountDownLatch来说，则表示计数值的大小。

### 阻塞线程 ###

接着来看await方法，直接调用了AQS的acquireSharedInterruptibly。

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

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

首先尝试获取共享锁，实现方式和独占锁类似，由CountDownLatch实现判断逻辑。

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

返回1代表获取成功，返回-1代表获取失败。如果获取失败，需要调用doAcquireSharedInterruptibly：

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

doAcquireSharedInterruptibly的逻辑和独占功能的acquireQueued基本相同，阻塞线程的过程是一样的。不同之处：

1.  创建的Node是定义成共享的（Node.SHARED）；
2.  被唤醒后重新尝试获取锁，不只设置自己为head，还需要通知其他等待的线程。（重点看后文释放操作里的setHeadAndPropagate）

### 释放操作 ###

1.  `public void countDown() {`
2.  `sync.releaseShared(1);`
3.  `}`
4.

countDown操作实际就是释放锁的操作，每调用一次，计数值减少1：

1.  `public final boolean releaseShared(int arg) {`
2.  `if (tryReleaseShared(arg)) {`
3.  `doReleaseShared();`
4.  `return true;`
5.  `}`
6.  `return false;`
7.  `}`
8.

同样是首先尝试释放锁，具体实现在CountDownLatch中：

1.  `protected boolean tryReleaseShared(int releases) {`
2.  `// Decrement count; signal when transition to zero`
3.  `for (;;) {`
4.  `int c = getState();`
5.  `if (c == 0)`
6.  `return false;`
7.  `int nextc = c-1;`
8.  `if (compareAndSetState(c, nextc))`
9.  `return nextc == 0;`
10. `}`
11. `}`
12.

死循环加上cas的方式保证state的减1操作，当计数值等于0，代表所有子线程都执行完毕，被await阻塞的线程可以唤醒了，下一步调用doReleaseShared：

1.  `private void doReleaseShared() {`
2.  `for (;;) {`
3.  `Node h = head;`
4.  `if (h != null && h != tail) {`
5.  `int ws = h.waitStatus;`
6.  `if (ws == Node.SIGNAL) {`
7.  `//1`
8.  `if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))`
9.  `continue; // loop to recheck cases`
10. `unparkSuccessor(h);`
11. `}`
12. `//2`
13. `else if (ws == 0 &&`
14. `!compareAndSetWaitStatus(h, 0, Node.PROPAGATE))`
15. `continue; // loop on failed CAS`
16. `}`
17. `if (h == head) // loop if head changed`
18. `break;`
19. `}`
20. `}`
21.

标记1里，头节点状态如果SIGNAL，则状态重置为0，并调用unparkSuccessor唤醒下个节点。

标记2里，被唤醒的节点状态会重置成0，在下一次循环中被设置成PROPAGATE状态，代表状态要向后传播。

1.  `private void unparkSuccessor(Node node) {`
2.  `int ws = node.waitStatus;`
3.  `if (ws < 0)`
4.  `compareAndSetWaitStatus(node, ws, 0);`
5.   
6.  `Node s = node.next;`
7.  `if (s == null || s.waitStatus > 0) {`
8.  `s = null;`
9.  `for (Node t = tail; t != null && t != node; t = t.prev)`
10. `if (t.waitStatus <= 0)`
11. `s = t;`
12. `}`
13. `if (s != null)`
14. `LockSupport.unpark(s.thread);`
15. `}`
16.

在唤醒线程的操作里，分成三步：

*  处理当前节点：非CANCELLED状态重置为0；
 *  寻找下个节点：如果是CANCELLED状态，说明节点中途溜了，从队列尾开始寻找排在最前还在等着的节点
 *  唤醒：利用LockSupport.unpark唤醒下个节点里的线程。

线程是在doAcquireSharedInterruptibly里被阻塞的，唤醒后调用到setHeadAndPropagate。

1.  `private void setHeadAndPropagate(Node node, int propagate) {`
2.  `Node h = head;`
3.  `setHead(node);`
4.   
5.  `if (propagate > 0 || h == null || h.waitStatus < 0 ||`
6.  `(h = head) == null || h.waitStatus < 0) {`
7.  `Node s = node.next;`
8.  `if (s == null || s.isShared())`
9.  `doReleaseShared();`
10. `}`
11. `}`
12.

setHead设置头节点后，再判断一堆条件，取出下一个节点，如果也是共享类型，进行doReleaseShared释放操作。下个节点被唤醒后，重复上面的步骤，达到共享状态向后传播。

要注意，await操作看着好像是独占操作，但它可以在多个线程中调用。当计数值等于0的时候，调用await的线程都需要知道，所以使用共享锁。

### 限定时间的await ###

CountDownLatch的await方法还有个限定阻塞时间的版本.

1.  `public boolean await(long timeout, TimeUnit unit)`
2.  `throws InterruptedException {`
3.  `return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));`
4.  `}`
5.

跟踪代码，最后来看doAcquireSharedNanos方法，和上文介绍的doAcquireShared逻辑基本一样，不同之处是加了time字眼的处理。

1.  `private boolean doAcquireSharedNanos(int arg, long nanosTimeout)`
2.  `throws InterruptedException {`
3.  `if (nanosTimeout <= 0L)`
4.  `return false;`
5.  `final long deadline = System.nanoTime() + nanosTimeout;`
6.  `final Node node = addWaiter(Node.SHARED);`
7.  `boolean failed = true;`
8.  `try {`
9.  `for (;;) {`
10. `final Node p = node.predecessor();`
11. `if (p == head) {`
12. `int r = tryAcquireShared(arg);`
13. `if (r >= 0) {`
14. `setHeadAndPropagate(node, r);`
15. `p.next = null; // help GC`
16. `failed = false;`
17. `return true;`
18. `}`
19. `}`
20. `nanosTimeout = deadline - System.nanoTime();`
21. `if (nanosTimeout <= 0L)`
22. `return false;`
23. `if (shouldParkAfterFailedAcquire(p, node) &&`
24. `nanosTimeout > spinForTimeoutThreshold)`
25. `LockSupport.parkNanos(this, nanosTimeout);`
26. `if (Thread.interrupted())`
27. `throw new InterruptedException();`
28. `}`
29. `} finally {`
30. `if (failed)`
31. `cancelAcquire(node);`
32. `}`
33. `}`
34.

进入方法时，算出能够执行多久的deadline，然后在循环中判断时间。注意到代码中间有句：

nanosTimeout > spinForTimeoutThreshold

1.  `static final long spinForTimeoutThreshold = 1000L;`
2.

spinForTimeoutThreshold写死了1000ns，这就是所谓的自旋操作。当超时在1000ns内，让线程在循环中自旋，否则阻塞线程。