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

左手的ㄟ右手 2021-09-28 01:48 413阅读 0赞

> 原文出自 :

--------------------

## 一、Condition接口简介 ##

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

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

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

![这里写图片描述][SouthEast]

Condition接口定义：

![这里写图片描述][SouthEast 1]

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

各接口的含义如下：

![这里写图片描述][SouthEast 2]

## 二、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();
            }
        }
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27

测试代码如下：

public class Run {
    
        public static void main(String[] args) throws InterruptedException {
    
            MyService service = new MyService();
            new ThreadA(service).start();
            Thread.sleep(3000);
            service.signal();
        }
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10

执行结果：

await时间为 1509246482943
    signal时间为 1509246485946
     
      
       1
       2

看过《[Java多线程编程-（4）-线程间通信机制的介绍与使用][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());
        }
    }
     
      
       1
       2
       3
       4
       5
       6
       7
       8
       9
       10
       11
       12
       13
       14
       15
       16
       17
       18
       19
       20
       21
       22
       23
       24
       25
       26
       27
       28
       29
       30
       31
       32
       33
       34
       35
       36
       37
       38
       39
       40
       41
       42
       43
       44
       45
       46
       47
       48
       49
       50
       51
       52
       53
       54
       55
       56
       57
       58
       59
       60
       61
       62
       63
       64
       65
       66
       67
       68
       69
       70
       71
       72
       73
       74
       75
       76
       77
       78
       79
       80
       81
       82
       83
       84
       85
       86
       87
       88
       89
       90
       91
       92
       93
       94
       95
       96
       97

## 三、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()` 方法，将会以当前线程构造节点，并将该节点从尾部加入到等待队列，等待队列的基本结构如下图：

![这里写图片描述][SouthEast 3]

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

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

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

![这里写图片描述][SouthEast 4]

又因为Condition的实现是AQS的内部类，因此每个Condition对象都可以访问AQS同步器提供的方法，相当于每个Condition都拥有所属同步器AQS的引用。

**2、等待的实现**

当前线程调用`Condition.await()` 方法的时候，相当于将当前线程从**同步队列**的首节点移动到Condition的等待队列中，并释放锁，同时线程变为等待状态。

当前线程加入到等待队列的过程如下：

![这里写图片描述][SouthEast 5]

可以看出同步队列的首节点并不是直接加入到等待队列的尾节点，而是封装成等待队列的节点才插入到等待队列的尾部的。

**3、通知的实现**

调用当前线程的`Condition.signal()` 方法，将会唤醒在等待队列中等待时间最长的节点也就是首节点，在唤醒节点之前，会将该节点移到同步队列中。

节点从等待队列加入到同步队列的过程如下：

![这里写图片描述][SouthEast 6]

通过调用同步器的方法将等待队列中的头结点线程安全的移到同步队列的尾节点，当前线程在使用LockSupport唤醒该节点的线程。

被唤醒后的线程，将会从`await（）` 方法中的while循环中退出，进而调用同步器的方法加入到获取同步状态的竞争中。

成功获取同步状态之后，被唤醒的线程从先前调用的await饭发个返回，此时该线程已经成功的获取了锁。

Condition的`signalAll()` 方法，相当于对等待队列中的每一个节点均执行一次`signal（）`方法，效果就是将等待队列中的所有节点全部移到同步队列中，并唤醒每个节点的线程。

参考文章：

1、部分内容和截图来自《Java并发编程的艺术》

2、[http://blog.csdn.net/ghsau/article/details/7481142][http_blog.csdn.net_ghsau_article_details_7481142]

3、[http://ifeve.com/understand-condition/][http_ifeve.com_understand-condition]

4、[http://www.cnblogs.com/zhengbin/p/6420984.html][http_www.cnblogs.com_zhengbin_p_6420984.html]

[SouthEast]: /images/20210923/22aa126206874751b89bf47c4a30bb63.png
[SouthEast 1]: /images/20210923/f0716b91eb694bbbbac81bd56e5a1ce3.png
[SouthEast 2]: /images/20210923/c83072d2d7d544719e6649eefec4adc1.png
[Java_-_4_-]: http://blog.csdn.net/xlgen157387/article/details/78195817
[SouthEast 3]: /images/20210923/a2f933d0fb5642fc86295c9d3101f32c.png
[SouthEast 4]: /images/20210923/afd6ef55bf304d4db508d354fb1ff826.png
[SouthEast 5]: /images/20210923/e2f06edae66741ab9418cafc6b8d820f.png
[SouthEast 6]: /images/20210923/cbf8be43bac9411cb62ccd434835cbcd.png
[http_blog.csdn.net_ghsau_article_details_7481142]: http://blog.csdn.net/ghsau/article/details/7481142
[http_ifeve.com_understand-condition]: http://ifeve.com/understand-condition/
[http_www.cnblogs.com_zhengbin_p_6420984.html]: http://www.cnblogs.com/zhengbin/p/6420984.html