深入剖析线程同步工具CountDownLatch原理

Posted 2022-03-02

0 CountDownLatch的作用

CountDownLatch作为一个多线程间的同步工具，它允许一个或多个线程等待其他线程（可以是多个）完成工作后，再恢复执行。

就像下面这样：

1 从一个Demo说起

我们直接拿源码中给出的Demo看一下，源码中的这个demo可以看做模拟一个赛跑的场景。 赛跑肯定有跑得快的运动员也有跑的慢的运动员，每个运动员就表示一个线程。 运动员听到枪声后开始起跑，而最后一个运动员到达终点后，标志的比赛的结束。 整个过程如下图所示：

CountDownLatch跑步模拟

源码如下所示

public class Race 

    private static final int N = 4;

    public static void main(String[] args) throws InterruptedException 
        CountDownLatch startSignal = new CountDownLatch(1);  // 鸣枪开始信号
        CountDownLatch doneSignal = new CountDownLatch(N);   // 等待N个运动员都跑完后，比赛结束（结束信号）

        for (int i = 0; i < N; ++i) // N个运动员准备就绪，等待枪声
            new Thread(new Runner(startSignal, doneSignal, i)).start();

        Thread.sleep(1000); // 等待所有运动员就绪
        System.out.println("所有运动员就绪");
        startSignal.countDown();      // 鸣枪，开赛
        System.out.println("比赛进行中...");
        doneSignal.await();           // 等待N个运动员全部跑完（等待doneSignal变为0）
        System.out.println("比赛结束");
    


class Runner implements Runnable 
    private final CountDownLatch startSignal;
    private final CountDownLatch doneSignal;
    private int number;

    Runner(CountDownLatch startSignal, CountDownLatch doneSignal, int number) 
        this.startSignal = startSignal;
        this.doneSignal = doneSignal;
        this.number = number;
    
    public void run() 
        try 
            // 等待枪声（等待开始信号startSignal变为0）
            System.out.println(number + "号运动员准备就绪");
            startSignal.await();
            // 赛跑
            System.out.println(number + "号运动员跑步中...");
            Thread.sleep(new Random().nextInt(10) * 1000);
            // 此运动员跑到终点
            System.out.println(number + "号运动员到达终点");
            doneSignal.countDown();
         catch (InterruptedException ex)  // return;
    

上面代码运行后，输出如下：

0号运动员准备就绪
3号运动员准备就绪
2号运动员准备就绪
1号运动员准备就绪
所有运动员就绪
比赛进行中...
0号运动员跑步中...
1号运动员跑步中...
2号运动员跑步中...
3号运动员跑步中...
2号运动员到达终点
1号运动员到达终点
0号运动员到达终点
3号运动员到达终点
比赛结束

下面，深入到代码细节，看一下CountDownLatch初始化、countDown方法、await方法是如何实现的。

2 CountDownLatch类图

通过下图来了解一下CountDownLatch的类继承关系

CountDownLatch类图
3 CountDownLatch的初始化
CountDownLatch只有一个构造方法：

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

他会初始化一个Sync，这是他的一个内部类，类似于ReentrantLock，Sync也继承于AbstractQueuedSynchronizer（AQS）。

AQS是个啥？可以参考笔者的另一篇文章：Java队列同步器（AQS）到底是怎么一回事

然后看一下Sync的源码

private static final class Sync extends AbstractQueuedSynchronizer 
    private static final long serialVersionUID = 4982264981922014374L;

    // 调用AQS的setState方法，将state赋值为count的值
    Sync(int count) 
        setState(count);
    

    // 获取AQS state的当前值
    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;
        
    

所以CountDownLatch的初始化，其实是将参数count的值赋值给AQS的state，依然是用state来控制同步状态。

4 await方法的实现
依然用上面赛跑的例子来说明这个问题。这里我们只考虑所有运动员等待枪声的情景。

回忆一下，赛跑的例子中，通过下面的方式创建了鸣枪信号：

CountDownLatch startSignal = new CountDownLatch(1);  // 鸣枪开始信号

然后创建了N个线程（表示N个运动员），并调用其start方法让其开始执行（运动员准备就绪，等待鸣枪开跑）。

然后通过在run方法中调用startSignal.await()，来实现等待鸣枪的动作（其实就是等startSignal的值降为0）。

我们来看一下他是怎么await的。

public void await() throws InterruptedException 
    sync.acquireSharedInterruptibly(1);

调用了AQS的acquireSharedInterruptibly方法

public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException 
    // 判断线程是否已经被中断
    if (Thread.interrupted())
        throw new InterruptedException();
    // 调用CountDownLatch.Sync的tryAcquireShared方法
    // 此方法判断count的值是否==0，如果==0，返回1，否则返回-1
    // 目前我们还没有执行countDown，所以count肯定不等于0，这里肯定返回-1
    // 所以会执行到AQS的doAcquireSharedInterruptibly方法中
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);


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

AQS.doAcquireSharedInterruptibly的实现如下

/
 * Acquires in shared interruptible mode.
 * @param arg the acquire argument
 */
// 此方法会在count>0时将当前线程加入到等待队列中
// 由于我们目前还没有执行countDown，所以count会保持>0
// 启动的N个线程会全部加入到队列中
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException 
    // 将当前线程添加到等待队列中（SHARED模式）
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try 
        // 自旋获取同步状态
        for (;;) 
            final Node p = node.predecessor();
            if (p == head) 
                // 依然调用CountDownLatch.Sync的tryAcquireShared方法判断
                // 如果count降为0，退出自旋
                int r = tryAcquireShared(arg);
                if (r >= 0) 
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                
            
            // 将node的waitStatus设置为-1（常量SIGNAL，表示需要唤醒），并阻塞
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        
     finally 
        if (failed)
            cancelAcquire(node);
    
**
```**

假设N=4，那么4个线程全部start后，会全部加入到队列中自旋等待，像下面这样：

CountDownLatch.await自旋等待
5 countDown方法的实现
countDown方法实际上就是将AQS中的state的值-1。然后判断当前state的值是否==0，如果等于0，说明所有线程都执行结束了，需要唤醒所有等待的线程。

依然继续上面的场景，鸣枪后，所有的运动员开跑。

鸣枪这个动作实际上就是在主线程中执行：

startSignal.countDown();

这就相当于向刚才队列中的所有线程发了一个恢复执行的信号，所有线程会被唤醒，继续执行await后面的代码。

countDown具体干了啥呢？

public void countDown()
sync.releaseShared(1);

他会调用AQS的releaseShared方法

public final boolean releaseShared(int arg)
// 调用CountDownLatch.Sync的tryReleaseShared方法
// 该方法尝试将count值-1，并判断-1后的count是否==0，如果==0，返回true，否则false
// 该方法的源码已经在Sync的源码中给出，可翻阅上文查看
if (tryReleaseShared(arg))
doReleaseShared();
return true;

return false;

由于startSignal中count的初始值==1，startSignal.countDown()后，count变为0。所以tryReleaseShared会返回true。

然后开始执行doReleaseShared，唤醒队列中的线程。

doReleaseShared是AQS中的方法。

/**

• Release action for shared mode -- signals successor and ensures
• propagation. (Note: For exclusive mode, release just amounts
• to calling unparkSuccessor of head if it needs signal.)
  /
  private void doReleaseShared()
  /
  • Ensure that a release propagates, even if there are other
  • in-progress acquires/releases. This proceeds in the usual
  • way of trying to unparkSuccessor of head if it needs
  • signal. But if it does not, status is set to PROPAGATE to
  • ensure that upon release, propagation continues.
  • Additionally, we must loop in case a new node is added
  • while we are doing this. Also, unlike other uses of
  • unparkSuccessor, we need to know if CAS to reset status
  • fails, if so rechecking.
    */
    for (;;)
    Node h = head;
    if (h != null && h != tail)
    int ws = h.waitStatus;
    // Node.SIGNAL == -1
    // 由上文可知，进入队列的线程的waitStatus都等于-1
    // 所以这里为true
    if (ws == Node.SIGNAL)
    // 尝试将waitStatus从-1改为0，如果修改成功，就恢复这个线程的执行状态
    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
    continue; // loop to recheck cases
    unparkSuccessor(h);
    
    else if (ws == 0 &&
    !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
    continue; // loop on failed CAS
    
    if (h == head) // loop if head changed
    break;
    
    

这里，被阻塞的线程又恢复执行，恢复到哪了呢？就是刚才自旋等待的那里。

把上面的源码直接拿下来，再说明一下（注释部分）

private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException 
    // 将当前线程添加到等待队列中（SHARED模式）
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try 
        // 线程被释放后，继续下一次循环
        for (;;) 
            final Node p = node.predecessor();
            if (p == head) 
                int r = tryAcquireShared(arg);
                // 获取头节点，从头结点开始释放，由于count已经降为0，所以r >= 0为true
                // 然后会将自己摘除当前队列，使下一个节点成为头节点
                // 等下一个节点也恢复过来后，同样执行上面的过程
                // 这样，队列中的所有线程就被释放了
                if (r >= 0) 
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                
            
            // 将node的waitStatus设置为-1（常量SIGNAL，表示需要唤醒），并阻塞
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        
     finally 
        if (failed)
            cancelAcquire(node);
    

6 总结

本文从源码层面详细说明了CountDownLatch是如何运作的。 CountDownLatch也是基于AQS实现，所以了解AQS的机制，对于理解本文至关重要。 其实，CountDownLatch最核心的就是通过控制AQS的state，来同步多个线程之间的状态。