AQS底层源码深度剖析-BlockingQueue

目录

AQS底层源码深度剖析-BlockingQueue

BlockingQueue定义

队列类型

队列数据结构

ArrayBlockingQueue

LinkedBlockingQueue

DelayQueue

BlockingQueue API

添加元素

检索(取出)元素

BlockingQueue应用队列总览图

AQS底层源码深度剖析-BlockingQueue【重点中的重点】

看源码前，先要明白Condition的含义：

BlockingQueue源码会涉及三个队列【重点掌握】：

put()源码剖析：

 take()源码剖析：

 总结：

---

AQS底层源码深度剖析-BlockingQueue

BlockingQueue定义

线程通信一个工具，在任意时刻，不管并发有多高，在单台JVM上，同一时间永远只能有一个线程能够对队列进行入队或者出队操作。

官方点说：BlockingQueue，是java.util.concurrent包提供的用于解决并发生产者-消费者问题的最有用的类，它的特性是在任意时刻只有一个线程可以进行take或put操作，并且BlockingQueue提供了超时return null的机制，在许多生产应用场景里都可以看到这个工具的身影。

应用场景：

线程池，springcloud-Eureka的三级缓存，Nacos，Netty，MQ

队列类型

1. 无限队列 （unbounded queue ） - 几乎可以无限增长
2. 有限队列 （ bounded queue ） - 定义了最大容量

队列数据结构

队列实质就是一种存储数据的结构

• 通常用链表或者数组实现
• 一般而言队列具备FIFO先进先出的特性，当然也有双端队列（Deque）优先级队列
• 主要操作：入队（EnQueue）与出队（Dequeue）

常见的4种阻塞队列

• ArrayBlockingQueue 由数组支持的有界队列
• LinkedBlockingQueue 由链接节点支持的可选有界队列
• PriorityBlockingQueue 由优先级堆支持的无界优先级队列
• DelayQueue 由优先级堆支持的、基于时间的调度队列

ArrayBlockingQueue

队列基于数组实现,容量大小在创建ArrayBlockingQueue对象时已定义好

数据结构如下图：

 队列创建：

BlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>();

应用场景

在线程池中有比较多的应用，生产者消费者场景

工作原理

基于ReentrantLock保证线程安全，根据Condition实现队列满时的阻塞

LinkedBlockingQueue

是一个基于链表的无界队列(理论上有界)

BlockingQueue<String> blockingQueue = new LinkedBlockingQueue<>();

上面这段代码中，blockingQueue的容量将设置为Integer.MAX_VALUE

向无限容量的队列中添加元素的所有操作都将永远不会阻塞，因此它可以增长到非常大的容量(注意 这里不是说不会加锁保证线程安全，同样会加锁来保证同一时刻只会有一个线程对队列添加元素或取出元素成功)

使用无限容量的BlockingQueue设计生产者-消费者模型时最重要的是消费者应该能够像生产者向队列添加消息一样快的消费消息。否则存储消息数据的内存可能会填满，然后得到一个OutOfMemory异常

DelayQueue

由优先级堆支持的、基于时间的调度队列，内部基于无界队列PriorityQueue实现，而无界队列基于数组的扩容实现。

队列创建：

BlockingQueue<String> blockingQueue = new DelayQueue();

要求：

入队的对象必须要实现Delayed接口,而Delayed集成自Comparable接口

应用场景：

电影票

工作原理：

队列内部会根据时间优先级进行排序。延迟类线程池周期执行。

BlockingQueue API

BlockingQueue 接口的所有方法可以分为两大类：负责向队列添加元素的方法和检索(取出)这些元素的方法。在队列满/空的情况下，来自这两个组的每个方法的行为都不同。

添加元素

方法	说明
add()	如果插入成功则返回 true，否则抛出 IllegalStateException 异常
put()	将指定的元素插入队列，如果队列满了，那么会阻塞直到有空间插入
offer()	如果插入成功则返回 true，否则返回 false
offer(E e, long timeout, TimeUnit unit)	尝试将元素插入队列，如果队列已满，那么会阻塞直到有空间插入

检索(取出)元素

方法	说明
take()	获取队列的头部元素并将其删除，如果队列为空，则阻塞并等待元素变为可用
poll(long timeout, TimeUnit unit)	检索并删除队列的头部，如有必要，等待指定的等待时间以使元素可用，如果超时，则返回 null

在构建生产者 - 消费者程序时，这些方法是 BlockingQueue 接口中最重要的构建块。

BlockingQueue应用队列总览图

应用队列：存储消息数据的队列。无论是consumer还是producer，想要对应用队列中的消息数据进行操作(存入或取出)时，必须先获取到锁对象。如果获取不到锁，则无法操作。

(1) 

极端情况下：

当producer把队列容量放满了，那么producer释放锁，producer阻塞，让consumer获取到锁 然后去消费消息数据

同理当consumer消费完队列中的消息数据，那么consumer会释放锁，consumer阻塞，让producer获取到锁，然后去生产并且加入消息数据到队列中

普通情况下：

当然应用队列中没有放满，consumer也可以消费取出数据 。队列数据没有被消费完时，producer也可以生产消息并且存入应用队列中。

(2) 当producer在同步队列中存入一个消息数据后，会进行通知consumer，consumer接收到通知，会从条件队列中转移到阻塞CLH队列，在CLH阻塞队列中的consumer会进行消费应用队列中存储的消息数据

 AQS底层源码深度剖析-BlockingQueue【重点中的重点】

以下会深度剖析BlockingQueue的put()和take()方法的底层实现源码，一步步走完后会进行总结。如果不看源码，就没有任何的说服性。

看源码前，先要明白Condition的含义：

Condition的实现ConditionObject：

ConditionObject是AQS类的内部类，在BlockingQueue底层的实现中主要功能有：等待队列，等待和通知。

等待队列，等待和通知：源码中会使用到notEmpty和notFull

notEmpty: 消费者对应的等待队列。有啥用？当应用队列中的数据被消费完毕后，最后一次消费数据的消费者会释放掉自己持有的锁，然后调用notEmpty.await()加入到notEmpty等待队列的尾部。并且会调用notFull.signal()表示唤醒生产者等待队列中的一个节点加入到CLH阻塞队列中去[因为队列中没有数据啦，所以要唤醒生产者加入到CLH阻塞队列中]。

notFull：生产者对应的等待队列。有啥用？当应用队列中的空间已经被数据占满后，最后一次生产数据的生产者会释放掉自己持有的锁，然后调用notFull.await()加入到notFull等待队列的尾部。并且会调用notEmpty.signal()表示唤醒消费者等待队列中的一个节点加入到CLH阻塞队列中去[因为队列的空间被数据占满啦，所以要唤醒消费者加入到CLH阻塞队列中]。

补充：

其实唤醒也不是说只有当极端情况下(队列被占满或队列为空)才会进行唤醒消费者或生产者，其实每一次往队列中进行加入或取出数据都会导致各自的唤醒操作。

eg：加入一条数据会导致唤醒一个消费者等待队列中的消费者加入到CLH队列。取出一条数据时同理即可。

BlockingQueue源码会涉及三个队列【重点掌握】：

1.应用队列：

存放消息数据，模拟出的一个虚拟队列概念，可以认为是一个虚拟的不存在的存储结构，底层是使用数组进行存储插入应用队列的数据，以此模拟出一个应用队列。

2.CLH双端阻塞队列：存储封装当前线程对象对应的Node节点，是底层真实存在的一个队列

3.条件等待队列：这个就是上面Condition中记录的：notEmpty和notFull

put()源码剖析：

API层面调用put()方法其实就是生产者加入一条数据到应用队列

1.

 2.

3.put方法解析

生产者插入一个数据到应用队列，以下为具体的分析过程：

对put()方法中的await()方法解析一下：

对put方法中的equeue()方法解析：

 take()源码剖析：

 API层面调用take()方法其实就是生产者从应用队列中取出一条数据

1.

2.

3.

 对take()中的await方法解析： 

 对take()中的dequeue方法解析：

 总结：

(1)记清楚三个队列

(2)明白Condition的含义

(3)画出图即可

其实也没啥，看懂了也挺简单的，关于这个源码的图，使用processon绘制：

 ProcessOn Flowchart