AQS条件队列源码详细剖析

AQS条件队列源码详细剖析

---

0.简介

欢迎来到我的博客：TWind的博客

最好先看过我博客中的 ReentrantLock的超详细源码解析 ，不然想要理解条件队列的源码会非常困难。

AQS中的条件队列相比同步队列略显简单，但依然优异且高效，复杂而严谨，是AQS的一大亮点

注：本文适合想要深究AQS的条件队列的实现和原理的人参考，文章较长

---

1.条件队列

对于AQS中的节点，其拥有多种状态值，比如-2就代表这个节点是属于条件队列的

也就是说，同步队列和条件队列中的节点都是同一个数据结构来保存的，那么，就应该能够互相转化吧？

一个条件队列是一个链表，里面的node同样是同步队列中的node的结构

其拥有两种主要方法：await和signal

await用来让一个已经获得锁的线程让出自己的锁，并把自己包装成condition，进入一个条件队列（一个条件队列由多个condition组成），然后挂起————同步队列转移到条件队列

signal用来从对应的条件队列中唤醒一个/全部的condition，condition被唤醒后会把自己转移到同步队列上，像一个正常的线程那样去抢锁————条件队列转移到同步队列

比如说：

当我执行了一次signal后：

这样Node 1 就会去竞争锁

这也就是条件队列的大致用法，但是这只是一个浅显的介绍，对于中断的处理，对于节点的转移才是条件队列的精华所在。

现在让我们来介绍一下AQS的条件队列的大致结构：

关于Node数据结构的介绍请参考上一篇文章

其中的Node多用到了一个nextWaiter属性：

这个属性在同步队列中用来标志是独享模式还是共享模式，而在条件队列中因为没有这种需求而改成了链接下一个节点

而且这是条件队列的节点的唯一一个个别的节点的联系

所以我们能知道，条件队列是一个单向链表，并不像同步队列那样是双向的

同时，其具有两个特殊值：

firstWaiter：

​ 链接到条件队列的第一个节点

lastWaiter

​ 链接到条件队列的最后一个节点

我们之前常常说，signal的唤醒是随机的，其实在一定程度上是有序的，你想，既然条件队列是由一个单向链表存储的，就不可能随机一个节点去删除。实际上，signal是唤醒firstWaiter来实现唤醒一个condition的

那么，signalAll就显而易见的是从头遍历到尾并唤醒其中每一个condition并将其转移到同步队列之中

那么，为了便于理解，我们就先来剖析一下signal类的代码：

---

signal（）

public final void signal() {if (!isHeldExclusively())	//判断是否为独享模式throw new IllegalMonitorStateException();//不是就抛异常Node first = firstWaiter;if (first != null)	//条件队列队首不为空就执行doSignal(first);
}

这里的isHeldExclusively是返回线程是否是独享模式，条件队列仅支持独享模式，如果是共享模式就会抛出异常

接着获取了条件队列第一个节点first，只要不为空就执行doSignal，跟入：

private void doSignal(Node first) {do {if ( (firstWaiter = first.nextWaiter) == null)	//把头节点设成原头结点的后继节点后判断lastWaiter = null;						//如果为空，说明原本队列只有一个头节点，那也把尾结点设空first.nextWaiter = null;} while (!transferForSignal(first) &&	//如果拿出来的头节点无法转移到同步队列就重复(first = firstWaiter) != null);
}

这里有一个do-while循环，会尝试对从头到尾的第一个节点进行转移，转移成功就退出，找不到也会退出

而且在循环过程中，一直在把头节点向后设置，所以执行完毕后，同步队列会缩减到第一个能转移的节点那里，前面的节点都会被删除

让我们来看看transferForSignal方法：

final boolean transferForSignal(Node node) {/** If cannot change waitStatus, the node has been cancelled.*/if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))	//CAS把条件节点的状态值设成0return false;/** Splice onto queue and try to set waitStatus of predecessor to* indicate that thread is (probably) waiting. If cancelled or* attempt to set waitStatus fails, wake up to resync (in which* case the waitStatus can be transiently and harmlessly wrong).*/Node p = enq(node);		//加入同步队列，返回前驱节点int ws = p.waitStatus;if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))LockSupport.unpark(node.thread);//如果前驱节点被取消或者无法将其状态设成-1（不是-1就不会唤醒后继），就会直接将其唤醒return true;						//从而防止死锁
}

这里会尝试把该节点的 waitStatus 通过CAS操作来修改成0

在上一篇我们知道，对于一个节点，-1代表这个节点会唤醒下一个节点，0代表无状态

那为什么不直接改成-1？原因在于我们还不知道这个节点具体的行为，就如同注释中所说：

[!NOTE]

*   CONDITION:  This node is currently on a condition queue.
*               It will not be used as a sync queue node
*               until transferred, at which time the status
*               will be set to 0. (Use of this value here has
*               nothing to do with the other uses of the
*               field, but simplifies mechanics.)

翻译过来就是，这个节点在转移到同步队列中会被设成0，这个0没有任何特殊的语意，只是单纯的是一个重新抢锁的节点

如果CAS失败无法设成0，就会返回false，这样doSignal会再次while尝试

接着执行Node p = enq(node);，跟入：

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;}}}
}

这里会从后向前遍历尝试把节点接在尾部

执行完Node p = enq(node);，就是

int ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))LockSupport.unpark(node.thread);
return true;

这里有一个很重要的地方：虽然你成功的把节点加入到了同步队列，但是有可能你的前继节点被取消或各种奇奇怪怪的错误

前继节点取消或无法将其状态设成-1（唤醒后继）的话我们就得手动将其唤醒，这样我们就会被唤醒（见下方await部分）执行acquireQueued（详见上一篇）从而由AQS的同步队列将其处理

执行到这里，让我们返回doSignal

private void doSignal(Node first) {do {if ( (firstWaiter = first.nextWaiter) == null)	//把头节点设成原头结点的后继节点后判断lastWaiter = null;						//如果为空，说明原本队列只有一个头节点，那也把尾结点设空first.nextWaiter = null;} while (!transferForSignal(first) &&	//如果拿出来的头节点无法转移到同步队列就重复(first = firstWaiter) != null);
}

这里要么是已经成功转移了要么是CAS失败再次重试（如果是后继节点出错之类的返回的是true，这种错误重试也没用），这里也就是唤醒一个条件节点并尝试将其转移到同步队列的全过程，其实相对同步队列来说相对简单,当然是在你学懂了同步队列的前提下

---

signalAll()

public final void signalAll() {if (!isHeldExclusively())throw new IllegalMonitorStateException();Node first = firstWaiter;if (first != null)doSignalAll(first);
}

显而易见的唤醒全部，直接步入doSignalAll：

private void doSignalAll(Node first) {lastWaiter = firstWaiter = null;do {Node next = first.nextWaiter;first.nextWaiter = null;transferForSignal(first);first = next;} while (first != null);
}

非常的显而易见的遍历所有节点并将其加入同步队列，就不再赘述了

---

awaitUninterruptibly()

接下来的接口都是将节点加入条件队列的接口，相对比较复杂

public final void awaitUninterruptibly() {Node node = addConditionWaiter();	//将本线程包装成一个condition加入int savedState = fullyRelease(node);	//既然要加入条件队列等待唤醒，那就先把自己拿到的锁全部释放boolean interrupted = false;while (!isOnSyncQueue(node)) {	//一直阻塞等待被唤醒，如果是被中断唤醒的就设置中断标记，如果节点在中同步队列说明是被头节点LockSupport.park(this);		//调用release唤醒的，那就退出然后执行下面的acquireQueued尝试抢锁if (Thread.interrupted())interrupted = true;}if (acquireQueued(node, savedState) || interrupted)	//如果抢锁过程被中断过或者在条件队列等待时被中断过，就会调用一下selfInterrupt();							//selfInterrupt，重新设置一下中断位，因为interrupted会清除中断位
}

这是最简单的把节点加入条件队列的方法，因为其不会抛出中断异常，省去了很多判断

会先把节点加入条件队列，然后将其堵塞，等待唤醒

这里要么会被中断唤醒,要么是被同步队列唤醒，如果此时已经在同步队列中的就可以尝试抢一次锁（acquireQueued），如果这个节点没抢到那就在条件队列阻塞，因为完全可能是还没有排到队头就被另外一个中断唤醒了

看一下fullyRelease：

final int fullyRelease(Node node) {boolean failed = true;try {int savedState = getState();	//获取当前锁重入次数if (release(savedState)) {		//释放锁failed = false;return savedState;} else {throw new IllegalMonitorStateException();}} finally {if (failed)node.waitStatus = Node.CANCELLED;}		//如果抛异常代表释放出错，就把该节点设成CANCELLED，让同步队列处理
}

因为必须要有锁才能await，所以这里大概率是能成功release掉锁的，release不掉就取消节点交给同步队列（release会唤醒同步队列的下一个节点）

看一看加入条件队列的代码：

private Node addConditionWaiter() {Node t = lastWaiter;// 获取最后一个节点if (t != null && t.waitStatus != Node.CONDITION) {unlinkCancelledWaiters();	//如果这个节点不合法，就会取消掉它t = lastWaiter;		//获取新的尾节点，unlinkCancelledWaiters会清除整个条件队列的不合法节点}	Node node = new Node(Thread.currentThread(), Node.CONDITION);if (t == null)firstWaiter = node;elset.nextWaiter = node;lastWaiter = node;	//单纯的新建一个节点加进去return node;
}

会尝试加在尾部，如果发现尾部节点不合法，会触发清理，再加上去

看一看unlinkCancelledWaiters是怎么工作的：

private void unlinkCancelledWaiters() {Node t = firstWaiter;Node trail = null;	//维护上一个合法的节点while (t != null) {Node next = t.nextWaiter;if (t.waitStatus != Node.CONDITION) {t.nextWaiter = null;if (trail == null)firstWaiter = next;elsetrail.nextWaiter = next;if (next == null)lastWaiter = trail;}elsetrail = t;t = next;}
}

非常的显而易见的从头遍历到尾，取消掉每一个不合法的节点（status不为CONDITION）

isOnSyncQueue就不贴了，就是一个重后向前遍历找节点的方法

---

await()

这是重头戏，引入了对中断的详细处理

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);	//阻塞if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)	//调用checkInterruptWhileWaiting来判断中断模式break;			//如果有中断就跳出来处理}if (acquireQueued(node, savedState) && interruptMode != THROW_IE)interruptMode = REINTERRUPT;	//如果获取锁阶段被打断而且打断模式不是THROW_IE，就会简单设置中断标记if (node.nextWaiter != null) //如果下一个条件队列节点为空，就会清理一遍队列unlinkCancelledWaiters();if (interruptMode != 0)		//根据对应的中断类型执行不同操作reportInterruptAfterWait(interruptMode);
}

这里会尝试加入条件队列并阻塞，并在唤醒后看看自己是不是被中断的，如果是的话就判断中断的类型，采取相应的操作，不像上面的awaitUninterruptibly直接忽略异常

让我们来看看checkInterruptWhileWaiting是怎么判断异常的：

private int checkInterruptWhileWaiting(Node node) {return Thread.interrupted() ?(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :0;
}

如果线程被中断(Thread.interrupted)就会判断transferAfterCancelledWait，是真返回THROW_IE（抛出异常），否则返回REINTERRUPT（重新设置中断位）

步入transferAfterCancelledWait看看

final boolean transferAfterCancelledWait(Node node) {if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {	//注意！如果这个CAS能成功说明这个节点还没有进入同步队列enq(node);										//也就是说还没有signal，那就直接将其移入同步队列然后等待获取锁后return true;									//将其清除}while (!isOnSyncQueue(node))		//如果到了这里，说明一开始的节点已经或正在进入同步队列，就是已经被signal了Thread.yield();					//那就不断的yield等待其成功进入同步队列再进行下一步操作return false;
}

这里会判断这个中断的发生时间：是在signal前面还是后面

因为如果signal没发生那CAS就能成功，那就手动加入同步队列等待其拿到锁后让AQS将其删除，随后抛出异常

不能草率的直接将其删除，因为后面你还得执行acquireQueued来拿回锁来统一处理

如果已经signal，就等到signal完成后重设一下中断位即可

接下来，就根据判断的结果调用reportInterruptAfterWait：

private void reportInterruptAfterWait(int interruptMode)throws InterruptedException {if (interruptMode == THROW_IE)throw new InterruptedException();else if (interruptMode == REINTERRUPT)selfInterrupt();
}

如果是THROW_IE，对应未调用signal，就抛出异常

否则，只是修改符号位

---

awaitNanos(),awaitUntil(Date deadline),await(long time, TimeUnit unit)

判断逻辑和await()没什么区别，唯一的不同是加了个超时机制

这里以awaitNanos示例：

public final long awaitNanos(long nanosTimeout)throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();Node node = addConditionWaiter();int savedState = fullyRelease(node);final long deadline = System.nanoTime() + nanosTimeout;	//设置超时时间int interruptMode = 0;while (!isOnSyncQueue(node)) {if (nanosTimeout <= 0L) {	//如果超时就取消该节点transferAfterCancelledWait(node);break;}if (nanosTimeout >= spinForTimeoutThreshold)LockSupport.parkNanos(this, nanosTimeout);if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)break;nanosTimeout = deadline - System.nanoTime();}if (acquireQueued(node, savedState) && interruptMode != THROW_IE)interruptMode = REINTERRUPT;if (node.nextWaiter != null)unlinkCancelledWaiters();if (interruptMode != 0)reportInterruptAfterWait(interruptMode);return deadline - System.nanoTime();
}

其他地方一模一样

---

总结

总的来说，条件队列相对简单，围绕着：包装成condition->转移到同步队列就能很容易理解

当然，这一切都建立在你熟悉同步队列的实现和原理的前提之下

后面会介绍分享模式，读写锁之类的，欢迎捧场！