Jdk的延时队列

Posted 2021-11-05 好大的月亮

工作上一般能用jdk自带功能的直接拿来用就可以，以免出现各种bug

DelayQueue
JDK 中提供了一组实现延迟队列的API，位于Java.util.concurrent包下DelayQueue。
DelayQueue是一个BlockingQueue（无界阻塞）队列，它本质就是封装了一个PriorityQueue（优先队列），PriorityQueue内部使用完全二叉堆（来实现队列元素排序，我们在向DelayQueue队列中添加元素时，会给元素一个Delay（延迟时间）作为排序条件，队列中最小的元素会优先放在队首。队列中的元素只有到了Delay时间才允许从队列中取出。队列中可以放基本数据类型或自定义实体类，在存放基本数据类型时，优先队列中元素默认升序排列，自定义实体类就需要我们根据类属性值比较计算了。
先简单实现一下看看效果，添加三个demo入队DelayQueue，分别设置在当前时间的5秒、10秒、15秒后取消。

整体思路和实现Runnable/Callable接口差不多，先定义自己的实体类，然后塞到队列里执行

定义一个实现Delayed接口的实体类

package com.fchan.mq.jdkDelay;

import com.fasterxml.jackson.annotation.JsonFormat;

import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;


public class MyDelay implements Delayed {

    /**
     * 延迟时间
     */
    @JsonFormat(locale = "zh", timezone = "GMT+8", pattern = "yyyy-MM-dd HH:mm:ss")
    private Long time;
    String name;


    public MyDelay(String name, Long time, TimeUnit unit) {
        this.name = name;
        this.time = System.currentTimeMillis() + (time > 0 ? unit.toMillis(time) : 0);
    }

    public Long getTime() {
        return time;
    }

    public String getName() {
        return name;
    }

    //获取延时时间
    @Override
    public long getDelay(TimeUnit unit) {
        return time - System.currentTimeMillis();
    }

    //对延时队列中的元素进行排序
    @Override
    public int compareTo(Delayed o) {
        MyDelay myDelay = ((MyDelay) o);
        return this.time.compareTo(myDelay.getTime());
    }
}

将延时任务塞进队列

package com.fchan.mq.jdkDelay;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.TimeUnit;

public class MyDelayDemo {

    public static void main(String[] args) throws InterruptedException {
        MyDelay myDelay1 = new MyDelay("MyDelay1", 5L, TimeUnit.SECONDS);
        MyDelay myDelay2 = new MyDelay("MyDelay2", 10L, TimeUnit.SECONDS);
        MyDelay myDelay3 = new MyDelay("MyDelay3", 15L, TimeUnit.SECONDS);

        DelayQueue<MyDelay> delayQueue = new DelayQueue<MyDelay> ();
        //add 和 put 后面都是调用的offer方法，内部使用了ReentrantLock
        //delayQueue.put();
        delayQueue.add(myDelay1);
        delayQueue.add(myDelay2);
        delayQueue.add(myDelay3);

        System.out.println("订单延迟队列开始时间:" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
        while (delayQueue.size() != 0) {
            /**
             * 取队列头部元素是否过期
             */
            //DelayQueue的put/add方法是线程安全的，因为put/add方法内部使用了ReentrantLock锁进行线程同步。
            // DelayQueue还提供了两种出队的方法 poll() 和 take() ，
            // poll() 为非阻塞获取，没有到期的元素直接返回null；
            // take() 阻塞方式获取，没有到期的元素线程将会等待。
            MyDelay task = delayQueue.poll();
            if (task != null) {
                System.out.format("任务:{%s}被取消, 取消时间:{%s}\\n", task.name, LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
            }
            Thread.sleep(1000);
        }


    }
}

参考了大佬的博文
https://juejin.cn/post/6844904150703013901