今天看公司代码，发现里面使用了 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();
}
}
}
}
以上仅仅是个人学习测试理解和记录，不一定对。
原文地址：https://www.cnblogs.com/wangpipi/p/11122416.html
时间： 2024-08-30 16:34:32