java并发队列之DelayQueue

Bertha 。 2022-01-15 04:47 235阅读 0赞

[为什么80%的码农都做不了架构师？>>> ][80_]  ![hot3.png][]

特点：无界队列，阻塞队列，具有延时性。用于存放实现了Delayed接口的对象，其中的对象只能在到期时才能从队列中取走。这种队列是有序的，即队头对象的延迟到期时间最长。其中有两个重要的方法compareTo和getDelay，第一个是比较两个任务的延迟时间进行排序，第二个方法用来获取延迟时间。DelayQueue内部是使用PriorityQueue实现的。注意：不能将null元素放置到这种队列中。

应用场景：比如数据要在某个时间点，或者多长时间后进行某些操作，比如淘宝下单后一个小时后取消订单。

代码案例：

package com.example.web.web;
    
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.concurrent.*;
    
    @RestController
    public class HomeController {
    
        @RequestMapping("/index")
        public String index() throws Exception {
    
            DelayQueue<DelayedElement> delayQueue = new DelayQueue<DelayedElement>();
    
            //生产者
            producer(delayQueue);
            //消费者
            consumer(delayQueue);
    
            TimeUnit.SECONDS.sleep(30);
    
            return "网站已得到响应";
        }
    
        //每1000毫秒创建一个对象，放入延迟队列，延迟时间10000毫秒
        private static void producer(final DelayQueue<DelayedElement> delayQueue) {
            new Thread(() -> {
                for (int i = 0; i < 5; i++) {
                    DelayedElement element = null;
                    //将第三个元素过期时间提前
                    if (i == 2) {
                        element = new DelayedElement(3000, "test" + i);
                    } else {
                        element = new DelayedElement(10000, "test" + i);
                    }
                    delayQueue.offer(element);
                    System.out.println(String.format("放入队列中一个数据:%s,长度：%s", element.msg, delayQueue.size()));
                    //进行下一个添加进行休停下,不然时间一样就区分不出来
                    try {
                        TimeUnit.MILLISECONDS.sleep(50);
                    } catch (Exception ex) {
                    }
                }
            }).start();
        }
    
    
        //消费者，从延迟队列中获得数据,进行处理
        private static void consumer(final DelayQueue<DelayedElement> delayQueue) {
            new Thread(() -> {
                while (true) {
                    DelayedElement element = null;
                    try {
                        //没有满足延时的元素 用poll返回 null
                        //没有满足延时的元素 用take会阻塞
                        System.out.println("准备开始读取数据....");
                        element = delayQueue.take();
                        System.out.println("读取数据操作完成....");
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                    System.out.println("获取到的数据信息" + element.msg);
                }
            }).start();
        }
    }
    
    class DelayedElement implements Delayed {
    
        //延迟时间
        private final long delay;
        //到期时间
        private final long expire;
        //队列数据
        public final String msg;
    
        public DelayedElement(long delay, String msg) {
            this.delay = delay;
            this.msg = msg;
            //到期时间 = 当前时间+延迟时间
            this.expire = System.currentTimeMillis() + delay;
        }
    
        //需要实现的接口，获得延迟时间   用过期时间-当前时间
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
    
        //用于延迟队列内部比较排序   当前时间的延迟时间 - 比较对象的延迟时间
        public int compareTo(Delayed o) {
            return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
        }
    }

转载于:https://my.oschina.net/uwith/blog/3056134

[80_]: https://my.oschina.net/u/2663968/blog/3051541
[hot3.png]: /images/20220115/04aa56d180cc4154b633be8193d785e3.png