22.Condition的功能和原理

Posted 2022-09-22 纵横千里，捭阖四方

前面我们比较完整的介绍了重入锁的原理和实现过程，本开始，我们梳理几个的比较重要的并发工具：Condition、CountDownLatch、Semaphone、CyclicBarrier和原子类。本章我们先看Condition。

Condition在业务代码中使用的并不算多，但是在很多开源框架的源码中有大量应用。它的作用和wait()/notify()方法相同，都是基于某个条件去等待和唤醒，所以可以认为两者作用基本一致。

Condition有两个方法，说明如下，这两个方法与wait()/notify()/notifyAll()是对应的。

• await()方法，让线程等待，并释放锁。

• signal()/signalAll()方法，唤醒被await()方法阻塞的线程。

1 Condition的基本应用

下面通过一个比较简单的案例讲解一下Condition的基本应用

public class ConditionExampleWait implements  Runnable

    private Lock lock;
    private Condition condition;
    public ConditionExampleWait(Lock lock, Condition condition)
        this.lock=lock;
        this.condition=condition;
    

    @Override
    public void run() 
        System.out.println("begin -ConditionExampleWait");
        try 
            lock.lock();
            condition.await();
            System.out.println("end - ConditionExampleWait");
         catch (InterruptedException e) 
            e.printStackTrace();
        finally 
            lock.unlock();
        
    

ConditionExampleWait()方法用来实现条件等待，其原理基本和wait()/notify()相同，区别在于如下两个方面：

• 同步锁采用JUC的Lock，调用await()方法前需要加锁。

• 让线程等待的方法变成了await()方法。

同样，调用了await()方法后会释放当前持有的锁，使得其他线程可以有机会抢占到锁资源。

我们再看一下signal的用法：

public class ConditionExampleSignal  implements  Runnable
    private Lock lock;
    private Condition condition;
    public ConditionExampleSignal(Lock lock, Condition condition)
        this.lock=lock;
        this.condition=condition;
    

    @Override
    public void run() 
        System.out.println("begin -ConditionExampleSignal");
        try 
            lock.lock();
            condition.signal();
            System.out.println("end - ConditionExampleSignal");
        finally 
            lock.unlock();
        
    

ConditionExampleSignal类调用condition.signal()方法唤醒被condition.await()阻塞的线程。如果await和signal要通信，就必须要持有同一把锁，也就是两个线程 Lock()锁是同一个示例。

调用方法：

public class ConditionTestMain 
    public static void main(String[] args) throws InterruptedException 
        Lock lock = new ReentrantLock();
        Condition condition = lock.newCondition();
        ConditionExampleWait conditionExampleWait = new ConditionExampleWait(lock, condition);
        ConditionExampleSignal conditionExampleSignal = new ConditionExampleSignal(lock, condition);
        new Thread(conditionExampleWait).start();
        Thread.sleep(10);
        new Thread(conditionExampleSignal).start();
    

我们可以看到这里与使用await()的一个很大区别是需要先创建Condition实例，Condition可以根据不同的场景设置多个不同的Condition，当调用 condition.signal()方法时，不需要唤醒所有线程，只需要唤醒指定Condition的线程即可，因此这样可以减少线程的无效竞争，提高效率。

2 等待过程的设计与实现

我们接下来开始分析await的设计过程和实现原理，首先抢占锁的逻辑可以基于重入锁来实现，而ReetrantLock又使用了AQS中的排他锁实现的，所以这里还是用到了AQS实现线程的同步机制。

其次，通过condition.await()方法阻塞的线程会释放锁，释放锁的这个线程未来还是需要通过再次竞争锁来恢复执行，所以condition中应该也有一个队列来保存这些被阻塞的线程。另外，其他线程调用condition.notify()方法时，应该从等待队列中的线程移动到AQS的CLH队列再去竞争锁，从而完成整体的流程。

因此，我们断定condition中还有一个等待队列，这个队列独立于与AQS的CLH队列，但是两者能协同工作。

2.1 await()方法

首先可以看到Condition是一个接口：

 具体实现类是ConditionObject，而且该类还是AQS的内部类，这是因为Condition的抢占逻辑等依赖AQS。

等待的过程我们从上面例子ConditionExampleWait的condition.await()开始。

public final void await() throws InterruptedException 
    if (Thread.interrupted())
        throw new InterruptedException();
    //创建一个新结点，状态为condition，采用的数据结构仍然是链表。
    Node node = addConditionWaiter();
    //释放当前的锁，得到锁的状态，并唤醒AQS队列中的一个线程
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    //如果当前结点没有再同步队列上，即还没有被signal，则将当前线程阻塞
    //判断这个队列是否在AQS队列中，第一次判断的结果是false，因为前面已经释放过了。
    while (!isOnSyncQueue(node)) 
    //第一次总是park自己，开始阻塞等待
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    
    //interruptMode != THROW_IE表示这个线程没有成功将node入队，但signal执行了enq()方法让其入队了。
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
        //线程被中断
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);

从上面看，condition.await()方法会做三件事，我们只需要根据这三件事找到源码实现即可。

• 把当前线程添加到等待队列中，在上述代码中是通过addConditionWaiter()方法来实现的。

• 释放锁，使用fullyRelease()方法来完成锁的释放。

• 阻塞当前线程，仍然采用 LockSupport.park()方法来阻塞自己。

2.2 addConditionWaiter

我们继续看上面的addConditionWaiter，我们前面说Condition要维护一个自己的队列，而这个队列就是在addConditionWaiter中构建的：

private Node addConditionWaiter() 
    Node t = lastWaiter;
    // 如果等待队列不为空，则尝试清理失效的结点
    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;

很明显，这里的主要作用就是把当前线程添加到等待队列中。

2.3 fullyRelease

当线程添加到队列后，就可以安心调用fullyRelease来释放锁。这里之所以要释放锁，是因为当前线程是阻塞的，再占用资源已经没有意义了。不过当前锁的释放和lock.unlock()还是有些区别的，fullyRelease是彻底释放锁。而重入锁因为可能多次重入，因此每次unlock()不一定能将state降到0，因此可能导致其他线程仍然不能获得锁。

 final int fullyRelease(Node node) 
    boolean failed = true;
    try 
    //获得当前锁的状态值
        int savedState = getState();
        //释放锁
        if (release(savedState)) 
            failed = false;
            return savedState;
         else 
            throw new IllegalMonitorStateException();
        
     finally 
        if (failed)
            node.waitStatus = Node.CANCELLED;
    

fullyRelease()方法中彻底释放锁的方式不难理解，就是得到当前线程的重入次数savedState，直接用state减去总的重入次数即可。注意saveState需要保存起来，后序线程被唤醒后，锁的重入次数必须要恢复到阻塞之前，否则会导致后续释放时出现异常。

2.4 isOnSyncQueue

判断这个节点是否在AQS的同步队列中，由于前面已经释放了锁，当前线程必然不在，所以第一次判断的结果是false。后续如果有其他线程调用了condition.signal()方法，那么该线程又会移动到AQS的同步队列中，所以这个条件的判断表示其他线程还没有调用signal()方法，需要通过LockSupport.park(this);方法阻塞当前线程，代码如下：

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

小结

当Condition.await()方法整体执行完成后，假如ThreadA调用condition.await()方法之后，会执行两个工作：

• 构建一个Condition等待队列，把ThreadA线程包装成Node结点并添加到该队列中。

• ThreadA释放锁之后，使得原本处于同步队列的ThreadB抢占锁。

3 signal过程设计与实现原理

3.1 signal()方法

调用signal()之后，会唤醒处于Condition等待队列中的线程。被唤醒的线程需要等到消费者线程调用lock.unlock()方法来释放锁之后才能真正执行，代码如下：

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

通过前面的分析，condition.signal()方法应该要把等待队列中等待最久的节点（头部节点）移动到AQS的CLH队列中，之所以要这么做，是因为该线程仍然需要去竞争锁，如果不这样做，那么锁的排他性就会被打破。

doSignal()方法在唤醒等待队列中的线程是采用了循环来保证等待队列中正常节点的成功转移，循环条件的逻辑如下：

• first是当前condition队列的头结点，通过调用transferForSignal()方法把first结点转移到AQS队列中，如果失败（结点是CANCELLED状态），则继续查找下一个节点。

• (first = firstWaiter) != null)表示当first节点不为空时才执行transfer操作。

3.2 transferForSignal()方法

我们继续看transferForSignal()方法，该方法表示转移等待队列的头结点到CLH队列中，并且唤醒该线程。

final boolean transferForSignal(Node node) 
//如果CAS失败，则说明当前结点状态为CANCELLED，此时需要继续查找等待队列中的下一个结点
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;

    Node p = enq(node);
    int ws = p.waitStatus;
    //如果上一节结点的状态被取消，或者尝试设置上一个结点的状态为SIGNAL失败了，SIGNAL表示next结点需要停止阻塞
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);//唤醒输入结点上的线程
    return true;

这里需要说明的是ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)这个条件，则唤醒当前线程。之所以这么做，是因为把等待队列中的结点移动到同步队列之后，在同步队列中触发锁竞争自然会被唤醒，可为什么要提前唤醒呢？

注意这里的

• 条件1：p结点是CLH队列中原来的tail结点，ws>0表示原tail结点的状态为CANCELLED状态。

• 条件2：compareAndSetWaitStatus修改原tail结点的状态为SIGNAL失败了。

满足两个条件的任意一个，在提前唤醒当前线程后， 我们再看一下await()方法的代码：

   while (!isOnSyncQueue(node)) 
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;

可以看到，当前线程如果被提前唤醒，就可以预先执行checkInterruptWhileWaiting()方法，然后通过acquireQueued()方法来抢占锁，这意味着当前线程就可以提前执行不需要涉及同步操作的代码。

这里之所以要满足两个条件才被唤醒，是因为当前结点状态为CANCELLED时，AQS会执行一次失效结点的清理工作。而当前从等待结点被转移到同步队列中的线程不需要等待这个清理过程完成后再执行，这可以提高性能。

在ThreadB执行完signal()方法之后，整个数据存储变化如下图所示：

• 先从condition队列中取出first结点。

• 然后把这个first结点转移到AQS的CLH队列中。

• ThreadB释放锁，接着ThreadA竞争到锁被唤醒，继续从await()阻塞的地方开始执行。

4 锁竞争成功的过程

从阻塞队列转移到AQS的CLH队列中的线程，在竞争到锁之后，通过await()方法被唤醒。

4.1 await方法

继续执行await代码：

public final void await() throws InterruptedException 
    if (Thread.interrupted())
        throw new InterruptedException();
    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);

被唤醒的线程通过if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)来检查线程的唤醒是因为中断，还是因为正常的signal()方法。

checkInterruptWhileWaiting()这个方法是用来判断被唤醒的线程是否因为interrupt()方法导致的，如果是，则调用transferAfterCancelledWait()方法判断后续处理应该是抛出异常还是重新中断.

private int checkInterruptWhileWaiting(Node node) 
    return Thread.interrupted() ?
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;

• 如果Thread.interrupted()方法如果返回true，表示被中断过，那么会调用transferAfterCancelledWait(node)方法。

• 否则，表示没有被触发中断，直接返回0。返回0之后，继续进入while(!isOnSyncQueue())循环中，此时因为前面已经调用过signal()，所以这里应该返回false，并跳出循环。

• 在transferAfterCancelledWait(node)中，如果返回true，则返回THROW_IE，表示抛出异常并唤醒线程，否则返回REINTERRUPT，后续重新进行中断。

4.2 transferAfterCancelledWait

final boolean transferAfterCancelledWait(Node node) 
//如果能修改成功，说明线程被中断时signal()方法还没有被调用
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) 
        enq(node);
        return true;
    
    while (!isOnSyncQueue(node))
        Thread.yield();
    return false;

这个方法主要用于判断被interrupt()中断是发生在signal()方法调用之前，还是调用之后。其中compareAndSetWaitStatus()方法用来判断在线程触发中断之前，signal()方法是否被调用。如果已经被调用过，那么node的状态不应该是CONDITION。

继续看await()方法：

if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);

interruptMode有三个返回值：

• THROW_IE(-1)，表示在触发中断之前，signal()方法还没有被调用，此时直接抛出异常。

• REINTERRUPT(1)，表示在signal()方法被调用之后执行，这意味着当前线程移到了AQS同步队列中，这是需要再次触发interrupt。

• 0，表示正常状态。

这里的逻辑是首先调用acquireQueued()方法，让当前线程去争抢同步锁。当然这个不一定能抢到，如果抢不到则继续在AQS同步队列中等待；如果抢到了，则直接唤醒ThreadA继续执行后续代码。

• node.nextWaiter!=null,如果为true，则调用unlinkCancelledWaiters()释放已取消的结点。

• reportInterruptAfterWait根据interruptMode判断是要抛出异常还是重新触发一次中断。

然后，ThreadA重新通过acquireQueued()方法竞争到锁之后 ，继续执行run()方法中的代码。