延迟队列的使用和理解——附代码

谁借莪１个温暖的怀抱￠ 2023-07-07 08:06 6阅读 0赞

package com.example.demo.ThreadPool;
    
    import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    
    /*** * 定义消息体，必须实现Delayed且重写方法 */
    public class DelayQueneMessage implements Delayed { 
        private String id;//消息id
        private String body; // 消息内容
        private long excuteTime;// 延迟时长，这个是必须的属性因为要按照这个判断延时时长。
        public String getId() { 
            return id;
        }
    
    
        public void setId(String id) { 
            this.id = id;
        }
    
        public void setBody(String body) { 
            this.body = body;
        }
    
        public void setExcuteTime(long excuteTime) { 
            this.excuteTime = excuteTime;
        }
    
        public String getBody() { 
    
            return body;
        }
    
        public long getExcuteTime() { 
            return excuteTime;
        }
    
        public DelayQueneMessage(String id, String body, long delayTime) { 
            this.id = id;
            this.body = body;
            this.excuteTime = TimeUnit.NANOSECONDS.convert(delayTime, TimeUnit.MILLISECONDS) + System.nanoTime();
        }
    
        /** * 延迟任务是否到时就是按照这个方法判断如果返回的是负数则说明到期否则还没到期 * @param unit * @return */
        @Override
        public long getDelay(TimeUnit unit) { 
            return unit.convert(this.excuteTime - System.nanoTime(), TimeUnit.NANOSECONDS);
        }
    
        @Override
        public int compareTo(Delayed o) { 
            // 比较、排序：对任务的延时大小进行排序，将延时时间最小的任务放到队列头部
            return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
        }
    }

package com.example.demo.ThreadPool;
    
    import java.util.concurrent.DelayQueue;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    
    /**延迟队列消费者（任务到期时需要自动执行的任务） * @author lisz */
    
    public class DelayQueueConsumer implements Runnable { 
        private final DelayQueue<DelayQueneMessage> delayQueue;
        /** * 构造函数 * @param delayQueue 延迟队列 */
        public DelayQueueConsumer(DelayQueue<DelayQueneMessage> delayQueue) { 
            this.delayQueue = delayQueue;
        }
        @Override
        public void run() { 
            while (true) { 
                try { 
                    // 从延迟队列的头部获取已经过期的消息
                    // 如果暂时没有过期消息或者队列为空，则take()方法会被阻塞，直到有过期的消息为止
                    DelayQueneMessage delayMessage = delayQueue.take();
                    System.out.println("Consumer received message: {}"+delayMessage.getBody());
                } catch (InterruptedException e) { 
                    e.printStackTrace();
                }
            }
    
        }
        }

测试

public static void main(String[] args) { 
            // 创建延时队列
            DelayQueue<DelayQueneMessage> queue = new DelayQueue<DelayQueneMessage>();
            // 添加延时消息,m1 延时3s
            DelayQueneMessage m1 = new DelayQueneMessage("324", "延迟3s", 3000);
            // 添加延时消息,m2 延时10s
            DelayQueneMessage m2 = new DelayQueneMessage("2314", "延迟10s", 10000);
            //将延时消息放到延时队列中
            queue.offer(m2);
            queue.offer(m1);
            // 启动消费线程 消费添加到延时队列中的消息，前提是任务到了延期时间
            ExecutorService exec = Executors.newFixedThreadPool(1);
            exec.execute(new DelayQueueConsumer(queue));
            exec.shutdown();
        }
        ```