JUC框架 CountDownLatch源码解析 JDK8

冷不防 2023-02-18 12:57 7阅读 0赞

### 文章目录 ###

*  前言
 *  实现核心
 *  构造器
 *  核心方法
 *   *  countDown()
     *  await()
     *  await(long timeout, TimeUnit unit)
     *  对比两个await方法返回时的情况
 *  分析两种线程
 *  总结

# 前言 #

`CountDownLatch`是一种常用的JUC框架构件，它用在一组线程任务需要等到条件满足后才能从同一起跑线开始执行的场景。比如当你去吃饭，你已经坐在桌上了但还是不能动筷子，因为各个领导还没来，只有领导都到齐了，大家才能开吃；再比如，参加了一场不能提前交卷的考试，学霸已经早早做完了但也不能提前走，只有当时间流逝直到考试结束时，大家才能离开考场。

`CountDownLatch`一般构造器给定一个大于0的数`n`，当调用了`n`次`CountDown`后，条件就将满足，所有阻塞在`await()`的线程将从同一起跑线开始执行。

[JUC框架 系列文章目录][JUC_]

# 实现核心 #

CountDownLatch向下依赖了AQS的共享锁部分，它使用了一个内部类继承了AQS。既然是使用的共享锁，那么肯定要实现`tryAcquireShared`和`tryReleaseShared`方法，因为共享锁的获取和释放流程都会依赖子类对这两个方法的实现。

private static final class Sync extends AbstractQueuedSynchronizer { 
            private static final long serialVersionUID = 4982264981922014374L;
    
            Sync(int count) { 
                setState(count);
            }
    
            int getCount() { 
                return getState();
            }
    
            protected int tryAcquireShared(int acquires) { 
                return (getState() == 0) ? 1 : -1;
            }
    
            protected boolean tryReleaseShared(int releases) { 
                // Decrement count; signal when transition to zero
                for (;;) { 
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }
        }
    
        private final Sync sync;

# 构造器 #

public CountDownLatch(int count) { 
            if (count < 0) throw new IllegalArgumentException("count < 0");
            this.sync = new Sync(count);
        }

构造器需要传入一个大于等于0的数，这个数将作为AQS的state的初始值。

# 核心方法 #

1.  countDown方法，每调用一次就将当前的计数器count减一，当count为0时，唤醒所有阻塞在await方法处的线程。
2.  await方法。当count不为0时，调用await的线程将阻塞。它有两种版本：普通版本、带超时的版本。两个版本都响应超时。

CountDownLatch具体的讲，依赖了AQS的共享锁模式的`sync queue`。

## countDown() ##

//CountDownLatch
        public void countDown() { 
            sync.releaseShared(1);
        }

countDown方法没有参数，因为每次调用只能将count减少1。

//AbstractQueuedSynchronizer
        public final boolean releaseShared(int arg) { 
            if (tryReleaseShared(arg)) {   //返回值很重要，直接影响能否执行doReleaseShared
                doReleaseShared();
                return true;
            }
            return false;
        }

`tryReleaseShared`的返回值很重要，直接影响能否执行`doReleaseShared`。而`doReleaseShared`的逻辑就是“唤醒所有阻塞在await方法处的线程”。

//CountDownLatch
            protected boolean tryReleaseShared(int releases) { 
                // Decrement count; signal when transition to zero
                for (;;) { 
                    int c = getState();
                    if (c == 0)
                        return false;
                    int nextc = c-1;
                    if (compareAndSetState(c, nextc))
                        return nextc == 0;
                }
            }

*  获得当前AQS的state
    
     *  如果state是0，那么不能再减了直接返回false，因为state最多能减到0。
     *  如果state不是0，需要CAS设置state。
        
         *  CAS操作成功了，返回`nextc == 0`。
         *  当然CAS操作可能失败，失败了就需要再次循环执行CAS，因为可能同时有多个线程在同时执行`countDown`。
 *  CAS操作成功返回的是`nextc == 0`，`nextc`为CAS设置成功后，state的新值。也就是说，只有那个将state从1设置为0的线程，才会返回true，其他调用`countDown`的线程都会返回false。

回到`releaseShared`的逻辑，只有`tryReleaseShared(arg)`返回了true，`doReleaseShared()`才会执行，才会去唤醒所有阻塞在await方法处的线程。关于[共享锁的流程][Link 1]之前已经讲过，我们只需要知道`doReleaseShared()`会唤醒`sync queue`中的head后继，而被唤醒的线程`tryAcquireShared`成功后在一定条件下也会去调用`doReleaseShared()`唤醒它的后继，这样可能会有多个线程同时执行`doReleaseShared()`。重点在于，唤醒线程的速度很快，几乎可以算是同时进行的。

一直在说唤醒`sync queue`中的head后继，那head后继的代表线程到底阻塞在`await()`执行过程的哪里了呢，带着这一问题，我们进入下一章。

## await() ##

首先它的效果和`Condition`接口的`await()`类似，一般调用之后线程就会陷入阻塞。而当**count为0**时，线程从`await()`处被唤醒而继续执行。从下面的函数声明就可以看出，CountDownLatch的`await()`是响应中断的。

public void await() throws InterruptedException { 
        sync.acquireSharedInterruptibly(1);
    }
    
    public final void acquireSharedInterruptibly(int arg) throws InterruptedException { 
        if (Thread.interrupted())  //上来先检查一下，当前线程的中断状态
            throw new InterruptedException();
        if (tryAcquireShared(arg) < 0)  //如果Acquire失败，就需要走循环阻塞的流程了
            doAcquireSharedInterruptibly(arg);
    }

我们来看看CountDownLatch的AQS子类是怎么实现`tryAcquireShared`的。

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

发现有点欺骗观众的感觉，因为传入的参数`acquires`根本没有使用，逻辑只是检查AQS的state是否为0。如果state为0返回大于0的数，如果state不为0返回小于0的数。

[共享锁的流程][Link 1]中我们提到过`tryAcquireShared`返回值的含义：

<table> 
 <thead> 
  <tr> 
   <th align="center">tryAcquireShared子类实现判断</th> 
   <th align="center">tryAcquireShared返回值</th> 
   <th align="center">tryAcquireShared返回值含义</th> 
   <th align="center">await流程</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">state为0</td> 
   <td align="center"><code>&gt; 0</code></td> 
   <td align="center">获取共享锁成功，并且后续获取也可能获取成功</td> 
   <td align="center">将返回，不阻塞</td> 
  </tr> 
  <tr> 
   <td align="center">-</td> 
   <td align="center"><code>= 0</code></td> 
   <td align="center">获取共享锁成功，但后续获取可能不会成功</td> 
   <td align="center">将返回，不阻塞</td> 
  </tr> 
  <tr> 
   <td align="center">state不为0</td> 
   <td align="center"><code>&lt; 0</code></td> 
   <td align="center">获取共享锁失败</td> 
   <td align="center">将阻塞</td> 
  </tr> 
 </tbody> 
</table>

所以，只要`tryAcquireShared`返回了 `> 0`的数，`acquireSharedInterruptibly`就直接返回了不会阻塞了，因为此时state已经为0了，说明已经调用了足够次数的`countDown`了。如果`tryAcquireShared`返回了 `< 0`的数，`acquireSharedInterruptibly`就需要调用下面的`doAcquireSharedInterruptibly`，将当前线程包装成node扔到`suyc queue`上去，走循环 抢锁->阻塞 的流程了。

private void doAcquireSharedInterruptibly(int arg)
            throws InterruptedException {   //不同之处1
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try { 
                /*boolean interrupted = false;*/    //不同之处2
                for (;;) { 
                    final Node p = node.predecessor();
                    if (p == head) { 
                        int r = tryAcquireShared(arg);
                        if (r >= 0) { 
                            setHeadAndPropagate(node, r);
                            p.next = null;
                            /*if (interrupted) //不同之处3 selfInterrupt();*/  
                            failed = false;
                            return;
                        }
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();  //不同之处4
                }
            } finally { 
                if (failed)
                    cancelAcquire(node);  //不同之处5
            }
        }

之前在[共享锁的流程][Link 1]中已经讲过了`doAcquireShared`，而`doAcquireSharedInterruptibly`和它也很类似，所以我们看看不同之处就好。

1.  函数声明是抛出异常的，因为现在的`doAcquireSharedInterruptibly`是响应中断的版本。
2.  此处在`doAcquireSharedInterruptibly`中被删除掉了，因为此函数在遇到中断时就直接抛出异常，所以不需要局部变量来保存是否有过中断的信息。
3.  此处在`doAcquireSharedInterruptibly`中被删除掉了，分析同上。被删掉的此处在`doAcquireShared`中的作用是返回用户代码前设置一下中断状态。
4.  线程阻塞在`parkAndCheckInterrupt`后，如果因为中断而被唤醒，`parkAndCheckInterrupt`会返回true，然后就直接抛出异常就好。
5.  此处其实一模一样。但是在`doAcquireShared`中此处永远不可能执行到，在`doAcquireSharedInterruptibly`中如果因为抛出异常而退出函数，则会执行此处。

回想之前提的问题，“head后继的代表线程到底阻塞在`await()`执行过程的哪里”，可以看到，线程在`tryAcquireShared`失败后，一定会阻塞在`parkAndCheckInterrupt`这里的。所有的调用`CountDownLatch.await()`阻塞的线程，都是阻塞在这里的。

而当那个调用`countDown`从而将count从1变成0的线程执行完`countDown`后，会唤醒`sync queue`中的head后继线程，已经说了线程是阻塞在`parkAndCheckInterrupt`这里，所以head后继线程会从`parkAndCheckInterrupt`处唤醒，然后继续下一次循环，执行`tryAcquireShared`会返回`>0`的数（此时count已经为0了），然后执行`setHeadAndPropagate`，在里面然后又会执行`doReleaseShared`。

现在好了，不仅调用`countDown`从而将count从1变成0的线程会执行`doReleaseShared`，调用`await()`的线程被唤醒后也会执行`doReleaseShared`，之后被唤醒的线程也会去执行`doReleaseShared`。这样不断唤醒，很快在`sync queue`上的所有线程都会被唤醒。之所以说它**快**，是因为每个尝试获取锁（即`tryAcquireShared`动作）的线程，在获取锁完毕后并退出`await()`时就已经唤醒了自己的后继，当然，这本来就是共享锁获取的流程。

## await(long timeout, TimeUnit unit) ##

`await(long timeout, TimeUnit unit)`的版本提供了超时功能。

public boolean await(long timeout, TimeUnit unit) throws InterruptedException { 
        return sync.tryAcquireSharedNanos(1, unit.toNanos(timeout));
    }
    
    public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException { 
        if (Thread.interrupted())
            throw new InterruptedException();
        return tryAcquireShared(arg) >= 0 || doAcquireSharedNanos(arg, nanosTimeout);
    }

同样的，如果`tryAcquireShared`返回了小于0的数，就只好调用`doAcquireSharedNanos`走循环 抢锁->阻塞 的过程。但注意，上面两个函数都是有返回值的。

<table> 
 <thead> 
  <tr> 
   <th align="center"><code>await(long timeout, TimeUnit unit)</code>返回值</th> 
   <th align="center">返回值含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">true</td> 
   <td align="center">在规定时间内，抢到了锁 ==&gt; 在规定时间内，count变成了0</td> 
  </tr> 
  <tr> 
   <td align="center">false</td> 
   <td align="center">在规定时间内，没抢到锁 ==&gt; 在规定时间内，count没变成0</td> 
  </tr> 
 </tbody> 
</table>

很明显，如果`tryAcquireShared`返回了大于等于0的数，就会直接返回true了；如果`tryAcquireShared`返回了小于0的数，具体返回的值就等于`doAcquireSharedNanos(arg, nanosTimeout)`的返回值了。

private boolean doAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException {   //不同之处1
        if (nanosTimeout <= 0L)  //不同之处2
            return false;
        final long deadline = System.nanoTime() + nanosTimeout;  //不同之处3
        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 true;
                    }
                }
                nanosTimeout = deadline - System.nanoTime();  //不同之处4
                if (nanosTimeout <= 0L)  //不同之处5
                    return false;
                if (shouldParkAfterFailedAcquire(p, node) &&
                    nanosTimeout > spinForTimeoutThreshold)  //不同之处6
                    LockSupport.parkNanos(this, nanosTimeout);  //不同之处7
                if (Thread.interrupted())  //不同之处8
                    throw new InterruptedException();
            }
        } finally { 
            if (failed)
                cancelAcquire(node);
        }
    }

直接看看`doAcquireSharedNanos(int arg, long nanosTimeout)`与`doAcquireSharedInterruptibly(int arg)`的不同之处吧。

1.  返回值是boolean的，含义解释过了。
2.  如果发现传来的时间段是小于0的，也就说明用户想要等待0的时间（反过来说，就是不需要等待），也就不用进入下面的循环了。进入`doAcquireSharedNanos`后还没有尝试抢过锁，所以返回false。
3.  计算出等待的截止时间。
4.  计算出离截止时间还有多久。
5.  如果发现当前时间已经超过了截止时间，则直接返回false，不用再去抢锁了。
6.  虽然没超过截止时间，但离截止时间已经很近了，就不要再阻塞了，直接自旋就好了。因为考虑到阻塞唤醒，时间太短反而无法控制。
7.  超时版本需要调用`LockSupport.parkNanos`。
8.  因为从上面的`LockSupport.parkNanos`处被中断唤醒时，不会带有信息（`LockSupport.parkNanos`没有返回值），所以需要检测一下中断状态。

简单的说，`doAcquireSharedNanos(int arg, long nanosTimeout)`正常返回时，有两种情况：因为获得锁而返回true，因为超时还没获得锁而返回false。

## 对比两个await方法返回时的情况 ##

<table> 
 <thead> 
  <tr> 
   <th align="center">返回原因</th> 
   <th align="center">等到了count变成0</th> 
   <th align="center">count未变成0的任何时候，(超时前)中断来临</th> 
   <th align="center">超时还没等到count为0</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center"><code>await()</code></td> 
   <td align="center">只要是正常返回的</td> 
   <td align="center">抛出中断异常</td> 
   <td align="center">-</td> 
  </tr> 
  <tr> 
   <td align="center"><code>await(long timeout, TimeUnit unit)</code></td> 
   <td align="center">正常返回的，且返回值为true</td> 
   <td align="center">抛出中断异常</td> 
   <td align="center">正常返回的，且返回值为false</td> 
  </tr> 
 </tbody> 
</table>

两个await方法，我们都可以从返回的情况，就可以知道返回的原因。

# 分析两种线程 #

现将`await()`和`await(long timeout, TimeUnit unit)`都统称为await方法，那么关于CountDownLatch有两种线程：

*  调用countDown的线程，肯定不会阻塞。
 *  调用await的线程，会阻塞。

对于CountDownLatch的使用者来说，是这样的使用场景：需要当某个条件满足后，才让某些任务从同一个起跑线开始执行，而“满足条件”则被量化为一个数字。

*  当更加接近这个条件时，让第一类线程调用countDown，也可以调用多次。
 *  需要放到同一个起跑线的线程任务，则调用await。它们将几乎同时被唤醒。

当然，这也不是绝对的，完全可以一个线程同时充当两个角色，那么它将先调用countDown，再调用await。如果每个线程都这样使用，那么使用CountDownLatch就相当于使用了一个一次性的CyclicBarrier。

# 总结 #

*  `CountDownLatch`的使用场景：当某个量化为数字的条件被满足后，几个线程任务才可以同时开始执行。
 *  调用`CountDownLatch#await`的线程将等待条件被满足，条件满足后，调用`CountDownLatch#await`的若干线程将从同一个时间点继续执行。
 *  调用`CountDownLatch#countDown`，让量化为数字的条件减一。
 *  调用`CountDownLatch#await`的线程，和调用`CountDownLatch#countDown`的线程，是两类线程，并无关系。
 *  唤醒阻塞在`await`的若干线程的过程很快，这是由于共享锁的特性导致：共享锁获取成功时，也会唤醒`sync queue`的后继节点线程，但独占锁可不这样。
 *  `CountDownLatch`依赖了共享锁模式节点的`sync queue`。
 *  `CountDownLatch`是一次性的，count为0无法改变。

[JUC_]: https://blog.csdn.net/anlian523/article/details/106598910
[Link 1]: https://blog.csdn.net/anlian523/article/details/106598739