J.U.C 组件 CountDownLatch 深度源码分析(一)

Dear 丶 2024-03-26 23:20 75阅读 0赞

### 一、介绍 ###

CountDownLatch 俗称为计数器，主要是一些线程阻塞直到所有的工作线程全部完成达到指定的一个数量(count)值才能被唤醒并执行后续的代码。这个计数是属于单次的，不能重新对这个 CountDownLatch 进行计数，如果要循环计数的话要使用到 J.U.C 下的另一个组件 CyclicBarrier，这个组件俗称循环屏障。CyclicBarrier 的深入底层源码分析会在后续文章给出，之前有文章只是粗略地介绍了 J.U.C 用法和概念，这篇文章会进行深入的源码分析，多余的话不说了，马上进行下面的深度源码分析。

### 二、CountDownLatch 源码解剖 ###

CountDownLatch 有两个重要的方法，await 和 countDown，await 带超时的方法类似，只是在原来的基础上加了超时.

#### 1.构造 ####

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

构造函数中传入一个 int 类型的 count 数值，通过 CountDownLatch 的静态内部类 Sync 同时这个类也是继承自 AQS，Sync 的构造函数如下：

Sync(int count) {
        setState(count);
    }

设置状态变量 state，同时这个变量是被 volatile 修饰的，保证线程之间的可见性

#### 2.await() 函数 ####

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

这里是调用了其内部类 Sync 的函数，但是由于其 Sync 并没有实现 AQS 的 acquireSharedInterruptibly 方法，所以是从 AQS 那边来实现这部分的逻辑的.

###### AQS.acquireSharedInterruptibly ######

public final void acquireSharedInterruptibly(int arg)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            if (tryAcquireShared(arg) < 0)
                doAcquireSharedInterruptibly(arg);
    }

首先判断是否有被中断过，有的话就抛出异常，如果没有就与 tryAcquireShared(arg) 返回的值做比较，具体实现如下：

###### Sync.tryAcquireShared ######

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

首先要明白这个 state 变量的意义，state 代表待达到条件的线程数，比如初始化为 5，表示待达到的条件的线程数为 5，每次调用 countDown() 函数都会减一，所以当没有达到条件也就是 state!=0 就会返回 -1，进行下面的函数调用，这个函数会比较长，具体的我们慢慢来分析，只要线程进来还没有达到 state==0 这个条件的话，都会封装为 Node 节点添加到链表中，这个链表是双向存在的，等到这个 state 为 0 时所有的线程都会从链表中一个个地进行唤醒.

###### AQS.doAcquireSharedInterruptibly ######

private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        // 添加共享状态的 Node 对象到队列中
        final Node node = addWaiter(Node.SHARED);
        boolean failed = true;
        try {
            for (;;) {
                // 获得所要添加的前驱节点
                final Node p = node.predecessor();
                if (p == head) {
                    // 获得 state 状态的结果
                    int r = tryAcquireShared(arg);
                    if (r >= 0) { // 如果 state=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);
        }
    }

###### AQS.addWaiter ######

我们先对 addWaiter 进行具体的剖析，看看线程如何变成节点被添加进去的，实现如下：

private Node addWaiter(Node mode) {
        // 将当前线程构造成一个 Node 对象
        Node node = new Node(Thread.currentThread(), mode);
        Node pred = tail;// 获得队列的尾节点
        if (pred != null) {
            node.prev = pred;// 将之前的尾节点赋值给当前节点的前置节点
            // 将当前节点设置为尾节点,如果 CAS 失败,则执行 enq() 方法进行补救
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        // 如果设置失败,说明是第一次创建,或者是之前有线程修改过了
        enq(node);
        return node;
    }

###### AQS.enq ######

private Node enq(final Node node) {
        for (;;) {
            Node t = tail;
            if (t == null) { // 第一个创建的节点为空
                if (compareAndSetHead(new Node()))
                    // 第一次头尾节点设置为同一节点
                    tail = head;
            } else {
                node.prev = t;
                if (compareAndSetTail(t, node)) {// 该节点一定会添加,只要设置成功后才会退出
                    t.next = node;
                    return t;
                }
            }
        }
    }

enq 就是保证当前线程封装成的 Node 一定会被添加到链表中。

到了这里，addWaiter 就分析完了，我们继续下面的分析工作，添加完节点以后如果 state 没有达到 =0 这个条件，就会执行

if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())
        throw new InterruptedException();

###### AQS.shouldParkAfterFailedAcquire ######

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus; // 获得节点的状态
        if (ws == Node.SIGNAL) // 表示节点可以运行
            return true; 
        if (ws > 0) {
            do {
                // 如果状态>0 说明该节点已不可用,这边是进行不可用的节点删除操作
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            // 状态 ==0 就将状态值修改为 SIGNAL 状态值
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

我们对 shouldParkAfterFailedAcquire 进行一个整体的描述，它去判断线程的状态 waitStatus，在唤醒该线程的过程中，去判别是不是应该忽略当前的线程，表明线程状态的分为以下几种：

*  CANCELLED(1)：在同步队列中等待的线程等待超时或被中断，需要从同步队列中取消该 Node 的节点，其节点的 waitStatus 为 CANCELLED，即结束状态，进入该状态后的节点将不会再变化
 *  SIGNAL(-1)：只要前置节点释放锁，就会通知标识为 SIGNAL 状态的后置节点的线程
 *  CONDITION(-2)：和 Condition 有关系
 *  PROPAGATE(-3)：共享模式下，PROPAGATE 状态处于可运行状态
 *  0：初始状态

只要 Node 节点的 waitStatus 值大于 0 的话就说明不需要唤醒与这个 Node 绑定的线程，新建的节点默认的状态是为 0.

shouldParkAfterFailedAcquire 是位于无限循环里面，这一点需要注意一般每个节点都要经历两次循环后然后被阻塞(**第一次是修改 waitStatus 值 为 -1，第二次是直接返回 true**)，当该函数返回 true 则会调用 parkAndCheckInterrupt 函数阻塞自身，到这里基本上每个调用 await() 函数的线程都会阻塞在这里，**同时唤醒的时候也是在这个地方.**

接下来我们来分析另外一个函数 countDown 所实现的功能和步骤

#### **3.countDown() 函数** ####

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

该函数也是委托其内部类 Sync 继承的 AQS 类进行实现的，注意这里的 arg 数值为 1.

###### AQS.tryReleaseShared ######

public final boolean c(int arg) {
        if (tryReleaseShared(arg)) { 
            doReleaseShared();
            return true;
        }
        return false;
    }

这里再看一下 tryReleaseShared 内部函数的实现.

###### Sync.tryReleaseShared ######

protected boolean tryReleaseShared(int releases) {
        // 减去数值,当为 0 的时候发出信号
        for (;;) {
            int c = getState();
            if (c == 0)
                return false;
            int nextc = c-1;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
    }

这个函数也告知了我们为什么 CountDownLatch 这个组件为什么只能执行一次，当 state=0 时这边不做任何的操作，也就是说计数器只能进行一次计数。言归正传，当 state 减去 1 以后的值为 0 才会返回 true并执行后面的结果，否则都一概忽略，假如已经达到了唤醒的条件，让我们看一下具体是如何被唤醒？

###### AQS.doReleaseShared ######

private void doReleaseShared() {
        for (;;) {
            Node h = head;// 获取头节点
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {// 头节点的状态为 SIGNAL,待唤醒的状态
                    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
                // 这里是否有疑问明明都有这个  Node h = head 为啥还要在判断一次？别着急后面单独对这个问题进行解释
                break;
        }
    }

执行唤醒函数的添加

private void unparkSuccessor(Node node) {
        int ws = node.waitStatus;// 获得 head 节点的状态
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);// 设置 head 节点状态为 0
        Node s = node.next;
        if (s == null || s.waitStatus > 0) {
            // 如果下一个节点为 null 或者 status>0 表示 CANCELLED 节点不可用状态
            // 通过从尾部节点开始扫描,找到距离 head 最近的一个 waitStatus<=0 的节点
            s = null;
            for (Node t = tail; t != null && t != node; t = t.prev)
                if (t.waitStatus <= 0)
                    s = t;
        }
        if (s != null) // next 节点不为空,直接唤醒这个线程即可
            LockSupport.unpark(s.thread);
    }

对于这个唤醒操作很好理解的，首先取得该节点的后继节点不为空且状态是 <0 的就进行唤醒，如果后继节点被取消，则从最后一次开始往前找，找一个满足条件的节点唤醒即可。

这边会有个疑问，要是如果有多个节点只在这进行一次唤醒工作吗？难道只需要唤醒一个吗？这边我们先找到阻塞的地方

[![alt][]

![alt][]][alt_ _alt]

关键就是在这里，线程阻塞在这里，唤醒后继续执行，由于满足条件，也就是 state==0 返回 1，满足 >=0 的条件，会进入到下面的一小段代码：

setHeadAndPropagate(node, r);
    p.next = null; // help GC
    failed = false;
    return;

###### AQS.setHeadAndPropagate ######

private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head;// 获得原先的 head 节点
        setHead(node);// 将当前节点设置为 head 节点
        // 下面这个 if 一定会进入,因为这个函数进来的时候 propagate 必须是 1 
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                // 重要的是这段代码, countDown 函数的时候也有调用这个
                doReleaseShared();
        }
    }

doReleaseShared() 这个函数就是在 countDown 函数里面进行调用的，在这里，被唤醒的线程再调用一次，依赖唤醒后续线程

[![alt][alt 1]

![alt 1][]][alt_alt 1_ _alt 1]

现在明白其唤醒机制了把，先唤醒一个线程(第一个阻塞的线程)，然后被唤醒的线程又会执行到这里唤醒线程，这样重复下去，最终所有线程都会被唤醒，其实这也是 AQS 共享锁的唤醒原理，自此完成了对 CountDownLatch 阻塞和唤醒源码的分析

#### 图解分析 ####

*  红色为 wait 函数执行的基本流程
 *  蓝色为最后一个 countDown 函数执行的基本流程

[![alt][alt 2]

![alt 2][]][alt_alt 2_ _alt 2]

借以文章记录研究 CountDownLatch 源码的学习过程，编写不易，有问题可以留言，觉得不错可以给个好评哈哈.

#### *相关* ####

[alt]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/95d1a94353aa4b0290866757afce97ed.png
[alt_ _alt]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2021/01/image-17.png?ssl=1
[alt 1]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/76c469d48dc147849edba0ff1f14ef59.png
[alt_alt 1_ _alt 1]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2021/01/image-19-1024x600.png?ssl=1
[alt 2]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/39f1903fdc5a413db0f438245e35b7d0.png
[alt_alt 2_ _alt 2]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2021/01/image-25-1024x996.png?ssl=1