AbstractQueuedSynchronizer（AQS） 源码细致分析 - Condition 条件队列流程分析

AQS 源码探究 Condition 条件队列 (手写一个入门的 BrokingQueue)

1、Condition 队列介绍

• AQS 中还有另一个非常重要的内部类 ConditionObject，它实现了 Condition 接口，主要用户实现条件锁。
• ConditionObject 中也维护了一个队列，这个队列主要用于等待条件的成立，当条件成立，其他线程将 signal 这个队列中的元素，将其移动到 AQS 的队列中，等待占有锁的线程释放锁后被唤醒。
• Condition 典型的应用是在 BrokingQueue 中的实现，当队列为空的时候，获取元素的线程阻塞在 notEmpty 条件上，一旦队列中添加了一个元素，将通知 notEmpty 条件，将其队列中的元素移动到 AQS 队列中等待被唤醒。

2、手写一个入门的 BrokingQueue

2.1、自定义 BrokingQueue 接口

/*** @author wcc* @date 2022/2/12 19:23*/
public interface BlockingQueue<T> {/*** 插入数据的接口*/void put(T element);/*** 获取数据的接口*/T take();
}

2.2、自定义 MiniArrayBrokingQueue 类

/*** @author wcc* @date 2022/2/12 19:24*/
public class MiniArrayBrokingQueue implements BlockingQueue{// 线程并发控制private Lock lock = new ReentrantLock();/*** 当生产者线程生产数据的时候，它会先检查当前queues是否已经满了，如果已经满了，需要将当前生产者线程调用notFull.await()* 进入到notFull 的条件队列中挂起，等待消费者线程消费一个数据的时候唤醒*/private Condition notFull = lock.newCondition();/*** 当消费者线程消费数据的时候，它会先检查当前queues中是否存在数据，如果没有数据。需要将当前消费者线程调用notEmpty.await()* 进入到notEmpty条件队列挂起，等待生产者线程生产数据的时候唤醒*/private Condition notEmpty = lock.newCondition();// 存储元素的数组private Object[] queues;// 数组长度private int size;/*** count：表示当前队列中可以被消费的数据量* putptr：记录生产者存放数据的下一次位置，每个生产者生产完一个数据后，会将 putptr++* takeptr：记录消费者消费数据的下一次位置，每个消费者生产完一个数据后，会将 takeptr++*/private int count, putptr, takeptr;public MiniArrayBrokingQueue(int size) {this.size = size;this.queues = new Object[size];}@Overridepublic void put(Object element) {lock.lock();try {// 第一件事？需要判断一下当前queues是否已经满了if(count == this.size){notFull.await();// 执行到这里，说明队列未满，可以向队列中存放数据了this.queues[putptr ++] = element;// 生产数据后更新count的值count ++;if(putptr == this.size){putptr = 0;}// 当向队列中成功放入一个元素之后，需要做什么呢？// 需要给notEmpty一个唤醒信号来唤醒一个消费者线程notEmpty.signal();}}catch (Exception e){e.printStackTrace();}finally {lock.unlock();}}@Overridepublic Object take() {Object element;lock.lock();try {// 第一件事？需要判断一下当前queues是否为空if (count == 0) {notEmpty.await();// 执行到这里，说明队列未满，可以向队列中存放数据了element = this.queues[takeptr++];// 取出数据后更新count 的值count--;if (takeptr == this.size) {takeptr = 0;}// 当向队列中成功消费一个元素之后，需要做什么呢？// 需要给notFull一个唤醒信号来唤醒一个生产者线程notFull.signal();return element;}}catch(Exception e){e.printStackTrace();}finally{lock.unlock();}}public static void main(String[] args) {MiniArrayBrokingQueue<Integer> queue = new MiniArrayBrokingQueue(10);Thread producer = new Thread(new Runnable() {@Overridepublic void run() {int i = 0;while (true){i ++;if(i == 10){i = 0;}try {System.out.println("生产数据："+ i);queue.put(Integer.valueOf(i));TimeUnit.MILLISECONDS.sleep(200);}catch (Exception e){e.printStackTrace();}}}});producer.start();Thread consumer = new Thread(new Runnable() {@Overridepublic void run() {while (true){try {Integer result = queue.take();System.out.println("消费者消费数据："+ result);TimeUnit.MILLISECONDS.sleep(200);}catch (Exception e){e.printStackTrace();}}}});consumer.start();}}

运行结果如下：

3、AQS Condition 条件队列源码分析

3.1、ConditionObject 内部类

这里先介绍一下 AQS 中的 Node 内部类中的关于 Condition 条件队列的两个属性：

// 下一个等待在条件上的节点（Condition锁的时候使用），注意，条件队列是一个单向链表
Node nextWaiter;
// node状态：可选值（0、SIGNAL(-1)、CANCELED(1)、CONDITION、PROPAGATE）
// waitStatus == 0 默认状态
// waitStatus > 0：取消状态（在ReentrantLock模式下）
// waitStatus == -1：表示当前node 如果是 head节点的时候，释放锁之后需要唤醒后继节点
// waitStatus == -2：表示当前node 是位于 Condition 条件队列上的节点
volatile int waitStatus;

上图是关于 AQS 阻塞队列的流程分析：

// 位于 ReentrantLock 类中：创建条件队列的方法
public Condition newCondition() {return sync.newCondition();
}// 位于 RenntrantLock 的静态内部类 Sync中
abstract static class Sync extends AbstractQueuedSynchronizer {final ConditionObject newCondition() {// 而 ConditionObject 是AQS 中的内部类return new ConditionObject();}
}public class ConditionObject implements Condition, java.io.Serializable {private static final long serialVersionUID = 1173984872572414699L;// 指向条件队列中的头结点private transient Node firstWaiter;// 指向条件队列中尾节点private transient Node lastWaiter;....
}

3.2、await() 方法解析

public final void await() throws InterruptedException {// 判断当前线程是否是中断状态，如果是则直接抛出中断异常if (Thread.interrupted())throw new InterruptedException();// 将调用await()方法的线程包装成node并且加入到条件队列中，并返回当前线程的nodeNode node = addConditionWaiter();// 完全释放掉当前线程对应的锁，将state置为0// 为什么要释放锁呢？ 加着锁挂起以后，谁还能将你唤醒呢？ 所以await之后要将锁释放掉long savedState = fullyRelease(node);// 0：在Condition 队列挂起期间未接收过中断信号// -1：在 Condition队列挂起期间接收到中断信号了// 1：在 Condition 队列挂起队列挂起期间未接收到中断信号，但是迁移到阻塞队列之后接收过中断信号int interruptMode = 0;// isOnSyncQueue返回 true：表示当前线程对应的 node 已经迁移到阻塞队列了//              返回false：表示当前线程仍然在条件队列中，需要继续parkwhile (!isOnSyncQueue(node)) {// 挂起条件队列中当前node对应的线程LockSupport.park(this);// 什么时候会被唤醒？// 1.常规路径，外部线程获取到lock后，调用了signal() 方法，转移条件队列的头结点到阻塞队列，这个节点获取到锁后，会唤醒// 2.转移至阻塞队列后，发现阻塞队列中的前驱节点状态是取消状态，此时会唤醒当前节点// 3.当前节点挂起期间被外部线程使用中断唤醒了...// checkInterruptWhileWaiting(node)：就算在 condition 队列挂起期间，线程发生中断了，对应的 node也会被迁移到阻塞队列if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)break;}// 执行到这里，就说明当前 node 已经迁移到阻塞队列了// acquireQueued(node, savedState)：竞争队列的逻辑 如果返回 true：表示在阻塞队列中挂起期间被中断唤醒了// 条件一：返回true：表示在竞争队列中被外部线程中断唤醒过// 条件二：interruptMode != THROW_IE（-1）成立：说明当前node在条件队列内未发生过中断// 设置 interruptMode 设置为重新中断，表示在阻塞队列中进行中断的if (acquireQueued(node, savedState) && interruptMode != THROW_IE)interruptMode = REINTERRUPT;// 考虑下 node.nextWaiter != null 什么时候成立：// 其实是node在条件队列内的时候如果被外部线程中断唤醒的时候，会加入到阻塞队列，但是并未设置nextWaiter = nullif (node.nextWaiter != null) // clean up if cancelled// 清理条件队列中取消状态的当前节点unlinkCancelledWaiters();// 条件成立：说明在挂起期间发生过中断（1.条件队列内的挂起，2.条件队列之外的挂起）if (interruptMode != 0)// 根据发生中断的位置不同进行不同的处理reportInterruptAfterWait(interruptMode);
}

3.3、addConditionWaiter() 方法：将当前线程包装成 node 节点加入到 Condition 队列中

/*** Adds a new waiter to wait queue.* @return its new wait node* 调用 await 方法的线程都是持锁状态的线程，也就是说 addConditionWaiter()这里不存在并发*/
private Node addConditionWaiter() {// 获取当前条件队列的尾节点的引用，保存到条件变量 t中Node t = lastWaiter;// If lastWaiter is cancelled, clean out.// 条件一：t != null 成立：说明当前队列中已经有 node 元素了，需要做追加操作了// 条件二：t.waitStatus != Node.CONDITION  注意，node 在 Condition队列中的时候，它的状态是 CONDITION(-2)//  成立：说明当前node发生中断了if (t != null && t.waitStatus != Node.CONDITION) {// 清理条件队列中所有取消状态的节点unlinkCancelledWaiters();// 更新局部变量t 为最新队列末尾节点，因为上面unlinkCancelledWaiters()可能会更改 lastWaiters引用t = lastWaiter;}// 为当前线程创建 node 节点，设置状态为 CONDITION 状态(-2)Node node = new Node(Thread.currentThread(), Node.CONDITION);// 条件成立：说明条件队列中没有任何元素，当前线程是第一个进入队列的元素if (t == null)firstWaiter = node;else// 说明当前条件队列已经有其他 node了，做追加操作t.nextWaiter = node;// 更新 lastWaiters为最新的条件队列尾节点lastWaiter = node;// 返回当前线程的nodereturn node;
}

3.4、fullyRelease(Node node) 方法：完全释放当前线程所占用的锁

// 完全释放当前线程所占用的锁
final long fullyRelease(Node node) {// 完全释放锁是否成功，当 failed 失败的时候，说明当前线程是未持有锁调用 await()方法的线程(错误写法)// 假设失败，在 finally 代码块中，会将刚刚加入到条件队列的当前线程对应的状态修改为取消状态// 后继线程就会将取消状态的节点给清理出去了boolean failed = true;try {// 获取当前线程所持有的 state 值long savedState = getState();// 绝大部分情况下：release 方法这里会返回 trueif (release(savedState)) {// 失败标记设置为 falsefailed = false;// 返回当前线程释放的 state 值// 为什么要返回 saveState值？// 因为在当你被迁移到阻塞队列后再次被唤醒，且当前node在阻塞队列中是head.next 而且// 当前 lock 状态是 state == 0的情况下，当前 node 可以获取到锁，此时需要将state 设置为 savedState（原先持有锁的状态）return savedState;} else {throw new IllegalMonitorStateException();}} finally {if (failed)node.waitStatus = Node.CANCELLED;}
}//位于AQS的静态内部类Sync中：真正释放锁的方法
//ReentrantLock.unlock() -> sync.release()
public final boolean release(int arg) {//tryRelease尝试释放锁//true:当前线程已经完全释放锁//false:当前线程尚未完全释放锁if (tryRelease(arg)) {//head什么情况下会被创建出来://当持锁线程未释放线程，且持锁期间有其他线程想要获取锁的时候，其他线程发现无法获取锁//且此时阻塞队列是空队列，此时后续线程会为当前持锁线程构建出一个head节点(将持锁线程封装入head)//然后后续线程会追加到head节点的后面(成为head的后驱)Node h = head;//条件1：h!=null成立：说明队列中的head节点已经初始化过了，ReentrantLock在使用期间，发生过多线程竞争//条件2：h.waitStatus!=0 成立，说明head后面一定插入过head节点if (h != null && h.waitStatus != 0)//唤醒后驱节点unparkSuccessor(h);return true;}return false;
}//位于AQS的静态内部类Sync中：尝试释放锁的方法
//true:当前线程已经完全释放锁 | false:当前线程尚未完全释放锁
protected final boolean tryRelease(int releases) {//state状态变量的值相减int c = getState() - releases;//如果条件成立：说明当前线程并未持锁-> 直接抛出异常if (Thread.currentThread() != getExclusiveOwnerThread())throw new IllegalMonitorStateException();//当前线程持有锁//当前线程是否已经完全释放锁，默认初始值为falseboolean free = false;//c==0 条件成立的时候：说明当前线程已经完全释放锁if (c == 0) {//free=true 当前线程已经完全释放锁free = true;//更新当前持锁线程为nullsetExclusiveOwnerThread(null);}//更新state (基于CAS)setState(c);//返回freereturn free;
}

3.5、isOnSyncQueue(Node node) 方法：判断当前 Condition队列 中的 node节点是否在阻塞队列中

// 判断当前 Condition队列 中的 node节点是否在阻塞队列中
final boolean isOnSyncQueue(Node node) {// 条件一：node.waitStatus == Node.CONDITION 成立：说明当前 node 一定是在条件队列中，因为signal方法迁移节点到阻塞队列前，会将node的状态设置为0// 条件二：node.prev == null 前置条件：当前node.waitStatus != Node.CONDITION ==> 1.node.waitStatus = 0（表示当前节点已经被 signal了）//  2.node.waitStatus = 1：当前线程是未持有锁调用 await方法的，最终会将其node修改为取消状态 CANCELED//  node.waitStatus == 0：为什么还要判断 node.prev == null//  因为 signal 方法是先修改状态再迁移if (node.waitStatus == Node.CONDITION || node.prev == null)return false;// 执行到这里，会是哪种情况？// node.waitStatus != CONDITION 且 node.prev != null ==> 可以排除掉 node.waitStatus == 1// 为什么可以排除掉取消状态? 因为signal方法不会把取消状态的node节点迁移走的// 设置 prev引用的逻辑是迁移阻塞队列的逻辑设置的  (enq)// 入队的逻辑：1.设置 node.prev = tail 2.CAS 当前 node 为阻塞队列 的tail 尾节点，成功才算真正进入到了阻塞队列中 3.pred.next = node// 可以推算出就算 prev != null，也不能说明当前node已经成功入队到阻塞队列了// 条件成立：说明当前节点已经成功入队到阻塞队列了，且当前节点后面已经有其他node了..（非尾节点的时候）if (node.next != null) // If has successor, it must be on queuereturn true;/*** 执行到这里，说明当前节点的状态为 node.prev != null且 node.waitStatus != CONDITION 而且可以排除 CANCELED 取消，且 node.waitStatus == 0* findNodeFromTail 从阻塞队列的尾节点开始向前遍历查找 node，如果查找到返回true，查找不到返回false* 当前node有可能正在signal过程中，正在迁移中..还未完成..*/return findNodeFromTail(node);
}

3.6、checkInterruptWhileWaiting(Node node)方法：判断判断当前线程是否被中断并且确定在哪个位置被中断的

private int checkInterruptWhileWaiting(Node node) {// Thread.interrupted()：返回当前线程的中断标记位，并且重置当前标记位为falsereturn Thread.interrupted() ?// transferAfterCancelledWait(node)：这个方法只有在线程是被中断唤醒的时候才会调用(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :0;
}// 判断当前线程节点是在条件队列中被中断还是迁移到阻塞队列中被中断的
final boolean transferAfterCancelledWait(Node node) {// compareAndSetWaitStatus(node, Node.CONDITION, 0)：条件成立：说明当前node一定是在条件队列内，因为signal迁移节点的时候会将节点状态修改为0if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {// 中断唤醒的node也会被加入到阻塞队列中enq(node);// 返回true就表示是在条件队列中被中断的return true;}// 执行到这里有几种情况？// 1.当前node已经被外部线程调用signal方法将其迁移到阻塞队列内了// 2.当前node正在被外部线程调用signal方法将其迁移到阻塞队列中但是还没有完全迁移完成while (!isOnSyncQueue(node))Thread.yield();// false：表示当前节点被中断唤醒的时候不在条件队列了return false;
}

3.7、unlinkCancelledWaiters() 方法：清理条件队列中取消状态的节点

// 清理条件队列中取消状态的节点
private void unlinkCancelledWaiters() {// 表示当前节点，从链表的第一个节点开始向后迭代处理Node t = firstWaiter;// 当前链表中上一个正常状态的node节点Node trail = null;while (t != null) {// 获取当前节点的下一个节点Node next = t.nextWaiter;if (t.waitStatus != Node.CONDITION) {// 更新nextWaiter为nullt.nextWaiter = null;// 条件成立：说明遍历到的节点还未碰到过正常节点if (trail == null)// 更新firstWaiter指针为下一个节点firstWaiter = next;else// 让上一个正常节点指向取消节点的下一个节点，中间有问题的节点就被跳过了trail.nextWaiter = next;// 条件成立：当前节点为队尾节点了，更新lastWaiter 指向最后一个正常节点if (next == null)lastWaiter = trail;}else // 条件不成立执行到else，说明当前节点是正常节点trail = t;t = next;}
}

3.8、reportInterruptAfterWait(int interruptMode)：根据中断发生的位置来确定之后的操作

private void reportInterruptAfterWait(int interruptMode)throws InterruptedException {// 条件成立：说明在条件队列内发生过中断，此时await()方法抛出中断异常if (interruptMode == THROW_IE)throw new InterruptedException();// 条件成立：说明在条件队列之外发生的中断，此时设置当前线程的中断标记位为true，中断处理交给你的业务层代码处理，如果你不处理，那么什么事也不会发生else if (interruptMode == REINTERRUPT)selfInterrupt();
}

3.9、signal() 方法：唤醒条件队列中的线程

public final void signal() {// 判断当前调用signal方法的线程是否是独占锁持有线程，如果不是，则直接抛出异常if (!isHeldExclusively())throw new IllegalMonitorStateException();// 拿到条件队列中第一个节点nodeNode first = firstWaiter;// 条件队列中的第一个节点不为null，则将第一个节点进行迁移到阻塞队列中的逻辑if (first != null)doSignal(first);
}private void doSignal(Node first) {do {// firstWaiter = first.nextWaiter：因为当前 firstWaiter 马上要出条件队列了// 所以更新 firstWaiter 为当前节点的下一个节点// 如果当前节点的下一个节点为 null，说明当前条件队列只有当前一个节点了，那么当前firstWaiter 出队后，就为null了// 所以需要更新 lastWaiter 也为 nullif ( (firstWaiter = first.nextWaiter) == null)lastWaiter = null;// 当前 first节点出条件队列，断开和下一个节点的关系first.nextWaiter = null;// transferForSignal(first)// boolean：true  表示当前first节点迁移到阻塞队列成功，false：迁移失败// while循环：(first = firstWaiter) != null 当前first迁移失败，则将first更新为first.next，继续尝试迁移...// 直至迁移某个节点成功，或者条件队列为空} while (!transferForSignal(first) &&(first = firstWaiter) != null);
}

3.10、transferForSignal(Node node)方法：迁移条件队列中的firstWaiter节点到阻塞队列中

// 迁移条件队列中的firstWaiter节点到阻塞队列中
final boolean transferForSignal(Node node) {/** If cannot change waitStatus, the node has been cancelled.*/// CAS 修改当前节点的状态为0，因为当前节点马上要迁移到阻塞队列了// 成功：表示当前节点在条件队列中状态正常// 失败：1.取消状态 （线程await的时候未持有锁，最终线程对应的node会被设置为CANCLED 状态）//      2：node对应的线程挂起期间，被其他线程使用中断信号唤醒过（也会主动进入到阻塞队列，也会修改状态为0）if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))return false;// enq(node)：返回当前节点的前置节点，并将当前节点加入到阻塞队列中 p 是当前节点在阻塞队列中前置节点Node p = enq(node);// 获取前置节点的状态int ws = p.waitStatus;// 条件一：ws > 0 成立：说明前驱节点的状态在阻塞队列中是取消状态，唤醒当前节点// 条件二：前置条件（ws <= 0）：compareAndSetWaitStatus(p, ws, Node.SIGNAL)//       返回true：表示设置前驱节点状态为signal状态成功，//       返回false：当前驱node对应的线程调用的是 lockInterrupt入队的node的时候，会响应中断的，外部线程给前驱线程中断信号的时候，前驱node会将修改为CANCELED//        并且执行出队逻辑// 前驱节点的状态只要不是 0 或者 -1，那么就唤醒当前线程if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))// 唤醒当前node对应的线程，之后的逻辑，回头再说LockSupport.unpark(node.thread);return true;
}
e p = enq(node);// 获取前置节点的状态int ws = p.waitStatus;// 条件一：ws > 0 成立：说明前驱节点的状态在阻塞队列中是取消状态，唤醒当前节点// 条件二：前置条件（ws <= 0）：compareAndSetWaitStatus(p, ws, Node.SIGNAL)//       返回true：表示设置前驱节点状态为signal状态成功，//       返回false：当前驱node对应的线程调用的是 lockInterrupt入队的node的时候，会响应中断的，外部线程给前驱线程中断信号的时候，前驱node会将修改为CANCELED//        并且执行出队逻辑// 前驱节点的状态只要不是 0 或者 -1，那么就唤醒当前线程if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))// 唤醒当前node对应的线程，之后的逻辑，回头再说LockSupport.unpark(node.thread);return true;
}