J.U.C 组件 CyclicBarrier 深度源码分析(二)

╰半橙微兮° 2024-03-26 23:20 80阅读 0赞

### 一、介绍 ###

cyclicBarrier 是可循环屏障，可以多次进行计数，当一个计数为 0 时，又会重新开始计数，可循环的一个组件，和 CountDownLatch 相比，它可以多次。在线程调用 await() 方法会阻塞停止运行，等到线程数量到达屏障数量的时候才会释放这些停止的线程，会调用 signalAll() 方法唤醒所有等待线程，这个方法相当于就是将条件队列的节点转换为同步队列节点来存储，下面会详细介绍这个方法.

### 二、CyclicBarrier 源码剖析 ###

CyclicBarrier 只有一个重要的方法，就是 await()，await(timeout) 只是在 await() 方法的基础上加上了超时，其他类似.

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

public CyclicBarrier(int parties) {// parties 是屏障数量,临界点
        this(parties, null);
    }

该构造只有一个参数，为 int 类型的屏障数量值，当线程数量到达这个临界点会触发自定义的执行代码，**自定义的执行代码还未确定好**

public CyclicBarrier(int parties, Runnable barrierAction) {
        if (parties <= 0) throw new IllegalArgumentException();
        this.parties = parties;
        this.count = parties;
        this.barrierCommand = barrierAction;// 传入到达屏障点要运行的动作
    }

该构造也是上一个构造的具体实现，此方法有一个可执行的动作 **barrierAction** ，用于到达屏障点来执行的代码，**这段执行代码是已经确定好的**

#### 2.成员变量 ####

private static class Generation {
        boolean broken = false;// 记录是否出现异常和是否进入下一代
    }
    /** await 方法入口所需锁 */
    private final ReentrantLock lock = new ReentrantLock();
    /** 等待直到屏障数量足够时 */
    private final Condition trip = lock.newCondition();
    /** 屏障数量 */
    private final int parties;
    /* 到达屏障点时要执行的动作 */
    private final Runnable barrierCommand;
    /** 当前代 */
    private Generation generation = new Generation();
    /** 记录还有多少线程数到达屏障点*/
    private int count;

#### 3.核心方法 await() ####

public int await() throws InterruptedException, BrokenBarrierException {
        try {
            return dowait(false, 0L);// false 代表不是超时等待
        } catch (TimeoutException toe) {
            throw new Error(toe); // 不可能发生
        }
    }

await() 方法有对应的重载方法，区别在于是否超时

###### CyclicBarrier.dowait ######

dowait 方法是 wait 方法具体的实现，代码如下：

private int dowait(boolean timed, long nanos)
            throws InterruptedException, BrokenBarrierException,
                   TimeoutException {
        final ReentrantLock lock = this.lock;
        lock.lock();// 获取锁
        try {
            final Generation g = generation;
            if (g.broken) // 判断状态,看当前代是否被破坏
                throw new BrokenBarrierException();
            if (Thread.interrupted()) {// 判断线程是否中断
                breakBarrier();// 中断屏障
                throw new InterruptedException();
            }
            // 线程既然调用了 await() 方法,如果当前线程没有中断或当前代没有被破坏,就会 count-1
            int index = --count;
            if (index == 0) { // 判断当前代的屏障数量是否足够 
                boolean ranAction = false;
                try {
                    final Runnable command = barrierCommand;// 检查是否有已确定好的执行动作
                    if (command != null)
                        command.run();// 执行确定好的代码
                    ranAction = true;
                    nextGeneration();// 进行下一代,方法内部有调用 signalAll() 唤醒所有阻塞的线程
                    return 0;
                } finally {
                    if (!ranAction)// 主要是防止服务宕机的情况
                        breakBarrier();
                }
            }
    
            // 循环直到屏障数量足够到达下一代,或者中断,或者 await(timeout) 方法超时
            for (;;) {
                try {
                    if (!timed)
                        trip.await();// 一直阻塞
                    else if (nanos > 0L)
                        nanos = trip.awaitNanos(nanos);// 带有超时的阻塞
                } catch (InterruptedException ie) {
                    // 如果线程被打断,并且当前代未被修改过
                    if (g == generation && ! g.broken) {
                        breakBarrier();// 那么要关闭当前代的执行,并且将已到达的屏障线程唤醒
                        throw ie;
                    } else {
                        // 否则:已经进入到下一代或者当前代状态不正常,那么会在下方清理线程状态
                        Thread.currentThread().interrupt();
                    }
                }
                if (g.broken)// 如果当前代被破坏,则抛出异常
                    throw new BrokenBarrierException();
                if (g != generation) // 已经进入到下一代了,则返回下一代未到达屏障数量的线程数,比如屏障数量为 5,已到达 3 个线程数,那么还剩下 2 个线程数未到达.
                    return index;
                if (timed && nanos <= 0L) { // 如果超时了,也要关闭屏障
                    breakBarrier();
                    throw new TimeoutException();
                }
            }
        } finally {
            lock.unlock();// 释放锁
        }
    }

通过上面的代码，总结归纳下 dowait() 方法的逻辑：

*  线程调用后，会检查当前代是否被破坏(**broken**)，当前调用线程是否被中断
 *  在初始化 CyclicBarrier 时，设置的屏障数量 count，会进行 count--
 *  假如有 10 个线程，前面 9 个线程执行 dowait() 方法时，由于 count!=0 会进入到 **for(;;)**，会执行 **trip.await();** 实现当前线程的阻塞，也就是 Condition 内部的 wait 方法，Condition 的 wait 方法是由 AQS 来实现的，**下面会具体分析该方法**
 *  阻塞结束的条件有：线程超时(**调用的是 await(timeout)**)、当前代被破坏、当前线程阻塞、当前线程被唤醒(**可能是进入到了下一代或者是程序运行完成**)
 *  当第 10 个线程执行 dowait() 方法时，由于 count==0，会先检查 command 是否有确定好的执行代码，有则执行，无则执行不确定的执行代码
 *  最后执行nextGeneration()，在内部调用 trip.signalAll() 唤醒所有trip.await() 的线程

###### CyclicBarrier.nextGeneration ######

该方法表示当前代已经结束(**count==0** )时，进入下一代，代码如下：

// 开启下一代,重置所有参数
    private void nextGeneration() {
        // 唤醒上一代所有等待的线程
        trip.signalAll();
        // 屏障数量重置
        count = parties;
        generation = new Generation();// 重新实例化当前代对象
    }

###### CyclicBarrier.breakBarrier ######

当前线程出现异常或当前代被破坏时，将中断当前代，并释放已经到达屏障点的所有其他线程，代码如下：

// 在屏障数量不够或出现异常时破坏/打断当前代
    private void breakBarrier() {
        generation.broken = true;
        count = parties;
        trip.signalAll();// 唤醒所有阻塞线程
    }

###### Condition.wait ######

在屏障数量不足够时，当有线程执行到 dowait() 方法时，会执行到 **trip.wait()** 方法，这个方法其实是调用了 Condition 接口下的 wait() 方法，代码如下：

void await() throws InterruptedException;

Condition 接口只是定义了方法，并没有做任何的实现，具体的实现是由 AQS 的内部类 **ConditionObject** 来完成的：

[![alt][]

![alt][]][alt_ _alt]

###### AQS.ConditionObject.wait() ######

public final void await() throws InterruptedException {
        if (Thread.interrupted())// 当前线程被中断抛出异常
            throw new InterruptedException();
        Node node = addConditionWaiter();// 向条件队列中添加节点,并返回添加后的节点
        int savedState = fullyRelease(node);// 释放当前线程获取的锁
        int interruptMode = 0;
        // 判断当前节点是否为同步队列节点,如果不是则会阻塞当前线程
        while (!isOnSyncQueue(node)) {
            LockSupport.park(this);
            // 说明 signalAll() 被调用了或者线程被中断,如果因为中断被唤醒,则跳出循环
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
        // while 循环结束,开始抢锁,表示当前线程已经被中断过并且之前有线程已经执行过了 signalAll 方法
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        // 如果当前节点在条件队列里面的引用还没有被移除,所以这边先对不可用的条件队列节点进入移除
        if (node.nextWaiter != null)
            unlinkCancelledWaiters();// 清理不为 CONDITION 状态的条件队列节点
        if (interruptMode != 0) // 判断该值看是否要抛出异常还是设置仅仅只是中断当前线程
            reportInterruptAfterWait(interruptMode);
    }

通过上面的代码，总结归纳下 wait() 方法执行的逻辑：

*  将当前线程封装为节点，添加到条件队列(ConditionQueue)的尾部
 *  释放当前线程获取的锁(通过操作 state 的值)
 *  判断当前节点是否在同步队列中(SyncQueue)中，不在的话使用 park 挂起
 *  循环结束之后，说明已经在同步队列中(SyncQueue)中了，后面等待获取到锁，继续执行即可.

在这里一定要把条件队列和同步队列进行区分：

*  条件队列/等待队列：即 Condition 的队列，是单向的队列实现
 *  同步队列：AQS 的队列，是双向队列的实现

下面是对 await 里面重要方法进行分析：

###### AQS.ConditionObject.addConditionWaiter ######

private Node addConditionWaiter() {
        Node t = lastWaiter;
        // 如果条件队列中尾节点不为空并且它的等待状态不为 CONDITION
        if (t != null && t.waitStatus != Node.CONDITION) {
            unlinkCancelledWaiters();// 清理掉条件队列节点状态不为 CONDITION 的节点
            t = lastWaiter;// 将清理过后的条件队列的尾节点赋给引用变量 t
        }
        // 实例化一个新的条件队列节点
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        if (t == null)// 如果尾节点为空
            firstWaiter = node;// 设置当前节点为头节点
        else // 尾节点不为空
            t.nextWaiter = node;// 将原尾节点的 next 引用指向当前节点 
        lastWaiter = node;// 把当前节点设置为尾节点
        return node;
    }

该方法的主要逻辑是：

*  先检查条件队列是否有需要清理的节点，有的话先进行清理操作
 *  构建好一个 CONDITION 状态的节点添加到等待队列中
 *  检查是否是第一个进入的节点，如果是就将头尾节点都指向当前构建好的节点
 *  不是第一个进入的节点就将原尾节点的 next 引用指向当前节点

###### AQS.fullyRelease ######

final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            int savedState = getState();// 获取当前线程的 state,这个状态是获取锁的标识,重入锁可能是 1 以上的值,获取到独占锁就是会为 1
            if (release(savedState)) { // 释放当前线程获取到的锁
                failed = false;
                return savedState;
            } else {
                throw new IllegalMonitorStateException();
            }
        } finally {
            if (failed) // 如果释放锁失败,就设置当前条件队列节点为 CANCELLED,待清理状态
                node.waitStatus = Node.CANCELLED;
        }
    }

###### AQS.release ######

public final boolean release(int arg) {
        if (tryRelease(arg)) {// 调用 ReentrantLock 释放锁
            Node h = head;
            if (h != null && h.waitStatus != 0)// 查看同步队列中是否有需要被唤醒的节点
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

释放当前线程锁的方法执行逻辑：

*  尝试释放当前线程获取到的锁，释放成功则执行 await() 下面的代码，失败则将当前节点状态修改为待清理状态

###### AQS.isOnSyncQueue ######

final boolean isOnSyncQueue(Node node) {
        if (node.waitStatus == Node.CONDITION || node.prev == null)// 同步队列节点是否有节点,无节点返回 true
            return false;
        if (node.next != null) // 查看是否有后继节点,有的话说明当前节点存在于同步队列中
            return true;
        // 从同步队列的尾节点开始扫描是否存在于当前节点的引用,有则返回 true
        return findNodeFromTail(node);
    }

查看当前节点是否是同步队列中的节点，该方法的主要逻辑是：

*  判断当前节点状态是否为 CONDITION 或当前节点的前置是否为空，其中一个是的话就返回 false，说明当前节点不是同步队列节点
 *  判断当前节点的 next 引用是否为空，为空则继续在整个同步队列中查找是否存在当前节点引用，否则返回 true

###### AQS.checkInterruptWhileWaiting、transferAfterCancelledWait ######

private int checkInterruptWhileWaiting(Node node) {
        // 判断线程是否被中断过,有则通过 CAS 尝试去将当前节点添加到同步队列中,CAS 成功返回 -1,失败返回 1
        // 返回 REINTERRUPT 表示已经执行了 signalAll() 方法
        return Thread.interrupted() ?
            (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
            0;
    }

final boolean transferAfterCancelledWait(Node node) {
        // CAS 尝试修改当前节点的状态
        if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
            enq(node);// CAS 成功就将当前节点添加到同步队列中
            return true;
        }
        /*
            CAS 失败表示当前节点可能不是 CONDITION 状态了,就是有线程执行了 signalAll 方法
            signalAll 方法中的 transferForSignal 方法将节点状态改为 SIGNAL 和将节点加入队列
            的这两个操作不是原子操作,所以可能存在并发问题,也就是说可能会存在更新节点状态的时候,还
            未将当前节点添加到同步队列中,下面就是考虑这种场景,才需要礼让当前线程的 CPU 资源等待 signalAll
            方法将节点添加到同步队列中
        */
        while (!isOnSyncQueue(node))
            Thread.yield();
        return false;// 表示有线程执行了 signalAll 方法,当前节点已经是同步队列节点
    }

将当前线程执行**(LockSupport.park(this);)**阻塞以后，在接着执行上面的这段代码，梳理一下以上代码的执行逻辑：

*  检查当前线程是否被中断过,如果已经被中断过，说明有方法执行了 **signalAll()** 方法，**signalAll() 方法执行的时候会将所有条件队列的节点转移到同步队列中**.
 *  transferAfterCancelledWait() 方法中会先通过 CAS 将当前节点状态修改为初始状态 0，成功的话就将其添加到同步队列中，失败的话说明全部的条件队列节点都已经修改状态为 0 或者已经移交到了同步队列
 *  上面代码注释有详细阐述，如果只是修改节点状态为 0 的话就说明还没有在条件队列中，这时候当前线程就会礼让一些 CPU 资源给到执行 signalAll() 方法的线程优先完成

###### AQS.ConditionObject.reportInterruptAfterWait ######

private void reportInterruptAfterWait(int interruptMode)
        throws InterruptedException {
        if (interruptMode == THROW_IE) // 未执行 signalAll 方法则抛出异常
            throw new InterruptedException();
        else if (interruptMode == REINTERRUPT)// 执行过 signalAll 方法则中断当前线程即可
            selfInterrupt();
    }

执行到这里的时候，线程之间会先去竞争锁，如果线程已经被中断并且之前有方法已经执行过了 signalAll() 方法，这时候为了保险起见再次将状态修改为中断状态，状态值如下：

/** Mode meaning to reinterrupt on exit from wait */
    private static final int REINTERRUPT =  1; // 从等待状态切换为中断状态
    /** Mode meaning to throw InterruptedException on exit from wait */
    private static final int THROW_IE    = -1; // 抛出异常标识

接着会判断当前节点是否有存在于条件队列中节点的 next 引用，如果还有的话从条件队列中进行当前节点的清理，执行的是 unlinkCancelledWaiters 方法

最后判断的是 interruptMode 的值，为 -1 抛出异常，为 1 则中断线程.

#### 4.核心方法 signalAll ####

###### AQS.ConditionObject.signalAll() ######

public final void signalAll() {
        if (!isHeldExclusively())// 如果当前线程不是独占锁线程的话抛出异常,不可能发生
            throw new IllegalMonitorStateException();
        Node first = firstWaiter;// 获取条件队列中的首节点
        if (first != null)
            doSignalAll(first); // 将条件队列节点转换为同步队列节点
    }

Condition 类下的 signal 和 signalAll 方法都是由 AQS.ConditionObject 内部类来实现的，该方法主要实现的就是将条件队列中所有节点转换为同步队列节点

###### AQS.ConditionObject.doSignalAll ######

private void doSignalAll(Node first) {
        lastWaiter = firstWaiter = null;// 将条件队列中的首尾节点指向 null,删除
        do {
            Node next = first.nextWaiter;// 获取首节点的下一个节点
            first.nextWaiter = null;// 将首节点的 next 引用指向 null 
            transferForSignal(first);// 将条件队列所有节点转换为同步队列节点并修改所有节点状态为 SIGNAL
            first = next;// 下一个节点变为首节点
        } while (first != null);// 直到条件队列中的节点都清空为止才退出循环
    }

该方法会把条件队列中的节点在循环里面进行清理，其中包含链表的 next 引用取消，移除一个节点也会同时向同步队列中添加一个节点并修改状态为 SIGNAL，SIGNAL 表示待唤醒的一个状态值.

###### AQS.transferForSignal ######

final boolean transferForSignal(Node node) {
        // 如果 CAS 失败,说明当前节点已经为待清理状态
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        // 走到这里说明该状态已经被修改成了初始状态 0,将其加入到同步队列尾部,并返回前一个节点
        Node p = enq(node);
        int ws = p.waitStatus;
        // ws>0：原来的 tail 节点如果已经是待清理状态
        // !compareAndSetWaitStatus(p, ws, Node.SIGNAL)↓
        // CAS 失败的情况就是并发场景下在执行 ws>0 是成功的,
        // 但执行 CAS 操作时原节点被移除了,这边是可以立即唤醒新的尾部节点线程,
        // 减少了 AQS.ConditionObject#await 方法执行次数,
        // 也就是会在大并发的场景下提升一定的性能,不可忽视
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

该方法执行的大致逻辑如下：

*  先修改当前的状态值为初始值 0，修改失败代表当前节点已经被清理或已经修改为了 0
 *  CAS 成功以后就把当前节点加入到同步队列中并返回当前节点的前置节点也就是之前的尾部节点
 *  接下来会判断这个尾节点是否待清理，在这个时候其实是做了优化处理的，在我们执行 await 方法时有一个地方同步队列中多个节点去竞争锁，在这里做的优化处理，如果这个节点在转移的时候之前的同步队列尾节点已经释放了锁，此时之前的尾节点状态必是待清理的且 CAS 肯定会失败，这个时候当前节点所持有的线程就会马上被唤醒不需要在去争抢这把锁，在大并发场景下这种优化是不可忽略的.

##### 总结 #####

到这里，对于整个 CyclicBarrier 循环屏障的相关源码做了大致的阅读和理解，整体来说：循环的每一次操作都是在当前这个代进行的，线程在不足够屏障数量的时候会执行 await 方法去构建成 CONDITION 状态的节点并添加到条件队列中，在线程数量达到所需要的屏障数量时这个当前线程会执行 signalAll() 方法去一个个将条件队列的节点转移到同步队列中，这个时候阻塞在 await() 方法代码处的线程就会继续向下执行，按照正常的争抢锁逻辑来执行，在当前代屏障数量的所有线程全部完成执行，这个时候就会执行我们自定义的执行代码(**这段代码是可以提前确定的**).

整个的执行流程就是按照这样子来完成的，这里面涉及到了 Codition 和 AQS 的操作，有大多的细节可能没有讲解到位，望体谅

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

[alt]: https://image.dandelioncloud.cn/pgy_files/images/2024/03/11/fbe751dff1f04dfcab0f4fa280e68072.png
[alt_ _alt]: https://i0.wp.com/www.vnjohn.com/wp-content/uploads/2021/01/image-27-1024x286.png?ssl=1