Java多线程--队列

红太狼 2023-03-13 06:23 13阅读 0赞

原文网址：[Java多线程--队列\_IT利刃出鞘的博客-CSDN博客][Java_--_IT_-CSDN]

# DelayQueue #

**其他网址**

> [Java同步数据结构之DelayQueue/DelayedWorkQueue - 莫待樱开春来踏雪觅芳踪 - 博客园][Java_DelayQueue_DelayedWorkQueue - _ -]

**简介**

> DelayQueue：只有当其指定的延迟时间到了，才能够从队列中获取到该元素。DelayQueue就是基于PriorityQueue实现的，DelayQueue队列实际上就是将队列元素保存到内部的一个PriorityQueue实例中的（所以也不支持插入null值），DelayQueue只专注于实现队列元素的延时出队。
> 
> 延迟队列DelayQueue是一个**无界阻塞队列**，它的队列元素只能在该元素的延迟已经结束（或者说过期）才能被出队。它怎么判断一个元素的延迟是否结束呢，原来DelayQueue队列元素必须是实现了Delayed接口的实例，该接口有一个getDelay方法需要实现，延迟队列就是通过实时的调用元素的该方法来判断当前元素是否延迟已经结束。
> 
> 既然**DelayQueue是基于优先级队列来实现的**，那肯定元素也要实现Comparable接口。因为Delayed接口继承了Comparable接口，所以实现Delayed的队列元素也必须要实现Comparable的compareTo方法。延迟队列就是以时间作为比较基准的优先级队列，这个时间即延迟时间，这个时间大都在构造元素的时候就已经设置好，随着程序的运行时间的推移，队列元素的延迟时间逐步到期，DelayQueue就能够基于延迟时间运用优先级队列并配合getDelay方法达到延迟队列中的元素在延迟结束时精准出队。

**实例**

> package org.example.a;
> 
> import com.oracle.jrockit.jfr.Producer;
> 
> import java.util.concurrent.*;
> 
> class DelayTask implements Delayed {
> 
> private long delay; //延迟多少纳秒开始执行
> private TimeUnit unit;
> 
> public DelayTask(long delay, TimeUnit unit){
> this.unit = unit;
> this.delay = TimeUnit.NANOSECONDS.convert(delay, unit);//统一转换成纳秒计数
> }
> 
> @Override
> public long getDelay(TimeUnit unit) {//延迟剩余时间，单位unit指定
> return unit.convert(delay - System.currentTimeMillis(), TimeUnit.NANOSECONDS);
> }
> 
> @Override
> public int compareTo(Delayed o) {//基于getDelay实时延迟剩余时间进行比较
> if(this.getDelay(TimeUnit.NANOSECONDS) < o.getDelay(TimeUnit.NANOSECONDS)) //都换算成纳秒计算
> return -1;
> else if(this.getDelay(TimeUnit.NANOSECONDS) > o.getDelay(TimeUnit.NANOSECONDS)) //都换算成纳秒计算
> return 1;
> else
> return 0;
> }
> 
> @Override
> public String toString() {
> return "DelayTask{" +
> "delay=" + delay +
> ", unit=" + unit +
> '}';
> }
> }
> 
> public class Demo {
> public static void main(String[] args) {
> DelayQueue<DelayTask> dq = new DelayQueue();
> 
> //入队四个元素，注意它们的延迟时间单位不一样。
> dq.offer(new DelayTask(5, TimeUnit.SECONDS));
> dq.offer(new DelayTask(2, TimeUnit.MINUTES));
> dq.offer(new DelayTask(700, TimeUnit.MILLISECONDS));
> dq.offer(new DelayTask(1000, TimeUnit.NANOSECONDS));
> 
> while(dq.size() > 0){
> try {
> System.out.println(dq.take());
> }catch (Exception e){
> e.printStackTrace();
> }
> }
> }
> }

> 执行结果
> 
>     DelayTask{delay=1000, unit=NANOSECONDS}
>     DelayTask{delay=700000000, unit=MILLISECONDS}
>     DelayTask{delay=5000000000, unit=SECONDS}
>     DelayTask{delay=120000000000, unit=MINUTES}

# ConcurrentLinkedQueue #

**其他网址**

> 另见《Java并发编程之美》=> 7.1 ConcurrentLinkedQueue 原理探究

## 简介 ##

**ConcurrentLinkedQueue与BlockingQueue区别**

> <table> 
> <tbody> 
> <tr> 
> <td style="width:104px;">项</td> 
> <td style="width:226px;">ConcurrentLinkedQueue</td> 
> <td style="width:396px;">BlockingQueue</td> 
> </tr> 
> <tr> 
> <td style="width:104px;">是否阻塞</td> 
> <td style="width:226px;">非阻塞（使用CAS）</td> 
> <td style="width:396px;">阻塞(使用锁）</td> 
> </tr> 
> <tr> 
> <td style="width:104px;">优点</td> 
> <td style="width:226px;"> 性能高。 （特别是多个消费者时） </td> 
> <td style="width:396px;"> 多线程操作共同的队列时不需要额外的同步。 队列会自动负载均衡，即那边（生产与消费两边）处理快了就会被阻塞掉，从而减少两边的处理速度差距。 </td> 
> </tr> 
> <tr> 
> <td style="width:104px;">使用场景</td> 
> <td style="width:226px;">弹出的元素是多个线程共享的。（线程安全）</td> 
> <td style="width:396px;"></td> 
> </tr> 
> </tbody> 
> </table>
> 
> 跟BlockingQueue基本一样。区别点：
> 
> <table> 
> <tbody> 
> <tr> 
> <td>方法</td> 
> <td style="width:373px;">ConcurrentLinkedQueue</td> 
> <td style="width:279px;">BlockingQueue</td> 
> </tr> 
> <tr> 
> <td>take()</td> 
> <td style="width:373px;">无此方法</td> 
> <td style="width:279px;">有此方法</td> 
> </tr> 
> <tr> 
> <td>size()</td> 
> <td style="width:373px;"> 轮询元素，复杂度为O(n)。 线程不安全（可能有其他线程offer、remove等） </td> 
> <td style="width:279px;"> 直接返回大小，复杂度为O(1)。 线程安全。 </td> 
> </tr> 
> <tr> 
> <td>isEmpty()</td> 
> <td style="width:373px;">有此方法</td> 
> <td style="width:279px;">无此方法</td> 
> </tr> 
> </tbody> 
> </table>

**方法**

> <table> 
> <tbody> 
> <tr> 
> <td colspan="1"> 构造方法摘要 </td> 
> </tr> 
> <tr> 
> <td> <a title="ConcurrentLinkedQueue" rel="nofollow">ConcurrentLinkedQueue</a>()&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;创建一个最初为空的&nbsp;ConcurrentLinkedQueue。 </td> 
> </tr> 
> <tr> 
> <td> <a title="ConcurrentLinkedQueue" rel="nofollow">ConcurrentLinkedQueue</a>(<a href="http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/Collection.html" title="Collection" rel="nofollow">Collection</a>&lt;? extends&nbsp;<a href="http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/concurrent/ConcurrentLinkedQueue.html" title="E" rel="nofollow">E</a>&gt;&nbsp;c)&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;创建一个最初包含给定 collection 元素的&nbsp;ConcurrentLinkedQueue，按照此 collection 迭代器的遍历顺序来添加元素。 </td> 
> </tr> 
> </tbody> 
> </table>
> 
> <table style="width:684px;"> 
> <tbody> 
> <tr> 
> <td style="width:97.5pt;"> 返回值 </td> 
> <td style="width:551px;"> 描述 </td> 
> </tr> 
> <tr> 
> <td style="vertical-align:top;width:97.5pt;"> boolean </td> 
> <td style="width:551px;"> <a title="add" rel="nofollow">add</a>(<a href="http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/concurrent/ConcurrentLinkedQueue.html" title="E" rel="nofollow">E</a>&nbsp;e)&nbsp;&nbsp;&nbsp;&nbsp;// 将指定元素插入此队列的尾部。 </td> 
> </tr> 
> <tr> 
> <td style="vertical-align:top;width:97.5pt;"> boolean </td> 
> <td style="width:551px;"> <a title="contains" rel="nofollow">contains</a>(<a href="http://tool.oschina.net/uploads/apidocs/jdk-zh/java/lang/Object.html" title="Object" rel="nofollow">Object</a>&nbsp;o)&nbsp;&nbsp;&nbsp;&nbsp;// 如果此队列包含指定元素，则返回&nbsp;true。 </td> 
> </tr> 
> <tr> 
> <td style="vertical-align:top;width:97.5pt;"> boolean </td> 
> <td style="width:551px;"> <a title="isEmpty" rel="nofollow">isEmpty</a>()&nbsp;&nbsp;&nbsp;&nbsp;// 如果此队列不包含任何元素，则返回&nbsp;true。 </td> 
> </tr> 
> <tr> 
> <td style="vertical-align:top;width:97.5pt;"> <a href="http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/Iterator.html" title="Iterator" rel="nofollow">Iterator</a>&lt;<a href="http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/concurrent/ConcurrentLinkedQueue.html" title="E" rel="nofollow">E</a>&gt; </td> 
> <td style="width:551px;"> <a title="iterator" rel="nofollow">iterator</a>()&nbsp;&nbsp;&nbsp;&nbsp;// 返回在此队列元素上以恰当顺序进行迭代的迭代器。 </td> 
> </tr> 
> <tr> 
> <td style="vertical-align:top;width:97.5pt;"> boolean </td> 
> <td style="width:551px;"> <a title="offer" rel="nofollow">offer</a>(<a href="http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/concurrent/ConcurrentLinkedQueue.html" title="E" rel="nofollow">E</a>&nbsp;e)&nbsp;&nbsp;&nbsp;&nbsp;// 将指定元素插入此队列的尾部。 </td> 
> </tr> 
> <tr> 
> <td style="vertical-align:top;width:97.5pt;"> <a href="http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/concurrent/ConcurrentLinkedQueue.html" title="E" rel="nofollow">E</a> </td> 
> <td style="width:551px;"> <a title="peek" rel="nofollow">peek</a>()&nbsp;&nbsp;&nbsp;&nbsp;// 获取但不移除此队列的头；如果此队列为空，则返回&nbsp;null。 </td> 
> </tr> 
> <tr> 
> <td style="vertical-align:top;width:97.5pt;"> <a href="http://tool.oschina.net/uploads/apidocs/jdk-zh/java/util/concurrent/ConcurrentLinkedQueue.html" title="E" rel="nofollow">E</a> </td> 
> <td style="width:551px;"> <a title="poll" rel="nofollow">poll</a>()&nbsp;&nbsp;&nbsp;&nbsp;// 获取并移除此队列的头，如果此队列为空，则返回&nbsp;null。 </td> 
> </tr> 
> <tr> 
> <td style="vertical-align:top;width:97.5pt;"> boolean </td> 
> <td style="width:551px;"> <a title="remove" rel="nofollow">remove</a>(<a href="http://tool.oschina.net/uploads/apidocs/jdk-zh/java/lang/Object.html" title="Object" rel="nofollow">Object</a>&nbsp;o)&nbsp;&nbsp;&nbsp;&nbsp;// 从队列中移除指定元素的单个实例（如果存在）。 </td> 
> </tr> 
> <tr> 
> <td style="vertical-align:top;width:97.5pt;"> int </td> 
> <td style="width:551px;"> <a title="size" rel="nofollow">size</a>()&nbsp;&nbsp;&nbsp;&nbsp;// 返回此队列中的元素数量。 </td> 
> </tr> 
> <tr> 
> <td style="vertical-align:top;width:97.5pt;"> <a href="http://tool.oschina.net/uploads/apidocs/jdk-zh/java/lang/Object.html" title="Object" rel="nofollow">Object</a>[] </td> 
> <td style="width:551px;"> <a title="toArray" rel="nofollow">toArray</a>()&nbsp;&nbsp;&nbsp;&nbsp;// 返回以恰当顺序包含此队列所有元素的数组。 </td> 
> </tr> 
> <tr> 
> <td style="width:97.5pt;"> &lt;T&gt; T[] </td> 
> <td style="width:551px;"> <a title="toArray" rel="nofollow">toArray</a>(T[]&nbsp;a)&nbsp;&nbsp;&nbsp;&nbsp;// 返回以恰当顺序包含此队列所有元素的数组；返回数组的运行时类型是指定数组的运行时类型。 </td> 
> </tr> 
> </tbody> 
> </table>

## 实例 ##

**其他网址**

> [ConcurrentLinkedQueue使用和方法介绍 - atyang - 博客园][ConcurrentLinkedQueue_ - atyang -]

**10000个人去饭店吃饭，10张桌子供饭，分别比较size() 和 isEmpty() 的耗时**

> public class Test01ConcurrentLinkedQueue {
> public static void main(String[] args) throws InterruptedException {
> int peopleNum = 10000;//吃饭人数
> int tableNum = 10;//饭桌数量
> 
> ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<>();
> CountDownLatch count = new CountDownLatch(tableNum);//计数器
> 
> //将吃饭人数放入队列（吃饭的人进行排队）
> for(int i=1;i<=peopleNum;i++){
> queue.offer("消费者_" + i);
> }
> //执行10个线程从队列取出元素（10个桌子开始供饭）
> System.out.println("-----------------------------------开饭了-----------------------------------");
> long start = System.currentTimeMillis();
> ExecutorService executorService = Executors.newFixedThreadPool(tableNum);
> for(int i=0;i<tableNum;i++) {
> executorService.submit(new Dinner("00" + (i+1), queue, count));
> }
> //计数器等待，知道队列为空（所有人吃完）
> count.await();
> long time = System.currentTimeMillis() - start;
> System.out.println("-----------------------------------所有人已经吃完-----------------------------------");
> System.out.println("共耗时：" + time);
> //停止线程池
> executorService.shutdown();
> }
> 
> private static class Dinner implements Runnable{
> private String name;
> private ConcurrentLinkedQueue<String> queue;
> private CountDownLatch count;
> 
> public Dinner(String name, ConcurrentLinkedQueue<String> queue, CountDownLatch count) {
> this.name = name;
> this.queue = queue;
> this.count = count;
> }
> 
> @Override
> public void run() {
> //while (queue.size() > 0){
> while (!queue.isEmpty()){
> //从队列取出一个元素 排队的人少一个
> System.out.println("【" +queue.poll() + "】----已吃完...， 饭桌编号：" + name);
> }
> count.countDown();//计数器-1
> }
> }
> }
> 
> **执行结果**
> 
> 使用size耗时：757ms
> 
> ![a9630ef203d6bcbdd5e7cd7fe01adc92.png][]
> 
> 使用isEmpty耗时：210
> 
> ![4d8205fd0ec403ad3e95d89316c32f00.png][]
> 
> 当数据量越大，这种耗时差距越明显。所以这种判断用isEmpty 更加合理

**返回在此队列元素上以恰当顺序进行迭代的迭代器**

> public static void main(String[] args) throws InterruptedException {
> 	ConcurrentLinkedQueue<String> queue = new ConcurrentLinkedQueue<String>();
> 	queue.offer("123");
> 	queue.offer("234");
> 	Iterator<String> iterator = queue.iterator();
> 	while (iterator.hasNext()){
> 		System.out.println(iterator.next());
> 	}
> }

# BlockingQueue #

**其他网址**

> [BlockingQueue的原理及使用方法\_Java\_Yolanda\_NuoNuo的专栏-CSDN博客Java并发（四）BlockingQueue的使用 - 摆渡者 - OSCHINALinkedBlockingQueue 的使用\_Java\_一只小棉花的博客-CSDN博客][BlockingQueue_Java_Yolanda_NuoNuo_-CSDN_Java_BlockingQueue_ - _ - OSCHINALinkedBlockingQueue _Java_-CSDN]
> 
> [线程池队列\_watson1360884839的博客-CSDN博客][watson1360884839_-CSDN]  
> [线程池的三种缓存队列\_Java\_nihaomabmt的专栏-CSDN博客][Java_nihaomabmt_-CSDN]

## 简介 ##

**队列类型及实例**

> <table> 
> <tbody> 
> <tr> 
> <td>队列类型</td> 
> <td>说明</td> 
> </tr> 
> <tr> 
> <td>ArrayBlockingQueue</td> 
> <td> &nbsp; &nbsp; &nbsp; &nbsp; 基于数组的FIFO队列；有界；创建时必须指定大小； &nbsp; &nbsp; &nbsp; &nbsp; 使用一个重入锁，默认使用非公平锁，入队和出队共用一个锁，互斥。 </td> 
> </tr> 
> <tr> 
> <td>LinkedBlockingQueue</td> 
> <td> &nbsp; &nbsp; &nbsp; &nbsp; 基于链表的FIFO队列；有/无界；默认大小是 Integer.MAX_VALUE（无界），可自定义（有界）； &nbsp; &nbsp; &nbsp; &nbsp; 两个重入锁分别控制元素的入队和出队，用Condition进行线程间的唤醒和等待。 &nbsp; &nbsp; &nbsp; &nbsp; 吞吐量通常要高于ArrayBlockingQueue。 &nbsp; &nbsp; &nbsp; &nbsp; 默认大小的LinkedBlockingQueue将导致在所有 corePoolSize 线程都忙时新任务在队列中等待。这样，创建的线程不会超过 corePoolSize。（因此，maximumPoolSize 的值也就无效了。）当每个任务完全独立于其他任务，即任务执行互不影响时，适合于使用无界队列；例如， 在 Web 页服务器中。这种排队可用于处理瞬态突发请求，当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线程具有增长的可能性。 </td> 
> </tr> 
> <tr> 
> <td>SynchronousQueue</td> 
> <td> &nbsp; &nbsp; &nbsp; &nbsp; 无缓存的等待队列；&nbsp; &nbsp; 无界；可认为大小为0。 &nbsp; &nbsp; &nbsp; &nbsp; 不保存提交任务，直接提交出去，超出corePoolSize个任务，直接创建新线程来执行任务，直到(corePoolSize＋新建线程)&gt; maximumPoolSize。 &nbsp; &nbsp; &nbsp; &nbsp; 此策略可以避免在处理可能具有内部依赖性的请求集时出现锁。直接提交通常要求无界 maximumPoolSizes 以避免拒绝新提交的任务。当命令以超过队列所能处理的平均数连续到达时，此策略允许无界线 程具有增长的可能性。 &nbsp; &nbsp; &nbsp; &nbsp; 吞吐量通常要高于LinkedBlockingQueue。 &nbsp; &nbsp; &nbsp; &nbsp; //也有地方说：是一个不存储元素的阻塞队列。每个插入操作必须等到另一个线程调用移除操作，否则插入操作一直处于阻塞状态 &nbsp; &nbsp; &nbsp; &nbsp; 详见下边的：CachedThreadPool的execute流程 </td> 
> </tr> 
> <tr> 
> <td> PriorityBlockingQueue </td> 
> <td> &nbsp; &nbsp; &nbsp; &nbsp; 基于链表的优先级队列；有/无界；默认大小是 Integer.MAX_VALUE，可自定义； &nbsp; &nbsp; &nbsp; &nbsp; 类似于LinkedBlockingQueue，但是其所含对象的排序不是FIFO，而是依据对象的自然顺序或者构造函数的Comparator决定。 </td> 
> </tr> 
> <tr> 
> <td>DelayedWorkQueue</td> 
> <td></td> 
> </tr> 
> </tbody> 
> </table>

**放入数据**

> <table style="width:595px;"> 
> <tbody> 
> <tr> 
> <td style="width:162px;">方法</td> 
> <td style="width:414px;">说明</td> 
> </tr> 
> <tr> 
> <td style="width:162px;">offer(E e)</td> 
> <td style="width:414px;"> 向队列尾部插入一个元素。 如果队列中有空闲则插入成功后返回 true 如果队列己满则丢弃当前元素然后返回false。 如果e元素为null则抛出NullPointerException异常。 该方法是非阻塞的。 </td> 
> </tr> 
> <tr> 
> <td style="width:162px;">offer(E o, long timeout, TimeUnit unit)</td> 
> <td style="width:414px;">可以设定等待的时间，若在指定的时间内，还不能往队列中加入BlockingQueue，则返回失败。</td> 
> </tr> 
> <tr> 
> <td style="width:162px;">add(E e)</td> 
> <td style="width:414px;"> 内部调用offer方法。 与直接调用offer的区别： add：失败时，抛出异常 offer：失败时，返回false </td> 
> </tr> 
> <tr> 
> <td style="width:162px;">put(E e)</td> 
> <td style="width:414px;"> 向队列尾部插入一个元素。 如果队列中有空闲则插入后直接返回。 如果队列己满则阻塞当前线程，直到队列有空闲插入成功后返回。 如果在阻塞时被其他线程设置了中断标志，则被阻塞线程会抛出InterruptedException异常而返回。 如果e元素为null则抛出NullPointerException异常。 </td> 
> </tr> 
> </tbody> 
> </table>

**获取数据**

> <table style="width:639px;"> 
> <tbody> 
> <tr> 
> <td style="width:138px;">方法</td> 
> <td style="width:488px;">说明</td> 
> </tr> 
> <tr> 
> <td style="width:138px;">poll()</td> 
> <td style="width:488px;"> 获取当前队列头部元素并从队列里面移除它。 如果队列为空则返回null。 </td> 
> </tr> 
> <tr> 
> <td style="width:138px;">poll(long timeout, TimeUnit unit)</td> 
> <td style="width:488px;"> 从BlockingQueue取出（会删除对象）一个队首的对象。 一旦在指定时间内有数据可取，则立即返回队列中的数据。 若直到时间超时还没有数据可取，返回失败。 </td> 
> </tr> 
> <tr> 
> <td style="width:138px;">take()</td> 
> <td style="width:488px;"> 获取当前队列头部元素并从队列里面移除它。 如果队列为空则阻塞当前线程直到队列不为空然后返回元素； 如果在阻塞时被其他线程设置了中断标志，则被阻塞线程会抛出InterruptedException异常而返回。 </td> 
> </tr> 
> <tr> 
> <td style="width:138px;">drainTo()</td> 
> <td style="width:488px;"> 一次性从BlockingQueue获取（会删除对象）所有可用的数据对象（可指定获取数据的个数）。 本方法可提升获取数据效率，不需要多次分批加锁或释放锁。 </td> 
> </tr> 
> </tbody> 
> </table>

**其他方法**

> <table style="width:500px;"> 
> <tbody> 
> <tr> 
> <td>方法</td> 
> <td>说明</td> 
> </tr> 
> <tr> 
> <td>remainingCapacity()</td> 
> <td>获取队列中剩余的空间</td> 
> </tr> 
> <tr> 
> <td>contains(Object o)</td> 
> <td>判断队列中是否拥有该值。</td> 
> </tr> 
> <tr> 
> <td>remove(Object o)</td> 
> <td>从队列中移除指定的值。</td> 
> </tr> 
> <tr> 
> <td>size()</td> 
> <td> 获得队列中有多少值。 （返回AtomicLong的值） </td> 
> </tr> 
> </tbody> 
> </table>

## **ArrayBlockingQueue** ##

**其他网址**

> [Java同步数据结构之ArrayBlockingQueue - 莫待樱开春来踏雪觅芳踪 - 博客园][Java_ArrayBlockingQueue - _ -]

**简介**

> ArrayBlockingQueue通过数组实现的FIFO**有界阻塞队列**，它的大小在实例被初始化的时候就被固定了，不能更改。
> 
> 该类支持一个可选的公平策略，用于被阻塞等待的线程获取独占锁的排序，因为ArrayBlockingQueue内部的操作都需要获取一个ReentrantLock锁，该锁是支持公平策略的，所以ArrayBlockingQueue的公平策略就直接作用于ReentrantLock锁，决定线程是否有公平获取锁的权利。默认情况下是非公平的，公平模式下队列按照FIFO顺序授予线程访问权。公平性通常会降低吞吐量，但会降低可变性并避免饥饿。

**ArrayBlockingQueue的缺陷**

> 通过源码可以看见，ArrayBlockingQueue内部的**几乎每一个操作方法都需要先获取同一个ReentrantLock独占锁**才能进行，这极大的**降低了吞吐量**，几乎每个操作都会阻塞其它操作，最主要是插入操作和取出操作相互之间互斥。所以ArrayBlockingQueue不适用于需要高吞吐量的高效率数据生成与消费场景。LinkedBlockingQueue就能弥补其低吞吐量的缺陷。

**实例**

> 创建一个corePoolSize为2，maximumPoolSize为3的线程池。其中ArrayBlockingQueue设置缓存2个任务。执行6个任务。ArrayBlockingQueue为有界队列：
> 
> 1.  任务1和2在核心线程中执行；
> 2.  任务3和4进来时，放到ArrayBlockingQueue缓存队列中，并且只能放2个（ArrayBlockingQueue设置的大小为2）；
> 3.  任务5和6进来的时候，任务5新建线程来执行任务，已经达到最大线程数3，所以任务6拒绝；
> 4.  当有线程执行完的时候，再将任务3和4从队列中取出执行

> 创建线程池代码如下：
> 
> /**
> * ArrayBlockingQueue
> */
> private static void arrayQueue() {
> System.out.println("\n\n =======ArrayBlockingQueue====== \n\n");
> Executor executors = new ThreadPoolExecutor(
> 2, 3, 30, TimeUnit.SECONDS,
> new ArrayBlockingQueue<Runnable>(2),
> new RejectHandler());
> execute(executors);
> }
> 
> 执行结果如下
> 
> 1 is running... 
> 2 is running... 
> 6 is rejected ^^ //6被拒
> 5 is running... //5新建线程执行
> 1 is end !!! 
> 2 is end !!! 
> 3 is running... //1和2执行完之后3和4才执行
> 4 is running... 
> 5 is end !!!

## **LinkedBlockingQueue** ##

**简介**

<table> 
 <tbody> 
 <tr> 
 <td style="width:96px;">项</td> 
 <td style="width:389px;">LinkedBlockingQueue</td> 
 <td style="width:371px;">ArrayBlockingQueue</td> 
 </tr> 
 <tr> 
 <td colspan="3">相同点</td> 
 </tr> 
 <tr> 
 <td style="width:96px;">队列类型</td> 
 <td colspan="2">不允许插入null值，FIFO队列。</td> 
 </tr> 
 <tr> 
 <td style="width:96px;">容量更改</td> 
 <td colspan="2">容量在实例被构造完成之后不允许被更改</td> 
 </tr> 
 <tr> 
 <td colspan="3">不同点</td> 
 </tr> 
 <tr> 
 <td style="width:96px;">大小指定</td> 
 <td style="width:389px;">实例化时可指定队列大小，也可以不指定大小（这时候默认就是Integer.MAX_VALUE）。</td> 
 <td style="width:371px;">实例化时必须指定大小。</td> 
 </tr> 
 <tr> 
 <td style="width:96px;">吞吐量</td> 
 <td style="width:389px;"> 大。 采用了“双锁队列” 算法，元素的入队和出队分别由putLock、takeLock两个独立的可重入锁来实现。 </td> 
 <td style="width:371px;"> 小。 几乎每一个方法都需要先获取同一个ReentrantLock独占锁才能进行。 </td> 
 </tr> 
 </tbody> 
</table>

**实例**

> 创建一个corePoolSize为2，maximumPoolSize为3的线程池。无界队列。同样执行6个任务
> 
> 1. 核心线程执行任务1和2，其它的任务3～6放到队列中
> 2. 执行完1和2，将3和4从队列中取出执行
> 3. 执行完3和4，将5和6从队列中取出
> 
> 创建线程池代码如下：
> 
> /**
> * LinkedBlockingQueue
> */
> private static void linkedQueue() {
> System.out.println("\n\n =======LinkedBlockingQueue====== \n\n");
> Executor executors = new ThreadPoolExecutor(
> 2, 3, 30, TimeUnit.SECONDS,
> new LinkedBlockingQueue<Runnable>(),
> new RejectHandler());
> execute(executors);
> }
> 
> 运行结果如下：
> 
> 1 is running... 
> 2 is running... //中间线程休眠 
> 2 is end !!! //10s之后才运行完
> 1 is end !!! 
> 3 is running... //任务3和4才执行
> 4 is running... 
> 4 is end !!! 
> 3 is end !!! 
> 6 is running... 
> 5 is running... 
> 5 is end !!! 
> 6 is end !!!

## **SynchronousQueue** ##

> 创建一个corePoolSize为2，maximumPoolSize为3的线程池。执行6个任务。根据我的理解，SynchrousQueue是个一个无缓存的队列（理论依据为SynchrousQueue源码可以看到：isEmpty()始终为true；size()始终返回0）
> 
> 根据参数设置应该只可以执行3个任务：
> 
> 1.  2个核心线程执行2个任务；
> 2.  第3个任务的时候，创建线程来执行任务3；
> 3.  当第4个任务来的时候，此时已经超过了maximumPoolSize，所以拒绝任务。

**代码**

> /**
> * SynchronousQueue
> */
> private static void syncQueue() {
> System.out.println("\n\n =======SynchronousQueue====== \n\n");
> Executor executors = new ThreadPoolExecutor(
> 2, 3, 30, TimeUnit.SECONDS,
> new SynchronousQueue<Runnable>(),
> new RejectHandler());
> execute(executors);
> }

**执行结果**

>     1 is running... 
>     4 is rejected ^^ //4被拒
>     2 is running... 
>     3 is running... 
>     5 is rejected ^^  //5被拒
>     6 is rejected ^^  //6被拒
>     3 is end !!! 
>     1 is end !!! 
>     2 is end !!!

## PriorityBlockingQueue ##

**简介**

> PriorityBlockingQueue是一个**无限容量的阻塞队列**，由于容量是无限的所以put等入队操作其实不存在阻塞，只要内存足够都能够立即入队成功，当然多个入队操作的线程之间还是存在竞争唯一锁的互斥访问。虽然PriorityBlockingQueue逻辑上是无界的，但是尝试添加元素时还是可能因为资源耗尽而抛出OutOfMemoryError。
> 
> 该队列也不允许放入null值，它使用与类java.util.PriorityQueue 相同的排序规则，也不允许放入不可比较的对象，这样做会导致ClassCastException。
> 
> **值得注意的是，虽然PriorityBlockingQueue叫优先级队列，但是并不是说元素一入队就会按照排序规则被排好序，而是只有通过调用take、poll方法出队或者drainTo转移出的队列顺序才是被优先级队列排过序的。所以通过调用 iterator() 以及可拆分迭代器 spliterator() 方法返回的迭代器迭代的元素顺序都没有被排序。如果需要有序遍历可以通过 Arrays.sort(pq.toArray()) 方法来排序。注意peek方法永远只获取且不删除第一个元素，所以多次调用peek都是返回同样的值。**
> 
> PriorityBlockingQueue其实是通过Comparator来排序的，要么入队的元素实现了Comparator接口（即所谓的自然排序），要么构造PriorityBlockingQueue实例的时候传入一个统一的Comparator实例，如果两者兼备那么以后者为准
> 
> PriorityBlockingQueue不保证具有相同优先级的元素顺序，但是你可以定义自定义类或比较器，通过辅助属性来决定优先级相同的元素的顺序，后文会举例说明。

## DelayedWorkQueue ##

**简介**

> 为什么不直接使用DelayQueue而要重新实现一个DelayedWorkQueue呢，可能是了方便在实现过程中加入一些扩展。

**使用场景**

>  *  重试机制实现。比如当调用接口失败后，把当前调用信息放入delay=10s的元素，然后把元素放入队列，那么这个队列就是一个重试队列，一个线程通过take方法获取需要重试的接口，take返回则接口进行重试，失败则再次放入队列，同时也可以在元素加上重试次数。
>  *  TimerQueue的内部实现

# 顺序消费实例 #

**其他网址**

> [Java并发（四）BlockingQueue的使用 - 摆渡者 - OSCHINA][Java_BlockingQueue_ - _ - OSCHINA]

**场景**

> 机器要对手机顺序做如下任务：生产、打包、发货。消费者等待收货。

**代码**

> package org.example.a;
> 
> import java.util.Random;
> import java.util.concurrent.ExecutorService;
> import java.util.concurrent.Executors;
> import java.util.concurrent.LinkedBlockingQueue;
> import java.util.concurrent.TimeUnit;
> 
> class Phone {
> /**
> * 手机的状态：
> * PRODUCED: 已生产
> * PACKED: 已打包
> * DELIVERED: 已发货
> * 手机的状态只能由PRODUCED->PACKED->DELIVERED转变
> */
> public enum Status {
> PRODUCED, PACKED, DELIVERED
> }
> private Status status = Status.PRODUCED;//默认状态为PRODUCED
> private final int id;
> public Phone(int id) { this.id = id;}
> public void pack() {status = Status.PACKED;}
> public void deliver() {status = Status.DELIVERED;}
> public Status getStatus() {return status;}
> public int getId() {return id;}
> public String toString() {
> return "Phone id: " + id + ", status: " + status;
> }
> }
> 
> class PhoneQueue extends LinkedBlockingQueue<Phone> {}
> 
> /**
> * 生产手机的任务。
> */
> class Producer implements Runnable {
> private PhoneQueue PhoneQueue;
> private int count = 0;
> private Random random = new Random(47);
> public Producer(PhoneQueue queue) {
> this.PhoneQueue = queue;
> }
> @Override
> public void run() {
> try {
> while(!Thread.interrupted()) {
> TimeUnit.MILLISECONDS.sleep(300 + random.nextInt(500));
> //生产一片手机，这些手机是有序的
> Phone Phone = new Phone(count++);
> System.out.println(Phone);
> //放到PhoneQueue中
> PhoneQueue.put(Phone);
> }
> } catch (InterruptedException e) {
> System.out.println("Producer interrupted.");
> }
> System.out.println("Producer off.");
> }
> }
> 
> /**
> * 打包的任务。
> */
> class Packer implements Runnable {
> private PhoneQueue producedQueue;
> private PhoneQueue PackedQueue;
> public Packer(PhoneQueue producedQueue, PhoneQueue PackedQueue) {
> this.producedQueue = producedQueue;
> this.PackedQueue = PackedQueue;
> }
> 
> @Override
> public void run() {
> try {
> while(!Thread.interrupted()) {
> //在取得下一个手机之前会一直阻塞
> Phone Phone = producedQueue.take();
> Phone.pack();
> System.out.println(Phone);
> PackedQueue.put(Phone);
> }
> } catch (InterruptedException e) {
> System.out.println("Packer interrupted.");
> }
> System.out.println("Packer off.");
> }
> }
> 
> /**
> * 发货的任务。
> */
> class Deliverer implements Runnable {
> private PhoneQueue butteredQueue;
> private PhoneQueue finishedQueue;
> public Deliverer(PhoneQueue butteredQueue, PhoneQueue finishedQueue) {
> this.finishedQueue = finishedQueue;
> this.butteredQueue = butteredQueue;
> }
> 
> @Override
> public void run() {
> try {
> while(!Thread.interrupted()) {
> //在取得下一个手机之前会一直阻塞
> Phone phone = butteredQueue.take();
> phone.deliver();
> System.out.println(phone);
> finishedQueue.put(phone);
> }
> } catch (InterruptedException e) {
> System.out.println("Deliverer interrupted.");
> }
> System.out.println("Deliverer off.");
> }
> }
> 
> /**
> * 买手机的人，消费者。
> */
> class Consumer implements Runnable {
> private PhoneQueue finishedQueue;
> private int count = 0;
> public Consumer(PhoneQueue finishedQueue) {
> this.finishedQueue = finishedQueue;
> }
> @Override
> public void run() {
> try {
> while(!Thread.interrupted()) {
> //在取得下一个手机之前会一直阻塞
> Phone phone = finishedQueue.take();
> //验证取得的手机是有序的，而且状态是DELIVERED的
> if (phone.getId() != count++ ||
> phone.getStatus() != Phone.Status.DELIVERED) {
> System.out.println("Error -> " + phone);
> System.exit(-1);
> } else {
> //使用手机
> System.out.println(phone + "->Use");
> }
> }
> } catch (InterruptedException e) {
> System.out.println("Consumer interrupted.");
> }
> System.out.println("Consumer off.");
> }
> }
> public class Demo {
> public static void main(String[] args) {
> PhoneQueue producedQueue = new PhoneQueue();
> PhoneQueue packedQueue = new PhoneQueue();
> PhoneQueue deliveredQueue = new PhoneQueue();
> ExecutorService exec = Executors.newCachedThreadPool();
> exec.execute(new Producer(producedQueue));
> exec.execute(new Packer(producedQueue, packedQueue));
> exec.execute(new Deliverer(packedQueue, deliveredQueue));
> exec.execute(new Consumer(deliveredQueue));
> try {
> TimeUnit.SECONDS.sleep(5);
> }catch (Exception e){
> e.printStackTrace();
> }
> exec.shutdownNow();
> }
> }

**执行结果**

>     Phone id: 0, status: PRODUCED
>     Phone id: 0, status: PACKED
>     Phone id: 0, status: DELIVERED
>     Phone id: 0, status: DELIVERED->Use
>     Phone id: 1, status: PRODUCED
>     Phone id: 1, status: PACKED
>     Phone id: 1, status: DELIVERED
>     Phone id: 1, status: DELIVERED->Use
>     Phone id: 2, status: PRODUCED
>     Phone id: 2, status: PACKED
>     Phone id: 2, status: DELIVERED
>     Phone id: 2, status: DELIVERED->Use
>     Phone id: 3, status: PRODUCED
>     Phone id: 3, status: PACKED
>     Phone id: 3, status: DELIVERED
>     Phone id: 3, status: DELIVERED->Use
>     Phone id: 4, status: PRODUCED
>     Phone id: 4, status: PACKED
>     Phone id: 4, status: DELIVERED
>     Phone id: 4, status: DELIVERED->Use
>     Phone id: 5, status: PRODUCED
>     Phone id: 5, status: PACKED
>     Phone id: 5, status: DELIVERED
>     Phone id: 5, status: DELIVERED->Use
>     Phone id: 6, status: PRODUCED
>     Phone id: 6, status: PACKED
>     Phone id: 6, status: DELIVERED
>     Phone id: 6, status: DELIVERED->Use
>     Phone id: 7, status: PRODUCED
>     Phone id: 7, status: PACKED
>     Phone id: 7, status: DELIVERED
>     Phone id: 7, status: DELIVERED->Use
>     Consumer interrupted.
>     Packer interrupted.
>     Packer off.
>     Deliverer interrupted.
>     Deliverer off.
>     Producer interrupted.
>     Producer off.
>     Consumer off.

[Java_--_IT_-CSDN]: https://knife.blog.csdn.net/article/details/106076450
[Java_DelayQueue_DelayedWorkQueue - _ -]: https://www.cnblogs.com/txmfz/p/10338334.html
[ConcurrentLinkedQueue_ - atyang -]: https://www.cnblogs.com/yangzhenlong/p/8359875.html
[a9630ef203d6bcbdd5e7cd7fe01adc92.png]: /images/20230312/d84ae56ff0764cb4b71769de73ed03bb.png
[4d8205fd0ec403ad3e95d89316c32f00.png]: /images/20230312/08ed2c0120f249808aba01f12bac99de.png
[BlockingQueue_Java_Yolanda_NuoNuo_-CSDN_Java_BlockingQueue_ - _ - OSCHINALinkedBlockingQueue _Java_-CSDN]: https://blog.csdn.net/baofengyu90/article/details/84564013
[watson1360884839_-CSDN]: https://blog.csdn.net/watson1360884839/article/details/82757715
[Java_nihaomabmt_-CSDN]: https://blog.csdn.net/nihaomabmt/article/details/81667481
[Java_ArrayBlockingQueue - _ -]: https://www.cnblogs.com/txmfz/p/10289507.html
[Java_BlockingQueue_ - _ - OSCHINA]: https://my.oschina.net/itblog/blog/515735