又见队列同步器——Condition接口的实现

你的名字 2023-07-17 14:14 5阅读 0赞

## Condition接口与Object监视器方法 ##

任意一个Java对象，都拥有一组监视器方法（定义在java.lang.Object上），主要包括wait()、wait(long timeout)、nofity()以及notifyAll()方法，这些方法与synchronized同步关键字配合，可以实现等待/通知模式。Condition接口也提供了类似Object的监视器方法，与Lock配合也可以实现等待/通知模式，但是这两者在使用方式以及功能特性上是有差别的。

在[Java线程通信与协作的解决方案——等待/通知机制][Java]这篇博文中有对二者实现等待/通知模式的详细分析。

Condition是个接口，基本的方法就是await()和signal()方法。Condition依赖于Lock接口，生成一个Condition的基本代码是lock.newCondition() 。 必须要注意的是，Condition的await()/signal() 使用都必须在lock保护之内，也就是说，必须在lock.lock()和lock.unlock之间才可以使用。

Condition接口相对于对象监视器强大的地方在于它能够精确的控制多线程的休眠与唤醒（注意是唤醒，唤醒只意味着进入了同步队列，不意味着一定能获得资源），例如有A、B、C、D四个线程共享Z资源，如果A占用了Z，并且调用了b\_condition.signal()就可以释放资源唤醒B线程，而Object的nofity就无法保证B、C、D中会被唤醒的是哪一个了。Condition接口的await/signal机制是设计用来代替监视器锁的wait/notify机制的。

通过对比Object的监视器方法与Condition接口，可以更详细地了解Codition的特性，对比结果如下：

![20200322121053837.png][]

Condition接口的定义如下：

public interface Condition { 
     
         // 当前线程进入等待状态直到被通知(signal)或被中断
        void await() throws InterruptedException;
        // 不响应中断等待，直到被通知(signal)
        void awaitUninterruptibly();
        // 等待指定时长直到被通知或中断或超时。
        long awaitNanos(long nanosTimeout) throws InterruptedException;
        // 等待指定时长直到被通知或中断或超时。
        boolean await(long time, TimeUnit unit) throws InterruptedException;
        // 当前线程进入等待状态直到被通知、中断或者到某个时间。如果没有到指定事件就被通知，方法返回true，否则false。 
        boolean awaitUntil(Date deadline) throws InterruptedException;
        // 唤醒一个等待在Condition上的线程，该线程从等待方法返回前必须获得与Condition相关联的锁 
        void signal();
        // 唤醒所有等待在Condition上的线程，该线程从等待方法返回前必须获得与Condition相关联的锁 
        void signalAll();
    }

先来看一个Java官方文档提供的使用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;
    
        // 生产者方法，往数组里面写数据
        public void put(Object x) throws InterruptedException { 
            lock.lock();
            try { 
                while (count == items.length)
                    //数组已满，没有空间时，挂起等待，直到数组不非满（notFull）
                    notFull.await(); 
                items[putptr] = x;
                if (++putptr == items.length) 
                    putptr = 0;
                ++count;
                // 因为放入了一个数据，数组肯定不是空的了
                // 此时唤醒等待这notEmpty条件上的线程
                notEmpty.signal(); 
            } finally { 
                lock.unlock();
            }
        }
    
        // 消费者方法，从数组里面拿数据
        public Object take() throws InterruptedException { 
            lock.lock();
            try { 
                while (count == 0)
                    // 数组是空的，没有数据可拿时，挂起等待，直到数组非空（notEmpty）
                    notEmpty.await(); 
                Object x = items[takeptr];
                if (++takeptr == items.length) 
                    takeptr = 0;
                --count;
                // 因为拿出了一个数据，数组肯定不是满的了
                // 此时唤醒等待这notFull条件上的线程
                notFull.signal();
                return x;
            } finally { 
                lock.unlock();
            }
        }
    }

以上是一个典型的生产者-消费者模型。这里在同一个lock锁上，创建了两个条件队列notFull、 notEmpty。当数组已满没有存储空间时，put方法在notFull条件上等待，直到数组又变得不满；当数组空了，没有数据可读时，take方法在notEmpty条件上等待，直到数组变得不空，而notEmpty.signal()和notFull.signal()则用来唤醒等待在这个条件上的线程。

注意，上面所说的，在notFull及notEmpty条件上等待，事实上是指线程在等待队列（condition queue，也叫条件队列）上等待，当该线程被相应的signal()方法唤醒后，将进入到同步队列中去争锁，争抢到了锁后才能能await()方法处返回。即唤醒只意味着进入了同步队列，不意味着一定能获得资源。

这里接牵涉到两种队列——等待队列（condition queue）和同步队列（sync queue），它们都定义在AbstractQueuedSynchronizer中。

## 同步队列与等待队列 ##

在[从源码角度理解ReentrantLock及队列同步器][ReentrantLock]这篇博文中，我们了解到，所有等待ReentrantLock独占锁的线程都会被包装成Node对象扔到一个同步队列中。该同步队列的结构如下：

![20200322124737306.png][]

sync queue是一个双向链表，我们使用prev、next属性来串联节点。但是在这个同步队列中，我们一直没有用到nextWaiter属性，即使是在共享锁模式下，这一属性也只作为一个标记，指向了一个空节点，因此，在sync queue中，我们不会用它来串联节点。

AQS对Condition这个接口的实现主要是通过ConditionObject类，上面已经说过，它的核心实现就是一个等待队列（condition queue）。如下面类图所示，ConditionObject是同步器AbstractQueuedSynchronizer的内部类，因为Condition的操作需要获取相关联的锁，所以作为同步器的内部类也较为合理。每个Condition对象都包含着一个等待队列，该队列是Condition对象实现等待/通知功能的关键。

![20200322124319760.png][]

ConditionObject的核心属性只有两个：

/** First node of condition queue. */
    private transient Node firstWaiter;
    /** Last node of condition queue. */
    private transient Node lastWaiter;

这两个属性分别代表了等待队列（condition queue）的队头和队尾，每当我们新建一个ConditionObject对象，都会对应一个等待队列。

等待队列（condition queue）也是一个FIFO的队列，在队列中的每个节点都包含了一个线程引用，该线程就是在ConditionObject对象上等待的线程，如果一个线程调用了condition.await()方法，那么该线程将会释放锁并构造成节点加入等待队列，进入等待状态。事实上，节点的定义复用了同步器中节点的定义，也就是说，同步队列和等待队列中节点类型都是同步器的静态内部类AbstractQueuedSynchronizer.Node，每创建一个CondtionObject对象就会对应一个等待队列，每一个调用了await()方法的线程都会被包装成Node扔进一个等待队列（condition queue）中，就像下图这样：

![20200322124812961.png][]

值得注意的是，condition queue是一个单向链表，在该链表中我们使用nextWaiter属性来串联链表。就像在sync queue中不会使用nextWaiter属性来串联链表一样，在condition queue中，也并不会用到prev和next属性，它们的值都为null。也就是说在等待队列中，Node节点真正用到的属性只有三个：

*  thread：代表当前正在等待某个条件的线程
 *  waitStatus：条件的等待状态
 *  nextWaiter：指向等待队列中的下一个节点

waitStatus变量的取值范围：

volatile int waitStatus;
    static final int CANCELLED =  1;
    static final int SIGNAL    = -1;
    static final int CONDITION = -2;
    static final int PROPAGATE = -3;

在等待队列中，我们只需要关注一个值即可——CONDITION。它表示线程处于正常的等待状态，而只要waitStatus不是CONDITION，我们就认为线程不再等待了，此时就要从等待队列中出队。

### 同步队列与等待队列的关系 ###

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

![20200322122551855.png][]

ConditionObject类是AQS的内部类，因此每个Condition实例都能够访问同步器提供的方法，相当于每个Condition都拥有所属同步器的引用。

![ad98ffdcb56a49608d3f062ef29648a2.png][]

## Condition接口方法的实现分析 ##

找到JDK中的java.util.concurrent.locks.AbstractQueuedSynchronizer.java源码，查看其内部类ConditionObject的代码，分析Condition接口方法的实现逻辑。

### 等待await() ###

public final void await() throws InterruptedException { 
        // 如果当前线程在调动await()方法前已经被中断了，则直接抛出InterruptedException
        if (Thread.interrupted())
            throw new InterruptedException();
        // 将当前线程封装成Node添加到条件队列
        Node node = addConditionWaiter();
        // 释放当前线程所占用的锁，保存当前的锁状态
        int savedState = fullyRelease(node);
        int interruptMode = 0;
        // 如果当前节点不在同步队列中，说明刚刚被await, 还没有人调用signal方法，则直接将当前线程挂起
        while (!isOnSyncQueue(node)) { 
            LockSupport.park(this); // 线程将在这里被挂起，停止运行
            // 能执行到这里说明要么是signal方法被调用了，要么是线程被中断了
            // 所以检查下线程被唤醒的原因，如果是因为中断被唤醒，则跳出while循环
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                break;
        }
        //运行到这里说明isOnSyncQueue(node)方法返回true，线程从上面循环中退出了，即被signal唤醒，下面就开始加入获取同步状态（锁）的竞争之中
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
            interruptMode = REINTERRUPT;
        if (node.nextWaiter != null) // clean up if cancelled
            unlinkCancelledWaiters();
        if (interruptMode != 0)
            reportInterruptAfterWait(interruptMode);
        
    }

把当前线程封装成Node扔进等待队列中的addConditionWaiter方法：

private Node addConditionWaiter() { 
        Node t = lastWaiter;
        // 如果尾节点被cancel了，则先遍历整个链表，清除所有被cancel的节点
        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;
    }

ConditionObject对象拥有首尾节点的引用，而新增节点只需要将原有的尾节点nextWaiter指向它，并且更新尾节点即可。上述节点引用更新的过程并没有使用CAS保证，首先思考存在两个不同的线程同时入队的情况吗？不存在。原因在于能调用await()方法的线程必然是已经获得了锁的线程，而获得了锁的线程只有一个，所以这里不存在并发，因此不需要CAS操作。

如果从队列（同步队列和等待队列）的角度看awit()方法，当调用await()方法时，相当于将同步队列的首节点（获取了锁的节点）移动到Codition的等待队列的末尾。同步队列首节点通过addConditionWaiter()方法加入等待队列的过程如下图所示：

![2020032213025437.png][]

### 通知signal() ###

public final void signal() { 
        //检查当前调用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) { 
        // 如果该节点在调用signal方法前已经被取消了，则直接跳过这个节点
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
    
       // 如果该节点在条件队列中正常等待，则利用enq方法将该节点添加至sync queue队列的尾部
        Node p = enq(node);
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        	//唤醒当前节点的线程
            LockSupport.unpark(node.thread);
        return true;
    }
    
    
    private Node enq(final Node node) { 
        for (;;) { 
            Node t = tail;
            if (t == null) {  // Must initialize
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else { 
                node.prev = t;
                if (compareAndSetTail(t, node)) { 
                    t.next = node;
                    return t;
                }
            }
        }
    }

调用该方法的前置条件是当前线程必须获取了锁，可以看到signal()方法进行了isHeldExclusively()检查，也就是当前线程必须是获取了锁的线程。doSignal()方法是一个do-while循环，目的是遍历整个条件队列，找到第一个没有被cancelled的节点，并将它添加到同步队列的末尾。

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

![20200322130550397.png][]

通过调用同步器的enq(Node node)方法，等待队列中的头节点线程安全地移动到同步队列。当节点移动到同步队列后，当前线程再使用LockSupport唤醒该节点的线程。被唤醒后的线程，将从await()方法中的while循环中退出，由于节点已经在同步队列中，isOnSyncQueue(Node node)方法将返回true，进而调用同步器的acquireQueued(node, savedState)方法加入到获取同步状态的竞争中。注意，这里传入的需要获取锁的重入数量是savedState，即之前释放了多少，这里就需要再次获取多少。

final boolean acquireQueued(final Node node, int arg) { 
        boolean failed = true;
        try { 
            boolean interrupted = false;
            for (;;) { 
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) { 
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt()) // 如果线程获取不到锁，则将在这里被阻塞住
                    interrupted = true;
            }
        } finally { 
            if (failed)
                cancelAcquire(node);
        }
    }

acquireQueued()方法是一个阻塞式的方法，获取到锁则退出，获取不到锁则会被挂起。该方法只有在最终获取到了锁后，才会退出。

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

Condition的signalAll()方法，相当于对等待队列中的每个节点均执行一次signal()方法，效果就是将等待队列中所有节点全部移动到同步队列中，并唤醒每个节点的线程。但是，这里尤其要注意的是，node是被一个一个转移过去的，哪怕我们调用的是signalAll()方法也是一个一个转移过去的，而不是将整个condition queue队列接在sync queue的末尾。

## Condition接口相对于synchronized的优势 ##

Lock解决了synchronized不能处理死锁的问题，并对功能进行了扩展，而Condition接口就是解决synchronized只能有一个条件变量的缺点。

同一个synchronize内置锁只对应一个wait set，即当线程调用wait方法时，会把当前线程放入到同一个wait set中，当我们需要根据某些特定的条件来唤醒符合条件的线程时，我们只能先从wait set里唤醒一个线程后，再看是否符合条件，如果不符合条件，则需要将此线程继续wait，然后再去wait set中获取下一个线程再判断是否满足条件。这样会导致许多无意义的cpu开销。所以Condition就是为了解决上面问题存在的。每一个锁都会对应多个Condition，每个Condtition都有一个容器来保存相应的等待线程，拿到锁的线程想唤醒某个等待特定条件的线程时，只需要去唤醒对应Condition容器中的线程即可。

[Java]: https://blog.csdn.net/fuzhongmin05/article/details/65448387
[20200322121053837.png]: /images/20230528/54c954978a704651b13c76deaa27f032.png
[ReentrantLock]: https://blog.csdn.net/fuzhongmin05/article/details/104721667
[20200322124737306.png]: /images/20230528/3a4d4d8db2214691a042b47f9bae47c6.png
[20200322124319760.png]: /images/20230528/7b410008096f4d7b91cf20a329d36b2e.png
[20200322124812961.png]: /images/20230528/a8f756d2b08142c2b2c399c040480fcf.png
[20200322122551855.png]: /images/20230528/4f84aff89f764003a3e82927a0f12d79.png
[ad98ffdcb56a49608d3f062ef29648a2.png]: /images/20230528/dd78bcabf1c847be8f0bb037e49299b7.png
[2020032213025437.png]: /images/20230528/a897036d270e4ed9bc66483cddafc4b5.png
[20200322130550397.png]: /images/20230528/32256f51a0ad47f8ab4f245279eaac0a.png