Java源码学习之高并发编程基础——AQS源码剖析之线程间通信之条件等待队列

1.前言&目录

前言：

在Java中，使用synchronized关键字构建的锁，线程间通信可以使用某对象实例的wait/notify机制完成。AQS同样也提供了一套线程间通信的解决方案——条件等待队列。

在AQS源码分析的两篇文章AQS源码分析（上）、AQS源码分析（下）中，我们知道了，无论是独占锁模式还是共享锁模式，AQS提供的能力是将获取不到锁的线程将它们封装成链表节点的形式组织起来进行同步等待。

AQS也提供了如wait/notity等机制，它就是条件等待队列，队列元素是Condition——条件，条件的实例对象是ConditionObject。AQS的条件等待队列是一个单向队列，它的节点和AQS同步等待队列的节点是同一个类，都是AbstractQueuedSynchronizer.Node，目的就是为了条件等待节点最终能并入AQS同步等待队列。

以下就是AQS同步队列与条件等待队列模型的关系图：

条件等待节点最终会并入AQS同步队列中，意味着当前在等待条件的线程将重新进入AQS同步队列排队竞争锁。接下来，还是会以源码讲解的形式深入理解AQS中的条件等待。

1.前言&目录

2.AQS条件使用场景

3.AQS条件源码剖析

3.1 ConditionObject条件实例

3.1 await()方法

3.2 signal()方法

3.3 AQS条件总结

4.简单案例

5.总结

2.AQS条件使用场景

AQS的条件是用作线程间通信的，一般来说多应用于生产者/消费者模型中，如果你使用的是如ReentrantLock等继承AQS实现的独占锁，若需要线程间通信就需要到条件。

在生产者/消费者模型中，生产者线程创建的商品不是无限制可以创建的，它们是受到库存容量的限制的，消费者线程消费的商品也是有限的，最多能消费生产出来的商品。

这种模型，在阻塞队列比较常见，如LinkedBlockingQueue、ArrayBlockingQueue、LinkedBlockingDeque，它们的某些增加、获取元素方法使用到了AQS的条件等待，目标就是实现一定条件下的”阻塞“等待。

3.AQS条件源码剖析

AQS条件的源码解读，主要分三部分：熟悉条件等待队列模型、掌握关键的等待、释放方法。

3.1 ConditionObject条件实例

工欲善其事，必先利其器，在学习掌握AQS条件的源码之前，我们必须先了解条件等待队列的模型——ConditionObject，它底层也是像AQS一样，由Node节点组成的队列，它是单向链表，AQS是双向链表。

ConditionObject有Node firstWaiter、Node lastWaiter两个成员变量，分别表示条件队列的头节点和尾节点。

public abstract class AbstractQueuedSynchronizerextends AbstractOwnableSynchronizer{public class ConditionObject implements Condition {// 队列头元素private transient Node firstWaiter;// 队列尾元素private transient Node lastWaiter;// 实现条件等待的方法public final void await() throws InterruptedException {...}// 唤醒条件等待节点的方法public final void signal() {...}}
}

并且它的两个await()、signal()等方法分别表示实现条件等待和唤醒条件等待，掌握这两个方法对理解条件等待节点是怎么并入AQS同步等待队列是非常重要的。

3.1 await()方法

await是AQS内部类ConditionObject的方法，它的作用用一句话概括就是，添加条件等待节点到队列，并将自己阻塞起来直到被唤醒，然后加入AQS同步等待队列。

往细一点说，一共是下面的步骤：

通过addConditionWaiter()方法，将当前线程封装为Node节点（下文称条件等待节点），该节点waitStatus是CONDITION（-2），接着将该条件等待节点添加到ConditionObject条件等待队列的尾部去。
调用fullyRelease(node)方法释放当前线程持有的独占锁，为什么需要在这里释放呢？原因是条件需要和独占锁配合使用，这种情况通常是生产者/消费者模型。
自旋检查当前条件等待节点是否在AQS的同步队列中，如果不是则说明此时的条件等待节点还没有并入、接入AQS同步队列中，会将该当前线程阻塞起来。唤醒的时机是同一个ConditionObject实例对象调用了signal()方法。
如果被唤醒了，则会进入acquireQueued方法，这个方法在AQS阻塞队列上文中介绍过，该方法是将获取不到独占锁的线程进行自旋操作：二次获取锁和经过最多两次阻塞预判会阻塞当前线程——会一直阻塞到其他持有独占锁的线程主动释放锁。

public abstract class AbstractQueuedSynchronizerextends AbstractOwnableSynchronizer{public class ConditionObject implements Condition{public final void await() throws InterruptedException {if (Thread.interrupted())throw new InterruptedException();// 往条件等待队列添加绑定了当前线程的等待节点，并返回它Node node = addConditionWaiter(); int savedState = fullyRelease(node); // 释放当前线程持有的独占锁int interruptMode = 0;// 只要当前条件等待节点没有存在于AQS同步队列时，就将其阻塞起来while (!isOnSyncQueue(node)) {LockSupport.park(this);if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)break;}// 当条件等待节点被唤醒后其实会被添加到同步队列尾部，因此在这里会进入acquireQueued// 方法重新自旋获取锁if (acquireQueued(node, savedState) && interruptMode != THROW_IE)interruptMode = REINTERRUPT;if (node.nextWaiter != null) // clean up if cancelledunlinkCancelledWaiters();if (interruptMode != 0)reportInterruptAfterWait(interruptMode);}}
}

经过上面的步骤划分，是否想起了熟悉的脉络？当某个线程拿到了独占锁后进入到同一时刻只能有一个线程访问的代码区域，如果处理背景不是生产者/消费者模型，则在最后的finally语句块释放锁即可。

但是对于生产者/消费者模型来说，其特点是，当库存商品已经达到上限了，就需要停止生产并通知消费者过来将库存消耗到上限值以下。当消费者将库存商品都一扫而空了，就需要停止消费并通知生产者重新投入生产中。

即在独占锁锁住的代码区域中处理的是生产者/消费者模型时，就需要通过一些手段或者机制将当前持有锁的线程进行阻塞（停止生产或消费）、唤醒阻塞线程（重写投入生产/消费）。

那么这里所讲的await方法就是应对于当库存商品已经达到上限了，需要暂时将生产者线程停止生产的情况，并通知消费者线程开始消费。

3.2 signal()方法

await方法用于阻塞当前线程并将其加入条件等待队列，signal方法就是负责将条件等待队列的节点接入到AQS的同步队列中。

signal是这么将条件等待队列节点并入到AQS的同步队列中的：

调用isHeldExclusively()方法，判断当前线程持有独占锁才能执行后面的并入操作。
将条件等待队列的头节点获取并调用doSignal方法，在这里会先做转移和删除无用的节点，什么算是无用节点呢？答案在transferForSignal方法里，当要转移的条件等待节点的waitStatus不是CONDITION（-2）时，说明它可能已经被取消了（CANCELLED）。
transferForSignal方法首先做的就是将条件等待节点的waitStatus从CONDITION转换为0，目的是将它视为正常加入AQS同步等待队列的节点一样（等锁节点初始化时waitStatus都是0）。然后，通过AQS#enq(node)方法将该条件等待节点通过自旋的CAS操作添加到AQS同步队列尾部，注意返回的变量p其实是刚添加节点的前驱节点，这里做了一个额外保障：如果其前驱节点被取消了或者无法通过CAS更新其waitStatus为SIGNAL，则会直接将该条件等待节点绑定的线程唤醒。

        public final void signal() {if (!isHeldExclusively()) // 判断当前执行signal()方法的线程是否是持有独占锁的线程throw new IllegalMonitorStateException();Node first = firstWaiter;if (first != null)// 找到队列头节点进行从条件节点并入AQS同步等待队列doSignal(first);}private void doSignal(Node first) {do {if ( (firstWaiter = first.nextWaiter) == null)lastWaiter = null;first.nextWaiter = null;// 找到可用的一个条件等待节点将其并入AQS同步队列时退出doSignal方法} while (!transferForSignal(first) &&(first = firstWaiter) != null);}final boolean transferForSignal(Node node) {// 条件等待节点的预期状态是CONDITION，不是的话则直接退出将其并入AQS同步队列if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))return false;// 自旋添加到AQS同步队列尾部 ，并返回其前驱节点Node p = enq(node);int ws = p.waitStatus;if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))// 如果刚添加节点的前驱节点被取消了或者无法通过CAS更新为SIGNAL，则// 直接唤醒新添加的节点去竞争锁   LockSupport.unpark(node.thread);return true;}

signal方法的使用场景就是当消费者线程消耗了库存商品，此时库存容量也空出来了，就通过该方法去通知生产者线程重新投入生产。

3.3 AQS条件总结

条件需要搭配独占锁使用，通常应用于生产者/消费者模型中，目的是保证生产总额和消耗总额的一个动态平衡。

根据前面两个小节，可以总结出AQS同步队列和条件等待队列的模型关系如下：

当持有独占锁的线程，假设它是生产者线程，它发现此时库存容量已经达到最大容量了，再生产的商品也堆积不下，此时它就需要停止生产，即将自己持有的独占锁释放并加入到条件等待队列中去，等待消费者去消费，把库存的商品数量减到能继续投入生产为止。

当消费者把库存都一扫而空了，就要通知生产者赶紧生产补货，同时消费者也将阻塞自己并让出持有的独占锁，其过程跟生产者如出一辙。

4.简单案例

在这里我们通过一个简单的生产者/消费者模型去复盘并验证上面章节所讲述的。

在下面简单的生产者/消费者模型中，定义了超市的最大库存量为3、有5个生产者线程、有2个消费者线程，按照上个章节的分析，当生产者线程生产的商品已经达到阈值时，就会进入阻塞状态，直到消费者线程消费了商品后才能通知生产者线程重新投入生产。

注意notifyConsumer方法和notifyProduce方法中使用的是signalAll()方法，其实signalAll是将条件等待队列中的所有节点都并入AQS同步队列中，signal每次只能操作一个条件等待节点。

那么让我们运行以下的main函数，看看结果是否符合预期？

public class MyAqsConditionTest {private static volatile int maxGoodCount = 3; // 任一时刻超时库存最大容量private static List<Integer> supermarket = new ArrayList<>();public static void main(String[] args) {ReentrantLock produceLock = new ReentrantLock(true);ReentrantLock consumeLock = new ReentrantLock(true);Condition produceCondition = produceLock.newCondition();Condition consumeCondition = consumeLock.newCondition();for (int i=0;i<5;i++){new Thread(new ProduceRunnable(i, produceLock ,produceCondition, consumeLock, consumeCondition)).start();}for (int i=0;i<5;i++){new Thread(new ConsumeRunnable(produceLock ,produceCondition, consumeLock, consumeCondition)).start();}}static class ProduceRunnable implements Runnable {private int goodsNum;private  ReentrantLock produceLock ;private  ReentrantLock consumeLock ;private  Condition produceCondition ;private  Condition consumeCondition ;// ...构造函数@Overridepublic void run() {produceLock.lock();try {int currentSize = supermarket.size();if (currentSize >= maxGoodCount){// 当前超时商品容量已经达到上限，无法继续生产，需要进行阻塞System.out.println("生产者：当前商品数量已经达到库存上限，需要等待消费者线程消费");produceCondition.await();}supermarket.add(goodsNum);System.out.println("生产者线程创建了商品"+goodsNum);}catch (Exception e){}finally {produceLock.unlock();}// 如果此时超时有货了就通知消费者线程if (supermarket.size()>0){notifyConsumer(consumeLock, consumeCondition);}}}static class ConsumeRunnable implements Runnable {private  ReentrantLock produceLock ;private  ReentrantLock consumeLock ;private  Condition produceCondition ;private  Condition consumeCondition ;// ...构造函数@Overridepublic void run() {consumeLock.lock();try {int currentSize = supermarket.size();if (currentSize == 0){// 此时超市商品卖光了，阻塞当前消费者线程直到生产者将商品生产出来consumeCondition.await();}int buyGoodsNum = supermarket.remove(supermarket.size()-1);System.out.println("消费者线程买到了商品"+buyGoodsNum);}catch (Exception e){}finally {consumeLock.unlock();}if (supermarket.size() < maxGoodCount){notifyProduce(produceLock, produceCondition);}}}// 由生产者线程调度，当此时库存商品超过上限了，通知消费者线程消费static void notifyConsumer(ReentrantLock consumeLock, Condition consumeCondition){try {consumeLock.lock();consumeCondition.signalAll();}finally {consumeLock.unlock();}}// 由消费者线程调度，当此时超时库存讲到阈值以下了，通知生产者线程投入生产static void notifyProduce(ReentrantLock produceLock, Condition produceCondition){try {produceLock.lock();produceCondition.signalAll();}finally {produceLock.unlock();}}
}

输出结果符合预期，具体如下所示，生产者线程的确最多只能同时生产3个商品，等消费者线程消费以后才能继续投入生产。即AQS的await()方法成功的将持有独占锁的线程阻塞了起来，signalAll方法也在合适的时机唤醒了在条件等待队列中的被阻塞的线程。

-------输出结果-------
生产者线程创建了商品0
生产者线程创建了商品2
生产者线程创建了商品3
生产者：当前商品数量已经达到库存上限，需要等待消费者线程消费
消费者线程买到了商品3
生产者线程创建了商品4
消费者线程买到了商品4
生产者线程创建了商品1
-------输出结果-------