AbstractQueuedSynchronizer实现Android版源码解析（三）

前两部分，已经剖析了独占模式和共享模式下的获取锁以及释放锁的过程，接下来就是AQS最后的一个实现部分，Condition的实现。

AQS框架在内部提供两一个ConditionObject类，给其它独占锁提供Condition支持。一个锁对象可以关联任意数目的条件对象，条件对象提供了await、signal和signalAll操作，还包括一些带有超时，interrupt的方法。ConditionObject要求当前线程持有锁并且条件对象属于该锁的时候，条件操作（await、signal和signalALl）才是合法的。

ConditionObject使用了和同步器一样的队列结点。不过每个ConditionObject在自己内部有一个单独的条件队列来维护结点，signal操作主要是把条件队列的结点转移到锁的等待队列中实现的。另外，条件队列在实现上只需要维护next域，因此只是一个单向链表。

具体实现解析

（1）await实现

await的实现同样有好几个变种，这里先从最简单的最简单实现入手，awaitUninterruptibly函数忽略线程interrupt。实现如下

public final void awaitUninterruptibly() {
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
boolean interrupted = false;
while (!isOnSyncQueue(node)) {
LockSupport.park(this);
if (Thread.interrupted())
interrupted = true;
}
if (acquireQueued(node, savedState) || interrupted)
selfInterrupt();
}

awaitUninterruptibly首先调用addConditionWaiter往自己的条件队列添加结点，然后调用fullyRelease释放当前锁状态，在while循环里，通过判断当前结点是否被转移到锁的等待队列中来进行自旋，否则就会进入阻塞状态。如果被唤醒后，isOnSyncQueue返回true（也就是signal的时候该结点移动到锁的等待队列），就会重新进入获取锁的自旋（acquireQueued）。另外如果acquireQueued返回true的话（表示在等待重新获取锁的时候发生中断），或者前面等待的时候线程被中断了，则需要重新设置中断状态。

addConditionWaiter函数如下

private Node addConditionWaiter() {
Node t = lastWaiter;
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}

函数要从尾结点插入新结点（FIFO），因此首先判断尾结点是否被取消（在条件队列中的结点waitStatus只会被设置为CANCELLED或者CONDITION，CANCELLED是在fullyRelease时由于抛出异常设置），如果被取消，则会调用unlinkCancelledWaiters遍历整个条件队列断开被取消结点的连接。然后创建新的CONDITION结点，添加到尾部，返回该结点。这里要注意的是，进行连接的变量是nextWaiter，这个变量是共享模式中用来设置一个共同的空结点的标记，由于条件队列只在独占模式中使用，因此可以利用这个变量来进行连接。

先来看看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;
else
trail.nextWaiter = next;
if (next == null)
lastWaiter = trail;
}
else
trail = t;
t = next;
}
}

整个函数的实现很简单，就是从头结点开始遍历到尾结点，把所有取消的结点都断开连接。该函数在结点wait的时候被取消或者当插入新结点时，尾结点被取消。遍历整个队列而不是仅仅针对修改某个结点，可以避免当超时或者interrupt带来的瞬间大量结点被取消，需要多次重复遍历的情况。

回到awaitUninterruptibly函数里，我们看看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;
}
}

函数调用getState获取之前的阻塞状态参数，然后调用release释放之前的状态，显然ConditionObject要求当前线程必须要获取了当前锁，所以release一定会成功返回true，如果返回了false，则证明当前线程没有获得了锁，则抛出IllegalMonitorStateException异常。同时，如果在release抛出异常，则在finally里必须把结点的状态设置为CANCELLED。

然后awaitUninterruptibly会调用isOnSyncQueue判断当前结点是否在锁的等待队列里。函数实现如下。

final boolean isOnSyncQueue(Node node) {
if (node.waitStatus == Node.CONDITION || node.prev == null)
return false;
if (node.next != null)
return true;
return findNodeFromTail(node);
}

首先waitStatus为CONDITION或者前继结点为null则证明该结点必定在条件队列，因此返回false。此时，waitStatus不等于CONDITION并且prev也不为null的时候，仍然需要进行下面判断。这是由于在signal的时候把结点转移到锁的等待队列过程中，先把该结点的nextWaiter变量修改为null，然后对结点的prev进行了修改（调用enq函数），在enq里会首先修改结点的prev域，才进行CAS添加到等待队列，CAS成功之后才会修改next域，因此如果node的next域不为null，则必定已经转移到等待队列。如果以上判断失败（不大可能发生的情况），这时候就需要进行遍历确认。

findNodeFromTail实现如下。

private boolean findNodeFromTail(Node node) {
Node t = tail;
for (;;) {
if (t == node)
return true;
if (t == null)
return false;
t = t.prev;
}
}

函数从tail往前遍历，如果找到结点返回true，找不到则返回false。当然，这时到最坏情况下的遍历，一般很少可能会发生。

我们重新看回awaitUninterruptibly的实现，如果signal的时候把结点转移到了锁的等待队列后，当调用signal的线程释放锁以后，假设唤醒的是当前线程，然后当前线程就会退出while循环，接下来便会调用acquireQueued进入等待队列的自旋同步机制，尝试获取锁。

接下来看看signal的实现。

（2）signal实现

signal包括signal、signalAll，也就是唤醒一个结点以及唤醒所有结点的区别。先来看看唤醒一个结点的实现。

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

首先是一个isHeldExclusive的判断，同样为了遵守条件对象必须获得锁的前提下才能使用的原则，如果isHeldExclusively返回false，则当前线程没有获取锁，函数会抛出IllegalMonitorStateException异常。然后当头结点为非null的时候，调用函数doSignal。

private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null)
lastWaiter = null;
first.nextWaiter = null;
} while (!transferForSignal(first) &&
(first = firstWaiter) != null);
}

doSignal的实现也是通过循环，把头结点调用transferForSignal转移到锁的等待队列，同时修改firstWaiter为前头结点的nextWaiter值，另外如果当前队列为空还注意把lastWaiter的值清空。并且把first.nextWaiter变为null。

transferForSignal实现如下。

final boolean transferForSignal(Node node) {
if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
return false;

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

首先把waitStatus修改为0（等待队列中的结点默认值）。如果CAS失败，则证明该结点如果被取消了，因此返回false，表明转移失败。成功修改状态后，调用enq把结点添加到锁的等待队列中，然后返回值（node再等待队列的前继结点）结点的waitStauts如果是大于0（CANCELLED）或者CAS更改SIGNAL失败，则重新唤醒该结点，这样线程提早进入acquireQueued里（默认情况下会在等待队列中等待下一次锁释放时唤醒），通过自旋重新修正状态。一切都完成后，返回true表明转移队列成功。

这样signal的大致流程就完结了。主要作用是把当前条件队列中的头结点转移到锁的等待队列中。

接下来我们再看看signalAll的实现。

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

signalAll的实现和signal一致，唯一不同的时候在最后调用的是doSignalAll函数。其实现如下。

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

由于signalAll要把条件队列中所有结点转移到等待队列中，因此lastWaiter和firstWaiter均置为null，然后遍历队列，调用transferForSignal把每个结点都转移到锁的等待队列中。

这样signal的实现分析也完结了，接下来看看await的几个变种的实现。

（3）await变种

首先看看可以被interrupt的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)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
}

函数大体实现结构与忽略interrupt版本类似，但有几个不同指出。

1.在自旋循环里，唤醒以后，增加了函数checkInterruptWhileWaiting调用，该函数主要判断如果在wait的时候线程被interrupt之后，要判断是否在signal前还是signal后，由于按照JSR133标准，如果中断发生在signal之前，await必须重新获取锁之后，抛出InterruptedException。如果中断发生在signal之后，await必须返回且不抛异常，同时设置线程的中断状态。而checkInterruptWhileWaiting通过返回不同值来表示中断是在signal之前还是之后（THROW_IE表示signal之前，REINTERRUPT表示signal之后）。

2.当acquireQueued返回（也就是线程重新获取锁之后），如果返回的是true，则表示获取锁的等待过程中发生中断，此时interruptMode不为THROW_IE的话（可能为0或者REINTERRUPT），则需要把interruptMode设为REINTERRUPT，保证下面能够重新设置中断状态。

3.然后如果此时该结点node.nextWaiter不为null，也就是没有被移动锁的等待队列，则该结点在等待过程中发生来中断，则需要调用unlinkCancelledWaiters断开被取消结点的连接。

4.最后，如果interruptMode不为0，调用reportInterruptAfterWait来决定抛出异常还是重新设置中断状态。

先来看看checkInterruptWhileWaiting的实现。

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

checkInterruptWhileWaiting实现很简单，返回0表示没有被中断；返回THROW_IE表示在signal之前发生中断，需要在await最后抛出异常；返回REINTERRUPT表示在signal之后发生中断，则需要在await最后重置中断状态。如果发生了中断，则调用transferAfterCancelledWait来判断是在signal之前，还是signal之后，然后返回对应值表示中断情况。判断在signal前被中断还是signal后中断其实比较简单，由于signal会把结点的waitStatus通过CAS更改为0，然后重新加入等待队列，因此判断是在signal前还是signal后只需要通过判断waitStatus即可。

final boolean transferAfterCancelledWait(Node node) {
if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
enq(node);
return true;
}
while (!isOnSyncQueue(node))
Thread.yield();
return false;
}

transferAfterCancelledWait函数通过返回true表明取消等待是在signal之前，返回false表示取消等待是在signal之后。其具体实现首先通过一个CAS判断当前结点的waitStatus能否从CONDITION转为0，如果成功，则把当前结点加入等待队列，等待重新获取锁，然后返回true表示在signal之前被中断。如果CAS失败，则表示该结点已经被signal更改来waitStatus，但此时还需要一个自旋调用isOnSyncQueue判断是否已经成功进入等待队列，循环里调用Thread.yield提示应该调度其它线程，当然理想情况下就是执行signal的线程。这样做的目的是因为不能够在signal完成enq的时候进行其它操作，在一个还没完成的转移队列操作中取消是很少有而且很短暂的，需要通过自旋来等待signal完成操作。

接着我们来看看reportInterruptAfterWait的实现。

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

函数实现也很简单，只是按照interrutMode的值来执行对应操作，THROW_IE则抛出异常，REINTERRUPT则重新设置中断状态。

于是，我们就分析完来添加了中断异常逻辑的await变种。接下来再来看看增加来超时机制的await变种。

public final boolean await(long time, TimeUnit unit)
throws InterruptedException {
if (unit == null)
throw new NullPointerException();
long nanosTimeout = unit.toNanos(time);
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
long lastTime = System.nanoTime();
boolean timedout = false;
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
if (nanosTimeout <= 0L) {
timedout = transferAfterCancelledWait(node);
break;
}
if (nanosTimeout >= spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
long now = System.nanoTime();
nanosTimeout -= now - lastTime;
lastTime = now;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
return !timedout;
}

函数实现大体上与之前无太大差异，最大区别就是添加来超时的逻辑判断，首先通过获取当前时间以及计算需要等待的时间之后，进入while自旋，同样，也设置来spinForTimeoutThreshold来增加超时的计算的准确性，另外，等nanosTimeout小于0的时候，也就是等待超时，需要调用transferAfterCancelledWait把当前结点转移到锁的等待队列上，然后如果返回true则表示是在signal之前超时，否则false表示在signal之后超时，把这个返回值作为await返回值表明当前await是在signal之前超时结束，还是signal之后顺利结束。

另外还有一个类似的超时await版本，awaitNanos，但该函数返回long来判断具体的超时的时间。还有一个等待绝对时间的超时版本，awaitUntil，通过赋予具体的等待时间，如果超过此时间点则表明超时，具体实现也类似，此处略去。

总结

到此，Condition的解析完结了，分开三个部分解析的AQS框架的具体部分也完结了。当然还有一些方法提供给子类方便判断当前锁状态，由于理解了以上三大部分之后，对于理解这些方法也就不难，因此这里也就不赘述了。这三大部分共同组成了AQS的总体结构，通篇下来，我们可以看到这是一个高效并且设计精湛的FIFO锁同步的实用框架，如果想更加从宏观上把握总体设计思路，包括主要需求定义以及总体设计思想，可以参见作者Doug Lea的文章

The java.util.concurrent Synchronizer Framework

当然如果觉得英文读起来很辛苦，可以参考

The java.util.concurrent Synchronizer Framework 中文翻译版

无论如何，通过这三个部分的分析，希望可以让大家可以深刻地理解这种多并发条件下等数据结构的维护与操作，并且，当然了，这又同时希望能够证明自己的能力所在，为以后的深入学习进行知识积累。