延迟队列 DelayQueue 使用与源码解析

小灰灰 2022-12-24 10:51 189阅读 0赞

### 文章目录 ###

*   *  DelayQueue结构
     *  继承结构
     *  工作原理
     *  核心属性
     *  核心方法 void take(E e)
     *  核心方法 void put(E e)
     *  总结

当内存中有一堆的客户信息，需要实时移除VIP到期的客户的特权时有以下做法。

*  做法1：  
    定时一段时间检查一遍所有的元素，如果客户的VIP到期日期小于当前，则取消客户的VIP特权。
 *  做法2：  
    利用[优先队列][Link 1]小顶堆的结构，将最快要过期的客户信息放置在堆顶，一个消费线程去poll堆顶元素并且处理，这样就不用去遍历全部的元素了。当日期不到时便让线程等待多长时间，当时间到时唤醒线程处理。

以上做法2有现成的队列可以使用，就是接下来要说的延迟队列。

## DelayQueue结构 ##

是不是跟优先队列很像。DelayQueue其内部就聚合了一个优先队列。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]

## 继承结构 ##

public class DelayQueue<E extends Delayed> extends AbstractQueue<E> implements BlockingQueue<E>

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 1]  
其实现了Queue，Collection，BlockingQueue。这里注意一点，因为DelayQueue其泛型被指定为需要继承Delayed接口。所以放入的元素也是要继承该接口的。先来看看Delayed接口

public interface Delayed extends Comparable<Delayed> { 
        /** * 返回与此对象的剩余延迟。 * * @param 时间单位 * @return 剩余的延迟；零或负值表示延迟已经超时 */
        long getDelay(TimeUnit unit);
    }

其还继承了Comparable接口

public interface Comparable<T> { 
        /** * 将当前对象this与指定对象进行比较。 * 返回负整数对象小于指定的对象，零等于指定的对象 正整数大于指定的对象 */
        public int compareTo(T o);
    }

存放的对象需实现上述两个接口，那么就要重写这两个方法。一个用于获取当前剩余的延迟时间，一个用于对象之间比较大小。

## 工作原理 ##

*  队列元素需要实现getDelay(TimeUnit unit)方法和compareTo(Delayed o)方法,  
    getDelay定义了剩余到期时间，compareTo方法定义了元素排序规则，内部存储结构聚合了[优先级队列][Link 1]。
 *  存放元素时元素存储交由优先级队列存放。存放元素时，将元素根据compareTo放入优先级队列。
 *  获取元素时，总是判断PriorityQueue队列的队首元素是否到期，若未到期就等待，线程进入等待状态等待唤醒。若到期则元素出队。
 *  其中涉及到leader/Follower线程模型，线程去获取堆顶元素时，元素并未到期，那么会去将自己设置为leader线程等待指定的时间后唤醒，如果leader已经有其他线程，则自己为Follower。

## 核心属性 ##

*  private final transient ReentrantLock lock = new ReentrantLock();  
    可重入锁
 *  private final PriorityQueue q = new PriorityQueue();  
    底层存储数据用的优先级队列
 *  private Thread leader = null;  
    Leader-follower线程模型
 *  private final Condition available = lock.newCondition();

## 核心方法 void take(E e) ##

检索并删除此队列的头部。如有元素没到过期时间需要等待直到该元素过期。  
先判断队列有元素，如果没有线程等待，如果有元素，获取队列元素的过期时间，如果到期就会执行出队操作。否则需要判断一下leader线程是否为Null，如果leader 不为空 说明已经有线程在等待堆顶元素过期，加入到等待队列中等待唤醒即当前线程变成了Follower，如果leader为空那么说明并没有线程在等待头节点过期，那么将当前线程设置为leder线程，并且等待头节点的超时时间后唤醒线程。

public E take() throws InterruptedException { 
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try { 
            for (;;) { 
                //获取堆顶元素
                E first = q.peek();
                // 队首为空，则阻塞当前线程
                if (first == null)
                    available.await();//第一个await
                else { 
                    // 获取队首元素的超时时间
                    long delay = first.getDelay(NANOSECONDS);
                    //如果延迟到期，直接元素出队(要先执行finally内的代码块)
                    if (delay <= 0)
                        return q.poll();
                    first = null; // don't retain ref while waiting
                    //这里用到了Leader/Followers模式
                    //如果leader 不为空 说明已经有线程在监听 在之前已经有线程进来检查到元素未到延迟时间
                    //释放当前线程获取的锁 加入到等待队列中 即 当前线程变成了Followers
                    if (leader != null)
                        available.await();//第二个await
                    else { 
                        //如果没有leader 说明没有线程在监听，目前线程为第一个检查头节点的线程
                        // 将当前线程置为leader线程 
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;
                        try { 
                            //使线程进入等待状态，到一定时间（堆顶节点的延迟时间）自动唤醒，来处理头节点
                            available.awaitNanos(delay);//第三个await
                        } finally { 
                            //awaitNanos这个线程被唤醒后，
                            //需要先将leader置为Null
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally { 
            // 出队后leader置为null，如果队列不为空，那就唤醒等待队列中的线程
            if (leader == null && q.peek() != null)
                available.signal();
            lock.unlock();
        }
    }

## 核心方法 void put(E e) ##

将指定的元素插入此延迟队列。由于队列是无界的，因此此方法将永远不会阻塞。但是指定的存放的元素需要实现 Delayed接口。  
将元素入队到优先队列中（底层存储结构为优先级队列），入队后需要检查一下刚刚插入的元素是否是优先级最高的（延迟最近一个到期的）如果刚刚存入元素就是延迟最近一个到期的需要先把leader线程置为Null，再唤醒一个take线程。那么线程就会在上述take方法中的第一第二第三个await时被唤醒，进入下次循环再执行take流程。

public void put(E e) { 
        offer(e);
    }
    public boolean offer(E e) { 
        final ReentrantLock lock = this.lock;
        //加锁
        lock.lock();
        try { 
            //将元素入队到优先队列中
            q.offer(e);
            //入队后需要检查一下刚刚插入的元素是否是优先级最高的（延迟最近一个到期的）
            如果刚刚存入元素就是延迟最近一个到期的需要手动唤醒一个take线程
            //并且需要先把leader线程置为Null。
            if (q.peek() == e) { 
                leader = null;
                available.signal();
            }
            return true;
        } finally { 
            lock.unlock();
        }
    }

## 总结 ##

1.  底层存储使用了优先级队列
2.  存放的元素需要实现getDelay()方法和compareTo()方法，用来确定元素是否到期，以及比较元素之间的优先级
3.  堆顶元素只有一个，使用了leader/Follower线程模型。获取元素时如果元素还未到过期时间，需要检查leader线程是否为空，为空的话说明还没有线程监听堆顶元素。需要将自己休眠指定时间后唤醒处理堆顶元素
4.  入队时需要检查一下刚刚插入的元素是否是优先级最高的（延迟最近一个到期的）是的话需要手动唤醒一个take线程， 并且把leader线程置为Null。被唤醒的线程需要去检查堆顶元素是否过期。

[Link 1]: https://blog.csdn.net/wsdfym/article/details/110145082
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70]: /images/20221120/e3f1fa6210cb4fe1896ab5c74b042f37.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70 1]: /images/20221120/9dfb9502e1314c8e8c6d5ab9614c9d5f.png