AQS之CyclicBarrier源码流程

2024-09-11 14:18:11   0  举报





CyclicBarrier是AQS框架下的一种同步机制，用于控制多个线程同时到达某个屏障点。它的核心内容包括：首先，通过构造函数设置屏障数，每次调用await()方法时，计数器减一，当计数器减至0时，唤醒所有等待线程。其次，CyclicBarrier支持重置操作，使得屏障可重复使用。此外，CyclicBarrier还提供了getNumberWaiting()方法来获取当前在屏障处等待的线程数量。在源码中，CyclicBarrier使用了AbstractQueuedSynchronizer（AQS）框架，底层通过volatile修饰的state变量和双向队列实现线程同步。

java 并发

CyclicBarrier原理

AQS源码

AQS循环屏障源码

作者其他创作

大纲/内容

trail = firstwaiter

lastwaiter

nextWaiter

conditionnode

同步等待队列

returnfindNodeFromTail(node)

当前线程为中断状态，去破坏屏障

线程被唤醒后去尝试获取锁

设置当前代的 broken 属性为 true

重置count 计数器为下一代屏障做准备

trail

lock.unlock()

fullyRelease(node)

firstwaiter

trip.signalAll()

AbstractQueuedSynchronizer.ConditionObject#await()

条件队列：node 的 nextWaiter 属性指向下一个 node同步队列：node 的 next 属性指向下一个 node

trail.nextwaiter = t.next

记录最新的尾节点

node 有 next node

条件队列转同步队列

finally

return 0

将当前 condition node 转到同步队列中

创建一个新的 Generation 实例，并将其赋值给 generation表示进入下一代屏障

cancellednode

isHeldExclusively()

interruptMode == REINTERRUPT

t.nextWaiter = node

CyclicBarrier#await()

尾节点不是 cancelled node

return true

unlock释放重新获取到的独占锁

AbstractQueuedSynchronizer.ConditionObject#doSignalAll

interruptMode != THROW_IE

Barrier 核心

记录条件等待队列的尾节点

所有参与的线程都已经到达了屏障点，此时屏障被触发

唤醒node对应的线程的目的是为了重新同步，设置正确的prev nodeprev node 如果是 cancelled node 则指向新的prev node，从同步队列中剔除 cancelled nodeprev node 的 waitStaus 更新为 SIGNAL

ws = p.waitStatus

release(savedState)

加锁

first

breakBarrier()

tail

如果非 condition 节点的 prev 不为 null，但还没有在队列中（可能是 CAS tail node 入队失败），则从队列尾部开始遍历，确认节点是否在队列中

唤醒同步等待队列中阻塞的线程

(interruptMode = checkInterruptWhileWaiting(node)) != 0

lastWaiter = firstWaiter = null

入队条件队列

校验当前代的屏障是否被破坏

尾节点是 cancelled node

不是最后到达屏障点的线程

lock加独占锁

BrokenBarrierException

CAS 成功说明线程在条件等待队列中等待期间被中断唤醒

IllegalMonitorStateException

获取锁失败

firstWaiter = node

AQS 维护了同步等待队列

AbstractQueuedSynchronizer#isOnSyncQueue

独占锁

unlinkCancelledWaiters()

重置count 计数器

first = next

LockSupport.unpark(node.thread)

Thread.interrupted()

这里都是 AQS 的一些通用方法，看下https://www.processon.com/diagraming/66cec2deb03fbb660e59a136 即可

线程在对应的node 被 Condition.signal() 方法转到同步等待队列中后才被中断唤醒则设置线程的中断标记

generation.broken = true

Node t = lastWaiter

如果 command.run() 执行失败则去破坏屏障

使用了超时的方法但是传入的超时时间小于等于0则抛出超时异常

条件等待队列中所有节点转换到同步等待队列

入队到条件等待队列末尾

node.next != null

runnable action触发屏障时执行 action

唤醒所有在当前屏障上等待的线程（条件等待队列转成同步等待队列）并进入下一代屏障

AbstractQueuedSynchronizer.ConditionObject#unlinkCancelledWaiters

g != generation

ReentrantLock.Sync

将条件队列中所有的节点转到同步队列中

解锁线程唤醒后获取到的锁

所以这里 ConditionObject 调用 AQS 的方法时实际是走的 ReentrantLock.Sync 对象的实现

ReentrantLock.Sync#tryRelease

释放的是这把锁

selfInterrupt()

判断当前节点是否在同步等待队列中

lock.lock()

这里重新加锁

设计两个指针：t - 当前处理的 node（初始值为 firstWaiter）trail - 记录上一个未取消的 node（初始值为 null）当 t 为 cancelled node 时需要将该 node 从队列中剔除- trail 为 null 时 firstwaiter 需要替换为 t.next- trail 非 null 时直接 trail.nextwaiter = t.next最终的队列就是由 trail 记录过的 node 组成的

Node p = enq(node)

unlock

ReentrantLock lock = this.lock

Y抛出异常

extends

当前屏障的代每次屏障被触发或重置时，代会改变

Node first = firstWaiter

count = parties

开始阻塞

判断当前线程是否是锁的持有者

trip.await()

释放 ReentrantLock锁

trail = null

如果node的waitStatus是CONDITION或者prev node为null

Runnable command = barrierCommand

unpark 唤醒线程或是线程因中断而唤醒检查线程中断情况

解锁最外面的lock

LockSupport.park(this)

如果 CyclicBarrier 创建时设置了Runnable action 那么执行该action

ranAction = true

记录条件队列的头节点

head

timed && nanos <= 0L

N创建 Condition Node

尾节点为空则初始化同步等待队列node 既是头节点也是尾节点

继续去唤醒同步队列中的下个节点

acquire lock重新获取独占锁

unlinkCancelledWaiters

while (!isOnSyncQueue(node))

ws 0（说明 prev node has bean cancelled）或者 CAS 失败则 unpark 当前唤醒当前线程

指针指到 next condition node执行转换操作

return Thread.interrupted() ? (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 0

signal(条件队列转到同步队列)

Thread.yield()

return false

enq(node)

AbstractQueuedSynchronizer#transferAfterCancelledWait

node.nextWaiter != null

抛出异常

t = lastWaiter

park

lastWaiter = node

doSignalAll(first)

interruptMode = REINTERRUPT

AbstractQueuedSynchronizer.ConditionObject#addConditionWaiter

!ranAction

清除条件队列中的 cancelled node 的设计原理流程就不画了，主要还是这个设计真的很巧妙

addConditionWaiter()

唤醒条件等待队列中所有的节点（条件等待队列中所有节点转换到同步等待队列）

return index

删除条件队列

command.run()

next generation开启下一代

node 不在同步队列park 阻塞当前线程

index == 0

Node next = first.nextWaiter

线程在条件等待队列中等待期间被中断唤醒则抛出中断异常

g.broken

判断线程中断情况

获取锁成功

release释放独占锁

AbstractQueuedSynchronizer#transferForSignal

返回前还会执行最外面的finally的lock.unlock()

interruptMode != 0如果线程因中断而被唤醒则需要进行一些操作

1. 线程在条件等待队列中等待期间被中断唤醒则返回 THROW_IE，代表后续需要抛出中断异常2. 线程在对应的node 被 Condition.signal() 方法转到同步等待队列中后才被中断唤醒则返回 REINTERRUPT，代表后续需要设置中断标记3. 如果线程是 unpark 唤醒的（不是中断唤醒），那么返回 0，代表线程没有中断

1. Condition.await() 会在入队条件队列完成后先 release 释放锁2. 所有参与线程到达屏障点触发屏障执行signal 条件队列转到同步队列后，等待 unpark 唤醒后会再次获取锁，最后该锁由 lock.unlock 释放

AbstractQueuedSynchronizer

N校验线程状态

当前线程是否是最后一个到达屏障点

node01

条件等待队列是单向链表结构

for (;;) { // 循环直到屏障触发、屏障被破坏、参与线程中断、等待超时}

从条件等待队列的头节点开始一个个的转到同步等待队列的末尾

ConditionObject

设计原理

CAS失败说明线程在对应的node 被 Condition.signal() 方法转到同步等待队列中后才被中断唤醒

如果 CAS 失败，说明该 node 已 cancelled