Java 并发 - Condition接口

转载：Java并发编程之锁机制之Condition接口

1. 前言

在前面的文章中，我曾提到过，整个Lock接口下实现的锁机制中AQS(AbstractQueuedSynchronizer，下文都称之为AQS)与Condition才是真正的实现者。也就说Condition在整个同步组件的基础框架中也起着非常重要的作用，既然它如此重要与犀利，那么现在我们就一起去了解其内部的实际原理与具体逻辑。

在阅读该文章之前，我由衷的建议先阅读《Java并发编程之锁机制之AQS》与《Java并发编程之锁机制之LockSupport工具》这两篇文章。因为整个Condtion的内部机制与逻辑都离不开以上两篇文章提到的知识点。

2. Condition接口方法介绍

在正式介绍Condtion之前，我们可以先了解其中声明的方法。具体方法声明，如下表所示：

2020-08-19-5SYvij

从该表中，我们可以看出其内部定义了等待（以await开头系列方法）与通知（以singal开头的系列方法)两种类型的方法，类似于Object对象的wait()与notify()/NotifyAll()方法来对线程的阻塞与唤醒。

3. ConditionObject介绍

在实际使用中，Condition接口实现类是AQS中的内部类ConditionObject。在其内部维护了一个FIFO(first in first out)的队列（这里我们称之为等待队列，你也可以叫做阻塞队列，看每个人的理解），通过与AQS中的同步队列配合使用，来控制获取共享资源的线程。

3.1 等待队列

等待队列是ConditionObjec中内部的一个FIFO(first in first out)的队列，在队列中的每个节点都包含了一个线程引用，且该线程就是在ConditionObject对象上阻塞的线程。需要注意的是，在等待队列中的节点是复用了AQS中Node类的定义。换句话说，在AQS中维护的同步队列与ConditionObjec中维护的等待队列中的节点类型都是AQS.Node类型。（关于AbstractQueuedSynchronizer.Node类的介绍，大家可以参看《Java并发编程之锁机制之AQS》文章中的描述）。

在ConditionObject类中也分别定义了firstWaiter与lastWaiter两个指针，分别指向等待队列中头部与尾部。当实际线程调用其以await开头的系列方法后。会将该线程构造为Node节点。添加等待队列中的尾部。关于等待队列的基本结构如下图所示：

2020-08-19-89c0mL

对于等待队列中节点添加的方式也很简单，将上一尾节点的nextWaiter指向新添加的节点，同时使lastWaiter指向新添加的节点。

3.2 同步队列与等待队列的对应关系

上文提到了整个Lock锁机制需要AQS中的同步队列与ConditionObject的等待队列配合使用，其对应关系如下图所示：

2020-08-19-uYh0O5

在Lock锁机制下，可以拥有一个同步队列和多个等待队列，与我们传统的Object监视器模型上，一个对象拥有一个同步队列和等待队列不同。lock中的锁可以伴有多个条件。

3.3 Condition的基本使用

为了大家能够更好的理解同步队列与等待队列的关系。下面通过一个有界队列BoundedBuffer来了解Condition的使用方式，该类是一个特殊的队列，当队列为空时，队列的获取操作将会阻塞当前"拿"线程，直到队列中有新增的元素，当队列已满时，队列的放入操作将会阻塞"放入"线程，直到队列中出现空位。具体代码如下所示：

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;

    /**
     * 添加一个元素
     * （1）如果当前数组已满，则把当前"放入"线程，加入到"放入"等待队列中，并阻塞当前线程
     * （2）如果当前数组未满，则将x元素放入数组中，唤醒"拿"线程中的等待线程。
     */
    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();
        }
    }

    /**
     * 拿一个元素
     * （1）如果当前数组已空，则把当前"拿"线程，加入到"拿"等待队列中，并阻塞当前线程
     * （2）如果当前数组不为空，则把唤醒"放入"等待队列中的线程。
     */
    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();
        }
    }
}

从代码中我们可以看出，在该类中我们创建了两个等待队列notFull与notEmpty。这两个等待队列的作用分别是，当请数组已满时，notFull用于存储阻塞的"放入"线程，notEmpty用于存储阻塞的"拿"线程。需要注意的是获取一个Condition必须通过Lock的newCondition()方法。关于ReentrantLock，在后续的文章中，我们会进行介绍。

4. 阻塞实现 await()

在了解了ConditionObject的内部基本结构和与AQS中内部的同步队列的对应关系后，现在我们来看看其阻塞实现。调用ConditionObject的await()方法（或者以await开头的方法），会使当前线程进入等待队列，并释放同步状态，需要注意的是当该方法返回时，当前线程一定获取了同步状态（具体原因是当通过signal()等系列方法，线程才会从await（）方法返回，而唤醒该线程后会加入同步队列）。这里我们以awati()方法为例，具体代码如下所示：

  public final void await() throws InterruptedException {
			//如果当前线程已经中断，直接抛出异常  
            if (Thread.interrupted())
                throw new InterruptedException();
            //(1)将当前线程加入等待队列
            Node node = addConditionWaiter();
            //(2)释放同步状态（也就是释放锁），同时将线程节点从同步队列中移除，并唤醒同步队列中的下一节点
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            //(3)判断当前线程节点是否还在同步队列中，如果不在则阻塞线程
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            //(4)当线程被唤醒后，重新在同步队列中与其他线程竞争获取同步状态
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

从代码整体来看，整个方法分为以下四个步骤：

• （1）通过 addConditionWaiter()方法将线程节点加入到等待队列中。

• （2）通过fullyRelease(Node node)方法释放同步状态（也就是释放锁），同时将线程节点从同步队列中移除，并唤醒同步队列中的下一节点。

• （3）通过isOnSyncQueue(Node node)方法判断当前线程节点是否在同步队列中，如果不在，则通过LockSupport.park(this);阻塞当前线程。

• （4）当线程被唤醒后，调用acquireQueued(node, savedState)方法，重新在同步队列中与其他线程竞争获取同步状态

因为每个步骤涉及到的逻辑都稍微有一点复杂，这里为了方便大家理解，分别对以上四个步骤涉及到的方法分别进行介绍。

4.1 addConditionWaiter()方法

该方法主要将同步队列中的需要阻塞的线程节点加入到等待队列中，关于addConditionWaiter()方法具体代码如下所示：

    private Node addConditionWaiter() {
            Node t = lastWaiter;
            // (1)如果当前尾节点中中对应的线程已经中断，
            //则移除等待队列中所有的已经中断或已经释放同步状态的线程节点
            if (t != null && t.waitStatus != Node.CONDITION) {
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
		    //(2)构建等待队列中的节点
            Node node = new Node(Node.CONDITION);
			
			//(3)将该线程节点添加到队列中，同时构建firstWaiter与lastWaiter的指向
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
            lastWaiter = node;
            return node;
        }

该方法的逻辑也比较简单，分为以下三个步骤：

• （1）获取等待队列中的尾节点，如果当前尾节点已经中断，那么则通过unlinkCancelledWaiters()方法移除等待队列中所有的已经中断或已经释放同步状态（也就是释放锁）的线程节点

• （2）构建等待队列中的节点，注意，是通过New的形式，那么就说明与同步队列中的线程节点不是同一个。（对Node状态枚举不清楚的小伙伴，可以参看Java并发编程之锁机制之AQS文章下的Node状态枚举介绍）。

• （3）将该线程节点添加到等待队列中去，同时构建firstWaiter与lastWaiter的指向，可以看出等待队列总是以FIFO(first in first out )的形式添加线程节点。

4.1.1 unlinkCancelledWaiters()方法

因为在addConditionWaiter()方法的步骤（1）中，调用了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的指向
                        lastWaiter = trail;
                }
                else
                    trail = t;
                t = next;
            }
        }

该方法具体流程如下图所示：

2020-08-19-mt8Qw1

4.2 fullyRelease(Node node)

在将阻塞线程加入到等待队列后，会将该线程节点从同步队列中移除，释放同步状态（也就是释放锁），并唤醒同步队列中的下一节点。具体代码如下所示：

  final int fullyRelease(Node node) {
        try {
            int savedState = getState();
            if (release(savedState))
                return savedState;
            throw new IllegalMonitorStateException();
        } catch (Throwable t) {
            node.waitStatus = Node.CANCELLED;
            throw t;
        }
    }

release(int arg)方法会释放当前线程的同步状态， 并唤醒同步队列中的下一线程节点，使其尝试获取同步状态，因为该方法已经在Java并发编程之锁机制之AQS文章下的unparkSuccessorNode node)方法的下分析过了，所以这里就不再进行分析了。希望大家参考上面提到的文章进行理解。

4.3 isOnSyncQueue(Node node)

该方法主要用于判断当前线程节点是否在同步队列中。具体代码如下所示：

  final boolean isOnSyncQueue(Node node) {
        //判断当前节点 waitStatus ==Node.CONDITION或者当前节点上一节点为空,则不在同步队列中
        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);
    }

如果你还记得AQS中的同步队列，那么你应该知道同步队列中的Node节点才会使用其内部的pre与next字段，那么在同步队列中因为只使用了nextWaiter字段，所以我们就能很简单的通过这两个字段是否为==null，来判断是否在同步队列中。当然也有可能有一种特殊情况。有可能需要阻塞的线程节点还没有加入到同步队列中，那么这个时候我们需要遍历同步队列来判断是该线程节点是否已存在。具体代码如下所示：

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

这里之所以使用同步队列tail（尾节点）来遍历，如果node.netx!=null，那么就说明当前线程已经在同步队列中。那么我们需要处理的情况肯定是针对node.next==null的情况。所以需要从尾节点开始遍历。

4.4 acquireQueued(final Node node, int arg)

当线程被唤醒后（具体原因是当通过signal()等系列方法，线程才会从await（）方法返回）会调用该方法将该线程节点加入到同步队列中。该方法我在《Java并发编程之锁机制之AQS》中具体描述过了。这里就不在进行过多的解析。

4.5 阻塞流程

在理解了整个阻塞的流程后，现在我们来归纳总结一下，整个阻塞的流程。具体流程如下图所示：

2020-08-19-f0hUDA

• （1）将该线程节点从同步队列中移除，并释放其同步状态。

• （2）构造新的阻塞节点，加入到等待队列中。

5. 唤醒实现 signal()

当需要唤醒线程时，会调用ConditionObject中的singal开头的系列方法，该系列方法会唤醒等待队列中的首个线程节点，在唤醒该节点之前，会先讲该节点移动到同步队列中。这里我们以singal()方法为例进行讲解，具体代码如下：

  public final void signal() {
		    //（1）判断当前线程是否获取到了同步状态（也就是锁）
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            //（2）获取等待队列中的首节点，然后将其移动到同步队列，然后再唤醒该线程节点
            if (first != null)
                doSignal(first);
        }

该方法主要逻辑分为以下两个步骤：

• （1）通过isHeldExclusively()方法，判断当前线程是否获取到了同步状态（也就是锁）。

• （2）通过doSignal(Node first)方法，获取等待队列中的首节点，然后将其移动到同步队列，然后再唤醒该线程节点。

下面我们会分别对上面涉及到的两个方法进行描述。

5.1 isHeldExclusively()方法

isHeldExclusively()方法是AQS中的方法，默认交给其子类实现，主要用于判断当前调用singal()方法的线程，是否在同步队列中，且已经获取了同步状态。具体代码如下所示：

    protected boolean isHeldExclusively() {
        throw new UnsupportedOperationException();
    }

5.2 doSignal(Node first)方法

那我们继续跟踪doSignal(Node first)方法，具体方法如下：

     private void doSignal(Node first) {
            do {
                //（1）将等待队列中的首节点从等待队列中移除，并重新制定firstWaiter的指向
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&
            //（2）将等待队列中的首节点，加入同步队列中，并重新唤醒该节点
                     (first = firstWaiter) != null);
        }

该方法也很简单，分为两个步骤：

1. 将等待队列中的首节点从等待队列中移除，并设置firstWaiter的指向为首节点的下一个节点。 为了方便大家理解该步骤所描述的逻辑，这里画了具体的图，具体情况如下图所示：

   

   2020-08-19-aX6NHz
2. 通过 transferForSignal(Node node)方法，将等待队列中的首节点，加入到同步队列中去，然后重新唤醒该线程节点。

5.2.1 transferForSignal(Node node)方法

因为步骤（2）中transferForSignal(Node node)方法较为复杂，所以会对该方法进行详细的讲解。具体代码如下所示：

    final boolean transferForSignal(Node node) {
       
        //（1）将该线程节点的状态设置为初始状态，如果失败则表示当前线程已经中断了
        if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
            return false;
        //（2）将该节点放入同步队列中，
        Node p = enq(node);
        int ws = p.waitStatus;
        //（3）获取当前节点的状态并判断，尝试将该线程节点状态设置为Singal，如果失败则唤醒线程
        if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

该方法分为三个步骤：

1. 将该线程节点的状态设置为初始状态，如果失败则表示当前线程已经中断了，直接返回。

2. 通过enq(Node node)方法，将该线程节点放入同步队列中。

3. 当将该线程节点放入同步队列后，获取当前节点的状态并判断，如果该节点的waitStatus>0或者通过compareAndSetWaitStatus(ws, Node.SIGNAL)将该节点的状态设置为Singal，如果失败则通过LockSupport.unpark(node.thread)唤醒线程。

上述步骤中，着重讲enq(Node node)方法，关于LockSupport.unPark(Thread thread)方法的理解，大家可以阅读《Java并发编程之锁机制之LockSupport工具》。下面我们就来分析enq(Node node)方法。具体代码如下所示：

   private Node enq(Node node) {
        for (;;) {
            //(1)获取同步队列的尾节点
            Node oldTail = tail;
            //(2)如果尾节点不为空，则将该线程节点加入到同步队列中
            if (oldTail != null) {
	            //将当前节点的prev指向尾节点
                U.putObject(node, Node.PREV, oldTail);
                //将同步队列中的tail指针，指向当前节点
                if (compareAndSetTail(oldTail, node)) {
                    oldTail.next = node;
                    return oldTail;
                }
            } else {
	            //(3)如果当前同步队列为空，则构造同步队列
                initializeSyncQueue();
            }
        }
    }

观察该方法，我们发现该方法通过死循环(当然你也可以叫做自旋)的方式来添加该节点到同步队列中去。该方法分为以下步骤：

1. 获取同步队列的尾节点

2. 如果尾节点不为空，则将该线程节点加入到同步队列中

3. 如果当前同步队列为空，则通过initializeSyncQueue();构造同步队列。

这里对Node enq(Node node)中的步骤（2）补充一个知识点。我们来看一下调用U.putObject(node, Node.PREV, oldTail);语句，内部是如何将当前的节点的prev指向尾节点的。在AQS（AbstractQueuedSynchronizer）中的Node类中有如下静态变量和语句。这里我省略了一下不重要的代码。具体代码如下所示：

private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
//省略部分代码
static final long PREV;
    static {
         try {
		    //省略部分代码
            PREV = U.objectFieldOffset
               (Node.class.getDeclaredField("prev"));
              } catch (ReflectiveOperationException e) {
               throw new Error(e);
            }
        }
    }

其中Node.class.getDeclaredField("prev")语句很好理解，就是获取Node类中pre字段，如果有则返回相应Field字段，反之抛出NoSuchFieldException异常。关于Unfase中的objectFieldOffset(Field f)方法，我曾经在《Java并发编程之锁机制之LockSupport工具》描述过类似的情况。这里我简单的再解释一遍。该方法用于获取某个字段相对 Java对象的“起始地址”的偏移量，也就是说每个字段在类对应的内存中存储是有“角标”的，那么也就是说我们现在的PREV静态变量就代表着Node中prev字段在内存中的“角标”。

当获取到"角标"后，我们再通过U.putObject(node, Node.PREV, oldTail);该方法第一个参数是操作对象，第二个参数是操作的内存“角标”，第三个参数是期望值。那么最后，也就完成了将当前节点的prev字段指向同步队列的尾节点。

当理解了该知识点后，剩下的将同步队列中的tail指针，指向当前节点与如果当前同步队列为空，则构造同步队列这两个操作就非常好理解了。由于篇幅的限制，在这里我就不在进行描述了。希望读者朋友们，能阅读源代码，举一反三。关于这两个方法的代码如下所示：

   private static final sun.misc.Unsafe U = sun.misc.Unsafe.getUnsafe();
    private static final long STATE;
    private static final long HEAD;
    private static final long TAIL;
    static {
        try {
            STATE = U.objectFieldOffset
                (AbstractQueuedSynchronizer.class.getDeclaredField("state"));
            HEAD = U.objectFieldOffset
                (AbstractQueuedSynchronizer.class.getDeclaredField("head"));
            TAIL = U.objectFieldOffset
                (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
        } catch (ReflectiveOperationException e) {
            throw new Error(e);
        }
        Class<?> ensureLoaded = LockSupport.class;
    }

    private final void initializeSyncQueue() {
        Node h;
        if (U.compareAndSwapObject(this, HEAD, null, (h = new Node())))
            tail = h;
    }

    private final boolean compareAndSetTail(Node expect, Node update) {
        return U.compareAndSwapObject(this, TAIL, expect, update);
    }

5.3 唤醒流程

在理解了唤醒的具体逻辑后，现在来总结一下，唤醒的具体流程。具体如下图所示：

2020-08-19-fSuRj3

• 将等待队列中的头节点线程，移动到同步队列中。

• 当移动到同步队列中后。唤醒该线程。让该线程参与同步状态的竞争。

整体流程其实不算太复杂，大家只需要注意，当我们将等待队列中的线程节点加入到同步队列之后，才会唤醒线程。

6. 参考

• 《Java并发编程的艺术》