DelayQueue延时队列

港控/mmm° 2022-09-10 07:19 225阅读 0赞

DelayQueue是一个实现了BlockingQueue接口的类，它也是阻塞队列，但是他还具有一些独特的功能，那就是在获取元素时，需要检测元素是否到达延时时间，如果没有到达延时时间是无法获取到元素。

队列里的元素需要实现Delay接口，这个接口有一个方法getDelay，当这个方法返回0或者负数，说明元素已经到达了延时时间，在获取元素时可以正常获取，否则无法获取。Delay接口继承了Comparable接口，重写它的compareTo方法，可以对队列中的元素排序，通常是按照延时时间从小到达排序，用于避免前面的元素未到达延时时间不能取出导致无法获取后面的元素。

java延迟队列提供了在指定时间才能获取队列元素的功能，队列头元素是最接近过期的元素。没有过期元素的话，使用poll()方法会返回null值，超时判定是通过getDelay(TimeUnit.NANOSECONDS)方法的返回值小于等于0来判断。延时队列不能存放空元素。

public interface Delayed extends Comparable<Delayed> {
    
        /**
         * Returns the remaining delay associated with this object, in the
         * given time unit.
         *
         * @param unit the time unit
         * @return the remaining delay; zero or negative values indicate
         * that the delay has already elapsed
         */
        long getDelay(TimeUnit unit);
    }

由Delayed定义可以得知，队列元素需要实现getDelay(TimeUnit unit)方法和compareTo(Delayed o)方法, getDelay定义了剩余到期时间，compareTo方法定义了元素排序规则，注意，元素的排序规则影响了元素的获取顺。

获取元素：非阻塞

public E poll() {
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
                E first = q.peek();
                if (first == null || first.getDelay(NANOSECONDS) > 0)
                    return null;
                else
                    return q.poll();
            } finally {
                lock.unlock();
            }
        }

获取元素：阻塞

public E take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                for (;;) {
                    E first = q.peek();
                    if (first == null)
                        available.await();
                    else {
                        long delay = first.getDelay(NANOSECONDS);
                        if (delay <= 0)
                            return q.poll();
                        first = null; // don't retain ref while waiting
                        if (leader != null)
                            available.await();
                        else {
                            Thread thisThread = Thread.currentThread();
                            leader = thisThread;
                            try {
                                available.awaitNanos(delay);
                            } finally {
                                if (leader == thisThread)
                                    leader = null;
                            }
                        }
                    }
                }
            } finally {
                if (leader == null && q.peek() != null)
                    available.signal();
                lock.unlock();
            }
        }

测试：

package com.tech.netty.test;
    
    import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author lw
     * @since 2021/9/6
     */
    public class Item implements Delayed {
    
        private int num;
        private long time;
    
        public Item(int num, long delay) {
            this.num = num;
            this.time = delay+System.currentTimeMillis();
        }
    
        @Override
        public long getDelay(TimeUnit unit) {
            return time-System.currentTimeMillis();
        }
        
        //设置按照延时时间从小到大排列
        @Override
        public int compareTo(Delayed o) {
            Item o1 = (Item) o;
            long diff=time-o1.time;
            if(diff<=0){
                return -1;
            }else{
                return 1;
            }
        }
    
        @Override
        public String toString() {
            return "Item{" +
                    "num=" + num +
                    ", time=" + time +
                    '}';
        }
    }

package com.tech.netty.test;
    
    import java.util.concurrent.DelayQueue;
    
    /**
     * @author lw
     * @since 2021/9/6
     */
    public class DQueue {
        public static void main(String[] args) throws InterruptedException {
            DelayQueue<Item> delayQueue=new DelayQueue<Item>();
            Item item1 = new Item(1, 3000);
            Item item2 = new Item(2, 6000);
            Item item3= new Item(3, 9000);
            delayQueue.put(item2);
            delayQueue.put(item1);
            delayQueue.put(item3);
            int size = delayQueue.size();
            for (int i = 0; i <size ; i++) {
                System.out.println(delayQueue.take());
            }
        }
    }

可以发现元素是按延时时间升序，每间隔3秒输出一个

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_20_color_FFFFFF_t_70_g_se_x_16][]

如果设置元素按到期时间从大到小排列，则会发现元素是按延时时间倒序排列，几乎同时输出。

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/7e987625a3624764a5c3fa7224c88f79.png