对于DelayQueue的理解

Posted 2022-03-09 wangpipi

今天看公司代码，发现里面使用了 DelayQueue，学习以后记录下来：

概念：DelayQueue是一个支持延时获取元素的无界阻塞队列。里面的元素全部都是“可延期”的元素，列头的元素是最先“到期”的元素，如果队列里面没有元素到期，是不能从列头获取元素的，哪怕有元素也不行。也就是说只有在延迟期到时才能够从队列中取元素。

我理解为：延迟队列用于需要延迟处理的场景：比如延迟会话关闭，如果某一会话1分钟后需要关闭，则可以使用延迟队列，再比如：订单超时取消

例子场景：订单超时

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
我们看到队列泛型需要继承 Delayed 接口。创建一个实体类，实现此接口

创建实体类：  OrderInfo

 1 package wyp.delayqueue;
 2 
 3 import lombok.Data;
 4 
 5 import java.io.Serializable;
 6 import java.text.ParseException;
 7 import java.text.SimpleDateFormat;
 8 import java.util.concurrent.Delayed;
 9 import java.util.concurrent.TimeUnit;
10 
11 /**
12  * @author : miles wang
13  * @date : 2019/7/2  3:39 PM
14  * DelayQueue<T>
15  *      延迟队列的泛型必须实现 Delayed接口
16  */
17 @Data
18 public class OrderInfo implements Serializable , Delayed 
19     private static final long serialVersionUID = 1L;
20     private String orderNo;// 订单号
21     private String status;// 订单状态
22     private String expTime;// 订单过期时间
23     private String createTime;//订单创建时间
24 
25     /**
26      * 用于延时队列内部比较排序：当前订单的过期时间 与 队列中对象的过期时间 比较
27      * 排序方法：
28      * 我理解为：从延迟队列中取出过期元素的顺序，就是由此排序方法控制
29      */
30     @Override
31     public int compareTo(Delayed o) 
32         SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
33         long nowThreadtime = 0;
34         long queueThreadtime = 0;
35         try 
36             nowThreadtime = formatter.parse(this.expTime).getTime();
37             queueThreadtime = formatter.parse(((OrderInfo)o).expTime).getTime();
38          catch (ParseException e) 
39             e.printStackTrace();
40         
41         return Long.valueOf(nowThreadtime).compareTo(Long.valueOf(queueThreadtime));
42     
43 
44 
45     /**
46      * 时间单位：秒
47      * 延迟关闭时间 = 过期时间 - 当前时间
48      * 跟你官方提供的DOC文档，我理解为：
49      * 当此方法的返回值为负数的时候，表示次方法可以从延迟队列中移出
50      * 然后此时就可以接受一系列的消费者处理
51      */
52     @Override
53     public long getDelay(TimeUnit unit) 
54         SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
55         long time = 0;
56         try 
57             time = formatter.parse(this.expTime).getTime();
58          catch (ParseException e) 
59             e.printStackTrace();
60         
61         return time - System.currentTimeMillis();
62     
63 
64 
65 

创建延迟队列监听类

package wyp.delayqueue;

import java.util.Objects;
import java.util.concurrent.DelayQueue;

/**
 * @author : miles wang
 * @date : 2019/7/2  3:42 PM
 *
 * 使用延时队列DelayQueue实现订单超时关闭
 * 后台守护线程不断的执行检测工作
 * 双检查模式实现单例模式
 * 后面我把线程池改为了单线程，所有是否为单利模式不重要
 */
public class OrderOverTimeClose  

    private volatile static OrderOverTimeClose oderOverTimeClose = null;

    private OrderOverTimeClose() 

    

    /**
     * 守护线程
     */
    private Thread mainThread;

    /**
     * 创建空延时队列
     */
    private  DelayQueue<OrderInfo> queue = new DelayQueue<OrderInfo>();

    /**
     * 单例模式，双检查锁模式，在并发环境下对象只被初始化一次
     */
    public static OrderOverTimeClose getInstance()
        if(oderOverTimeClose == null )
            synchronized(OrderOverTimeClose.class)
                oderOverTimeClose =  new OrderOverTimeClose();
            
        
        return oderOverTimeClose;
    

    /**
     * 启动方法
     */
    public void init()
        mainThread =  new Thread(()->execute());
        mainThread.setDaemon(true);
        mainThread.setName("守护线程-->");
        mainThread.start();
    

    private void execute() 
        while (true) 
            try 
                if(queue.size() > 0)
                    //从队列里获取超时的订单
                    OrderInfo orderInfo = queue.take();
                    // 检查订单状态，是否已经成功，成功则将订单从队列中删除。
                    if (Objects.equals(orderInfo.getStatus(), "成功")) 
                        System.out.println("线程："+Thread.currentThread().getName()+"，订单号："
                                + orderInfo.getOrderNo() + "，订单状态："
                                + orderInfo.getStatus() + "，订单创建时间："
                                + orderInfo.getCreateTime()
                                + "，订单超时时间：" + orderInfo.getExpTime()+"，当前时间："+OrderPay.getTime(System.currentTimeMillis()));
                        Thread.sleep(2000);
                     else 
                        System.out.println("线程："+Thread.currentThread().getName()+"，订单号："
                                + orderInfo.getOrderNo() + "，变更订单状态为：超时关闭"
                                + "，订单创建时间："
                                + orderInfo.getCreateTime()
                                + "，订单超时时间：" + orderInfo.getExpTime()+"，当前时间："+OrderPay.getTime(System.currentTimeMillis()));
                        Thread.sleep(2000);
                    
                
             catch (InterruptedException e) 
                e.printStackTrace();
            
        
    

    /**
     * 插入订单到超时队列中
     */
    public void orderPutQueue(OrderInfo orderInfo, String createTime,
                              String overTime) 
        System.out.println("订单号：" + orderInfo.getOrderNo() + "，订单创建时间："
                + createTime + "，订单过期时间：" + overTime);
      //  queue.add(orderInfo);
          queue.put(orderInfo);
    

测试 主要是向延迟队列中插入元素

package wyp.delayqueue;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.Random;

/**
 * @author : miles wang
 * @date : 2019/7/2  3:46 PM
 */
public class OrderPay 
    static String[] str = new String[]"成功","支付中","订单初始化";

    public static String getTime(long time)
        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
        Date date = new Date(time);
        String currentTime = formatter.format(date);
        return currentTime;
    

    public static void main(String[] args) throws InterruptedException 
        OrderOverTimeClose.getInstance().init();
        for (int i = 0; i < 20; i++) 
            // 创建初始订单
            long createTime = System.currentTimeMillis();
            String currentTime = getTime(createTime);
            String overTime = getTime(createTime + 10000);// 十秒后超时
            String orderNo = String.valueOf(new Random().nextLong());
            OrderInfo order = new OrderInfo();
            order.setOrderNo(orderNo);
            order.setExpTime(overTime);
            int random_index = (int) (Math.random()*str.length);
            order.setStatus(str[random_index]);// 随机分配
            order.setCreateTime(currentTime);
            OrderOverTimeClose.getInstance().orderPutQueue(order, currentTime, overTime);
            try 
                Thread.sleep(3000);
             catch (InterruptedException e) 
                e.printStackTrace();
            
        
    

以上仅仅是个人学习测试理解和记录，不一定对。