DelayQueue讲解

快来打我* 2023-06-03 14:53 5阅读 0赞

# DelayQueue讲解 #

DelayQueue 是一个带延时功能的阻塞队列，可以通过它轻松的实现定时任务、延时任务，比如重试、异步提醒、定时通知等等。  
那DelayQueue为何有这样的能力？他是如何做的呢？

## JDK给他的定位是什么 ##

在看一个类时，先看作者给他的定位是什么，怎么看？通过类定义和类关系图来了解。  
**类定义**

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

**类关系图**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDI1MDcw_size_16_color_FFFFFF_t_70]  
看他直接关系，发现它是一个Queue，有BlockingQueue的能力，但有个限制，往它里面放的元素必须实现Delayed接口。

--------------------

## 作者的实现思路 ##

了解一个东西要先从表面看，一点一点深入，而不是直接来看，第一步怎么看？先看它有哪些成员变量。

// 锁：控制线程安全
        private final transient ReentrantLock lock = new ReentrantLock();
    
    	// 内部容器，PriorityQueue是一个二分小顶堆
        private final PriorityQueue<E> q = new PriorityQueue<E>();
        
    	// 等待一定时间的线程，除了leader线程外，其他线程都一直休息，直到被通知唤醒
        private Thread leader;
    
    	// 从队列里取内容失败是需要在这个变量里等待
        private final Condition available = lock.newCondition();

可以看到只有四个成员变量，每个变量的作用已注释，大概可以猜想出 DelayQueue 是在 PriorityQueue(优先队列)的基础上封装了一层（1.要求队列内元素必须实现Delayed能力；2.控制线程安全和阻塞等待）

我们思路已经有了，首先得有排序的能力，保证延迟少的在队列首部，延迟大的在后面，这样才能保证每次取到的是最先要过期的元素，由于PriorityQueue已经实现排序的能力，因此我们直接使用这个类即可。

其次，我们需要保证线程安全，因为 PriorityQueue 是线程不安全的，我们便引入了Lock，只要在调用方法前进行加锁，调用后解锁即可。

最重要的是控制线程的等待和唤醒，而这个能力Condition已经实现，引入它即可。

最后，我们定义一个`Thread leader`，只需保证一个线程在等即可（等待时间为第一个元素的过期时间），即所有执行take或者poll方法的线程中仅仅一个来等指定时间，其他线程由这个 leader 线程执行`available.single()`来唤醒。

--------------------

## 具体实现 ##

思路我们已经有了，接下来看具体怎么实现的。

小白看源码时候特别容易翻车，why？很可能是因为太过于纠结细节，我们第一遍的时候没必要纠结每一行代码什么意思，跟着我思路走：先看常用的，如构造函数，put、take，其他的先忽略。  
**构造函数**  
没什么特别的，和其他容器类似，提供设置初始大小，通过已有容器创建等。

### put ###

往里放时候最常用的方法。

public void put(E e) { 
    	        offer(e);
    	}
    	// 发现它其实调用了 offer，来看 offer
    	
        public boolean offer(E e) { 
        // 先加锁
            final ReentrantLock lock = this.lock;
            lock.lock();
            try { 
            // 放进优先队列
                q.offer(e);
                // 因为优先队列 q 是有顺序的，因此放进去立马拿，不一定是放的元素
                if (q.peek() == e) { 
               	// 如果放进去取出来还是e，只有两种情况，见下文解释。
                    leader = null;
                    available.signal();
                }
                return true;
            } finally { 
            // 释放锁
                lock.unlock();
            }
        }

可以看到由于内部使用的其实是 PriorityQueue ，而PriorityQueue 是没有大小限制的（Integer的最大值），因此加元素的方法肯定不会阻塞，一定返回true，不信可以看带阻塞时间的offer，可以看到，等待时间参数被忽略了。

public boolean offer(E e, long timeout, TimeUnit unit) { 
            return offer(e);
        }

回到正题，放元素的方法我们只需要关注一个点，发现它在放入之后又看了一下队首的元素，有个if判断两者是否相等，我们知道 PriorityQueue 是自动排序的，因此放进去立马拿，不一定是刚刚放的元素。

if 条件成立有以下两种可能性：

*  放之前队列里是空的
 *  刚放入的 e 的过期时间是当前所有元素中最小的

第一情况下，就需要唤醒等待者，别等了，尝试拿走吧。

而第二种情况，刚放了又是最小的，说明不需要等待原来预计的等待时间了，将提前执行，而修改等待时间的代码实现在take中，本线程只需要告诉等着的线程一声，而不是帮它做了。

### take ###

public E take() throws InterruptedException { 
    		// 加锁
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try { 
            	// 死循环：不拿到元素，绝不罢休
                for (;;) { 
                	// 1. 先看看队列里有元素吗，没有就去歇着，等着别人来叫我我再干活
                    E first = q.peek();
                    if (first == null)
                        available.await();
                    else { 
                    // 2. 发现队列里有元素
                        long delay = first.getDelay(NANOSECONDS);
                        // 2.1 看看元素是不是过期了，过期了我就拿走了
                        if (delay <= 0L)
                            return q.poll();
                        first = null; // don't retain ref while waiting 既然没过期，我还不能拿，那我就不要它了。
                        // 2.2 看看是否有其他线程来管啥时候能拿不，有的话，我就直接一直休息就行了，直到有人通知我
                        if (leader != null)
                            available.await();
                        else { 
                        // 2.3 既然没有线程来负责这件事，那我就当leader，负责这件事
                            Thread thisThread = Thread.currentThread();
                            leader = thisThread;
                            try { 
                            	// 我不能和2.2那样的线程不负责，当前第一个任务过期了，我再来干活，休息一定的时间，而不是一直休息。
                                available.awaitNanos(delay);
                            } finally { 
                            	// 我都醒了，这次leader我就先让让，再次循环尝试拿咯
                                if (leader == thisThread)
                                    leader = null;
                            }
                        }
                    }
                }
                // 释放锁
            } finally { 
            	// 如果没有lead了，而且队列里还有要执行的，那我就通知一下其他等待的线程，别歇着了，来领任务了~
                if (leader == null && q.peek() != null)
                    available.signal();
                lock.unlock();
            }
        }

注释的比较详细了已经，主要分以下几种情况

*  队列中没有任何元素
    
     *  一直等着，等别人叫醒我
 *  队列里有元素，而且过期了
    
     *  既然可以拿走，我就拿走咯，return 返回~
 *  队列里有元素，但是还没过期
    
     *  leader已经有人当了
        
         *  一直等着，等别人叫醒我
     *  leader没人当
        
         *  我当leader

这里核心点主要是，等待者有两种角色，一种是一直等着，一种是leader，而leader有什么特别的呢？

## DelayQueue核心 ##

DelayQueue核心即划分了两种等待者的角色

*  普通等待者 线程
 *  leader 线程

普通的比较简单，一句话就能概括，一直等着，没人叫我我不醒。  
leader 角色相对来说负责一些，队列中第一个元素什么时候过期，我啥时候醒，当然也可能提前醒来：新加元素的过期时间比之前预计的醒来时间还早，这时候重新进入循环，尝试取元素。

--------------------

## 用法示例 ##

DelayQueue 思想和原理我们都懂了，那怎么用呢？  
留白等催更。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM1NDI1MDcw_size_16_color_FFFFFF_t_70]: /images/20230601/aadfd71e6b4144b3afc3d9503211fa3b.png