阻塞队列之DelayQueue

谁践踏了优雅 2023-02-26 14:24 134阅读 0赞

## 阻塞队列之DelayQueue ##

DelayQueue是一个支持延时获取元素的无界阻塞队列，在队列底层使用priorityqueue实现，DelayQueue队列中的元素必须实现delayed接口，该接口定义了在创建元素时该元素的延迟时间，在内部通过为每个元素的操作加锁来保障数据的一致性，只有在延迟时间到后才能从队列中提取元素，我们可以将DelayQueue运用以下场景中  
缓存系统的设计：可以用DelayQueue保存缓存的有效性，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素，则表示缓存的有效期到了  
定时任务调度：使用DelayQueue保存即将执行的任务和执行时间，一旦从DelayQueue中获取元素，则表示任务开始执行，java中timerqueue就是通过DelayQueue是实现的

下面我们看下demo

package com.liu.demo09;
    
    import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @outhor liu
     * @dare 2020/7/14 16:34
     */
    public class CDelayQueue implements Delayed {
        
    
        private int id;
        //消息内容
        private String body;
        //延迟时间
        private long Time;
    
        public CDelayQueue(int id, String body, long time) {
        
            this.id = id;
            this.body = body;
            this.Time = TimeUnit.NANOSECONDS.convert(time,TimeUnit.MILLISECONDS)+System.nanoTime();
        }
    
        public int getId() {
        
    
            return id;
        }
    
        public void setId(int id) {
        
            this.id = id;
        }
    
        public String getBody() {
        
            return body;
        }
    
        public void setBody(String body) {
        
            this.body = body;
        }
    
        public long getTime() {
        
            return Time;
        }
    
        public void setTime(long time) {
        
            Time = time;
        }
    
    
        @Override
        public long getDelay(TimeUnit unit) {
        
            return unit.convert(this.Time - System.nanoTime(),TimeUnit.NANOSECONDS);
        }
    
        //自定义方法
        @Override
        public int compareTo(Delayed o) {
        
            CDelayQueue c = (CDelayQueue)o;
            return Integer.valueOf(this.id)>Integer.valueOf(c.id)? 1
                    :(Integer.valueOf(this.id)<Integer.valueOf(c.id)? -1 :0);
        }
    }

package com.liu.demo09;
    
    import java.util.concurrent.DelayQueue;
    
    /**
     * @outhor liu
     * @dare 2020/7/14 17:02
     */
    public class Consumer implements Runnable {
        
    
        //设置延时队列，消费者从其中获取消息进行消费
        private DelayQueue<CDelayQueue> queues;
    
        public Consumer(DelayQueue<CDelayQueue> queues) {
        
            this.queues = queues;
        }
    
        @Override
        public void run() {
        
            while (true){
        
                try {
        
                    CDelayQueue take = queues.take();
                    System.out.println("消费者id:" + take.getId() + "消费体："+ take.getBody());
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
        }
    }

package com.liu.demo09;
    
    import java.util.concurrent.DelayQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /**
     * @outhor liu
     * @dare 2020/7/14 16:39
     */
    public class CTest {
        
        public static void main(String[] args) {
        
            //创建延时队列
            DelayQueue<CDelayQueue> cd = new DelayQueue<>();
            //添加延时消息，3s
            CDelayQueue w = new CDelayQueue(1, "world", 3000);
            //添加延时消息，5s
            CDelayQueue h = new CDelayQueue(2, "hello", 5000);
            //将延时消息放入队列中
            cd.offer(h);
            cd.offer(w);
            //启动消费线程
            ExecutorService e = Executors.newFixedThreadPool(1);
            e.execute(new Consumer(cd));
            e.shutdown();
        }
    }

输出：

消费者id:1消费体：world
    消费者id:2消费体：hello