延迟队列DelayQueue研究

深碍√TFBOYSˉ_ 2021-09-15 15:14 433阅读 0赞

1.什么是DelayQueue

DelayQueue 是 Delayed 元素的一个无界阻塞队列，只有在延迟期满时才能从中提取元素。该队列的头部 是延迟期满后保存时间最长的 Delayed 元素。如果延迟都还没有期满，则队列没有头部，并且 poll 将返回 null。当一个元素的 getDelay(TimeUnit.NANOSECONDS) 方法返回一个小于等于 0 的值时，将发生到期。即使无法使用 take 或 poll 移除未到期的元素，也不会将这些元素作为正常元素对待。例如，size 方法同时返回到期和未到期元素的计数。此队列不允许使用 null 元素。

总结一下如下：

1、DelayQueue队列中的元素必须是Delayed接口的实现类，该类内部实现了getDelay()和compareTo()方法，第一个方法是比较两个任务的延迟时间进行排序，第二个方法用来获取延迟时间。   
     2、DelayQueue队列没有大小限制，因此向队列插数据不会阻塞   
     3、DelayQueue中的元素只有当其指定的延迟时间到了，才能够从队列中获取到该元素。否则线程阻塞。   
     4、DelayQueue中的元素不能为null。  
     5、DelayQueue内部是使用PriorityQueue实现的。compareTo()比较后越小的越先取出来。

2.使用场景

DelayQueue能做什么？

在我们的业务中通常会有一些需求是这样的：

1、订单业务:下单之后如果三十分钟之内没有付款就自动取消订单。

2、订餐通知:下单成功后60s之后给用户发送短信通知。

那么这类业务我们可以总结出一个特点:需要延迟工作。  
由此的情况，就是我们的DelayQueue应用需求的产生。

3.简单实例

下面通过一个简单实例来了解一用法

public static void main(String[] args) {
        DelayQueue<DelayedElement> delayQueue = new DelayQueue<DelayedElement>();
    
        //生产者
        producer(delayQueue);
    
        //消费者
        consumer(delayQueue);
    
        while (true) {
            try {
                TimeUnit.HOURS.sleep(1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    
    /**  * 每100毫秒创建一个对象，放入延迟队列，延迟时间1毫秒  * @param delayQueue  */
    private static void producer(final DelayQueue<DelayedElement> delayQueue) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(100);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
    
                    DelayedElement element = new DelayedElement(1000, "test");
                    delayQueue.offer(element);
                }
            }
        }).start();
    
        /**      * 每秒打印延迟队列中的对象个数      */
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    try {
                        TimeUnit.MILLISECONDS.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("delayQueue size:" + delayQueue.size());
                }
            }
        }).start();
    }
    
    /**  * 消费者，从延迟队列中获得数据,进行处理  * @param delayQueue  */
    private static void consumer(final DelayQueue<DelayedElement> delayQueue) {
        new Thread(new Runnable() {
            @Override
            public void run() {
                while (true) {
                    DelayedElement element = null;
                    try {
                        element = delayQueue.take();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    System.out.println("now=" + System.currentTimeMillis() + "---" + element);
                }
            }
        }).start();
    }

class DelayedElement implements Delayed {
    
    private final long   delay; //延迟时间
    private final long   expire;  //到期时间
    private final String msg;   //数据
    private final long   create; //创建时间
    
    public DelayedElement(long delay, String msg) {
        this.delay = delay;
        this.msg = msg;
        expire = System.currentTimeMillis() + delay;    //到期时间 = 当前时间+延迟时间
        create = System.currentTimeMillis();
    }
    
    /**  * 需要实现的接口，获得延迟时间   用过期时间-当前时间  * @param unit  * @return  */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }
    
    /**  * 用于延迟队列内部比较排序   当前时间的延迟时间 - 比较对象的延迟时间  * @param o  * @return  */
    @Override
    public int compareTo(Delayed o) {
        return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
    }
    
    @Override
    public String toString() {
        final StringBuilder sb = new StringBuilder("DelayedElement{");
        sb.append("delay=").append(delay);
        sb.append(", expire=").append(expire);
        sb.append(", msg='").append(msg).append('\'');
        sb.append(", create=").append(create);
        sb.append('}');
        return sb.toString();
    }

运行结果：

![640_wx_fmt_png][]

4.源码分析

下面来看看其主要实现的源码分析

（1）、从队列中取元素

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(TimeUnit.NANOSECONDS);  
                    if (delay <= 0)  
                        return q.poll();  
                    else 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();  
        }  
    }

可以看到，在这段代码里，在第一个元素的延迟时间还没到的情况下：   
a.如果当前没有其他线程等待，则阻塞当前线程直到延迟时间。   
b.如果有其他线程在等待，则阻塞当前线程。

（2）、向队列中放入元素

public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.offer(e);
            if (q.peek() == e) {
                leader = null;
                available.signal();
            }
            return true;
        } finally {
            lock.unlock();
        }
    }

在放入元素的时候，会唤醒等待中的读线程。

5.进阶-实现一个延迟缓存队列

实现一个带延迟缓存队列

（1）、定义delayIetm对象

public class DelayItem<T> implements Delayed  {
    
    //创建时间
    private long MILLISECONDS_ORIGIN = System.currentTimeMillis();
    
    //元素
    private T item;
    
    //元素的存活时间,单位为毫秒 (unit:milliseconds)
    private long liveMilliseconds;
    
    public DelayItem(T item, long liveMilliseconds) {
        this.liveMilliseconds = liveMilliseconds;
        this.item = item;
    }
    
    private final long now() {
        return System.currentTimeMillis() - MILLISECONDS_ORIGIN;
    }
    
    public void setMilliseconds(long milliseconds) {
        MILLISECONDS_ORIGIN = System.currentTimeMillis();
        this.liveMilliseconds = milliseconds;
    }
    
    /**  * 如果超时，或者Map缓存中已经没有该元素，都会导致失效  *  * @param unit  * @return  */
    public long getDelay(TimeUnit unit) {
        long d = unit.convert(liveMilliseconds - now(), TimeUnit.MILLISECONDS);
    
        //        LOGGER.debug("=============key:" + item + ",time:" + milliseconds + " , now:" + now() + ",times:{}", checkTimesLeft);
        return d;
    }
    
    
    @Override
    public boolean equals(Object obj) {
        if (obj instanceof com.github.lin.DelayItem) {
            return item.equals(((com.github.lin.DelayItem) obj).getItem());
        } else {
            return false;
        }
    }
    
    public int compareTo(Delayed o) {
        if (o == this) {
            return 0;
        }
    
        //根据距离下次超时时间的长短来排优先级，越接近下次超时时间的优先级越高
        long d = (getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS));
        return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
    }
    
    public T getItem() {
        return item;
     }
    }

（2）、定义缓存工厂

public class CacheFactory<K, V> {
    
    /**缓存map*/
    private ConcurrentHashMap<K, V> concurrentHashMap = new ConcurrentHashMap<K, V>();
    
    /**延迟队列*/
    private DelayQueue<com.github.lin.DelayItem<K>> delayQueue = new DelayQueue<com.github.lin.DelayItem<K>>();
    
    /**过期检查队列*/
    private Thread expireCheckThread;
    
    public CacheFactory() {
        //定时清理过期缓存
        expireCheckThread = new Thread() {
            @Override
            public void run() {
                dameonCheckOverdueKey();
            }
        };
        expireCheckThread.setDaemon(true);
        expireCheckThread.start();
    }
    
    /**  * 放入带有过期时间的元素  * @param key  * @param value  * @param liveMilliseconds  */
    public void put(K key, V value, long liveMilliseconds) {
        V oldValue = concurrentHashMap.put(key, value);
        if (oldValue != null) {
            //todo 这个地方性能比较差，DelayQueue删除元素慢
            boolean result = delayQueue.remove(new com.github.lin.DelayItem<K>(key, 0L));
        }
    
        com.github.lin.DelayItem delayItem = new com.github.lin.DelayItem(key, liveMilliseconds);
        delayQueue.put(delayItem);
    }
    
    /**  * 取元素  * @param key  * @return  */
    public V get(K key) {
        return concurrentHashMap.get(key);
    }
    
    /**  * 检查过期的key，从cache中删除  */
    private void dameonCheckOverdueKey() {
        DelayItem<K> delayedItem;
        while (true) {
            try {
                delayedItem = delayQueue.take();
                if (delayedItem != null) {
                    concurrentHashMap.remove(delayedItem.getItem());
                    System.out.println(System.nanoTime() + " remove " + delayedItem.getItem() + " from cache");
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
    }

其实就是一个map和一个queue来实现，后台有一个守护线程，如果检查到有过期的key。就从queue里取出来，再从map删除。

（3）、测试用例

public class CacheFactoryTest {
    
    public static void main(String[] args) throws InterruptedException {
        CacheFactory<String,String> cacheFactory = new CacheFactory<String, String>();
        //存活1s
        cacheFactory.put("key1","value1",1000);
    
        //存活10s
        cacheFactory.put("key2","value2",10000);
    
        System.out.println("begin get key1:" + cacheFactory.get("key1"));
        System.out.println("begin get key2:" +cacheFactory.get("key2"));
    
        //等待2s
        Thread.sleep(2000);
    
        System.out.println("after 2s:" +cacheFactory.get("key1"));
        System.out.println("after 2s:" +cacheFactory.get("key2"));
    
        //等待10s
        Thread.sleep(10000);
    
        System.out.println("after 10s:" +cacheFactory.get("key1"));
        System.out.println("after 10s:" +cacheFactory.get("key2"));
    
    }
    }

执行结果

![640_wx_fmt_png 1][]

![15b57730-550d-11e8-8023-41215de53e1b][]

扫码向博主提问

![3_zhangxing52077.jpg][]

zhangxing52077

非学，无以致疑；非问，无以广识

[640_wx_fmt_png]: /images/20210811/016d9ca96dfd4ef28dcbd0506cbce5f8.png
[640_wx_fmt_png 1]: /images/20210811/23f295eaa25f48c885f63e4f9fc32ed1.png
[15b57730-550d-11e8-8023-41215de53e1b]: /images/20210811/6e04bd76a6d345959df65d47c2b63f41.png
[3_zhangxing52077.jpg]: /images/20210811/4dd714f6b49248099b8109018afc21de.png