【JUC源码】阻塞队列：DelayQueue 源码分析

ゝ一纸荒年。 2022-12-11 06:24 220阅读 0赞

> 阻塞队列系列：
> 
>  *  [【JUC源码】阻塞队列：继承关系及接口分析][JUC]
>  *  [【JUC源码】阻塞队列：ArrayBlockingQueue 源码分析][JUC_ArrayBlockingQueue]
>  *  [【JUC源码】阻塞队列：LinkedBlockingQueue 源码分析][JUC_LinkedBlockingQueue]
>  *  [【JUC源码】阻塞队列：DelayQueue 源码分析][JUC_DelayQueue]
>  *  [【JUC源码】阻塞队列：SynchronousQueue 源码分析（非公平模式）][JUC_SynchronousQueue]
>  *  [【JUC源码】阻塞队列：SynchronousQueue 源码分析（公平模式）][JUC_SynchronousQueue 1]
>  *  [【JUC源码】阻塞队列：总结 --特性对比&应用场景举例][JUC_ --]
>  *  [【JUC源码】阻塞队列：各种阻塞队列面试题合集][JUC 1]

在看 DelayQueue 的源码实现前，可能很多同学还没有用过 DelayQueue，先来看个使用 DelayQueue 示例的把

public class DelayQueueDemo {
        
        
      @Data
      // 队列元素，实现了 Delayed 接口
      static class DelayedDTO implements Delayed{
        
        Long s;
        Long beginTime;
        public DelayedDTO(Long s,Long beginTime) {
        
          this.s = s;
          this.beginTime = beginTime;
        }
    
        // 重写getDelay,获取过期时间
        public long getDelay(TimeUnit unit) {
        
          return unit.convert(s - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
    
        // compareTo
        public int compareTo(Delayed o) {
        
          return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
        }
    
    
        public void run() {
        
            int sec = (int) ((System.currentTimeMillis() - beginTime) / 1000);
            System.out.println("延迟了" + sec +  "秒钟才执行");
        }
      }
        
        
      // 生产者
      static class Product implements Runnable {
        
        // DelayQueue是BlockQueue的实现类，因此这里使用接口更灵活
        private final BlockingQueue queue;
        public Product(BlockingQueue queue) {
        
          this.queue = queue;
        }
        
        @Override
        public void run() {
        
          try {
        
            System.out.println("begin put");
            long beginTime = System.currentTimeMillis();
            // 放入队列，延迟 2 秒执行
            queue.put(new DelayedDTO(System.currentTimeMillis() + 2000L,beginTime));
            // 延迟 5 秒执行
            queue.put(new DelayedDTO(System.currentTimeMillis() + 5000L,beginTime));
            // 延迟10秒执行
            queue.put(new DelayedDTO(System.currentTimeMillis() + 1000L * 10,beginTime));
            System.out.println("end put");
          } catch (InterruptedException e) {
        
              System.out.println("" + e);
          }
        }
      }
        
      // 消费者
      static class Consumer implements Runnable {
        
        private final BlockingQueue queue;
        public Consumer(BlockingQueue queue) {
        
          this.queue = queue;
        }
    
        @Override
        public void run() {
        
          try {
        
            System.out.println("Consumer begin");
            // 从队列中取，take 会阻塞
            ((DelayedDTO) queue.take()).run();
            ((DelayedDTO) queue.take()).run();
            ((DelayedDTO) queue.take()).run();
            System.out.println("Consumer end");
          } catch (InterruptedException e) {
        
              System.out.println("" + e);
          }
        }
      }
    
      // Main
      public static void main(String[] args) throws InterruptedException {
        
        BlockingQueue q = new DelayQueue();
        // 将delayQueue传入生产者与消费者
        DelayQueueDemo.Product p = new DelayQueueDemo.Product(q);
        DelayQueueDemo.Consumer c = new DelayQueueDemo.Consumer(q);
        new Thread(c).start();
        new Thread(p).start();
      }
    }

执行结果如下：

Consumer begin
    begin put
    end put
    延迟了2秒钟才执行
    延迟了5秒钟才执行
    延迟了10秒钟才执行
    Consumer end

可以看到，延迟队列就是，只有当到队列中的元素到达设定的过期时间后，队列中的元素元素才能被消费。下面我们就来看看它底层是怎么实现的。

## 1.DelayQueue 结构 ##

DelayQueue 继承关系，核心成员变量及主要构造函数：

// 队列中的元素都要实现Delayed接口，实现后每个实体对象就有过期时间了
    public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
    	implements BlockingQueue<E>{
        
    	
    	// 组合PriorityQueue进行队列操作
        private final PriorityQueue<E> q = new PriorityQueue<E>();
    	// reentrantLock保证线程安全
    	private final transient ReentrantLock lock = new ReentrantLock();
    	// 出队休眠Condition。其实很容易理解：当有线程来取元素，但队列中为空或者没有元素过期，所以要进入条件队列等待
    	private final Condition available = lock.newCondition();
        // 条件队列中剩余等待时间最短的线程，一般是先来的线程
        // 注：虽然引入了leader，但是也并非公平的（1.非公平锁 2.当新线程能获取元素时就直接返回了）
    	private Thread leader = null;
    	
    	//--------------------------------构造函数-------------------------------
    	public DelayQueue() {
        }
    		
        public DelayQueue(Collection<? extends E> c) {
        
            this.addAll(c);
        }
    	
    }

**Delayed 接口**

`<E extends Delayed>` 即队列中的所有元素都应该实现 Delayed 接口

public interface Delayed extends Comparable<Delayed> {
        
        long getDelay(TimeUnit unit);
    }

从上面的源码可以看出，Delayed 还继承了 Comparable 接口，所以队列中的元素应该实现以下两个方法：

*  getDelay：获取过期时间。所以这里一定要清楚，每个元素的过期时间是在放入队列前就确定的，而不是在放入队列时作为参数传进来。
 *  compareTo：用 getDelay 的过期时间对所有元素进行排序，是为了在 PriortyQueue 中进行上浮下沉比较用到

**PriorityQueue**

PriorityQueue 中文叫做优先级队列，在此处的作用就是可以根据过期时间做优先级排序，让先过期的可以先执行。具体请参考[【Java容器源码】PriorityQueue源码分析][Java_PriorityQueue]。

复用一直是个常见且重要的话题，就比如此处 DelayQueue 复用 PriorityQueue 的能力， 还有 LinkedHashMap 复用 HashMap 的能力，Set 复用 Map 的能力。

> PS：小结一下，如果想要复用需要做到哪些：
> 
>  *  需要把能遇见可复用的功能尽量抽象，并开放出可扩展的地方，比如说 HashMap 在操作数组的方法中，都给 LinkedHashMap >开放出很多 after 开头的方法，便于 LinkedHashMap 进行排序、删除等等；
>  *  采用组合或继承两种手段进行复用，比如 LinkedHashMap 采用的继承、 Set 和 DelayQueue 采用的组合，组合的意思就是把可复用的类给依赖进来。

## 2.入队 ##

### put() ###

public void put(E e) {
        
        	// 调用offer
            offer(e);
    }

### offer() ###

offer 的逻辑其实很简单

1.  调用 PriorityQueue 的 offer 方法将新元素添加进来。因为 PriorityQueue 的底层是**最小堆**，所以会对新元素进行上浮
2.  若这个新元素刚好成了队首（堆顶），那么就立刻随机唤醒一个等待的线程，然后删除现在的 leader，让新线程有机会成为 leader。
    
     *  随机唤醒一个线程的目的是重新休眠，不过，这次休眠时会设置休眠时间为该放入元素的过期时间
     *  删除当前 leader 是因为，leader 的线程必须是条件队列中剩余等待时间最短的线程（等待时间=队首过期时间），而现在已经有了等待时间更小的线程，所以，当前 leader 已经不行了。对这句话不理解的话，对比下面的take方法就一目了然了。

public boolean offer(E e) {
        
        final ReentrantLock lock = this.lock;
        lock.lock(); // 上锁
        try {
        
            // 直接调用 PriorityQueue 的 offer 方法，从而使用它的排序和扩容等能力
            q.offer(e);
            // 如果恰好刚放进去的元素正好在队列头
            // 立马唤醒一个在条件队列等待线程，并将现在leader置为null，让那个新唤醒的线程有机会成为leader
            if (q.peek() == e) {
        
                leader = null;
                available.signal();
            }
            return true;
        } finally {
        
            lock.unlock(); // 释放锁
        }
    }

## 3.出队 ##

### take() ###

take 方法的过程如下：

*  拿到锁的线程可以进入自旋
 *  取出队首元素 first，分为以下两种情况：
    
     *  情况 1：first=null，队列空，休眠当前线程
     *  情况 2：first != null，即队列中有元素，那么又要看该元素是否过期
        
         *  情况 2.1：过期时间 delay<0，即队头已经过期，可以取出，返回
         *  情况 2.2 ：队头没过期，那么就要将当前线程放入条件队列休眠等待，但这时又要考虑当前线程能否当leader
            
             *  情况 2.2.1：若当前线程是第一个来的线程就将其存入leader，然后定时休眠
             *  情况 2.2.2：已经有leader了，则当前线程直接休眠

> 两点注意：
> 
>  *  这里其实很好理解，因为如果所有线程都进入了阻塞了，并且没有一个计时，那么谁去唤醒？所以，必须有一个线程是计时休眠（计时且最短的线程是leader），待醒后拿到元素后，再去随机唤醒一个新线程
>  *  另外，从这就能清楚的看到 DelayQueue 是不公平的了，因为新线程一来就有可能拿到数据，而且 put 还会随机唤醒线程给它更短的休眠时间

public E take() throws InterruptedException {
        
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly(); // 加锁，可中断
            try {
        
                // 自旋 保证一定能成功
                for (;;) {
        
                    // 拿出队首first
                    E first = q.peek();
                    // 情况 1：first=null，队列为空，休眠当前线程
                    if (first == null)
                        available.await();
                    // 情况 2：队列非空
                    else {
        
                        long delay = first.getDelay(NANOSECONDS); // 获取队首元素（first）的过期时间delay
                        // 情况 2.1：first已经过期，则可以出队
                        if (delay <= 0)
                            return q.poll();
                        first = null; // 将first引用置为 null ，便于gc
                        // 情况 2.2：first未过期
                        // 情况 2.2.1：leader!=null，即在当前线程之前已经有线程等待取出了，所以直接休眠当前线程
                        if (leader != null)
                            available.await();
                        // 情况 2.2.2：leader == null，即当前线程当前线程是第一个要来取出的线程，存入leader
                        else {
        
                            Thread thisThread = Thread.currentThread();
                            leader = thisThread;
                            try {
        
                                // 给leader计时休眠，释放锁
                                // 到时自动醒来，大概率取得队首
                                available.awaitNanos(delay);
                            } finally {
        
                                if (leader == thisThread)
                                    leader = null;
                            }
                        }
                    }
                }
            } finally {
        
                if (leader == null && q.peek() != null)
                    available.signal();
                lock.unlock();  // 释放锁
            }
    }

### poll() ###

相较于 take，poll 会在队列为空或者队首未过期时直接返回 null，而不是阻塞在那里

public E poll() {
        
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
        
                E first = q.peek();
                // ！！！如果队列为空，或队首没过期，直接放回null
                if (first == null || first.getDelay(NANOSECONDS) > 0)
                    return null;
                else
                    return q.poll();
            } finally {
        
                lock.unlock();
            }
    }

另外，**如果不想一直无限阻塞，可以调用poll设置阻塞时间**  
![在这里插入图片描述][20200929023837933.png_pic_center]

## 4.获取队首：peek ##

public E peek() {
        
        	// 虽然只是查看队首，但也要获取锁，防止在查看时被修改
            final ReentrantLock lock = this.lock;
            lock.lock();
            try {
        
                // 调用priortyQueue的peek
                return q.peek();
            } finally {
        
                lock.unlock();
            }
    }

[JUC]: https://blog.csdn.net/weixin_43935927/article/details/113952868
[JUC_ArrayBlockingQueue]: https://blog.csdn.net/weixin_43935927/article/details/108833582
[JUC_LinkedBlockingQueue]: https://blog.csdn.net/weixin_43935927/article/details/108853656
[JUC_DelayQueue]: https://blog.csdn.net/weixin_43935927/article/details/108858129
[JUC_SynchronousQueue]: https://blog.csdn.net/weixin_43935927/article/details/108866902
[JUC_SynchronousQueue 1]: https://blog.csdn.net/weixin_43935927/article/details/113964294
[JUC_ --]: https://blog.csdn.net/weixin_43935927/article/details/108888294
[JUC 1]: https://blog.csdn.net/weixin_43935927/article/details/108857735
[Java_PriorityQueue]: https://blog.csdn.net/weixin_43935927/article/details/108858762
[20200929023837933.png_pic_center]: /images/20221123/d003a68709f64cb8878bab26e2a2ba3e.png