AQS双向队列状态迁移源码剖析

AQS双向队列状态迁移源码剖析前言AQS双向队列状态迁移源码剖析一、 AQS 双队列架构概述二、 核心流程从 await() 到 signal() 的节点转移三、 OpenJDK8源码详细逐行剖析1. 节点的产生与入队Condition Queue2. 彻底释放锁AQS State 归零3. 唤醒与节点转移的桥梁signal()4. 转移的核心实现transferForSignal(Node)5. 同步队列的尾插法enq(Node)6. 自旋检测isOnSyncQueue(Node)四、 核心设计思想与工程细节1. **为什么在 transferForSignal 中改变状态要用 CAS**2. **前驱节点状态转化为 Node.SIGNAL 的深意**3. **无锁化Lock-Free的双向队列尾插安全**前言本文旨在记录近期研读Java源码的学习心得与疑难问题。由于个人理解水平有限文中内容难免存在疏漏恳请读者不吝指正。AQS双向队列状态迁移源码剖析在Java并发包java.util.concurrent中AbstractQueuedSynchronizerAQS是构建锁和其他同步组件的核心基石。作为软件开发工程师深入理解其双向队列Sync Queue即同步队列/CLH队列与单向条件队列Condition Queue之间的节点转移机制是掌握高并发底层原理的必经之路。下面我们将结合OpenJDK8的具体源码深度剖析当线程调用Condition.await()和Condition.signal()时AQS 内部节点Node是如何在两个队列之间进行流转与状态切换的。一、 AQS 双队列架构概述AQS 内部维护了两种互补的队列结构同步队列Sync Queue / CLH Queue结构双向链表。通过head和tail指针维护。节点关系节点间通过prev和next指针双向链接。作用管理等待获取锁的线程。条件队列Condition Queue结构单向链表。由 AQS 的内部类ConditionObject维护通过firstWaiter和lastWaiter指针管理。节点关系节点间仅通过nextWaiter指针单向链接。作用管理因未满足特定条件而调用await()进入等待状态的线程。这两个队列使用的都是 AQS 的静态内部类Node但节点的关键属性和状态waitStatus在不同队列中承载着不同的含义waitStatus Node.CONDITION (-2)代表节点存在于条件队列中。waitStatus Node.SIGNAL (-1)代表该节点的后继节点需要被唤醒存在于同步队列中。waitStatus 0初始化状态。二、 核心流程从await()到signal()的节点转移当一个持有锁的线程调用condition.await()时它会释放锁并进入条件队列挂起当另一个线程调用condition.signal()时处于条件队列首部的节点将被转移到同步队列中等待重新获取锁。这一过程的底层转移逻辑由以下几个核心步骤组成[持有锁的线程] - 调用 await() - 释放锁 | v (封装为 Node.CONDITION) [Condition Queue (单向)] - 线程挂起 (LockSupport.park) | | (另一个线程调用 signal()) v [transferForSignal] - CAS修改状态为0 - 移出 Condition Queue | v (调用 enq() 尾插) [Sync Queue (双向)] - 前驱节点状态设为 SIGNAL - 等待/唤醒三、 OpenJDK8源码详细逐行剖析1. 节点的产生与入队Condition Queue当线程调用await()时首先会调用addConditionWaiter()将当前线程封装为Node并放入条件队列尾部。// java.util.concurrent.locks.AbstractQueuedSynchronizer.ConditionObjectpublicfinalvoidawait()throwsInterruptedException{if(Thread.interrupted())thrownewInterruptedException();// 1. 将当前线程封装为 Node加入到 Condition 队列尾部NodenodeaddConditionWaiter();// 2. 释放当前线程持有的所有锁AQS state并返回释放前的 state以便后续重入intsavedStatefullyRelease(node);intinterruptMode0;// 3. 自旋判断如果当前节点不在同步队列中则挂起线程while(!isOnSyncQueue(node)){LockSupport.park(this);// 线程在此处阻塞挂起// 如果在挂起期间或被唤醒时发生中断则跳出循环处理中断if((interruptModecheckInterruptWhileWaiting(node))!0)break;}// 4. 此时节点已被转移到同步队列调用 acquireQueued 尝试在同步队列中阻塞获取锁if(acquireQueued(node,savedState)interruptMode!THROW_IE)interruptModeREINTERRUPT;if(node.nextWaiter!null)// 清理条件队列中被取消的节点unlinkCancelledWaiters();if(interruptMode!0)reportInterruptAfterWait(interruptMode);}/** * 将当前线程作为等待者添加到条件队列的尾部 */privateNodeaddConditionWaiter(){NodetlastWaiter;// 如果尾部节点已被取消waitStatus 不是 CONDITION则进行清理if(t!nullt.waitStatus!Node.CONDITION){unlinkCancelledWaiters();tlastWaiter;}// 创建一个 waitStatus 为 CONDITION (-2) 的新节点NodenodenewNode(Thread.currentThread(),Node.CONDITION);if(tnull)firstWaiternode;// 队列为空初始化头部elset.nextWaiternode;// 单向链表尾插lastWaiternode;// 更新尾指针returnnode;}2. 彻底释放锁AQS State 归零进入条件队列后线程必须完全释放所持有的锁否则会造成死锁。// java.util.concurrent.locks.AbstractQueuedSynchronizer/** * 释放当前节点持有的所有锁还原 state */finalintfullyRelease(Nodenode){booleanfailedtrue;try{intsavedStategetState();// 释放独占锁若释放失败如未持有锁调用 await会抛出 IllegalMonitorStateExceptionif(release(savedState)){failedfalse;returnsavedState;}else{thrownewIllegalMonitorStateException();}}finally{if(failed)// 释放失败则将节点状态置为 CANCELLED 废弃node.waitStatusNode.CANCELLED;}}3. 唤醒与节点转移的桥梁signal()当另一个线程调用signal()时AQS 开始将条件队列头部的有效节点转移至同步队列中。// java.util.concurrent.locks.AbstractQueuedSynchronizer.ConditionObjectpublicfinalvoidsignal(){// 1. 验证当前线程是否持有独占锁if(!isHeldExclusively())thrownewIllegalMonitorStateException();NodefirstfirstWaiter;if(first!null)// 2. 唤醒条件队列中的第一个节点doSignal(first);}/** * 循环遍历条件队列找到第一个未取消的节点并将其转移到同步队列中 */privatevoiddoSignal(Nodefirst){do{// 将 firstWaiter 指向下一个节点断开当前节点的单向链接if((firstWaiterfirst.nextWaiter)null)lastWaiternull;first.nextWaiternull;// 帮助 GC}while(!transferForSignal(first)// 核心转移方法如果转移失败说明该节点已取消则继续尝试下一个(firstfirstWaiter)!null);}4. 转移的核心实现transferForSignal(Node)transferForSignal是整个 AQS 双向队列与条件队列交互最精妙的地方。它利用无锁 CAS 操作安全地改变节点状态并将节点拼接到同步队列尾部。// java.util.concurrent.locks.AbstractQueuedSynchronizer/** * 将节点从条件队列转移到同步队列。若成功则返回 true。 */finalbooleantransferForSignal(Nodenode){/* * 1. 尝试将节点状态由 CONDITION (-2) 通过 CAS 修改为初始状态 0。 * 如果修改失败说明该节点在被唤醒前由于超时或中断已经被取消CANCELLED。 */if(!compareAndSetWaitStatus(node,Node.CONDITION,0))returnfalse;/* * 2. 调用 enq 将该节点尾插到同步队列Sync Queue中。 * 注意enq 方法是死循环自旋的 CAS 插入并返回该节点在同步队列中的【前驱节点 p】。 */Nodepenq(node);intwsp.waitStatus;/* * 3. 核心优化 * 如果前驱节点 p 已经处于取消状态 (ws 0) * 或者尝试将前驱节点 p 的状态修改为 SIGNAL (-1) 失败说明前驱节点在此期间状态发生了改变如取消 * 那么为了安全起见我们必须立刻调用 LockSupport.unpark 唤醒当前节点的线程。 * 被唤醒的线程会在 await() 中的 acquireQueued() 流程中重新同步、自适应修正状态。 * * 如果 CAS 设置前驱为 SIGNAL 成功则当前线程可以继续保持 park 状态 * 因为前驱节点在释放锁时release 流程会负责将当前线程唤醒。 */if(ws0||!compareAndSetWaitStatus(p,ws,Node.SIGNAL))LockSupport.unpark(node.thread);returntrue;}5. 同步队列的尾插法enq(Node)enq方法负责在多线程并发竞争下安全地将节点插入到双向同步队列的尾部并建立起双向链接关系。// java.util.concurrent.locks.AbstractQueuedSynchronizer/** * 将节点插入同步队列尾部必要时进行初始化。 * return 返回插入节点的前驱节点 */privateNodeenq(finalNodenode){for(;;){// 死循环自旋确保 CAS 成功Nodettail;if(tnull){// 1. 队列为空进行初始化if(compareAndSetHead(newNode()))// CAS 创建哑节点 (Dummy Node) 作为 headtailhead;// 此时 head 和 tail 指向同一个哑节点}else{// 2. 队列不为空将新节点的前驱指针 prev 指向当前的 tailnode.prevt;// 3. CAS 尝试将 tail 指向新节点 nodeif(compareAndSetTail(t,node)){// 4. 关键点CAS 成功后将原尾节点的 next 指针指向新节点完成双向链表的闭环t.nextnode;returnt;// 返回前驱节点}}}}6. 自旋检测isOnSyncQueue(Node)在await()的挂起循环中线程通过isOnSyncQueue判断自己是否已经被成功转移到同步队列中。只有当节点存在于同步队列时它才有资格去竞争锁。// java.util.concurrent.locks.AbstractQueuedSynchronizer/** * 判断节点是否已经在同步队列Sync Queue中 */finalbooleanisOnSyncQueue(Nodenode){// 1. 如果状态还是 CONDITION或者前驱指针 prev 仍为 null则一定还在条件队列中if(node.waitStatusNode.CONDITION||node.prevnull)returnfalse;// 2. 如果 next 指针不为空说明它一定在同步队列中因为条件队列只用 nextWaiterif(node.next!null)returntrue;/* * 3. 如果 node.prev ! null 但 node.next null说明该节点可能正在执行 enq 入队。 * 在 enq() 中是先执行 node.prev t再执行 CAS tail最后才执行 t.next node。 * 此时需要从尾部向前遍历同步队列做最后的确认。 */returnfindNodeFromTail(node);}/** * 从同步队列尾部向前遍历寻找是否存在该节点 */privatebooleanfindNodeFromTail(Nodenode){Nodettail;for(;;){if(tnode)returntrue;if(tnull)returnfalse;tt.prev;}}四、 核心设计思想与工程细节作为软件开发工程师我们可以从这段优雅的代码中总结出 AQS 极具工业级水准的底层设计细节1.为什么在transferForSignal中改变状态要用 CAS在多线程环境下一个处于条件队列中挂起的线程可能在被signal()唤醒的同时遭遇了线程中断或超时。如果发生了中断线程自身会调用transferAfterCancelledWait尝试将自己移入同步队列。这就会与signal()线程产生并发冲突。AQS 通过compareAndSetWaitStatus(node, Node.CONDITION, 0)这一行 CAS 操作精妙地充当了“仲裁者”。谁 CAS 成功谁就负责将该节点放入同步队列。失败的一方则避开重复入队从而保证了线程状态的一致性。2.前驱节点状态转化为Node.SIGNAL的深意在同步队列中AQS 采用了一种被动唤醒机制一个节点在同步队列中被挂起后它不需要不断自旋而是靠它的前驱节点在释放锁时来唤醒它。因此在transferForSignal成功将节点尾插到同步队列后必须将前驱节点的状态改为Node.SIGNAL (-1)。这相当于新节点对前驱节点说“老哥我排在你后面了你待会儿释放锁的时候记得叫醒我。”如果前驱节点已经取消或者修改状态失败当前线程就会立刻被unpark唤醒防止其在同步队列中永久“失联”。3.无锁化Lock-Free的双向队列尾插安全在enq(Node node)中双向链表的构建分为三步node.prev t;线程本地安全操作CAS 改变tail指针原子分水岭t.next node;多线程可见这种顺序导致了在极短的瞬间第 2 步完成第 3 步未完成时从 head 顺着next指针往后找可能找不到新节点但从 tail 逆着prev指针往前找是一定能找到的。这也是为什么isOnSyncQueue的退化检查方法findNodeFromTail必须从后往前from tail遍历的原因。