AQS内部ConditionObject原理剖析前言ConditionObject原理剖析一、 AQS 双队列底层数据结构与内存布局1. Node 内部类的双重角色定义2. 双队列关键特性对比二、 ConditionObject.await() 源码深度剖析1. await() 整体流程图解2. 详细源码与逐行注释(1) await() 主方法(2) addConditionWaiter()条件队列入队(3) fullyRelease(Node node)锁全量释放(4) isOnSyncQueue(Node node)跨队列状态校验三、 ConditionObject.signal() / signalAll() 源码深度剖析1. signal() 源码解析2. signalAll() 源码解析3. transferForSignal(Node node)节点转移核心逻辑四、 双队列转移Transfer与并发竞争状态机1. 中断响应与 Signal 竞争状态机2. 并发竞争路径胜负判定矩阵3. reportInterruptAfterWait(int interruptMode)五、 关键边缘场景与设计细节1. unlinkCancelledWaiters()无锁条件队列的垃圾清理机制2. 为什么 fullyRelease() 必须采用“一次性全量释放”六、 系统工程设计哲学总结前言本文旨在记录近期研读Java源码的学习心得与疑难问题。由于个人理解水平有限文中内容难免存在疏漏恳请读者不吝指正。ConditionObject原理剖析一、 AQS 双队列底层数据结构与内存布局在 OpenJDK 8的AbstractQueuedSynchronizer(AQS) 设计中管程Monitor模型的实现依托于两套协作但解耦的队列系统同步队列Sync Queue / CLH 变体队列与条件队列Condition Queue。两套队列均复用AbstractQueuedSynchronizer.Node静态内部类但其指针拓扑形态、线程安全保证及节点状态迁移规则存在本质差异。 1. 同步队列 Sync Queue (双向链表带 Dummy Head 节点CAS 无锁并发维护) ------------ ------------ ------------ | Dummy Head | | Node | | Tail Node | | | | Thread A | | Thread B | ------------ next ------------ next ------------ waitStatus: -1 waitStatus: -1 waitStatus: 0 (SIGNAL) (SIGNAL) 2. 条件队列 Condition Queue (单向链表无 Dummy 节点独占锁保护无需 CAS) ------------ nextWaiter ------------ nextWaiter ------------ | First | ------------- | Node | ------------- | Last | | Thread C | | Thread D | | Thread E | ------------ ------------ ------------ waitStatus: -2 waitStatus: -2 waitStatus: -2 (CONDITION) (CONDITION) (CONDITION)1.Node内部类的双重角色定义Node结构体包含了支撑双队列所需的全部指针字段与状态标识staticfinalclassNode{/** 模式标记共享模式与独占模式 */staticfinalNodeSHAREDnewNode();staticfinalNodeEXCLUSIVEnull;/** 节点等待状态 waitStatus 的枚举值 */staticfinalintCANCELLED1;// 线程因超时或中断被取消终态staticfinalintSIGNAL-1;// 表示后继节点处于 park 挂起状态当前节点释放锁时必须唤醒后继staticfinalintCONDITION-2;// 节点位于 Condition 条件队列中staticfinalintPROPAGATE-3;// 共享模式下无条件向后传播唤醒状态/** * 节点的状态 * 在 Sync Queue 中可为 CANCELLED, SIGNAL, PROPAGATE 或 0 * 在 Condition Queue 中只能为 CONDITION 或 CANCELLED。 */volatileintwaitStatus;/** 同步队列 (Sync Queue) 的双向指针 */volatileNodeprev;volatileNodenext;/** 当前节点绑定的物理线程 */volatileThreadthread;/** * 关键复用指针 nextWaiter * 1. 在 Condition Queue 中作为单向链表的下一个节点指针 * 2. 在 Sync Queue 中用于标记节点模式SHARED 节点或 EXCLUSIVE 节点。 */NodenextWaiter;finalbooleanisShared(){returnnextWaiterSHARED;}finalNodepredecessor()throwsNullPointerException{Nodepprev;if(pnull)thrownewNullPointerException();elsereturnp;}Node(){}// 用于创建 Dummy Head 或 SHARED 标记节点Node(Threadthread,Nodemode){// 用于 Sync Queue 入队this.nextWaitermode;this.threadthread;}Node(Threadthread,intwaitStatus){// 用于 Condition Queue 入队this.waitStatuswaitStatus;this.threadthread;}}2. 双队列关键特性对比物理与逻辑维度同步队列 (Sync Queue)条件队列 (Condition Queue)拓扑结构双向链表带 Dummy 头节点解决边界极值单向链表不带 Dummy 节点直接指向真实节点指针域prev,nextnextWaiter头尾指针AQS 域head,tail(被volatile修饰)ConditionObject域firstWaiter,lastWaiter并发控制机制无锁 CAS 机制(入队/出队可能面临多线程竞争)锁保护机制(调用方必已持有独占锁无需 CAS)节点 waitStatusCANCELLED(1),SIGNAL(-1),PROPAGATE(-3),0CONDITION(-2),CANCELLED(1)线程阻塞原因抢占独占锁或共享许可失败被迫排队主动放弃锁等待业务逻辑信号条件Signal成立二、ConditionObject.await()源码深度剖析await()是条件变量中最复杂的逻辑其内部涵盖了条件入队、锁资源全量释放、挂起自旋检测、双队列转移以及中断状态修复等多个原子与非原子阶段。1.await()整体流程图解------------------ | 调用 await() | ------------------ | v [ 检查线程中断标记 ] ----(已被中断)---- 抛出 InterruptedException | v [ addConditionWaiter() ] 封装 Node(waitStatus-2) 并插入 Condition 队列尾部 | v [ fullyRelease(node) ] 获取当前 state一次性释放所有重入锁唤醒 Sync Queue 后继 | v ------------------------------------------------------- | 循环检测: while (!isOnSyncQueue(node)) | | 1. LockSupport.park(this) 阻塞当前线程 | | 2. 被唤醒后检查中断: checkInterruptWhileWaiting() | ------------------------------------------------------- | (节点已被转移至 Sync Queue) | v [ acquireQueued(node, savedState) ] 在 Sync Queue 中阻塞竞争锁恢复原 state 重入次数 | v [ unlinkCancelledWaiters() ] 若存在取消节点清理 Condition 队列 | v [ reportInterruptAfterWait() ] 根据中断时机抛出异常或补发中断信号2. 详细源码与逐行注释(1)await()主方法publicfinalvoidawait()throwsInterruptedException{// 1. 响应响应式中断如果进入 await 前线程已被中断按规范直接抛出异常if(Thread.interrupted())thrownewInterruptedException();// 2. 将当前线程封装为 CONDITION 状态的 Node追加到 Condition 队列末尾NodenodeaddConditionWaiter();// 3. 【关键操作】完全释放当前线程持有的独占锁解决可重入锁问题如重入 N 次则释放 N// savedState 保存释放前的锁重入次数以便被唤醒后精确还原锁状态intsavedStatefullyRelease(node);intinterruptMode0;// 记录中断模式0 表示无中断THROW_IE (-1)REINTERRUPT (1)// 4. 【挂起与转移检测循环】如果节点不在 Sync 队列中说明还没有被 signal需持续挂起while(!isOnSyncQueue(node)){LockSupport.park(this);// 线程挂起在此处交出 CPU 执行权// 线程被唤醒可能由 unpark、interrupt 或伪唤醒触发后校验是否在挂起期间收到过中断if((interruptModecheckInterruptWhileWaiting(node))!0)break;// 若发生中断打破循环强制向 Sync Queue 转移}// 5. 此时节点已进入 Sync 队列调用 acquireQueued 在 Sync 队列中自旋抢锁// 注意savedState 作为参数传入成功获取锁时 state 将被重置为 savedStateif(acquireQueued(node,savedState)interruptMode!THROW_IE)interruptModeREINTERRUPT;// 6. 清理逻辑如果在转移过程中 node.nextWaiter 不为空说明该节点是通过中断方式退出的// 其 nextWaiter 指针未被 signal() 清理执行一次尾部/全队列无用节点剪枝if(node.nextWaiter!null)unlinkCancelledWaiters();// 7. 处理中断根据 interruptMode 决定抛出异常还是重新设置线程中断标志if(interruptMode!0)reportInterruptAfterWait(interruptMode);}(2)addConditionWaiter()条件队列入队privateNodeaddConditionWaiter(){NodetlastWaiter;// 如果尾节点不为空且状态不是 CONDITION说明尾节点在 Condition 队列中被取消了如超时或中断if(t!nullt.waitStatus!Node.CONDITION){// 触发遍历清理 Condition 队列中的所有非 CONDITION 节点unlinkCancelledWaiters();tlastWaiter;// 重新获取清理后的尾节点}// 构造新节点waitStatus 显式置为 Node.CONDITION (-2)NodenodenewNode(Thread.currentThread(),Node.CONDITION);// 插入单向链表尾部此过程无需 CAS因为调用者必定持有了独占锁if(tnull)firstWaiternode;elset.nextWaiternode;lastWaiternode;returnnode;}(3)fullyRelease(Node node)锁全量释放finalintfullyRelease(Nodenode){booleanfailedtrue;try{intsavedStategetState();// 获取当前锁重入次数// 调用 AQS 模板方法 release() 释放全部锁状态if(release(savedState)){failedfalse;returnsavedState;// 返回重入值用于唤醒后还原}else{// 如果 release 返回 false说明当前线程并未真正持有锁或释放失败thrownewIllegalMonitorStateException();}}finally{if(failed)// 如果释放失败例如非锁持有者调用 await将刚压入 Condition 队列的节点标记为 CANCELLEDnode.waitStatusNode.CANCELLED;}}(4)isOnSyncQueue(Node node)跨队列状态校验该方法用于判断一个节点是否已经成功从Condition Queue转移到了Sync Queue中。finalbooleanisOnSyncQueue(Nodenode){// 状态为 CONDITION (-2) 或前驱指针 prev 为 null必定不在 Sync Queue 中// 注意节点入 Sync Queue (enq) 时第一步就是设置 node.prev tailif(node.waitStatusNode.CONDITION||node.prevnull)returnfalse;// 如果 next 指针不为 null必定在 Sync Queue 中Condition Queue 绝不使用 next 指针if(node.next!null)returntrue;/* * 如果 node.prev ! null 但 node.next null * 说明该节点正处于 enq(node) 的入队过程中 * CAS 刚成功将 node 设置为 tail (Step 2)但尚未执行 pred.next node (Step 3)。 * 此时需要从 tail 往前反向遍历 Sync Queue精确确认 node 是否在 Sync Queue 中。 */returnfindNodeFromTail(node);}privatebooleanfindNodeFromTail(Nodenode){Nodettail;for(;;){if(tnode)returntrue;if(tnull)returnfalse;tt.prev;}}三、ConditionObject.signal()/signalAll()源码深度剖析signal()的物理本质是将 Condition 队列队头排在最前面的未取消节点从 Condition 队列中剥离并利用 CAS 重新挂载到 Sync 队列的尾部由 Sync 队列掌控其后续唤醒机制。1.signal()源码解析publicfinalvoidsignal(){// 1. 验证调用权限必须是独占锁的持有者否则抛出异常if(!isHeldExclusively())thrownewIllegalMonitorStateException();NodefirstfirstWaiter;if(first!null)// 2. 触发转移逻辑doSignal(first);}privatevoiddoSignal(Nodefirst){do{// 将 firstWaiter 指向下一个节点断开旧节点的单向引用if((firstWaiterfirst.nextWaiter)null)lastWaiternull;first.nextWaiternull;// 尝试转移节点。如果 transferForSignal 返回 false说明该节点已被取消ws ! CONDITION// 循环继续寻找 Condition 队列中的下一个有效节点直到转移成功一个或队列为空}while(!transferForSignal(first)(firstfirstWaiter)!null);}2.signalAll()源码解析publicfinalvoidsignalAll(){if(!isHeldExclusively())thrownewIllegalMonitorStateException();NodefirstfirstWaiter;if(first!null)doSignalAll(first);}privatevoiddoSignalAll(Nodefirst){// 清空 Condition 队列引用firstWaiterlastWaiternull;// 循环将 Condition 队列中的所有节点依次转移到 Sync 队列中do{Nodenextfirst.nextWaiter;first.nextWaiternull;transferForSignal(first);firstnext;}while(first!null);}3.transferForSignal(Node node)节点转移核心逻辑此方法是跨队列转换的关键桥梁包含了一个极有深度的优化点避免不必要的立即唤醒Unpark Avoidance。finalbooleantransferForSignal(Nodenode){/* * 1. 使用 CAS 尝试将 waitStatus 从 Node.CONDITION (-2) 修改为 0。 * 如果修改失败说明该节点已被取消例如在 await 挂起期间发生超时或收到中断 * 直接返回 false调用方 doSignal 会继续尝试转移下一个节点。 */if(!compareAndSetWaitStatus(node,Node.CONDITION,0))returnfalse;/* * 2. 调用 enq(node) 将该节点无锁 CAS 压入 Sync 队列尾部。 * 返回值 p 是 node 在 Sync 队列中的【前驱节点】。 */Nodepenq(node);intwsp.waitStatus;/* * 3. 优化策略 * 前驱节点 p 的 waitStatus 代表了前驱对后继 node 的唤醒承诺。 * 我们尝试将前驱 p 的 waitStatus 用 CAS 设置为 Node.SIGNAL (-1)。 * * 如果前驱节点已经被取消 (ws 0)或者 CAS 将前驱设置为 SIGNAL 失败 * 说明前驱节点状态不可靠可能正在释放或取消此时必须【直接显式唤醒】node 的物理线程 * 被唤醒的线程将在 acquireQueued() 的自旋中主动纠正其前驱节点状态。 * * 如果 CAS 设置 SIGNAL 成功则**不需要唤醒 node 线程** * 线程继续在 Sync Queue 中保持 park 状态等待其前驱节点释放锁时顺理成章地唤醒它。 */if(ws0||!compareAndSetWaitStatus(p,ws,Node.SIGNAL))LockSupport.unpark(node.thread);returntrue;}四、 双队列转移Transfer与并发竞争状态机在高并发场景下线程可能在await()挂起时同时收到外部的signal()信号与Thread.interrupt()中断信号。此时 AQS 依靠严格的状态机 CAS 竞争来决定线程的响应顺序。1. 中断响应与 Signal 竞争状态机在await()的while (!isOnSyncQueue(node))循环中被唤醒的线程会调用checkInterruptWhileWaiting(node)privateintcheckInterruptWhileWaiting(Nodenode){returnThread.interrupted()?(transferAfterCancelledWait(node)?THROW_IE:REINTERRUPT):0;}核心判定逻辑封装在transferAfterCancelledWait(Node node)中finalbooleantransferAfterCancelledWait(Nodenode){/* * CAS 判定 1尝试将 waitStatus 从 CONDITION (-2) 改为 0。 * * 成功说明该 CAS 抢在 signal() 的 transferForSignal() 之前执行了 * 即【中断发生在 signal 之前】。 * 此时由中断线程自己负责调用 enq(node) 将节点塞入 Sync 队列 * 并返回 true后续抛出 InterruptedException。 */if(compareAndSetWaitStatus(node,Node.CONDITION,0)){enq(node);returntrue;}/* * 失败说明 CAS 竞争输给了 signal() 的 transferForSignal()。 * 即【signal 抢先发生中断发生在 signal 之后】。 * * 此时 transferForSignal() 正在或已经将节点放入 Sync 队列。 * 这里必须通过自旋 yield 等待直到节点完全完成 enq 入队即 node 已经在 Sync 队列中。 */while(!isOnSyncQueue(node))Thread.yield();returnfalse;// 返回 false后续补发中断标志位}2. 并发竞争路径胜负判定矩阵----------------------------------- | 线程在 await() 挂起中同时面临竞争 | ----------------------------------- | ---------------------------------------------- | | v v 【路径 ASignal 抢先】 【路径 BInterrupt 抢先】 1. signal() 执行 CAS(CONDITION-0) 成功 1. interrupt() 触发挂起唤醒 2. node 被 transferForSignal() 挂载至 Sync Queue 2. transferAfterCancelledWait 执行 CAS(CONDITION-0) 成功 3. 中断在此后到达CAS(CONDITION-0) 失败 3. 中断线程主动调用 enq(node) 将自己放入 Sync Queue 4. checkInterruptWhileWaiting 返回 REINTERRUPT (1) 4. checkInterruptWhileWaiting 返回 THROW_IE (-1) 5. 结果不抛出异常获取锁后重新补发中断信号 5. 结果获取锁后抛出 InterruptedException3.reportInterruptAfterWait(int interruptMode)privatevoidreportInterruptAfterWait(intinterruptMode)throwsInterruptedException{if(interruptModeTHROW_IE)thrownewInterruptedException();elseif(interruptModeREINTERRUPT)selfInterrupt();// 重置当前线程的中断标志位 Thread.currentThread().interrupt()}五、 关键边缘场景与设计细节1.unlinkCancelledWaiters()无锁条件队列的垃圾清理机制在条件队列中如果某个节点因为超时或中断退出了等待该节点会残留在单向链表firstWaiter中。如果不加以清理会导致内存泄漏或无效遍历。AQS 通过unlinkCancelledWaiters()实现了针对单向链表的高效剪枝算法privatevoidunlinkCancelledWaiters(){NodetfirstWaiter;Nodetrailnull;// 用于记录当前最新的有效 (CONDITION) 节点while(t!null){Nodenextt.nextWaiter;// 如果当前节点状态不是 CONDITION说明已被取消if(t.waitStatus!Node.CONDITION){t.nextWaiternull;// 斩断取消节点的 nextWaiter 引用便于 GC 垃圾回收if(trailnull)firstWaiternext;// 若头节点就是取消节点更新 firstWaiterelsetrail.nextWaiternext;// 跳过当前取消节点重新拼接链表if(nextnull)lastWaitertrail;// 若尾节点被清理更新 lastWaiter}else{trailt;// 更新上个有效节点}tnext;}}2. 为什么fullyRelease()必须采用“一次性全量释放”ReentrantLock支持锁的嵌套重入。如果线程重入锁N NN次直接调用release(1)只会递减一次states t a t e N − 1 state N - 1stateN−1此时当前线程仍然持有独占锁。如果在此状态下挂起由于锁未被彻底释放其他线程包括试图调用signal()的线程将永远无法获取独占锁造成死锁。因此fullyRelease必须通过getState()读取全量savedState并一次性全部释放intsavedStategetState();if(release(savedState)){// 释放成功锁计数归零唤醒 Sync Queue 中的下一个排队线程}在重新获取锁时调用acquireQueued(node, savedState)传入之前保存的savedState从而原汁原味地还原了线程在进入await()前的重入深度。六、 系统工程设计哲学总结--------------------------------------------------------------------------------------------------- | AQS Condition 整体架构设计哲学 | --------------------------------------------------------------------------------------------------- | 1. 状态分离 (State Separation) | | - 同步队列负责“互斥锁资源竞争”条件队列负责“业务条件就绪等待”。 | | | | 2. 性能极致控制 (Unpark Avoidance Lockless Transitions) | | - 条件队列的插入与裁剪依托独占锁无需 CAS减少总线高并发下的 Cache Line 失效 | | - signal() 转移节点时仅改 CAS 状态尽可能不直接 Unpark 线程让线程继续在 Sync Queue 中沉睡 | | 完全靠排队锁释放机制唤醒减少系统上下文切换Context Switch损耗。 | | | | 3. 强鲁棒性状态机 (Robust State Machine) | | - 巧妙复用 Node 内存实体通过 CAS 修改 waitStatus (CONDITION - 0) 作为强一致原子裁决点 | | 完美解决了 Signal 信号通知与 Interrupt 中断响应并发交织时的死锁与信号丢失问题。 | ---------------------------------------------------------------------------------------------------