延迟队列DelayQueue

怼烎@ 2024-04-20 22:07 61阅读 0赞

## 1.什么是[DelayQueue][]（延时队列） ##

DelayQueue 是一个通过[PriorityBlockingQueue][]实现延迟获取元素的无界队列无界阻塞队列，其中添加进该队列的元素必须实现Delayed接口（指定延迟时间），而且只有在延迟期满后才能从中提取元素。

为了更好理解,给大家说一个实际场景;

我们在购物平台买商品的时候,下单后,如果在30min内没有付款就会取消订单,这个就用到了延时队列;

#### DelayQueue主要用于两个方面：清掉缓存中超时的缓存数据- 任务超时处理 ####

## 2.DelayQueue主要方法 ##

### 1.offer添加元素 ###

public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            //调用优先队列
            q.offer(e);
            //检验元素是否为队首，是则设置 leader 为null, 并唤醒一个消费线程  
            if (q.peek() == e) {
                leader = null;
                available.signal();
            }
            return true;
        } finally {
            lock.unlock();
        }
    }

> 也可以使用 put() add() 方法 但他们还是调用了offer方法

### 2.take获取元素 ###

public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            for (;;) {
                //从优先队列中获取第一个元素，peek方法不会删除元素
                E first = q.peek();
                //如果获取不到数据，则调用available.await()进入阻塞状态
                if (first == null)
                    available.await();
                else {
                    //获取当前延时对象是否到期
                    long delay = first.getDelay(NANOSECONDS);
                    //到期那么返回这个延时对象
                    if (delay <= 0)
                        return q.poll();
                    first = null; // 
                    //leader不为空，表明已经有其他线程在等待这个延时对象了
                    //为什么不available.awaitNanos(delay)呢？这将会导致大量的线程在同一时间点被唤醒，然后去竞争
                    //这个到期的延时任务，影响性能，还不如直接将他们无时间限制的wait，leader线程或者其他新进来的线程获取到延时对象后，去唤醒
                    //让他们去竞争下一个延时对象
                    if (leader != null)
                        available.await();
                    else {
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try {
                            //指定纳秒级别线程阻塞时间，当前wait住的线程被唤醒后有可能与其他线程竞争失败，就会进入了同步队列阻塞，那个抢到锁的线程就会取走这个延时对象
                            available.awaitNanos(delay);
                        } finally {
                            //leader线程被唤醒并获取到锁之后会将leader设置为空
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            //leader为空并且队列不为空，那么唤醒正在等待的线程
            if (leader == null && q.peek() != null)
                available.signal();
            lock.unlock();  //释放锁
        }
    }

> 从优先队列中取值，如果取到的延时节点已经已经到期，那么直接返回，如果还没有到期并且已经有其他线程在执行delay时间等待了（也就是leader线程），那么挂起自己（避免延时 相同时间造成大量线程同时唤醒）， leader线程在指定delay时间后主动唤醒，然后取竞争锁，如果竞争成功，那么很大概率可以获取到延时节点，如果竞争失败，将被阻塞。

### 3.remove删除元素 ###

public boolean remove(Object o) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            return q.remove(o);
        } finally {
            lock.unlock();
        }
    }

## 3.使用实例 ##

实现Delayed接口：

public class MyDelay<T> implements Delayed {
    
        long delayTime; //延时时间
        long expire; //过期时间
        T data;
    
        public MyDelay(long delayTime, T data) {
            this.delayTime = delayTime;
            this.expire = System.currentTimeMillis() + delayTime;
            this.data = data;
        }
    
        /**
         * 剩余时间 = 到期时间 - 当前时间
         */
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
    
        /**
         * 优先级规则 两个任务比较 , 时间短的优先执行
         * @param o
         * @return
         */
        @Override
        public int compareTo(Delayed o) {
            long f = this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
            return (int)f;
        }
    
        @Override
        public String toString() {
            return "delayTime=" + delayTime +
                    ", expire=" + expire +
                    ", data=" + data;
        }
    
    
    }

测试用例如下：

public class DelayQueueDemo {
    
    
        static BlockingQueue<Delayed> queue = new DelayQueue();
    
        public static void main(String[] args) throws InterruptedException {
            queue.add(new MyDelay(800, "第一次添加任务"));
            queue.add(new MyDelay(300, "第二次添加任务"));
            queue.add(new MyDelay(50, "第三次添加任务"));
    
            while (!queue.isEmpty()) {
                Delayed delayed = queue.take();
                System.out.println(delayed);
            }
        }
    
    }

输出如下:

delayTime=50, expire=1673419474898, data=第三次添加任务
    delayTime=300, expire=1673419475148, data=第二次添加任务
    delayTime=800, expire=1673419475648, data=第一次添加任务

## 4.总结 ##

DelayQueue其实采用了装饰器模式，在对PriorityQueue进行包装下增加了延时时间获取元素的功能，其主要特点归纳如下：

1.DelayQueue是一个无界阻塞队列，队列内部使用PriorityQueue来实现。

2.进入队列的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素，只有在延迟期满时才能从中提取元素；

3.该队列头部是延迟期满后保存时间最长的Delayed元素；

4.如果没有延迟未过期元素，且队列没有头部，并且poll将返回null；

5.当一个元素的getDelay（TimeUnit.NANOSECONDS）方法返回一个小于等于0的值时，表示该元素已过期；

6.无法使用poll或take移除未到期的元素，也不会将这些元素作为正常元素对待；例如：size方法返回到期和未到期元素的计数之和。

7.此队列不允许使用null元素。

[DelayQueue]: https://so.csdn.net/so/search?q=DelayQueue&spm=1001.2101.3001.7020
[PriorityBlockingQueue]: https://so.csdn.net/so/search?q=PriorityBlockingQueue&spm=1001.2101.3001.7020