学习JUC源码（3）——Condition等待队列（源码分析结合图文理解）

前言

　　在Java多线程中的wait/notify通信模式结尾就已经介绍过，Java线程之间有两种种等待/通知模式，在那篇博文中是利用Object监视器的方法（wait()，notify()、notifyAll()）实现的，然而在实际生产环境中不推荐使用此方法，建议使用condition的等待通知模式，JUC包中很多核心实现也确实证实了这点，所以这必然是学习JUC包源码的基础。

　　如果之前阅读过前不久介绍同步队列的博文学习JUC源码（1）——AQS同步队列（源码分析结合图文理解），就能更好理解Condition等待队列了，都是基于AQS.Node实现的队列，两者是同步器实现的核心所在！

　　主要参考资料《Java并发编程艺术》（有需要的小伙伴可以找我，我这里只有电子PDF）同时结合ReentranLock、AQS、ArrayBlockingQueue等源码。

一、Condition等待队列介绍

1、对比Object监视器方法与Condition方法

以下对比图来源于《Java并发编程艺术》，可以清楚看到Condition比Object监视器更加灵活，支持中断响应等。

2、Condition方法使用介绍

我们先看下阻塞队列ArrayBlockingQueue中关于condition的经典应用，这里使用就是condition的等待通知模式实现有界阻塞队列，即简单总结：当队列满时，阻塞插入线程，队列空时，获取元素的线程等待。

/** Condition for waiting takes */
    private final Condition notEmpty;

    /** Condition for waiting puts */
    private final Condition notFull;

    public void put(E e) throws InterruptedException {
        checkNotNull(e);
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            // 队列满时等待
            while (count == items.length)
                notFull.await();
            enqueue(e);
        } finally {
            lock.unlock();
        }
    }
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            // 队列空时进入Condition等待队列等待
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally {
            lock.unlock();
        }
    }

从中我们看出两点：

  • Condition要结合Lock对象实现，两者是同时存在的。准确来说先有Lock对象，然后再创建Condition对象
  • 线程调用这些方法时候，需要提前获取到Condition对象关联的锁，Condition对象是由Lock对象（Lock.newConditoin()）创建出来的，也就是说Condition是依赖Lock对象的。

同样地，我们可以从中抽取出等待-通知模式，然后编写Demo如下，其中标红的可以理解为通用的等待/通知模式

public class ConditionDemo {
    ReentrantLock lock = new ReentrantLock();
    // 一般Condition都是作为成员变量
    Condition condition = lock.newCondition();

    public  void waitCondition() throws InterruptedException {
        lock.lock();
        try {
            // 当前线程等待
            // 调用signal方法后返回
            System.out.println(Thread.currentThread()+" is waiting..., now: "+new Date());
            Thread.sleep(2000);
            condition.await(); // 注意这里是await,而不是wait
            System.out.println(Thread.currentThread()+" return..., now: "+new Date());
        } finally {
            lock.unlock();
        }
    }
    public void signalCondition() throws InterruptedException {
        lock.lock();
        try {
            // 唤醒获得condition上的等待锁
            System.out.println(Thread.currentThread()+" is signaling..., now: "+new Date());
            condition.signal();
        } finally {
            lock.unlock();
        }
    }

    public static void main(String[] args) {
        ConditionDemo condition = new ConditionDemo();
            new Thread(()->{
                try {
                    condition.waitCondition();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
            new Thread(()->{
                try {
                    condition.signalCondition();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }).start();
    }

}

运行Demo，可以看到等待线程确实等待了2s之后从await()方法返回。

Thread[Thread-0,5,main] is waiting..., now: Wed Dec 23 21:35:53 CST 2020
Thread[Thread-1,5,main] is signaling..., now: Wed Dec 23 21:35:55 CST 2020
Thread[Thread-0,5,main] return..., now: Wed Dec 23 21:35:55 CST 2020

接下来就是深入源码理解Condition等待队列是如何实现的

二、Condition等待队列的实现分析（源码分析）

1、Condition等待队列介绍

（1）概念认识

先引出简单的认识，其实对比同步队列来说，很好理解，实际上更加简单

  • 等待队列是一个单向FIFO队列，队列每个节点都包含了一个线程引用，该线程是在Condition对象上等待的线程；
  • 实际上这里的等待队列和AQS中的同步队列，都是采用AQS.Node静态内部类；
  • 一个ConditionObject拥有首节点(fisrtWaiter)和尾节点(lastWaiter)；
  • 如果一个线程调用了Condition.await()方法，那么该线程将会释放锁（从同步队列中移除），构造成节点加入等待队列，等待被唤醒；
  • 如果一个线程调用了Condition.signal()方法，那么该线程将会被唤醒（从等待队列中移除），构造成节点加入同步队列，尝试重新获取同步状态；

（2）等待队列结构图

实际上，Condition的实现是在AQS中内部类ConditionObject实现Condition具体实现的：

　　　　　　　　

等待队列的结构图如下图，相比较同步队列而言：

  • 等待队列来说更加简单，是单向FIFO队列；
  • Condition拥有首尾节点引用，新增节点直接nextWaiter指向即可，这个过程不需要CAS保证，因为调用Condition.await()方法肯定是获取了锁的线程，也就是说该过程是来保证线程安全的。

　　　　　　

实际上，AQS同步器只拥有一个同步队列，但却有多个Condition等待队列，如下图。

　　　　　　　　

　　　　　　　　　　　　　　

2、await()方法实现解析

当调用await()方法时，相当于同步队列的首节点（获取了锁的节点）移动到了Condition的等待队列中。

　　　

更具体来说是首先调用await()方法之前肯定是能获取到同步状态的线程，也就是同步队列中首节点，之后调用await()方法由将释放锁，进入等待队列。

分析源码（重点部分都已经注释，结合图应该更好理解）：

1）调用await()方法，通过addConditionWaiter()方法加入等待线程，然后释放全部同步状态

2）进入while循环，判断是否已经移动到同步队列中，如果已经被移动到同步队列中则说明线程已经被唤醒（signal）；

3）接下来尝试获取竞争同步状态，即调用acquireQueue方法

　　public final void await() throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            // addConditionWaiter()方法当前线程加入等待队列
            Node node = addConditionWaiter();
            // 调用await()方法后释放锁（同步状态）
            int savedState = fullyRelease(node);
            int interruptMode = 0;
            // 检查node是否在同步队列中，不是的话说明已经获取到锁
            //  LockSupport.unpark唤醒线程后，从这里返回，此时已经在SyncQueue同步队列中，退出循环
            // 从这里也可以看出，也是经典的等待/通知模式
            while (!isOnSyncQueue(node)) {
                // 阻塞当前线程
                LockSupport.park(this);
                // 在调用signal前抛出中断异常，或者调用之后中断，都退出循环
                // THROW_IE if interrupted  before signalled, REINTERRUPT if after signalled
                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);
        }

结构流程如下图：

3、signal()方法实现解析

调用signal方法将会唤醒等待队列中等待时间最长的节点（首节点），在唤醒节点之前，会将节点移动到同步队列中

　　　　

具体来说当前线程获取到了锁，接着获取等待队列的首节点，将其移动到同步队列中，并且唤醒节点中的线程。

分析源码（重点部分都已经注释，结合图应该更好理解）：

1）进入signal()方法，调用doSignal(Node node)方法移动到同步队列中，并唤醒节点中线程

　　public final void signal() {
            // 判断当前线程是否持有获得锁
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                // 移动到同步队列中，并且唤醒节点中的线程
                doSignal(first);
        }
　　private void doSignal(Node first) {
            do {
                // 如果等待队列中只有一个节点（即首节点），则唤醒首节点后lastWaiter置空
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
                // 否则获取等待队列中的首节点，即next域断开置空
                first.nextWaiter = null;
            } while (!transferForSignal(first) &&
                     (first = firstWaiter) != null);
        }

2）doSignal(Node node)方法中调用transferForSignal(Node node)，通过调用enq(Node node)方法(这里其实就是同步队列的入队enq(Node node)方法)，等待队列中的头结点线程安全地移动到同步队列，当节点移动到同步队列后，当前线程将会被唤醒（LockSupport.unpark(node.thread)）。

final boolean transferForSignal(Node node) {
        /*
         * If cannot change waitStatus, the node has been cancelled.
         */
        // 如果没有正确设置等待状态为初始状态准备加入同步队列中，则返回，当前节点状态为Node.CONDITION
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        // 将等待队列中的头结点移动到同步队列中, 返回已经加入的当前node在同步队列中前节点
        Node p = enq(node);
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            // 唤醒当前node线程，返回while(isOnSynQueue(Node node))处，退出循环
            LockSupport.unpark(node.thread);
        return true;
    }

流程结构如下图：　

赞 (0)

相关推荐