Condition接口

Condition接口

任意一个Java对象都有一组监视器方法，这些方法定义在所有类的共同超类Obejct中，主要包括wait()、wait(long timeout)、notify()和notifyAll()，这些方法与synchronized同步关键字配合，可以实现等待/通知模式。

Condition接口也提供了类似Object的监视器方法，与Lock配合可以实现等待/通知模式。

Object的监视器和Condition监视器的对比

对比项 Object监视器 Condition
前置条件 获取对象的锁 调用Lock.lock获取锁，调用Lock.newCondition获取Condition对象
调用方式 直接调用，比如object.notify() 直接调用，比如condition.await()
等待队列的个数 一个 多个
当前线程释放锁进入等待状态 支持 支持
当前线程释放锁进入等待状态，在等待状态中不中断 不支持 支持
当前线程释放锁并进入超时等待状态 支持 支持
当前线程释放锁并进入等待状态直到将来的某个时间 不支持 支持
唤醒等待队列中的一个线程 支持 支持
唤醒等待队列中的全部线程 支持 支持

Condition接口中定义的主要方法

方法名称 描述
await() 当前线程进入等待状态直到被通知（signal）或者中断；当前线程进入运行状态并从await()方法返回的场景包括：（1）其他线程调用相同Condition对象的signal/signalAll方法，并且当前线程被唤醒；（2）其他线程调用interrupt方法中断当前线程；
awaitUninterruptibly() 当前线程进入等待状态直到被通知，在此过程中对中断信号不敏感，不支持中断当前线程
awaitNanos(long) 当前线程进入等待状态，直到被通知、中断或者超时。如果返回值小于等于0，可以认定就是超时了
awaitUntil(Date) 当前线程进入等待状态，直到被通知、中断或者超时。如果没到指定时间被通知，则返回true，否则返回false
signal() 唤醒一个等待在Condition上的线程，被唤醒的线程在方法返回前必须获得与Condition对象关联的锁
signalAll() 唤醒所有等待在Condition上的线程，能够从await()等方法返回的线程必须先获得与Condition对象关联的锁

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

下面通过一个实例来了解一下Condition的使用方式：

public class BoundedQueue<T> {

    private Object[] items;
    private int addIndex; // 将要被添加元素的下标
    private int removeIndex; // 要被删除的元素下标
    private int count; // 数组中当前元素的数量

    private Lock lock = new ReentrantLock();
    //创建于lock对象相关联的Condition对象
    private Condition notEmpty = lock.newCondition();
    private Condition notFull = lock.newCondition();


    public BoundedQueue(int size) {
        items = new Object[size];
    }

    /**
     * 添加一个元素，如果数组已满，那么线程进入等待状态，直到有空位
     * @param t
     * @throws InterruptedException
     */
    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();
        }
    }

    /**
     * 从队列头部删除一个元素，如果数组为空，则删除线程进入等待状态，直到有新添加元素
     * @return
     * @throws InterruptedException
     */
    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();
        }
    }
}

Condition的实现分析

newCondition方法返回的是一个ConditionObject对象，ConditionObject类Condition接口的实现类，位于AbstractQueuedSynchronizer的内部。因为Condition的操作需要获取相关联的锁，所以作为同步器的内部类也比较合理。每个Condition对象都包含着一个队列，称为等待队列，该队列是Condition对象实现等待/通知功能的关键。

1.等待队列

等待队列是一个FIFO队列，在队列中的每个节点都包含了一个线程引用，引用的线程就是在Condition对象上等待的线程，如果一个线程调用了Condition.await()方法，那么该线程将会释放锁、构造成节点加入等待队列并进入等待状态。该节点的定义复用了同步器中节点的定义，也就是说，同步队列和等待队列中的节点类型都是同步器的静态内部类java.util.concurrent.locks.AbstractQueuedSynchronizer.Node。等待队列的基本结构如下图所示：

image-20210609153742775

一个Condition包含一个等待队列，Condition拥有首节点（firstWaiter）和尾节点（lastWaiter）。当前线程调用Condition.await()方法将会以当前线程构造节点，并将节点从尾部加入等待队列。

当有新的节点加入时，只需要将原有的尾节点指向它，然后更新Condition中的尾节点指针即可。此处，节点引用更新的过程中，没有CAS保证，这是因为调用await方法的线程必定是获取了锁的线程，也就是说节点引用更新过程中的线程安全是由锁来保证的。

在Object的监视器模型上，一个对象拥有一个同步队列和等待队列，而在并发包中的一个Lock对象（更确切地说是一个同步器）拥有一个同步队列和多个等待队列（因为一个lock对象可以创建多个Condition对象）。其对应关系如图所示：

image-20210609154950904

2.等待

调用Contidion的以await开头的方法，会使当前线程进入等待队列并释放锁，同时线程状态变为等待状态。当从await方法返回时，当前线程一定获取了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;
    
    // 通过isOnSyncQueue(Node) 方法不断自省地检查node节点是否在同步队列中，如果不在，则说明该线程还不具备竞争锁的资格，则继续等待
    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) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

调用该await方法的线程是成功获取了锁的线程，也就是同步队列中的首节点，该方法会将当前线程构造节点并加入等待队列中，然后释放同步状态，唤醒同步队列中的后继节点，然后当前线程会进入等待状态。

如果从同步队列和等待队列的角度来看await方法，当调用await方法时，相当于同步队列的首节点（获取了锁的节点）移动到Condition的等待队列中，过程如下图所示：

image-20210609163048984

可以看出，同步队列的首节点并不会直接加入等待队列，而是通过addConditionWaiter()方法把当前线程构造成一个新的节点并将其加入到等待队列中。addConditionWaiter方法的源码如下：

private Node addConditionWaiter() {
    
    // 获取等待队列的尾节点
    Node t = lastWaiter;
    
    // 尾节点如果不是CONDITION状态，则表示该节点不处于等待状态，需要调用unlinkCancelledWaiters清理节点
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    // 根据当前线程（调用该方法的线程）创建Node节点
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    
    // 将该节点加入等待队列的末尾
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

3.通知

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

public final void signal() {
    // 判断是否是当前线程获取了锁
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    // 唤醒等待队列的首节点，也就是等待时间最长的节点
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}

调用该方法的前提条件是当前线程必须获取了锁，代码中通过调用isHeldExclusively方法来进行判断当前线程是否是获取了锁的线程。然后获取等待队列的首节点，将这个首节点移动到同步队列并使用LockSupport唤醒线程，doSignal方法执行线程的唤醒任务：

private void doSignal(Node first) {
    do {
        // 首先将首节点从等待队列移除
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

然后执行transferForSignal方法将首节点移动到同步队列的，其源码如下所示：

final boolean transferForSignal(Node node) {
	// 尝试将该节点的状态从CONDITION修改为0，如果无法更改状态值，则该节点已被取消。
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

	// 将节点线程安全地加入到同步队列尾部，返回该节点的前驱节点
    Node p = enq(node);
    int ws = p.waitStatus;
    // 如果前驱节点的状态为CANCELLED或者修改waitStatus失败，则直接唤醒当前线程
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}

节点从等待队列移动到同步队列的过程如下图所示：

image-20210609170317895

将节点线程安全地加入到同步队列尾部后，返回该节点的前驱节点。如果前驱节点的状态为CANCELLED或者将前驱节点的状态修改为SIGNAL时失败，则直接唤醒当前线程，否则先返回。

被唤醒后的线程，将从await()方法中的while循环中退出（因为此时 isOnSyncQueue(Node) 方法返回 true），进而调用acquireQueued()方法加入到获取同步状态的竞争中。

成功获取了锁之后，被唤醒的线程将从先前调用的await()方法返回，此时，该线程已经成功获取了锁。

Condition的signalAll()方法，相当于对等待队列的每个节点均执行一次signal() 方法，效果就是将等待队列中的所有节点移动到同步队列中。

参考资料：

《Java并发编程的艺术》

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/yxym2016/p/14880252.html