Condition

3、Condition 接口

前文已经对Lock、ReentrentLock接口进行了简单的分析，但是还有一部分遗漏，就是Condition接口。

摘自JDK官方定义：
Condition直译过来为“条件”，也可以叫做“条件队列”或“条件变量”，以下统称“条件变量” Condition将Object监视器方法（wait、notify、notifyAll）分解成截然不同的对象，以便通过将这些对象与任意Lock实现组合使用，可以为多个对象实现等待/通知机制。其中，Lock替代了synchronized方法和语句的使用、Condition替代了Object监视器方法的使用。

条件变量为线程提供了一个含义，可以在另一个线程通知它之前，一直挂起该线程。因为访问此共享状态信息发生在不同的线程中，所以它必须受保护，因此要将某种形式的锁与该条件相关联。等待提供一个条件的主要属性是：以原子方式释放相关的锁，并挂起当前线程，就像Object.wait()做的那样。

Condition 实例实质上被绑定到一个锁上。要为特定Lock实例获得Condition实例，请使用其newCondition()方法。

Condition使用示例：

public class BoundedBuffer {
    final Lock lock = new ReentrantLock();
    final Condition notFull = lock.newCondition();
    final Condition notEmpty = lock.newCondition();

    final Object[] items = new Object[100];
    int putptr, takeptr, count;

    public void put(Object x) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length)
                notFull.await();
            items[putptr] = x;
            if (++putptr == items.length) putptr = 0;
            ++count;
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }

    public Object take() throws InterruptedException {
        lock.lock();
        try {
            while (count == 0)
                notEmpty.await();
            Object x = items[takeptr];
            if (++takeptr == items.length) takeptr = 0;
            --count;
            notFull.signal();
            return x;
        } finally {
            lock.unlock();
        }
    }
}

BoundedBuffer代表了一个有界缓冲区，提供了put和get两个方法。若缓冲区已满，put时，线程将阻塞，直至take方法被调用；若缓冲区为空，take时，线程将阻塞，直至put方法被调用。通过等待/通知机制，完成两个线程之间的通信，极大的优化了线程之间的等待时间。

方法摘要：

返回值	方法名	说明
void	await()	当前线程在接到信号或被中断之前一直处于等待状态。
boolean	await(long time, TimeUnit unit)	当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态
long	awaitNanos(long nanosTimeout)	当前线程在接到信号、被中断或到达指定等待时间之前一直处于等待状态
void	awaitUninterruptibly()	当前线程在接到信号之前一直处于等待状态
boolean	awaitUntil(Date deadline)	当前线程在接到信号、被中断或到达指定最后期限之前一直处于等待状态
void	signal()	一个等待线程
void	signalAll()	所有等待线程

上文的介绍，基本上来自官方文档，下面结合官方有界缓冲示例，分析Condition接口的源码。

3.1 await()

public final void await() throws InterruptedException {
    // 若线程被中断，抛出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) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

（中断的处理，放到最后分析）

• 将当前线程构造成Node节点并添加到条件队列尾部。

private Node addConditionWaiter() {
    // Node 与AQS中的节点是同一个内部类
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    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;
}

条件队列的Node和AQS中节点的Node是完全相同的。结合上面的分析，Condition是依赖于Lock的，或者说Condition必须与Lock绑定到一起。Lock自身维护了一个同步队列，而Condition自身也维护了一个条件队列。

image.png

并且Condition队列中的节点会转移到Lock队列中。下面会有详细的介绍。

• 释放并记录节点的同步状态。释放同步状态，可以使其他线程获取锁；记录同步状态，以备后面再次获取锁的时候使用（后面还要释放锁，所以必须再次获取锁）。

• 阻塞当前线程。这一步就开始涉及到节点由条件队列中转到同步队列先关的内容了（该操作实际发生在signal()方法中） 首先调用!isOnSyncQueue(node)判断节点是否已经转移到了同步队列上，如果没有，阻塞当前线程；判断线程是否被中断、是否要退出while循环。那什么时候可以正常退出while循环呢？针对上文提到的例子。T1执行put方法，发现BoundedBuffer已满，阻塞；假设10秒后，T2执行take方法，则会唤醒T1。T1被唤醒后继续调用!isOnSyncQueue(node)判断，并决定是否退出循环。isOnSyncQueue方法初看之下，不容易理解，因为它跟signal()方法相关连。所以下面的分析也要跟signal()方法同时进行。

final boolean isOnSyncQueue(Node node) {
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    if (node.next != null) // If has successor, it must be on queue
        return true;
    return findNodeFromTail(node);
}

* `node.waitStatus == Node.CONDITION` 若节点的等待状态为CONDITION，说明节点未被转移到同步队列上，因为在`signal()`方法调用`enq(node)`将节点加入同步队列之前，会对节点的状态合法性进行判断、并将节点的状态改为0。
* `node.prev == null` 节点的前驱节点为null，说明节点未被转移到同步队列上，因为`enq(node)`方法未被执行。
* `node.next != null` 若节点存在后继节点，则节点一定被转移到了同步队列上。
* 若以上条件均为满足，则通过`findNodeFromTail(node)`方法再次从末尾遍历同步队列进行确认。因为此时不能单纯的通过`node.prev != null（综合以上判断，到此时，节点必满足此条件）`来判断节点是否已经被转移到了同步队列上。因为在`enq(node)`方法中，先设置了前驱节点值，后进行了CAS操作，那么很有可能CAS设置队尾正在进行中、也有可能CAS设置队尾失败，需要再次设置。这种情况下，就必须通过`findNodeFromTail(node)`方法确认，节点是否真正被转移到了同步队列上。
* 在分析以上代码之前，最好先分析一下`signal()`方法和`enq(node)`方法。

• 再次获取同步状态（获取锁），并根据中断模式，做出相应的处理。获取同步状态的方法前文多已经分析过，下面着重分析对中断的处理。对线程中断的处理，贯穿了整个wait()方法。
  • 首先，在进入await()方法之后，第一步就对线程的中断做了判断，如果中断了，则抛出异常。
  • 其次，在while循环等待期间，通过(interruptMode = checkInterruptWhileWaiting(node)) != 0再次对等待期间的线程中断状态做处理。并提供了两种中断处理模式。REINTERRUPT，在退出wait方法前，再次将线程中断；THROW_IE，在退出wait方法前，抛出中断异常。
  • 最后，根据第二步得到的中断状态做出相应的处理
  • 篇幅有限，这些代码就不一一粘贴了
• 清理“非法”状态节点，如果等待队列中的节点存在下一个节点，那么就需要尝试清理掉那些“非法”状态的节点。

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;
    }
}

3.2 signal()

await方法分析完之后，再来看signal方法。当然，这两个方法要结合起来分析，则能起到事半功倍的效果。

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

// 执行通知
private void doSignal(Node first) {
    do {
        // 处理节点，主要为了将节点回收（置空）
        // 如果节点的下一个
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } 
    // 执行转换
    while (!transferForSignal(first) && (first = firstWaiter) != null);
}

// 节点转换
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;
}

方法栈依次调用signal、doSignal、transferForSignal三个方法：

1. signal
   1. 执行signal方法的前提是当前线程持有锁（锁的获取过程在await方法中），若当前线程未持有锁，抛出IllegalMonitorStateException()异常。
   2. 先进先出，取头结点，即在队列中等待时间最长的节点；若头节点为空，则条件队列为空，无需任何操作；若头节点非空，则调用doSignal发起通知。
2. doSignal
   1. doSignal方法采用了do/while循环的方法体。这段代码虽短，但是特别精巧，先分析do、再分析while
   2. do代码块负责将节点依次向前移动，并回收已空节点。
      1. 通过(firstWaiter = first.nextWaiter)将节点向前移动一步，如果该表达式为null，则说明队列中有且只有节点，需要将lastWaiter置空。
      2. 通过first.nextWaiter = null，置空当前节点的下一个节点，注意表达式里判断的是first而不是firstWaiter。
   3. while代码块负责循环判断条件
      1. !transferForSignal(first) 判断节点转换是否成功，转换细节下文单独分析。
      2. (first = firstWaiter) != null，将firstWaiter指向first（注意：此时的firstWaiter是first.nextWaiter，即要唤醒节点的下一个节点），并判断是否为空。
3. transferForSignal 该方法负责节点转换，将节点由条件队列移动至同步队列。返回true，表示转移成功；返回false，表示节点状态被置为取消。
   1. if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))，尝试将同步状态由CONDITION（等待）改为0，其目的是为了检查入队节点的状态合法性。若合法，则调用enq方法将节点入队（同步队列，enq方法前文已有分析，不再赘述）；若不合法（或CAS失败）返回false。（注意：enq方法返回节点是入队节点的前驱节点）
   2. 接下来的if判断非常关键。试想一下既然enq方法已经完成了将节点从条件队列转移到了同步队列，那么代码逻辑应该结束了，后续操作的目的是什么呢？
      1. 注意：上文的enq方法返回的是入队节点的前驱节点；unpark方法唤醒的是当前节点。
      2. ws > 0，节点状态大于0的情况只有一种即CANCELLED，节点被取消。
      3. !compareAndSetWaitStatus(p, ws, Node.SIGNAL)，将前驱节点状态设置为等待通知失败。
      4. 如果2或者3的条件成立，则说明前驱节点的状态不合法了，那么此时可以通过LockSupport.unpark(node.thread)提前唤醒当前节点，即唤醒了在await方法中阻塞的线程，这样一来就使得await方法得以继续执行。
      5. 这么做的目的可以简单理解为对多线程并发场景下执行效率的优化。

3.3 延展阅读，ArrayBlockingQueue简析

前文分析了Condition接口的源码，下面做个延展，分析一下ArrayBlockingQueue，来看看Condition在Java中的实际应用。

ArrayBlockingQueue是一个由数组支持的有界阻塞队列（一提到有界、阻塞，现在是不是马上能联想到Condition呢？）此队列按 FIFO（先进先出）原则对元素进行排序。队列的头部是在队列中存在时间最长的元素。队列的尾部是在队列中存在时间最短的元素。新元素插入到队列的尾部，队列获取操作则是从队列头部开始获得元素。

这是一个典型的“有界缓存区”，固定大小的数组在其中保持生产者插入的元素和使用者提取的元素。一旦创建了这样的缓存区，就不能再增加其容量。试图向已满队列中放入元素会导致操作受阻塞；试图从空队列中提取元素将导致类似阻塞。

此类支持对等待的生产者线程和使用者线程进行排序的可选公平策略。默认情况下，不保证是这种排序。然而，通过将公平性 (fairness) 设置为 true 而构造的队列允许按照 FIFO 顺序访问线程。公平性通常会降低吞吐量，但也减少了可变性和避免了“不平衡性”。

对于数组、集合这样的类，似乎都离不开“增、删、改、查”等操作，下面就针对这些最常见的操作，结合一个例子，分析ArrayBlockingQueue的实现。

public class QueueTest {

    // 创建阻塞队列，大小固定，不可改变
    static ArrayBlockingQueue<String> queue = new ArrayBlockingQueue(5);

    public static void main(String[] args) throws InterruptedException {

        // 出队，此时队列为空，需要等待元素进队
        new Thread(() -> {
            try {
                String element = queue.take();
                System.out.println("取出第一个元素：" + element);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "take").start();


        // 入队，入队后通知等待执行出队的线程
        new Thread(() -> {
            queue.add("张三");
            System.out.println("放入第一个元素");
        }, "add").start();
    }
}

结合上文的例子，分析一下take和add方法。在此之前，先来看一下ArrayBlockingQueue一些重要的变量定义。

public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
        
    /** 队列元素集合 */
    final Object[] items;
    /** 下一个出队元素索引 */
    int takeIndex;
    /** 下一个进队元素索引 */
    int putIndex;
    /** 队列元素总数 */
    int count;
    
    // 上面都是队列中普遍使用一些变量定义，不多赘述。
    
    /** 锁（Condition必须与锁共同使用） */
    final ReentrantLock lock;
    /** 等待出队条件 */
    private final Condition notEmpty;
    /** 等待入队条件 */
    private final Condition notFull;
}

3.3.1 队列实例化

public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}


public ArrayBlockingQueue(int capacity, boolean fair) {
    if (capacity <= 0)
        throw new IllegalArgumentException();
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull =  lock.newCondition();
}

ArrayBlockingQueue提供了两种实例化方式，capacity参数指定队列大小，fair指定是否使用公平锁，并在构造函数中实例化了锁和两个Condition条件。

3.3.2 take

获取并移除此队列的头部，在元素变得可用之前一直等待（如果有必要）。

public E take() throws InterruptedException {
    // 加锁
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        // 不满足条件（队列为空），等待
        while (count == 0)
            notEmpty.await();
        // 条件满足，将第一个元素出队
        return dequeue();
    } finally {
        // 解锁
        lock.unlock();
    }
}

take方法代码的整体逻辑与前文的提到的有界缓冲区一致，注意两点：

• Condition的signal时间节点要发生在元素出队之后，所以在take方法里没有看到signal。
• 加锁的过程不忽略中断，一旦线程被中断，则抛出InterruptedException异常。

队列作为一种基本的数据结构，相信大家都不陌生，下面简单看一下元素出队的过程：

private E dequeue() {
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    // 根据出队索引，取出出队元素
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    // 重新计算出队索引，队列满则将出队索引清零
    if (++takeIndex == items.length)
        takeIndex = 0;
    // 将队列元素总数减一
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    // 通知    
    notFull.signal();
    return x;
}

3.3.2 add

将指定的元素插入到此队列的尾部（如果立即可行且不会超过该队列的容量），在成功时返回 true，如果此队列已满，则抛出 IllegalStateException。

public boolean add(E e) {
    // 入队成功，返回true
    if (offer(e))
        return true;
    // 入队失败（队列已满），抛出IllegalStateException异常
    else
        throw new IllegalStateException("Queue full");
}

public boolean offer(E e) {
    // 入队元素不能为空
    checkNotNull(e);
    // 加锁
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        // 队列已满，返回false
        if (count == items.length)
            return false;
        // 入队成功，返回true
        else {
            enqueue(e);
            return true;
        }
    } finally {
        // 解锁
        lock.unlock();
    }
}

private void enqueue(E x) {
    // 将元素入队至putIndex索引处
    final Object[] items = this.items;
    items[putIndex] = x;
    // 重新计算入队索引，队列满则将入队索引清零
    if (++putIndex == items.length)
        putIndex = 0;
    // 队列元素总数加1
    count++;
    // 通知
    notEmpty.signal();
}

3.3.2 小节

关于ArrayBlockingQueue就简单的介绍这些，感兴趣的可以去看一下全量源码。阻塞队列中的一些小技巧还是值得我们学习的，如takeIndex、putIndex，非常简单的解决了队满后，元素再次出队、入队存储索引的问题，代码非常简洁、清晰。