Java多线程--队列(Queue)--使用/教程/实例

╰半夏微凉° 2023-10-02 13:48 5阅读 0赞

原文网址：[Java多线程--队列(Queue)--使用/教程/实例\_IT利刃出鞘的博客-CSDN博客][Java_--_Queue_--_IT_-CSDN]

### 简介 ###

本文用示例来介绍Java中的队列的用法。包括：DelayQueue，ConcurrentLinkedQueue，BlockingQueue。

文章最后有一个常见的面试题代码实例：机器要对手机按顺序做如下任务：生产、打包、发货。消费者等待收货，如何使用BlockingQueue完成？

## DelayQueue ##

**简介**

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 ##

### 简介 ###

**ConcurrentLinkedQueue与BlockingQueue区别**

<table> 
 <tbody> 
 <tr> 
 <td> 项 </td> 
 <td> ConcurrentLinkedQueue </td> 
 <td> BlockingQueue </td> 
 </tr> 
 <tr> 
 <td> 是否阻塞 </td> 
 <td> 非阻塞（使用CAS） </td> 
 <td> 阻塞(使用锁） </td> 
 </tr> 
 <tr> 
 <td> 优点 </td> 
 <td> 性能高。 （特别是多个消费者时） </td> 
 <td> 多线程操作共同的队列时不需要额外的同步。 队列会自动负载均衡，即那边（生产与消费两边）处理快了就会被阻塞掉，从而减少两边的处理速度差距。 </td> 
 </tr> 
 <tr> 
 <td> 使用场景 </td> 
 <td> 弹出的元素是多个线程共享的。（线程安全） </td> 
 <td> </td> 
 </tr> 
 </tbody> 
</table>

ConcurrentLinkedQueue与BlockingQueue**方法的区别**

<table> 
 <tbody> 
 <tr> 
 <td> 方法 </td> 
 <td> ConcurrentLinkedQueue </td> 
 <td> BlockingQueue </td> 
 </tr> 
 <tr> 
 <td> take() </td> 
 <td> 无此方法 </td> 
 <td> 有此方法 </td> 
 </tr> 
 <tr> 
 <td> size() </td> 
 <td> 轮询元素，复杂度为O(n)。 线程不安全（可能有其他线程offer、remove等） </td> 
 <td> 直接返回大小，复杂度为O(1)。 线程安全。 </td> 
 </tr> 
 <tr> 
 <td> isEmpty() </td> 
 <td> 有此方法 </td> 
 <td> 无此方法 </td> 
 </tr> 
 </tbody> 
</table>

**方法**

<table> 
 <tbody> 
 <tr> 
 <td colspan="1"> 构造方法摘要 </td> 
 </tr> 
 <tr> 
 <td> ConcurrentLinkedQueue() &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;创建一个最初为空的&nbsp;ConcurrentLinkedQueue。 </td> 
 </tr> 
 <tr> 
 <td> ConcurrentLinkedQueue(Collection&lt;? extends&nbsp;E&gt;&nbsp;c)&nbsp; &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;创建一个最初包含给定 collection 元素的&nbsp;ConcurrentLinkedQueue，按照此 collection 迭代器的遍历顺序来添加元素。 </td> 
 </tr> 
 </tbody> 
</table>

<table> 
 <tbody> 
 <tr> 
 <td> 返回值 </td> 
 <td> 描述 </td> 
 </tr> 
 <tr> 
 <td> boolean </td> 
 <td> add(E&nbsp;e)&nbsp;&nbsp;&nbsp;&nbsp;// 将指定元素插入此队列的尾部。 </td> 
 </tr> 
 <tr> 
 <td> boolean </td> 
 <td> contains(Object&nbsp;o)&nbsp;&nbsp;&nbsp;&nbsp;// 如果此队列包含指定元素，则返回&nbsp;true。 </td> 
 </tr> 
 <tr> 
 <td> boolean </td> 
 <td> isEmpty()&nbsp;&nbsp;&nbsp;&nbsp;// 如果此队列不包含任何元素，则返回&nbsp;true。 </td> 
 </tr> 
 <tr> 
 <td> Iterator&lt;E&gt; </td> 
 <td> iterator()&nbsp;&nbsp;&nbsp;&nbsp;// 返回在此队列元素上以恰当顺序进行迭代的迭代器。 </td> 
 </tr> 
 <tr> 
 <td> boolean </td> 
 <td> offer(E&nbsp;e)&nbsp;&nbsp;&nbsp;&nbsp;// 将指定元素插入此队列的尾部。 </td> 
 </tr> 
 <tr> 
 <td> E </td> 
 <td> peek()&nbsp;&nbsp;&nbsp;&nbsp;// 获取但不移除此队列的头；如果此队列为空，则返回&nbsp;null。 </td> 
 </tr> 
 <tr> 
 <td> E </td> 
 <td> poll()&nbsp;&nbsp;&nbsp;&nbsp;// 获取并移除此队列的头，如果此队列为空，则返回&nbsp;null。 </td> 
 </tr> 
 <tr> 
 <td> boolean </td> 
 <td> remove(Object&nbsp;o)&nbsp;&nbsp;&nbsp;&nbsp;// 从队列中移除指定元素的单个实例（如果存在）。 </td> 
 </tr> 
 <tr> 
 <td> int </td> 
 <td> size()&nbsp;&nbsp;&nbsp;&nbsp;// 返回此队列中的元素数量。 </td> 
 </tr> 
 <tr> 
 <td> Object[] </td> 
 <td> toArray()&nbsp;&nbsp;&nbsp;&nbsp;// 返回以恰当顺序包含此队列所有元素的数组。 </td> 
 </tr> 
 <tr> 
 <td> &lt;T&gt; T[] </td> 
 <td> toArray(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 ##

见：[Java多线程--阻塞队列(BlockingQueue)--使用/教程/实例\_IT利刃出鞘的博客-CSDN博客][Java_--_BlockingQueue_--_IT_-CSDN]

## **其他网址** ##

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

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

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

[Java_--_Queue_--_IT_-CSDN]: https://knife.blog.csdn.net/article/details/121110813
[ConcurrentLinkedQueue_ - atyang -]: https://www.cnblogs.com/yangzhenlong/p/8359875.html
[a9630ef203d6bcbdd5e7cd7fe01adc92.png]: https://img-blog.csdnimg.cn/img_convert/a9630ef203d6bcbdd5e7cd7fe01adc92.png
[4d8205fd0ec403ad3e95d89316c32f00.png]: https://img-blog.csdnimg.cn/img_convert/4d8205fd0ec403ad3e95d89316c32f00.png
[Java_--_BlockingQueue_--_IT_-CSDN]: https://knife.blog.csdn.net/article/details/122795966
[Java_DelayQueue_DelayedWorkQueue - _ -]: https://www.cnblogs.com/txmfz/p/10338334.html
[Java_BlockingQueue_ - _ - OSCHINA]: https://my.oschina.net/itblog/blog/515735