使用延时队列DelayQueue实现订单超时关闭
2018-02-07 本文已影响0人
双城漂泊
DelayQueue是什么？
DelayQueue 实现了 BlockingQueue 接口，是一个无界的、线程安全的队列，持有的对象需要实现Delayed接口，重写compareTo方法与getDelay方法。
DelayQueue 延时队列试用那些场景？
1、商城订单超时关闭
2、支付结果异步通知功能
3、异步短信发送功能
/**
* 订单对象
*/
public class OrderInfo implements Serializable, Delayed {
private static final long serialVersionUID = 1L;
private String orderNo;// 订单号
private String status;// 订单状态
private String expTime;// 订单过期时间
private String createTime;//订单创建时间
/**
* 用于延时队列内部比较排序：当前订单的过期时间 与 队列中对象的过期时间 比较
*/
@Override
public int compareTo(Delayed o) {
SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
long nowThreadtime = 0;
long queueThreadtime = 0;
try {
nowThreadtime = formatter.parse(this.expTime).getTime();
queueThreadtime = formatter.parse(((OrderInfo)o).expTime).getTime();
} catch (ParseException e) {
e.printStackTrace();
}
return Long.valueOf(nowThreadtime).compareTo(Long.valueOf(queueThreadtime));
}
/**
* 时间单位：秒
* 延迟关闭时间 = 过期时间 - 当前时间
*/
@Override
public long getDelay(TimeUnit unit) {
SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
long time = 0;
try {
time = formatter.parse(this.expTime).getTime();
} catch (ParseException e) {
e.printStackTrace();
}
return time - System.currentTimeMillis();
}
public String getOrderNo() {
return orderNo;
}
public void setOrderNo(String orderNo) {
this.orderNo = orderNo;
}
public String getExpTime() {
return expTime;
}
public void setExpTime(String overTime) {
this.expTime = overTime;
}
public String getStatus() {
return status;
}
public void setStatus(String status) {
this.status = status;
}
public String getCreateTime() {
return createTime;
}
public void setCreateTime(String createTime) {
this.createTime = createTime;
}
}
/**
* [使用延时队列DelayQueue实现订单超时关闭]
* [后台守护线程不断的执行检测工作]
* [双检查模式实现单例模式]
*/
public class OrderOverTimeClose {
private volatile static OrderOverTimeClose oderOverTimeClose = null;
private OrderOverTimeClose() {
}
/**
* 单例模式，双检查锁模式，在并发环境下对象只被初始化一次
*/
public static OrderOverTimeClose getInstance(){
if(oderOverTimeClose == null ){
synchronized(OrderOverTimeClose.class){
oderOverTimeClose = new OrderOverTimeClose();
}
}
return oderOverTimeClose;
}
/**
* 守护线程
*/
private Thread mainThread;
/**
* 启动方法
*/
public void init(){
mainThread = new Thread(() -> execute());
mainThread.setDaemon(true);
mainThread.setName("守护线程-->");
mainThread.start();
}
/**
* 创建空延时队列
*/
private DelayQueue<OrderInfo> queue = new DelayQueue<OrderInfo>();
/**
* 读取延时队列，关闭超时订单
*/
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);
}
}
/**
* 模拟订单支付，创建订单
*/
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();
ExecutorService service = Executors.newFixedThreadPool(10);
for (int i = 0; i < 20; i++) {
Runnable run = new Runnable() {
@Override
public void run() {
// 创建初始订单
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);
}
};
service.execute(run);
try {
Thread.sleep(3000);
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
}
