2023-08-15  阅读(12)
原文作者:Ressmix 原文地址:https://www.tpvlog.com/article/39

一、DelayQueue简介

DelayQueue是JDK1.5时,随着J.U.C包一起引入的一种阻塞队列,它实现了BlockingQueue接口,底层基于已有的 PriorityBlockingQueue 实现:

202308152204494461.png

DelayQueue也是一种比较特殊的阻塞队列,从类声明也可以看出,DelayQueue中的所有元素必须实现Delayed接口:

    /**
     * 一种混合风格的接口,用来标记那些应该在给定延迟时间之后执行的对象。
     * <p>
     * 此接口的实现必须定义一个 compareTo 方法,该方法提供与此接口的 getDelay 方法一致的排序。
     */
    public interface Delayed extends Comparable<Delayed> {
    
        /**
         * 返回与此对象相关的剩余有效时间,以给定的时间单位表示.
         */
        long getDelay(TimeUnit unit);
    }

可以看到,Delayed接口除了自身的getDelay方法外,还实现了 Comparable 接口。getDelay方法用于返回对象的剩余有效时间,实现Comparable接口则是为了能够比较两个对象,以便排序。

也就是说,如果一个类实现了Delayed接口,当创建该类的对象并添加到DelayQueue中后, 只有当该对象的getDalay方法返回的剩余时间≤0时才会出队

另外,由于DelayQueue内部委托了PriorityBlockingQueue对象来实现所有方法,所以能以堆的结构维护元素顺序,这样剩余时间最小的元素就在堆顶, 每次出队其实就是删除剩余时间≤0的最小元素

DelayQueue的特点简要概括如下:

  1. DelayQueue是无界阻塞队列;
  2. 队列中的元素必须实现Delayed接口,元素过期后才会从队列中取走;

二、DelayQueue示例

为了便于理解DelayQueue的功能,我们先来看一个使用DelayQueue的示例。

队列元素

第一节说了,队列元素必须实现Delayed接口,我们先来定义一个 Data 类,作为队列元素:

    public class Data implements Delayed {
        private static final AtomicLong atomic = new AtomicLong(0);
        private static final DateTimeFormatter formatter = DateTimeFormatter.ofPattern("HH:mm:ss-n");
    
        // 数据的失效时间点
        private final long time;
    
        // 序号
        private final long seqno;
    
        /**
         * @param deadline 数据失效时间点
         */
        public Data(long deadline) {
            this.time = deadline;
            this.seqno = atomic.getAndIncrement();
        }
    
        /**
         * 返回剩余有效时间
         *
         * @param unit 时间单位
         */
        @Override
        public long getDelay(TimeUnit unit) {
            return unit.convert(this.time - System.nanoTime(), TimeUnit.NANOSECONDS);
        }
    
        /**
         * 比较两个Delayed对象的大小, 比较顺序如下:
         * 1. 如果是对象本身, 返回0;
         * 2. 比较失效时间点, 先失效的返回-1,后失效的返回1;
         * 3. 比较元素序号, 序号小的返回-1, 否则返回1.
         * 4. 非Data类型元素, 比较剩余有效时间, 剩余有效时间小的返回-1,大的返回1,相同返回0
         */
        @Override
        public int compareTo(Delayed other) {
            if (other == this)  // compare zero if same object
                return 0;
    
            if (other instanceof Data) {
                Data x = (Data) other;
    
                // 优先比较失效时间
                long diff = this.time - x.time;
                if (diff < 0)
                    return -1;
                else if (diff > 0)
                    return 1;
    
                else if (this.seqno < x.seqno)    // 剩余时间相同则比较序号
                    return -1;
                else
                    return 1;
            }
    
            // 一般不会执行到此处,除非元素不是Data类型
            long diff = this.getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS);
            return (diff < 0) ? -1 : (diff > 0) ? 1 : 0;
        }
    
        @Override
        public String toString() {
            return "Data{" +
                "time=" + time +
                ", seqno=" + seqno +
                "}, isValid=" + isValid();
        }
    
        private boolean isValid() {
            return this.getDelay(TimeUnit.NANOSECONDS) > 0;
        }
    }

关于队列元素 Data 类,需要注意以下几点:

  1. 每个元素的time字段保存失效时间点)的纳秒形式(构造时指定,比如当前时间+60s);
  2. seqno字段表示元素序号,每个元素唯一,仅用于失效时间点一致的元素之间的比较。
  3. getDelay方法返回元素的剩余有效时间,可以根据入参的 TimeUnit 选择时间的表示形式(秒、微妙、纳秒等),一般选择纳秒以提高精度;
  4. compareTo方法用于比较两个元素的大小,以便在队列中排序。由于DelayQueue基于优先级队列实现,所以内部是“堆”的形式,我们定义的规则是先失效的元素将先出队,所以先失效元素应该在堆顶,即compareTo方法返回结果<0的元素优先出队;

生产者-消费者

还是以“生产者-消费者”模式来作为DelayQueued的示例:

生产者

    public class Producer implements Runnable {
        private final DelayQueue<Data> queue;
    
        public Producer(DelayQueue<Data> queue) {
            this.queue = queue;
        }
    
        @Override
        public void run() {
            while (true) {
    
                long currentTime = System.nanoTime();
                long validTime = ThreadLocalRandom.current().nextLong(1000000000L, 7000000000L);
    
                Data data = new Data(currentTime + validTime);
                queue.put(data);
    
                System.out.println(Thread.currentThread().getName() + ": put " + data);
    
                try {
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

消费者

    public class Consumer implements Runnable {
        private final DelayQueue<Data> queue;
    
        public Consumer(DelayQueue<Data> queue) {
            this.queue = queue;
        }
    
        @Override
        public void run() {
            while (true) {
                try {
                    Data data = queue.take();
                    System.out.println(Thread.currentThread().getName() + ": take " + data);
    
                    Thread.yield();
    
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

调用

    public class Main {
        public static void main(String[] args) {
            DelayQueue<Data> queue = new DelayQueue<>();
    
            Thread c1 = new Thread(new Consumer(queue), "consumer-1");
            Thread p1 = new Thread(new Producer(queue), "producer-1");
    
            c1.start();
            p1.start();
        }
    }

执行结果:

producer-1: put Data{time=73262562161592, seqno=0}, isValid=true
producer-1: put Data{time=73262787192726, seqno=1}, isValid=true
producer-1: put Data{time=73265591291171, seqno=2}, isValid=true
producer-1: put Data{time=73266850330909, seqno=3}, isValid=true
consumer-1: take Data{time=73262562161592, seqno=0}, isValid=false
consumer-1: take Data{time=73262787192726, seqno=1}, isValid=false
producer-1: put Data{time=73267928737184, seqno=4}, isValid=true
producer-1: put Data{time=73265083111776, seqno=5}, isValid=true
producer-1: put Data{time=73268729942809, seqno=6}, isValid=true
consumer-1: take Data{time=73265083111776, seqno=5}, isValid=false

上面示例中,我们创建了一个生产者,一个消费者,生产者不断得入队元素,每个元素都会有个截止有效期;消费者不断得从队列者获取元素。从输出可以看出,消费者每次获取到的元素都是有效期最小的,且都是已经失效了的。(因为DelayQueue每次出队只会删除有效期最小且已经过期的元素)

三、DelayQueue原理

介绍完了DelayQueued的基本使用,读者应该对该阻塞队列的功能有了基本了解,接下来我们看下Doug Lea是如何实现DelayQueued的。

构造

DelayQueued提供了两种构造器,都非常简单:

    /**
     * 默认构造器.
     */
    public DelayQueue() {
    }
    /**
     * 从已有集合构造队列.
     */
    public DelayQueue(Collection<? extends E> c) {
        this.addAll(c);
    }

可以看到,内部的 PriorityQueue 并非在构造时创建,而是对象创建时生成:

    public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
        implements BlockingQueue<E> {
    
        private final transient ReentrantLock lock = new ReentrantLock();
        private final PriorityQueue<E> q = new PriorityQueue<E>();
    
        /**
         * leader线程是首个尝试出队元素(队列不为空)但被阻塞的线程.
         * 该线程会限时等待(队首元素的剩余有效时间),用于唤醒其它等待线程
         */
        private Thread leader = null;
    
        /**
         * 出队线程条件队列, 当有多个线程, 会在此条件队列上等待.
         */
        private final Condition available = lock.newCondition();
    
        //...
    
    }

上述比较特殊的是leader字段,我们之前已经说过,DelayQueue每次只会出队一个过期的元素,如果队首元素没有过期,就会阻塞出队线程,让线程在available这个条件队列上无限等待。

为了提升性能,DelayQueue并不会让所有出队线程都无限等待,而是用leader保存了第一个尝试出队的线程,该线程的等待时间是队首元素的剩余有效期。这样,一旦leader线程被唤醒(此时队首元素也失效了),就可以出队成功,然后唤醒一个其它在available条件队列上等待的线程。之后,会重复上一步,新唤醒的线程可能取代成为新的leader线程。这样,就避免了无效的等待,提升了性能。这其实是一种名为“Leader-Follower pattern”的多线程设计模式。


入队——put

put 方法没有什么特别,由于是无界队列,所以也不会阻塞线程。

    /**
     * 入队一个指定元素e.
     * 由于是无界队列, 所以该方法并不会阻塞线程.
     */
    public void put(E e) {
        offer(e);
    }
    
    public boolean offer(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            q.offer(e);             // 调用PriorityQueue的offer方法
            if (q.peek() == e) {    // 如果入队元素在队首, 则唤醒一个出队线程
                leader = null;
                available.signal();
            }
            return true;
        } finally {
            lock.unlock();
        }
    }

需要注意的是当首次入队元素时,需要唤醒一个出队线程,因为此时可能已有出队线程在空队列上等待了,如果不唤醒,会导致出队线程永远无法执行。

    if (q.peek() == e) {    // 如果入队元素在队首, 则唤醒一个出队线程
        leader = null;
        available.signal();
    }
    

出队——take

整个 take 方法在一个自旋中完成,其实就分为两种情况:

1.队列为空

这种情况直接阻塞出队线程。(在available条件队列等待)

2.队列非空

队列非空时,还要看队首元素的状态(有效期),如果队首元素过期了,那直接出队就行了;如果队首元素未过期,就要看当前线程是否是第一个到达的出队线程(即判断leader是否为空),如果不是,就无限等待,如果是,则限时等待。

    /**
     * 队首出队元素.
     * 如果队首元素(堆顶)未到期或队列为空, 则阻塞线程.
     */
    public E take() throws InterruptedException {
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try {
            for (; ; ) {
                E first = q.peek();     // 读取队首元素
                if (first == null)      // CASE1: 队列为空, 直接阻塞
                    available.await();
                else {                  // CASE2: 队列非空
                    long delay = first.getDelay(NANOSECONDS);
                    if (delay <= 0)                             // CASE2.0: 队首元素已过期
                        return q.poll();
    
                    // 执行到此处说明队列非空, 且队首元素未过期
                    first = null;
                    if (leader != null)                         // CASE2.1: 已存在leader线程
                        available.await();      // 无限期阻塞当前线程
                    else {                                      // CASE2.2: 不存在leader线程
                        Thread thisThread = Thread.currentThread();
                        leader = thisThread;    // 将当前线程置为leader线程
                        try {
                            available.awaitNanos(delay);        // 阻塞当前线程(限时等待剩余有效时间)
                        } finally {
                            if (leader == thisThread)
                                leader = null;
                        }
                    }
                }
            }
        } finally {
            if (leader == null && q.peek() != null)             // 不存在leader线程, 则唤醒一个其它出队线程
                available.signal();
            lock.unlock();
        }
    }

需要注意,自旋结束后如果leader == null && q.peek() != null,需要唤醒一个等待中的出队线程。
leader == null && q.peek() != null的含义就是——没有leader线程但队列中存在元素。我们之前说了,leader线程作用之一就是用来唤醒其它无限等待的线程,所以必须要有这个判断。

四、总结

DelayQueue 是阻塞队列中非常有用的一种队列,经常被用于缓存或定时任务等的设计。

考虑一种使用场景:

异步通知的重试,在很多系统中,当用户完成服务调用后,系统有时需要将结果异步通知到用户的某个URI。由于网络等原因,很多时候会通知失败,这个时候就需要一种重试机制。

这时可以用DelayQueue保存通知失败的请求,失效时间可以根据已通知的次数来设定(比如:2s、5s、10s、20s),这样每次从队列中take获取的就是剩余时间最短的请求,如果已重复通知次数超过一定阈值,则可以把消息抛弃。

后面,我们在讲J.U.C之executors框架的时候,还会再次看到DelayQueue的身影。JUC线程池框架中的ScheduledThreadPoolExecutor.DelayedWorkQueue就是一种延时阻塞队列。


Java 面试宝典是大明哥全力打造的 Java 精品面试题,它是一份靠谱、强大、详细、经典的 Java 后端面试宝典。它不仅仅只是一道道面试题,而是一套完整的 Java 知识体系,一套你 Java 知识点的扫盲贴。

它的内容包括:

  • 大厂真题:Java 面试宝典里面的题目都是最近几年的高频的大厂面试真题。
  • 原创内容:Java 面试宝典内容全部都是大明哥原创,内容全面且通俗易懂,回答部分可以直接作为面试回答内容。
  • 持续更新:一次购买,永久有效。大明哥会持续更新 3+ 年,累计更新 1000+,宝典会不断迭代更新,保证最新、最全面。
  • 覆盖全面:本宝典累计更新 1000+,从 Java 入门到 Java 架构的高频面试题,实现 360° 全覆盖。
  • 不止面试:内容包含面试题解析、内容详解、知识扩展,它不仅仅只是一份面试题,更是一套完整的 Java 知识体系。
  • 宝典详情:https://www.yuque.com/chenssy/sike-java/xvlo920axlp7sf4k
  • 宝典总览:https://www.yuque.com/chenssy/sike-java/yogsehzntzgp4ly1
  • 宝典进展:https://www.yuque.com/chenssy/sike-java/en9ned7loo47z5aw

目前 Java 面试宝典累计更新 400+ 道,总字数 42w+。大明哥还在持续更新中,下图是大明哥在 2024-12 月份的更新情况:

想了解详情的小伙伴,扫描下面二维码加大明哥微信【daming091】咨询

同时,大明哥也整理一套目前市面最常见的热点面试题。微信搜[大明哥聊 Java]或扫描下方二维码关注大明哥的原创公众号[大明哥聊 Java] ,回复【面试题】 即可免费领取。

阅读全文