Java通过Delayed延时队列来实现关闭已超时的订单

青旅半醒 2022-11-15 01:17 283阅读 0赞

## 1、定义延时订单类 ##

定义唯一的订单编号、超时时间、类型等字段  
类需要实现Delayed, Serializable 接口

import lombok.Data;
    
    import java.io.Serializable;
    import java.util.Date;
    import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    
    /** * @program: bike-lease * @description: 订单延时处理vo * @author: Jack.Fang * @date:2021-01-13 1101 **/
    @Data
    public class OrderDelayedVo implements Delayed, Serializable { 
    
        /** * 订单号（主键） */
        private String orderNo;
    
        /** * 订单种类（1-北斗 2-春藤 3-申城动力 4-智租） */
        private Integer orderCategory;
    
        /** * 订单过期时间（毫秒） */
        private long expTime;
    
        @Override
        public long getDelay(TimeUnit unit) { 
            return unit.convert(this.expTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
        }
    
        @Override
        public int compareTo(Delayed o) { 
            return Long.valueOf(this.expTime).compareTo(Long.valueOf(((OrderDelayedVo) o).expTime));
        }
    
        /** * 重写用于删除 */
        @Override
        public boolean equals(Object obj) { 
            if (this == obj) { 
                return true;
            }
            if (obj == null) { 
                return false;
            }
            OrderDelayedVo order = (OrderDelayedVo) obj;
            if (order.getOrderNo() == null) { 
                return false;
            } else if (!orderNo.equals(order.getOrderNo())) { 
                return false;
            }
            return true;
        }
    }

## 2、处理超时订单实现类 ##

实现了：  
1、启动项目时，查询已经超时的订单并关闭。  
2、查询未超时的订单到队列中。  
3、定义线程池去根据处理这些订单队列。

import com.bdxh.core.commons.utils.DateUtil;
    import com.bdxh.core.json.resp.BaseResp;
    import com.bdxh.userlease.entity.UserOrder;
    import com.bdxh.userleaseapi.restservice.UserOrderRestService;
    import com.bdxh.userleaseapi.vo.OrderDelayedVo;
    import com.google.common.util.concurrent.ThreadFactoryBuilder;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.InitializingBean;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Service;
    import org.springframework.util.CollectionUtils;
    
    import java.text.SimpleDateFormat;
    import java.util.ArrayList;
    import java.util.Date;
    import java.util.List;
    import java.util.concurrent.*;
    import java.util.concurrent.atomic.AtomicBoolean;
    
    /** * @program: bike-lease * @description: 用户订单延时服务类 * @author: Jack.Fang * @date:2021-01-13 1118 **/
    @Slf4j
    @Service
    public class UserOrderDelayService implements InitializingBean { 
    
        @Autowired
        private UserOrderRestService userOrderRestService;
        @Autowired
        private CTControlService ctControlService;
    
        private static DelayQueue<OrderDelayedVo> delayQueue = new DelayQueue<OrderDelayedVo>();
        private static AtomicBoolean start = new AtomicBoolean(false);
        // 线程池
        private static ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("userOrderDelay-pool-%d").build();
        private static ExecutorService executorService = new ThreadPoolExecutor(
                3, // 核心线程池大小
                6, // 最大线程池大小
                60L, // 线程最大空闲时间
                TimeUnit.MILLISECONDS, // 时间单位(毫秒)
                new LinkedBlockingQueue<Runnable>(),  // 线程等待队列
                threadFactory, // 线程创建工厂
                new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
        );
    
        @Override
        public void afterPropertiesSet() throws Exception { 
            start();
        }
    
        public <T> void start() { 
            if (start.get()) { 
                return;
            }
            start.set(true);
            //处理已超时订单
            List<UserOrder> dealList = userOrderRestService.dealOverTimeUserOrder();
            if(!CollectionUtils.isEmpty(dealList)){ 
                for(UserOrder userOrder:dealList){ 
                    //通知春藤关闭订单
                    if(UserOrder.OrderCategory.chunteng.getValue().equals(userOrder.getOrderCategory())){ 
                        BaseResp baseResp = ctControlService.setOrderStateInvalid(userOrder.getOrderNo());
                        log.info("关闭xx超时订单====>> 订单号：{},是否调用成功：{}",userOrder.getOrderNo(),baseResp.isRs());
                    }
                }
                log.info("已处理超时订单共{}条",dealList.size());
            }
    
            //查询未超时订单 放入队列
            List<UserOrder> orderList = userOrderRestService.getOverTimeUserOrder();
            if(!CollectionUtils.isEmpty(orderList)){ 
                List<OrderDelayedVo> list = new ArrayList<>();
                for(UserOrder userOrder:orderList){ 
                    OrderDelayedVo vo = new OrderDelayedVo();
                    vo.setOrderNo(userOrder.getOrderNo());
                    vo.setOrderCategory(userOrder.getOrderCategory());
                    vo.setExpTime(userOrder.getAddTime().getTime() + (30 * 60 * 1000));//超时时间:30分钟后
                    list.add(vo);
                }
                log.info("加入超时订单队列，共{}条。",list.size());
                list.forEach(item -> delayQueue.add(item));
            }
            executorService.submit(() -> { 
                // 取消订单
                while (start.get()) { 
                    try { 
                        OrderDelayedVo order = delayQueue.take();
                        if (order != null) { 
                            // 处理订单过期
                            boolean flag = userOrderRestService.updateOverTimeUserOrder(order.getOrderNo());
                            if(flag && UserOrder.OrderCategory.chunteng.getValue().equals(order.getOrderCategory())){ 
                                BaseResp baseResp = ctControlService.setOrderStateInvalid(order.getOrderNo());
                                log.info("超时订单延时队列====>>> 【通知xx关闭超时订单】====>> 订单号：{},是否成功调用:{}",order.getOrderNo(),baseResp.isRs());
                            }else{ 
                                log.info("超时订单延时队列====>>> 订单号为[{}],是否执行了关闭操作:{}。",order.getOrderNo(),flag);
                            }
                        }
                    } catch (Exception e) { 
                        e.printStackTrace();
                        log.info("超时订单延时队列====>>> 处理队列出现异常.");
                    }
                }
            });
        }
    
        public void add(OrderDelayedVo order) { 
            if (!start.get()) { 
                throw new RuntimeException("订单服务还未启动");
            }
            log.info("加入超时订单队列，订单号：{}", order.getOrderNo());
            delayQueue.add(order);
        }
    
        public void delete(String orderNo) { 
            if (!start.get()) { 
                throw new RuntimeException("订单服务还未启动");
            }
            log.info("移除超时订单队列，订单号：{}", orderNo);
            delayQueue.remove(orderNo);
        }
    
        public int getSize() { 
            if (!start.get()) { 
                throw new RuntimeException("订单服务还未启动");
            }
            return delayQueue.size();
        }
    
    }

## 3、在创建订单方法中加入延时队列 ##

/** * 下单 */
        @RequestMapping("/createOrder")
        public BaseResp createOrder(@Validated @RequestBody UserOrderVo userOrderVo){ 
        	//省略系统相关业务...
        	
        	//订单超时延迟处理
            OrderDelayedVo vo = new OrderDelayedVo();
            vo.setOrderNo("NO123456789");
            vo.setOrderCategory("1");
            vo.setExpTime(new Date().getTime() + (30 * 60 * 1000));//30分钟后超时
            delayService.add(vo);
            return BaseResp.success("创建订单成功");
        }