Java 并发编程之 Condition 接口

本文部分摘自《Java 并发编程的艺术》

概述

任意一个 Java 对象，都拥有一个监视器方法，主要包括 wait()、wait(long timeout)、notify() 以及 notifyAll() 方法，这些方法与 synchronized 同步关键字配合，可以实现等待 – 通知模式。Condition 接口也提供了类似 Object 的监视器方法，与 Lock 配合可以实现等待 – 通知模式

Object 的监视器方法与 Condition 接口的对比：

接口示例

Condition 定义了等待 – 通知两种类型的方法，当前线程调用这些方法时，需要提前获取到 Condition 对象关联的锁。Condition 对象是由 Lock 对象（调用 Lock 对象的 newCondition() 方法）创建，换句话说，Condition 是依赖 Lock 对象的

public class ConditionUserCase {Lock lock = new ReentrantLock();Condition condition = lock.newCondition();public void conditionWait() throws InterruptedException {lock.lock();try {condition.await();} finally {lock.unlock();}}public void conditionSignal() {lock.lock();try {condition.signal();} finally {lock.unlock();}}}

当调用 await() 方法后，当前线程会释放锁并在此等待，而其他线程调用 Condition 对象的 signal() 方法，通知当前线程后，当前线程才从 await() 方法返回，并且在返回前已经获取了锁

Condition 的部分方法以及描述：

下面通过一个有界队列的示例来深入理解 Condition 的使用方式

public class BoundedQueue<T> {private Object[] items;// 添加的下标，删除的下标和数据当前数量private int addIndex, removeIndex, count;private Lock lock = new ReentrantLock();private Condition notEmpty = lock.newCondition();private Condition notFull = lock.newCondition();public BoundedQueue(int size) {items = new Object[size];}/*** 添加一个元素，如果数组满，则添加线程进入等待状态，直到有空位*/public void add(T t) throws InterruptedException {lock.lock();try {while (count == items.length) {notFull.await();}items[addIndex] = t;if (++addIndex == items.length) {addIndex = 0;}++count;notEmpty.signal();} finally {lock.unlock();}}/*** 由头部删除一个元素，如果数组空，则删除线程进入等待状态，直到有新元素添加*/@SuppressWarnings("unchecked")public T remove() throws InterruptedException {lock.lock();try {while (count == 0) {notEmpty.await();}Object x = items[removeIndex];if (++removeIndex == items.length) {removeIndex = 0;}--count;notFull.signal();return (T) x;} finally {lock.unlock();}}}

实现分析

ConditionObject 是同步器 AbstractQueuedSynchronizer 的内部类，每个 Condition 对象都包含着一个队列（等待队列），该队列是 Condition 对象实现等待 – 通知功能的关键

1. 等待队列

等待队列是一个 FIFO 队列，在队列中的每个节点都包含了一个线程引用，该线程就是在 Condition 对象上等待的线程，如果一个线程调用了 Condition.await() 方法，那么该线程就会释放锁，构造成节点并加入等待队列并进入等待状态

一个 Condition 包含一个等待队列，Condition 拥有首尾节点的引用，新增节点只需要将原有的尾节点 nextWaiter 指向它，并更新尾节点即可。节点引用更新的过程并没有使用 CAS 来保证，原因在于调用 await() 方法的线程必定是获取了锁的线程，也就是该过程是由锁来保证线程安全的

在 Object 的监视器模型上，一个对象拥有一个同步队列和等待队列，而并发包中的 Lock 拥有一个同步队列和多个等待队列，其对应关系如图所示：

2. 等待

调用 Condition 的 await() 方法，会使当前线程进入等待队列并释放锁，同时线程状态变为等待状态。当从 await() 方法返回时，当前线程一定获取了 Condition 相关联的锁

Condition 的 await() 方法如下所示：

public final void await() throws InterruptedException {// 检测线程中断状态if (Thread.interrupted())throw new InterruptedException();// 当前线程包装为 Node 并加入等待队列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);}

3. 通知

调用 Condition 的 signal() 方法，将会唤醒在等待队列中等待时间最长的节点（首节点），在唤醒节点之前，会将节点移到同步队列中

Condition 的 signal() 方法代码如下所示：

public final void signal() {// 检查当前线程是否获取了锁if (!isHeldExclusively())throw new IllegalMonitorStateException();// 获取等待队列首节点，移动到同步队列并唤醒Node first = firstWaiter;if (first != null)doSignal(first);}

Condition 的 signAll() 方法，相当于对等待队列中的每个结点均执行一个 signal() 方法，效果就是将等待队列中所有节点全部移动到同步队列中，并唤醒每个节点的线程