Java中的Queue之概述

今天药忘吃喽~ 2023-07-21 11:22 27阅读 0赞

> 对技术还是得有敬畏之心，总觉得Queue好像没啥，其实只是没有仔细去了解过。

不过自从上次认真地看了线程池的源代码之后，发现Queue是一个很神奇的集合类。Queue的形式有无界、有界，还有堵塞、非堵塞。初略想想，这个实现可能就不简单。

## 一个问题 ##

在线程池中，自定义线程池时，放入什么样的队列可以让线程数达不到`maximumPoolSize`？

按照线程池的源码，线程增长有两个阶段：**一个阶段**是在达到`corePoolSize`之前，**第二个阶段**是在达到`corePoolSize`之后并且阻塞队列已经满了，才会继续增加线程，直到`maximumPoolSize`。

所以解决方案有如下两种：

①如果队列是无界的，那么将不会到达第二阶段的增长，也就无法到达`maximumPoolSize`；

②还有可以让`corePoolSize`在足够大的值，同时限定堆大小，保证在有限的任务数下，引发内存不足也算是一种解决方案。

在解决方案①的基础上，我给出了`LinkedBlockingQueue`这个答案，但是它是一个“有界”的堵塞队列，只是它的capacity默认是`Integer.MAX_VALUE`。因为一开始我得知它是一个无界队列，因为未做详细、深入的了解，但是后面我被指正后，引发了我对Queue的思考，因为不常用，所以了解的确不是很多；同时，Queue不仅仅是一个FIFO，还有同步操作在里面，就是\*\*\*生产者-消费者\*\*\*的同步问题的一个解决方案。综上，这让我对堵塞队列产生了强烈的兴趣。

## 概览 ##

找了一下BlockingQueue大致的实现类，它们的类图如下。

![format_png][]

从中可以看出，它们都继承自`AbstractQueue`，并且都实现了`BlockingQueue`接口，只有`LinkedTransferQueue`还实现了`TransferQueue`。加上对Queue的一些操作并不是特别熟悉，比如offer、peek、poll之类，所以从Queue接口开始，逐个了解其大致的功能。

### Queue ###

这个接口定义了6个方法，功能分成3个，分别是**插入**、**删除**、**获取队首元素**，每个功能有2种实现，对应如下表中的关系。这两种实现的区别也很明显

<table> 
 <thead> 
  <tr> 
   <th></th> 
   <th>Throws exception</th> 
   <th>Returns special value</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>Insert</td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#add%28E%29" rel="nofollow"><code>add(e)</code></a></td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#offer%28E%29" rel="nofollow"><code>offer(e)</code></a></td> 
  </tr> 
  <tr> 
   <td>Remove</td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#remove%28%29" rel="nofollow"><code>remove()</code></a></td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#poll%28%29" rel="nofollow"><code>poll()</code></a></td> 
  </tr> 
  <tr> 
   <td>Examine</td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#element%28%29" rel="nofollow"><code>element()</code></a></td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#peek%28%29" rel="nofollow"><code>peek()</code></a></td> 
  </tr> 
 </tbody> 
</table>

*  [`add(e)`][add_e] 和 [`offer(e)`][offer_e]：添加失败时，add抛异常，offer返回false。
 *  [`remove()`][remove] 和 [`poll()`][poll] ：没有元素时，remove抛异常，poll返回null。
 *  [`element()`][element] 和 [`peek()`][peek]：没有元素时，remove抛异常，peek返回null。

**还有一点需要注意**，虽然有的Queue的实现（`LinkedList`）允许插入null值，但是Queue通常不允许插入null值，因为null对peek、poll来说，是queue中没有元素。

### BlockingQueue ###

继承自`Queue`接口，同时在`Queue`的基础上，新增了堵塞插入、堵塞获取并删除、可堵塞一定时长的插入、可堵塞一定时长的获取并删除。相应关系对应下表。

<table> 
 <thead> 
  <tr> 
   <th align="left"></th> 
   <th>Throws exception</th> 
   <th>Special value</th> 
   <th>Blocks</th> 
   <th>Times out</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">Insert</td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/BlockingQueue.html#add%28E%29" rel="nofollow"><code>add(e)</code></a></td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/BlockingQueue.html#offer%28E%29" rel="nofollow"><code>offer(e)</code></a></td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/BlockingQueue.html#put%28E%29" rel="nofollow"><code>put(e)</code></a></td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/BlockingQueue.html#offer%28E,long,java.util.concurrent.TimeUnit%29" rel="nofollow"><code>offer(e, time, unit)</code></a></td> 
  </tr> 
  <tr> 
   <td align="left">Remove</td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#remove%28%29" rel="nofollow"><code>remove()</code></a></td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#poll%28%29" rel="nofollow"><code>poll()</code></a></td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/BlockingQueue.html#take%28%29" rel="nofollow"><code>take()</code></a></td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/BlockingQueue.html#poll%28long,java.util.concurrent.TimeUnit%29" rel="nofollow"><code>poll(time, unit)</code></a></td> 
  </tr> 
  <tr> 
   <td align="left">Examine</td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#element%28%29" rel="nofollow"><code>element()</code></a></td> 
   <td><a href="https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#peek%28%29" rel="nofollow"><code>peek()</code></a></td> 
   <td>not applicable</td> 
   <td>not applicable</td> 
  </tr> 
 </tbody> 
</table>

*  [`put(e)`][put_e] 和 [`offer(e, time, unit)`][offer_e_ time_ unit]：put一直堵塞直到空间可用，offer(e, time, unit)只堵塞特定时长。
 *  [`take()`][take] 和 [`poll(time, unit)`][poll_time_ unit]： take一直堵塞直到元素可用，poll(time, unit)只堵塞特定时长。

BlockingQueue设计主要被用来处理\*\*\*生产者-消费者\*\*\*问题。**还有一点需要注意**，`BlockingQueue`不接受null元素。插入操作\*\*\*happen-before\*\*\*读取、移除。

### AbstractQueue ###

抽象类，继承自`AbstractCollection`，并实现了`Queue`接口，但未给出具体实现，相当于引入`Queue`相应方法，在自身类中使用，然后靠子类实现`Queue`的相关方法。

它实现的add、remove、element方法，分别依靠offer、poll、peek，具体如下：

public boolean add(E e) {
        
        if (offer(e))
            return true;
        else
            throw new IllegalStateException("Queue full");
    }
    
    public E remove() {
        
        E x = poll();
        if (x != null)
            return x;
        else
            throw new NoSuchElementException();
    }
    
    public E element() {
        
        E x = peek();
        if (x != null)
            return x;
        else
            throw new NoSuchElementException();
    }
    
    public void clear() {
        
        while (poll() != null)
            ;
    }

正好对应于上面Queue方法的区别。

## 常见实现类 ##

Java中线程安全的内置队列如下表所示。队列的底层一般分成三种：**数组**、**链表**和**堆**。其中，堆一般情况下是为了实现带有优先级特性的队列。

<table> 
 <thead> 
  <tr> 
   <th align="left">队列</th> 
   <th align="left">有界性</th> 
   <th align="left">锁</th> 
   <th align="left">数据结构</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="left">ArrayBlockingQueue</td> 
   <td align="left">bounded</td> 
   <td align="left">加锁</td> 
   <td align="left">arraylist</td> 
  </tr> 
  <tr> 
   <td align="left">LinkedBlockingQueue</td> 
   <td align="left">optionally-bounded</td> 
   <td align="left">加锁</td> 
   <td align="left">linkedlist</td> 
  </tr> 
  <tr> 
   <td align="left">ConcurrentLinkedQueue</td> 
   <td align="left">unbounded</td> 
   <td align="left"><strong>无锁</strong></td> 
   <td align="left">linkedlist</td> 
  </tr> 
  <tr> 
   <td align="left">LinkedTransferQueue</td> 
   <td align="left">unbounded</td> 
   <td align="left"><strong>无锁</strong></td> 
   <td align="left">linkedlist</td> 
  </tr> 
  <tr> 
   <td align="left">PriorityBlockingQueue</td> 
   <td align="left">unbounded</td> 
   <td align="left">加锁</td> 
   <td align="left">heap</td> 
  </tr> 
  <tr> 
   <td align="left">DelayQueue</td> 
   <td align="left">unbounded</td> 
   <td align="left">加锁</td> 
   <td align="left">heap</td> 
  </tr> 
 </tbody> 
</table>

**基于数组线程安全的队列**，比较典型的是`ArrayBlockingQueue`，它主要通过**加锁**的方式来保证线程安全；

**基于链表的线程安全队列**，分成`LinkedBlockingQueue`和`ConcurrentLinkedQueue`两大类，前者也通过**锁的方式**来实现线程安全，而后者以及上面表格中的`LinkedTransferQueue`都是通过**原子变量CAS**这种**不加锁**的方式来实现的。

通过不加锁的方式实现的队列都是**无界**的（无法保证队列的长度在确定的范围内）；而加锁的方式，可以实现**有界队列**。

在稳定性要求特别高的系统中，为了防止生产者速度过快，导致内存溢出，只能选择**有界队列**；同时，为了减少Java的垃圾回收对系统性能的影响，会尽量选择array/heap格式的数据结构。

**Reference：**

https://tech.meituan.com/2016/11/18/disruptor.html

https://www.cnblogs.com/WangHaiMing/p/8798709.html

https://www.cnblogs.com/stateis0/p/9062076.html

[format_png]: /images/20230528/98ff0d98519146b8af44c5ad0b783d01.png
[add_e]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#add%28E%29
[offer_e]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#offer%28E%29
[remove]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#remove%28%29
[poll]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#poll%28%29
[element]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#element%28%29
[peek]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/Queue.html#peek%28%29
[put_e]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/BlockingQueue.html#put%28E%29
[offer_e_ time_ unit]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/BlockingQueue.html#offer%28E,long,java.util.concurrent.TimeUnit%29
[take]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/BlockingQueue.html#take%28%29
[poll_time_ unit]: https://docs.oracle.com/en/java/javase/11/docs/api/java.base/java/util/concurrent/BlockingQueue.html#poll%28long,java.util.concurrent.TimeUnit%29