第7章 Java并发包中并发队列原理剖析

朱雀 2023-02-14 06:49 22阅读 0赞

**JDK中提供了一系列场景的并发安全队列。总的来说，按照实现方式的不同可分为阻塞队列和非阻塞队列，前者使用锁实现，而后者则使用CAS非阻塞算法实现。**

**Java并发包中的并发队列**：

*  **`ConcurrentLinkedQueue`是线程安全的无界非阻塞队列**
 *  **独占锁实现的有界阻塞队列`LinkedBlockingQueue`**
 *  **有界数组方式实现的阻塞队列`ArrayBlockingQueue`**
 *  **`PriorityBlockingQueue`是带优先级的无界阻塞队列**
 *  **`DelayQueue`并发队列是一个无界阻塞延迟队列**

#### **7.1 ConcurrentLinkedQueue原理探究** ####

**`ConcurrentLinkedQueue`是线程安全的无界非阻塞队列**，其底层数据结构使用单向链表实现，对于入队和出队操作使用CAS来实现线程安全。

##### 7.1 类图结构 #####

![format_png][]

`ConcurrentLinkedQueue`内部的队列使用单向链表方式实现，其中有两个volatile类型的Node节点分别用来存放队列的首、尾节点。**默认头、尾节点都是指向item为null的哨兵节点。新元素会被插入队列末尾，出队时从队列头部获取一个元素。**

public ConcurrentLinkedQueue() {
        
      head = tail = new Node<E>(null);
    }

在Node节点内部则维护一个使用volatile修饰的变量item，用来存放节点的值；next用来存放链表的下一个节点，从而链接为一个单向无界链表。其内部则使用UNSafe工具类提供的CAS算法来保证出入队时操作链表的原子性。

##### 7.1.2 ConcurrentLinkedQueue原理介绍 #####

*  `offer(E e)`：offer操作是在队列末尾添加一个元素，如果传递的参数是null则抛出NPE异常，否则由于ConcurrentLinkedQueue是无界队列，该方法一直会返回true。另外，由于使用CAS无阻塞算法，因此该方法不会阻塞挂起调用线程。
 *  `add(E e)`：add操作是在链表末尾添加一个元素，其实在内部调用的还是offer操作。
 *  `poll()`：poll操作是在队列头部获取并移除一个元素，如果队列为空则返回null。
 *  `peek()` ：peek操作是获取队列头部一个元素（只获取不移除），如果队列为空则返回null。下面看下其实现原理。
 *  `size()`：size操作计算当前队列元素个数，在并发环境下不是很有用，因为CAS没有加锁，所以从调用size函数到返回结果期间有可能增删元素，导致统计的元素个数不精确。
 *  `remove(Object o)`：remove操作如果队列里面存在该元素则删除该元素，如果存在多个则删除第一个，并返回true，否则返回false。
 *  `contains(Object o)`：contains操作判断队列里面是否含有指定对象，由于是遍历整个队列，所以像size操作一样结果也不是那么精确，有可能调用该方法时元素还在队列里面，但是遍历过程中其他线程才把该元素删除了，那么就会返回false。

##### 7.1.3 小结 #####

`ConcurrentLinkedQueue`的底层**使用单向链表数据结构来保存队列元素，每个元素被包装成一个Node节点。队列是靠头、尾节点来维护的，创建队列时头、尾节点指向一个item为null的哨兵节点。第一次执行peek或者first操作时会把head指向第一个真正的队列元素。由于使用非阻塞CAS算法，没有加锁，所以在计算size时有可能进行了offer、poll或者remove操作，导致计算的元素个数不精确，所以在并发情况下size函数不是很有用**。

如下图所示，入队、出队都是操作使用volatile修饰的tail、head节点，要保证在多线程下出入队线程安全，只需要保证这两个Node操作的可见性和原子性即可。由于volatile本身可以保证可见性，所以只需要保证对两个变量操作的原子性即可。

![format_png 1][]

`offer`操作是在tail后面添加元素，也就是调用tail.casNext方法，而这个方法使用的是CAS操作，只有一个线程会成功，然后失败的线程会循环，重新获取tail，再执行casNext方法。

`poll`操作也通过类似CAS的算法保证出队时移除节点操作的原子性。

#### 7.2 LinkedBlockingQueue原理探究 ####

前面介绍了使用CAS算法实现的非阻塞队列`ConcurrentLinkedQueue`，下面我们来介绍使用**独占锁实现的阻塞队列`LinkedBlockingQueue`**。

##### 7.2.1 类图结构 #####

![format_png 2][]

由类图可以看到，**LinkedBlockingQueue也是使用单向链表实现的**，其也有两个Node，分别用来存放首、尾节点，并且还有一个初始值为0的原子变量count，用来记录队列元素个数。**另外还有两个ReentrantLock的实例，分别用来控制元素入队和出队的原子性，其中takeLock用来控制同时只有一个线程可以从队列头获取元素，其他线程必须等待，putLock控制同时只能有一个线程可以获取锁，在队列尾部添加元素，其他线程必须等待。另外，notEmpty和notFull是条件变量，它们内部都有一个条件队列用来存放进队和出队时被阻塞的线程，其实这是生产者—消费者模型。**

/** 执行take、poll等操作时需要获取该锁 */
    private final ReentrantLock takeLock = new ReentrantLock();
    
    /** 当队列为空时，执行出队操作的线程会被放入这个条件队列进行等待 */
    private final Condition notEmpty = takeLock.newCondition();
    
    /** 执行put、offer等操作时需要获取该锁 */
    private final ReentrantLock putLock = new ReentrantLock();
    
    /** 当队列满时，执行入队操作的线程会被放入这个条件队列进行等待 */
    private final Condition notFull = putLock.newCondition();

> 当调用线程在LinkedBlockingQueue实例上执行take、 poll等操作时需要获取到takeLock锁，从而保证同时只有一个线程可以操作链表头节点。另外由于条件变量notEmpty内部的条件队列的维护使用的是takeLock的锁状态管理机制，所以在调用notEmpty的await和signal方法前调用线程必须先获取到takeLock锁，否则会抛出IllegalMonitorStateException异常。notEmpty内部则维护着一个条件队列，当线程获取到takeLock锁后调用notEmpty的await方法时，调用线程会被阻塞，然后该线程会被放到notEmpty内部的条件队列进行等待，直到有线程调用了notEmpty的signal方法。
> 
> 在LinkedBlockingQueue实例上执行put、offer等操作时需要获取到putLock锁，从而保证同时只有一个线程可以操作链表尾节点。同样由于条件变量notFull内部的条件队列的维护使用的是putLock的锁状态管理机制，所以在调用notFull的await和signal方法前调用线程必须先获取到putLock锁，否则会抛出IllegalMonitorStateException异常。notFull内部则维护着一个条件队列，当线程获取到putLock锁后调用notFull的await方法时，调用线程会被阻塞，然后该线程会被放到notFull内部的条件队列进行等待，直到有线程调用了notFull的signal方法。

##### 7.2.2 LinkedBlockingQueue原理介绍 #####

*  `offer(E e)`：向队列尾部插入一个元素，**如果队列中有空闲则插入成功后返回true，如果队列已满则丢弃当前元素然后返回false**。如果e元素为null则抛出NullPointerException异常。另外，该方法是非阻塞的。
 *  `put(E e)`：向队列尾部插入一个元素，如果队列中有空闲则插入后直接返回，**如果队列已满则阻塞当前线程，直到队列有空闲插入成功后返回**。如果在阻塞时被其他线程设置了中断标志，则被阻塞线程会抛出InterruptedException异常而返回。另外，如果e元素为null则抛出NullPointerException异常。
 *  `poll()`：从队列头部获取并移除一个元素，**如果队列为空则返回null，该方法是不阻塞的**。
 *  `peek()`：获取队列头部元素但是不从队列里面移除它，**如果队列为空则返回null。该方法是不阻塞的**。
 *  `take()`：获取当前队列头部元素并从队列里面移除它。**如果队列为空则阻塞当前线程直到队列不为空然后返回元素**，如果在阻塞时被其他线程设置了中断标志，则被阻塞线程会抛出InterruptedException异常而返回。
 *  `remove(Object o)`：删除队列里面指定的元素，有则删除并返回true，没有则返回false。
 *  `size()`：获取当前队列元素个数。

##### 7.2.3 小结 #####

LinkedBlockingQueue的内部是通过单向链表实现的，使用头、尾节点来进行入队和出队操作，也就是入队操作都是对尾节点进行操作，出队操作都是对头节点进行操作。如下图所示，对头、尾节点的操作分别使用了单独的独占锁从而保证了原子性，所以出队和入队操作是可以同时进行的。另外对头、尾节点的独占锁都配备了一个条件队列，用来存放被阻塞的线程，并结合入队、出队操作实现了一个生产消费模型。

![format_png 3][]

#### 7.3 ArrayBlockingQueue原理探究 ####

上节介绍了使用有界链表方式实现的阻塞队列LinkedBlockingQueue，本节来研究使用**有界数组方式实现的阻塞队列`ArrayBlockingQueue`** 的原理。

##### 7.3.1 类图结构 #####

ArrayBlockingQueue的内部有一个数组items，用来存放队列元素，putindex变量表示入队元素下标，takeIndex是出队下标，count统计队列元素个数。从定义可知，这些变量并没有使用volatile修饰，这是因为访问这些变量都是在锁块内，而加锁已经保证了锁块内变量的内存可见性了。另外有个独占锁lock用来保证出、入队操作的原子性，这保证了同时只有一个线程可以进行入队、出队操作。另外，notEmpty、notFull条件变量用来进行出、入队的同步。

![format_png 4][]

##### 7.3.2 ArrayBlockingQueue原理介绍 #####

`offer(E e)`：向队列尾部插入一个元素，**如果队列有空闲空间则插入成功后返回true，如果队列已满则丢弃当前元素然后返回false**。如果e元素为null则抛出NullPointerException异常。另外，该方法是不阻塞的。

`put()`：向队列尾部插入一个元素，**如果队列有空闲则插入后直接返回true，如果队列已满则阻塞当前线程直到队列有空闲并插入成功后返回true**，如果在阻塞时被其他线程设置了中断标志，则被阻塞线程会抛出InterruptedException异常而返回。另外，如果e元素为null则抛出NullPointerException异常。

`poll()`：从队列头部获取并移除一个元素，**如果队列为空则返回null，该方法是不阻塞的**。

`take()`：获取当前队列头部元素并从队列里面移除它。**如果队列为空则阻塞当前线程直到队列不为空然后返回元素**，如果在阻塞时被其他线程设置了中断标志，则被阻塞线程会抛出InterruptedException异常而返回。

`peek()`：获取队列头部元素但是不从队列里面移除它，**如果队列为空则返回null，该方法是不阻塞的**。

`size()`：计算当前队列元素个数。

##### 7.3.3 小结 #####

ArrayBlockingQueue通过使用全局独占锁实现了同时只能有一个线程进行入队或者出队操作，这个锁的粒度比较大，有点类似于在方法上添加synchronized的意思。其中offer和poll操作通过简单的加锁进行入队、出队操作，而put、take操作则使用条件变量实现了，如果队列满则等待，如果队列空则等待，然后分别在出队和入队操作中发送信号激活等待线程实现同步。另外，相比LinkedBlockingQueue, ArrayBlockingQueue的size操作的结果是精确的，因为计算前加了全局锁。

![format_png 5][]

#### 7.4 PriorityBlockingQueue原理探究 ####

##### 7.4.1 介绍 #####

**`PriorityBlockingQueue`是带优先级的无界阻塞队列**，**每次出队都返回优先级最高或者最低的元素**。其内部是使用平衡二叉树堆实现的，所以直接遍历队列元素不保证有序。默认使用对象的compareTo方法提供比较规则，如果你需要自定义比较规则则可以自定义comparators。

##### 7.4.2 PriorityBlockingQueue类图结构 #####

PriorityBlockingQueue内部有一个数组queue，用来存放队列元素，size用来存放队列元素个数。allocationSpinLock是个自旋锁，其使用CAS操作来保证同时只有一个线程可以扩容队列，状态为0或者1，其中0表示当前没有进行扩容，1表示当前正在扩容。由于这是一个优先级队列，所以有一个比较器comparator用来比较元素大小。lock独占锁对象用来控制同时只能有一个线程可以进行入队、出队操作。notEmpty条件变量用来实现take方法阻塞模式。这里没有notFull条件变量是因为这里的put操作是非阻塞的，为啥要设计为非阻塞的，是因为这是无界队列。

![format_png 6][]

##### 7.4.3 原理介绍 #####

*  `offer(E e)`：队列中插入一个元素，由于是无界队列，所以一直返回true。
 *  `poll()`：获取队列内部堆树的根节点元素，如果队列为空，则返回null。
 *  `put(E e)`：put操作内部调用的是offer操作，由于是无界队列，所以不需要阻塞。
 *  `take()`：获取队列内部堆树的根节点元素，如果队列为空则阻塞。
 *  `size()`：计算队列元素个数。

##### 7.4.5 小结 #####

PriorityBlockingQueue队列在内部使用二叉树堆维护元素优先级，使用数组作为元素存储的数据结构，这个数组是可扩容的。当当前元素个数>=最大容量时会通过CAS算法扩容，出队时始终保证出队的元素是堆树的根节点，而不是在队列里面停留时间最长的元素。使用元素的compareTo方法提供默认的元素优先级比较规则，用户可以自定义优先级的比较规则。

PriorityBlockingQueue类似于ArrayBlockingQueue，在内部使用一个独占锁来控制同时只有一个线程可以进行入队和出队操作。另外，前者只使用了一个notEmpty条件变量而没有使用notFull，这是因为前者是无界队列，执行put操作时永远不会处于await状态，所以也不需要被唤醒。而take方法是阻塞方法，并且是可被中断的。当需要存放有优先级的元素时该队列比较有用。

![format_png 7][]

#### 7.5 DelayQueue原理探究 ####

**`DelayQueue`并发队列是一个无界阻塞延迟队列**，队列中的每个元素都有个过期时间，当从队列获取元素时，只有过期元素才会出队列。队列头元素是最快要过期的元素。

##### 7.5.1 DelayQueue类图结构 #####

DelayQueue内部使用PriorityQueue存放数据，使用ReentrantLock实现线程同步。另外，队列里面的元素要实现Delayed接口，由于每个元素都有一个过期时间，所以要实现获知当前元素还剩下多少时间就过期了的接口，由于内部使用优先级队列来实现，所以要实现元素之间相互比较的接口。

条件变量available与lock锁是对应的，其目的是为了实现线程间同步。

其中leader变量的使用基于Leader-Follower模式的变体，用于尽量减少不必要的线程等待。当一个线程调用队列的take方法变为leader线程后，它会调用条件变量available.awaitNanos（delay）等待delay时间，但是其他线程（follwer线程）则会调用available.await（）进行无限等待。leader线程延迟时间过期后，会退出take方法，并通过调用available.signal（）方法唤醒一个follwer线程，被唤醒的follwer线程被选举为新的leader线程。

![format_png 8][]

##### 7.5.2 主要函数原理讲解 #####

*  `offer(E e)`：插入元素到队列，如果插入元素为null则抛出NullPointerException异常，否则由于是无界队列，所以一直返回true。插入元素要实现Delayed接口。
 *  `take()`：获取并移除队列里面延迟时间过期的元素，如果队列里面没有过期元素则等待。
 *  `poll()`：获取并移除队头过期元素，如果没有过期元素则返回null。
 *  `size()`：计算队列元素个数，包含过期的和没有过期的。

##### 7.5.4 小结 #####

DelayQueue队列其内部使用PriorityQueue存放数据，使用ReentrantLock实现线程同步。另外队列里面的元素要实现Delayed接口，其中一个是获取当前元素到过期时间剩余时间的接口，在出队时判断元素是否过期了，一个是元素之间比较的接口，因为这是一个有优先级的队列。

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-esyWRrV5-1590996469822)(https://note.youdao.com/yws/public/resource/e74bb4019e6355a7745aa4de061090bf/xmlnote/WEBRESOURCE67c016fd4874e7b34f25b864d2510f09/37639)\]

[format_png]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9ub3RlLnlvdWRhby5jb20veXdzL3B1YmxpYy9yZXNvdXJjZS9lNzRiYjQwMTllNjM1NWE3NzQ1YWE0ZGUwNjEwOTBiZi94bWxub3RlL0E4NjFBRDYxNUNGNjRBMjA4RTQxNUM1OTlGNDI2QjYxLzM3NDE5?x-oss-process=image/format,png
[format_png 1]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9ub3RlLnlvdWRhby5jb20veXdzL3B1YmxpYy9yZXNvdXJjZS9lNzRiYjQwMTllNjM1NWE3NzQ1YWE0ZGUwNjEwOTBiZi94bWxub3RlLzE5RTM2QzgyM0RCQTQyMjA5NDRCQTFCNDM4NUEwMENELzM3NDYz?x-oss-process=image/format,png
[format_png 2]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9ub3RlLnlvdWRhby5jb20veXdzL3B1YmxpYy9yZXNvdXJjZS9lNzRiYjQwMTllNjM1NWE3NzQ1YWE0ZGUwNjEwOTBiZi94bWxub3RlL1dFQlJFU09VUkNFYjVkYTM0MzNlOTlkZDQwZTAyMjJjZDI1YzdlODU3OTYvMzc1MTE?x-oss-process=image/format,png
[format_png 3]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9ub3RlLnlvdWRhby5jb20veXdzL3B1YmxpYy9yZXNvdXJjZS9lNzRiYjQwMTllNjM1NWE3NzQ1YWE0ZGUwNjEwOTBiZi94bWxub3RlL1dFQlJFU09VUkNFYzJiNTY3ZmFhODY0MmIxMTZjZmFhM2FhNTJhM2I5NjUvMzc1MTM?x-oss-process=image/format,png
[format_png 4]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9ub3RlLnlvdWRhby5jb20veXdzL3B1YmxpYy9yZXNvdXJjZS9lNzRiYjQwMTllNjM1NWE3NzQ1YWE0ZGUwNjEwOTBiZi94bWxub3RlL1dFQlJFU09VUkNFN2RlMzVlMDI5NmY5M2JjZmRhYmQxYjg4Mjg1MmUzYmQvMzc1NDg?x-oss-process=image/format,png
[format_png 5]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9ub3RlLnlvdWRhby5jb20veXdzL3B1YmxpYy9yZXNvdXJjZS9lNzRiYjQwMTllNjM1NWE3NzQ1YWE0ZGUwNjEwOTBiZi94bWxub3RlL1dFQlJFU09VUkNFYTRmZjE1ZDBjN2VjMzRkZDBhYzQ2YzVkYjI4ODhmNGYvMzc1NDY?x-oss-process=image/format,png
[format_png 6]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9ub3RlLnlvdWRhby5jb20veXdzL3B1YmxpYy9yZXNvdXJjZS9lNzRiYjQwMTllNjM1NWE3NzQ1YWE0ZGUwNjEwOTBiZi94bWxub3RlL1dFQlJFU09VUkNFNzEzODU2OGE2MDIxNWEwZDQzNDEzNDMxMDUzZTg4NjYvMzc1NTc?x-oss-process=image/format,png
[format_png 7]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9ub3RlLnlvdWRhby5jb20veXdzL3B1YmxpYy9yZXNvdXJjZS9lNzRiYjQwMTllNjM1NWE3NzQ1YWE0ZGUwNjEwOTBiZi94bWxub3RlL1dFQlJFU09VUkNFNzY4YzM0OWJhMjkxZjkyMTZlOWUxZjBmODRkNWQ5Y2IvMzc1NTU?x-oss-process=image/format,png
[format_png 8]: https://imgconvert.csdnimg.cn/aHR0cHM6Ly9ub3RlLnlvdWRhby5jb20veXdzL3B1YmxpYy9yZXNvdXJjZS9lNzRiYjQwMTllNjM1NWE3NzQ1YWE0ZGUwNjEwOTBiZi94bWxub3RlL1dFQlJFU09VUkNFNDIwZDExYjA3ZDk3MWMxY2I4ZjgwNTAzZjNlY2M2NDEvMzc2Mjk?x-oss-process=image/format,png