线程条件队列ConditionObject源码解读

小记
好久没更博，窗外光芒万丈，冬日的晚晨，多么美好，就不浪费了，循着键盘上的点点星辰，开工！

啥子是条件队列？

我们都知道，在万类之祖Object里面定义了几个监视器方法：wait()，notify
()，notifyAll()，配合synchronized语义来控制线程的一些状态，在JDK1.5之后，由Lock替代了synchronized，而这几个监视器由条件队列Condition来实现，以便在某个状态条件现在可能为 true 的另一个线程通知它之前，一直挂起该线程（即让其“等待”），以原子的方式释放锁，并挂起当前线程，所以，也可以叫它为线程的条件队列(自创的)。

来看一个应用示例

在以前刚学习Java的时候写过一个题，题干大概是这样的：开启三个线程依次轮流打印出75个数，且次序不能乱。下面是代码：

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;public class CountTo_75 {public static void main(String[] args) {final Box box = new Box();new Thread(new Runnable() { // 线程1@Overridepublic void run() {for (int i = 1; i <= 5; i++) {box.main_1();}}},"Thread-1").start();new Thread(new Runnable() { // 线程2@Overridepublic void run() {for (int i = 1; i <= 5; i++) {box.main_2();}}},"Thread-2").start();new Thread(new Runnable() { // 线程3@Overridepublic void run() {for (int i = 1; i <= 5; i++) {box.main_3();}}},"Thread-3").start();}static class Box {Lock lock = new ReentrantLock();Condition condition_1 = lock.newCondition();Condition condition_2 = lock.newCondition();Condition condition_3 = lock.newCondition();private volatile int flag = 1;public static int count = 0;   //用于计数的变量public void main_1() {lock.lock();try {while(flag != 1){try {condition_1.await();} catch (InterruptedException e) {e.printStackTrace();}}for (int j = 0; j < 5; j++) {count++;System.out.println(Thread.currentThread().getName() +" "+ count);}flag = 2;condition_2.signal();} finally {lock.unlock();}}public void main_2() {lock.lock();try {while(flag != 2){try {condition_2.await();} catch (InterruptedException e) {e.printStackTrace();}}for (int j = 0; j < 5; j++) {count++;System.out.println(Thread.currentThread().getName() +" " + count);}flag = 3;condition_3.signal();} finally {lock.unlock();}}public void main_3() {lock.lock();try {while(flag != 3){try {condition_3.await();} catch (InterruptedException e) {e.printStackTrace();}}for (int j = 0; j < 5; j++) {count++;System.out.println(Thread.currentThread().getName() + " " + count);}flag = 1;condition_1.signal();} finally {lock.unlock();}}}
}

抛开当时混乱的逻辑和性能考虑不足不谈，这段丑陋的代码，终归是实现了功能，而且是运用了Lock和Condition来实现的，用在这里来说明Condition的语义再好不过了。代码中初始化了3个条件队列，分别来控制3个线程的挂起状态，flag变量则控制它们之间的关系。

与Lock之间的实现关系

Condition是一个接口，其实现类只有两个：AQS和AQLS，都以内部类的形式存在，内部类叫做ConditionObject，这里有点纳闷，既然这个类是为Lock专属定制的，为什么不在ReentrantLock里面来实现呢？放在AQS不会太臃肿吗？不知道Doug Lea上神当时是怎么考虑的。
由于在AQS中已经实现，因此在ReentrantLock里面对其操作也是很简单的，创建一个条件队列：

    public Condition newCondition() {return sync.newCondition();}

sync中的实现：

    abstract static class Sync extends AbstractQueuedSynchronizer {private static final long serialVersionUID = -5179523762034025860L;final ConditionObject newCondition() {return new ConditionObject();}}

很简单吧？呵呵，下面来看ConditionObject

ConditionObject实现

首先定义了两个核心成员变量，条件队列的头节点和尾节点：

/** First node of condition queue. */
private transient Node firstWaiter;/** Last node of condition queue. */
private transient Node lastWaiter;

1、核心方法：await() 不可中断的条件等待实现

public final void await() throws InterruptedException {//当前线程被中断则抛异常if (Thread.interrupted()) throw new InterruptedException();//添加进条件队列Node node = addConditionWaiter();//释放当前线程持有的锁int savedState = fullyRelease(node);int interruptMode = 0;//在这里一直查找创建的Node节点在不在Sync队列，不在就一直禁用当前线程while (!isOnSyncQueue(node)) {//park当前线程，直到唤醒LockSupport.park(this);//如被中断也跳出循环if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) break;}//此时已被唤醒，获取之前被释放的锁，if (acquireQueued(node, savedState) && interruptMode != THROW_IE)   interruptMode = REINTERRUPT;//移除节点，释放内存if (node.nextWaiter != null) // clean up if cancelledunlinkCancelledWaiters();//被中断后的操作if (interruptMode != 0)reportInterruptAfterWait(interruptMode);
}

简而言之，await()方法其实就是为当前线程创建一个Node节点，加入到Condition队列并释放锁，之后就一直查看这个节点是否在Sync队列中了(signal()方法将它移到Sync队列)，如果在的话就唤醒此线程，重新获取锁。此外，awaitNanos(long nanosTimeout) 方法和await(long time, TimeUnit unit) 方法的实现大同小异，只是在何时跳出while循环的时候加了一个超时罢了。
另外还有几个相关的方法也看一下：

/***  添加一个Node节点到Condition队列中*/
private Node addConditionWaiter() {Node t = lastWaiter;//如果尾节点被取消，就清理掉if (t != null && t.waitStatus != Node.CONDITION) {unlinkCancelledWaiters();t = lastWaiter;}//新建状态为的CONDITION节点，并添加在尾部Node node = new Node(Thread.currentThread(), Node.CONDITION);if (t == null)firstWaiter = node;elset.nextWaiter = node;lastWaiter = node;return node;
}

    /***  释放当前线程的state，实际还是调用tryRelease方法*/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;}}

    /***  检查当前节点在不在Sync队列*/final boolean isOnSyncQueue(Node node) {//如果当前节点状态为CONDITION，一定还在Condition队列//如果Sync队列的前置节点为null，则表明当前节点一定还在Condition队列if (node.waitStatus == Node.CONDITION || node.prev == null)return false;//有后继节点，也有前置节点，那么一定在Sync队列if (node.next != null) // If has successor, it must be on queuereturn true;//倒查Node节点，前置节点不能为null，第一个if已经做了判断，其前置节点为non-null,但是当前节点也不在Sync也是可能的,因为CAS操作将其加入队列也可能失败，所以我们需要从尾部开始遍历确保其在队列return findNodeFromTail(node);}

2、核心方法：signal() 将等待时间最长的线程（如果存在）从Condition队列中移动到拥有锁的Sync队列

   public final void signal() {//当前线程非独占线程，报非法监视器状态异常if (!isHeldExclusively())throw new IllegalMonitorStateException();//头节点是等待时间最长的节点Node first = firstWaiter;if (first != null)doSignal(first);}

   private void doSignal(Node first) {do {//如果头节点的下一节点为null，则将Condition的lastWaiter置为nullif ( (firstWaiter = first.nextWaiter) == null)lastWaiter = null;//将头结点的下一个节点设为nullfirst.nextWaiter = null;//被唤醒并且头节点不为null则结束循环} while (!transferForSignal(first) &&(first = firstWaiter) != null);}

   final boolean transferForSignal(Node node) {//如果无法改变节点状态，说明节点已经被唤醒if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))return false;//将当前节点添加到Sync队列尾部，并设置前置节点的waitStatus为SIGNAL，表明后继有节点（可能）将被唤醒，如果取消或者设置waitStatus失败，会唤醒重新同步操作，这时候waitStatus是瞬时的，出现错误也是无妨的Node p = enq(node);int ws = p.waitStatus;if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))LockSupport.unpark(node.thread);return true;}

3、核心方法：signalAll() 将所有线程从Condition队列移动到拥有锁的Sync队列中。

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

    /*** 遍历所有节点，加入到拥有锁的Sync队列*/private void doSignalAll(Node first) {lastWaiter = firstWaiter = null;do {Node next = first.nextWaiter;first.nextWaiter = null;transferForSignal(first);first = next;} while (first != null);}