Java多线程编程-（16）-等待/通知模式接口Condition接口深入分析

前几篇：

Java多线程编程-（12）-Java中的队列同步器AQS和ReentrantLock锁原理简要分析

Java多线程编程-（15）-读写锁ReentrantReadWriteLock深入分析

一、Condition接口简介

在上述两篇文章中讲解AQS的时候，我们已经知道了同步队列AQS的内部类ConditionObject实现了Condition接口，使用ReentrantLock和ReentrantReadWriteLock的内部类Sync我们可以通过newCondition() 方法创建一个或多个ConditionObject对象。

在使用synchronized作为同步的时候，可以使用任意的Java对象作为锁，这是因为任意的一个Java对象，都拥有一组监视器方法，这些监视器方法是定义在超类Object中的，主要包括：wait、notify、notifyAll这些方法，这些方法与synchronized关键字配合实现等待/通知模式。

Condition接口也提供了类似object的监视器方法，与Lock配合使用也可以实现等待/通知模式，虽然如此，但是两者在使用方式以及功能功能上还是有些许差别的，主要差别如下：

Condition接口定义：

可以看到基本和Object超类中定义的差不多。

各接口的含义如下：

二、Condition接口使用案例

示例代码如下:

public class MyService {

    private Lock lock = new ReentrantLock();
    public Condition condition = lock.newCondition();

    public void await() {
        try {
            lock.lock();
            System.out.println(" await时间为 " + System.currentTimeMillis());
            condition.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            lock.unlock();
        }
    }

    public void signal() {
        try {
            lock.lock();
            System.out.println("signal时间为 " + System.currentTimeMillis());
            condition.signal();
        } finally {
            lock.unlock();
        }
    }
}

测试代码如下：

public class Run {

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

        MyService service = new MyService();
        new ThreadA(service).start();
        Thread.sleep(3000);
        service.signal();
    }
}

执行结果：

 await时间为 1509246482943
signal时间为 1509246485946

看过《Java多线程编程-（4）-线程间通信机制的介绍与使用》这篇文章的小伙伴都应该还记得，我们使用wait和notify实现了一个阻塞队列，现在我们使用Condition对象搞一些事情，使用Condition把这个阻塞队列重写一下，代码如下：

public class MyQueue {

    //1、需要一个承装元素的集合
    private final LinkedList<Object> list = new LinkedList<>();
    //2、需要一个计数器
    private final AtomicInteger count = new AtomicInteger(0);
    //3、需要指定上限和下限
    private final int maxSize = 5;
    private final int minSize = 0;

    //5、初始化锁对象
    private final Lock lock = new ReentrantLock();
    private Condition notEmpty = lock.newCondition();
    private Condition notFull = lock.newCondition();

    /**
     * put方法
     */
    public void put(Object obj) {
        lock.lock();
        //达到最大无法添加，进入等到
        while (count.get() == maxSize) {
            try {
                notFull.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        list.add(obj); //加入元素
        count.getAndIncrement(); //计数器增加
        System.out.println(" 元素 " + obj + " 被添加 ");
        notEmpty.signal(); //通知另外一个阻塞的线程方法
        lock.unlock();
    }

    /**
     * get方法
     */
    public Object get() {
        lock.lock();
        Object temp;
        //达到最小，没有元素无法消费，进入等到
        while (count.get() == minSize) {
            try {
                notEmpty.await();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        count.getAndDecrement();
        temp = list.removeFirst();
        System.out.println(" 元素 " + temp + " 被消费 ");
        notFull.signal();
        lock.unlock();
        return temp;
    }

    private int size() {
        return count.get();
    }

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

        final MyQueue myQueue = new MyQueue();
        initMyQueue(myQueue);

        Thread t1 = new Thread(() -> {
            myQueue.put("h");
            myQueue.put("i");
        }, "t1");

        Thread t2 = new Thread(() -> {
            try {
                Thread.sleep(2000);
                myQueue.get();
                Thread.sleep(2000);
                myQueue.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }, "t2");

        t1.start();
        Thread.sleep(1000);
        t2.start();

    }

    private static void initMyQueue(MyQueue myQueue) {
        myQueue.put("a");
        myQueue.put("b");
        myQueue.put("c");
        myQueue.put("d");
        myQueue.put("e");
        System.out.println("当前元素个数：" + myQueue.size());
    }
}

三、Condition接口实现原理

ConditionObject实现了Condition接口，是AQS的内部类，因为Condition的操作需要获取相关联的锁，所以作为同步器的内部类是一个比较合理的方式。每一个Condition对象都包含一个等待队列，该队列是Condition实现等待通知机制的关键。

和synchronized一样，在调用wait和notify等方法之前都必须要先获取锁，同样使用Condition对象的await和signal方法的时候也是要先获取到锁！

1、等待队列

等待队列是一个FIFO的队列，在队列中的每一个节点都包含一个线程的引用，该线程就是在Condition对象上等待的线程，如果一个线程调用了Condition.await() 方法，那么该线程将会释放锁，构造成节点加入等待队列并进入等待状态。这里的节点Node使用的是AQS中定义的Node。也就是说AQS中的同步队列和Condition的等待队列使用的节点类型都是AQS中定义的Node内部类（AbstractQueuedSynchronizer.Node）。

一个Condition对象包含一个等待队列，Condition拥有首节点和尾节点。当前线程调用Condition.await() 方法，将会以当前线程构造节点，并将该节点从尾部加入到等待队列，等待队列的基本结构如下图：

如上图可知，Condition拥有首尾节点的引用，而新增节点只需要将原有的尾节点nextWaiter指向它，并且更新尾节点即可。上述节点引用更新的过程并没有使用到CAS保证，这是因为当前线程调用await（） 方法的时候必定是获取了锁的线程，也就是说该过程是由锁来保证线程安全的。

我们知道在使用synchronized的时候，是使用的对象监视器模型的，即在Object的监视器模型上，一个对象拥有一个同步队列和等待队列，而Lock可以拥有一个同步队列和多个等待队列，这是因为通过lock.newCondition() 可以创建多个Condition条件，而这多个Condition对象都是在同一个锁的基础上创建的，在同一时刻也只能由一个线程获取到该锁。

Lock模式下同步队列和等待队列的对应关系如下图：