十三. 等待通知机制之Condition接口-蒲公英云

前言

Condition接口定义了一组方法用于配合Lock实现等待/通知模式，与之作为对比的是，用于配合synchronized关键字实现等待/通知模式的定义在java.lang.Object上的监视器方法wait()和notify()等。

正文

通常基于Lock的newCondition()方法创建Condition对象并作为对象成员变量来使用，如下所示。

public class MyCondition {
    private Lock lock = new ReentrantLock();
    private Condition condition = lock.newCondition();
    ......
}

队列同步器AbstractQueuedSynchronizer的内部类ConditionObject实现了Condition接口，后续将基于ConditionObject的实现进行讨论。首先给出Condition接口定义的方法。

public interface Condition {
    void await() throws InterruptedException;
    void awaitUninterruptibly();
    long awaitNanos(long nanosTimeout) throws InterruptedException;
    boolean await(long time, TimeUnit unit) throws InterruptedException;
    boolean awaitUntil(Date deadline) throws InterruptedException;
    void signal();
    void signalAll();
}

上述方法的说明如下表所示。

方法	说明
`await()`	调用此方法的线程进入等待状态，响应中断，也可以被`signal()`和`signalAll()`方法唤醒并返回，唤醒并返回前需要获取到锁资源。
`awaitUninterruptibly()`	同`await()`，但不响应中断。
`awaitNanos()`	同`await()`，并可指定等待时间，响应中断。该方法有返回值，表示剩余等待时间。
`awaitUntil()`	同`await()`，并可指定等待截止时间点，响应中断。该方法有返回值，true表示没有到截止时间点就被唤醒并返回。
`signal()`	唤醒等待队列中的第一个节点。
`signalAll()`	唤醒等待队列中的所有节点。

针对上面的方法再做两点补充说明：

等待队列是Condition对象内部维护的一个FIFO队列，当有线程进入等待状态后会被封装成等待队列的一个节点并添加到队列尾；
从等待队列唤醒并返回的线程一定已经获取到了与Condition对象关联的锁资源，Condition对象与创建Condition对象的锁关联。

下面将结合ConditionObject类的源码来对等待/通知模式的实现进行说明。await()方法的实现如下所示。

public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 基于当前线程创建Node并添加到等待队列尾
    // 这里创建的Node的等待状态为CONDITION，表示等待在等待队列中
    Node node = addConditionWaiter();
    // 释放锁资源
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // Node从等待返回后会被添加到同步队列中
    // Node成功被添加到同步队列中则退出while循环
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    // 让Node进入自旋状态，竞争锁资源
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    // 遍历等待队列，将已经取消等待的节点从等待队列中去除链接
    if (node.nextWaiter != null)
        unlinkCancelledWaiters();
    // Node如果是被中断而从等待返回，则抛出中断异常
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

理解await()方法的整个执行流程前，先看一下等待队列的一个示意图，如下所示。

在这里插入图片描述

Condition对象分别持有等待队列头节点和尾节点的引用，新添加的节点会添加到等待队列尾，同时lastWaiter会指向新的尾节点。

现在回到await()方法，在await()方法中，会做如下事情。

首先，会基于当前线程创建Node并添加到等待队列尾，创建Node有两个注意点：1. 这里创建的Node复用了同步队列中的Node定义；2. 在创建Node前会判断等待队列的尾节点是否已经结束等待（即等待状态不为Condition），如果是则会遍历等待队列并将所有已经取消等待的节点从等待队列中去除链接；
然后，当前线程会释放锁资源，并基于LockSupport.park()进入等待状态；
再然后，当前线程被其它线程唤醒，或者当前线程被中断，无论哪种方式，当前线程对应的Node都会被添加到同步队列尾并进入自旋状态竞争锁资源，注意，此时当前线程对应的Node还存在于等待队列中；
再然后，判断当前线程对应的Node是否是等待队列尾节点，如果不是则触发一次清除逻辑，即遍历等待队列，将已经取消等待的节点从等待队列中去除链接，如果是等待队列尾节点，那么当前线程对应的Node会在下一次创建Node时从等待队列中被清除链接；
最后，判断当前线程从等待返回的原因是否是因为被中断，如果是，则抛出中断异常。

上面讨论了等待的实现，下面再结合源码看一下通知的实现。首先是signal()方法，如下所示。

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

由signal()方法可知，调用signal()方法的线程需要持有锁，其次signal()方法会唤醒等待队列的头节点，即可以理解为唤醒等待时间最久的节点。下面再看一下signalAll()方法，如下所示。

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

可以发现，signalAll()与signal()方法大体相同，只不过前者最后会调用doSignalAll()方法来唤醒所有等待节点，后者会调用doSignal()方法来唤醒头节点，下面以doSignal()方法进行说明。

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

实际就是在transferForSignal()方法中将头节点添加到同步队列尾，然后再调用LockSupport.unpark()进行唤醒。

总结

《Java并发编程的艺术》5.6小节对两者的差异进行了对比和总结，这里直接贴过来作参考。

对比项	Object Monitor Methods	Condition
当前线程释放锁并进入等待状态	支持	支持
当前线程释放锁并进入等待状态，等待过程中不响应中断	不支持	支持
当前线程释放锁并进入超时等待状态	支持	支持
当前线程释放锁并等待至将来某个时间点	不支持	支持
唤醒队列中的一个线程	支持	支持
唤醒队列中的多个线程	支持	支持