Java使用延时队列搞定超时订单处理的场景

import lombok.Getter; import lombok.Setter; import java.util.Date; import java.util.concurrent.Delayed; import java.util.concurrent.TimeUnit; @Setter @Getter public class ItemDelayed implements Delayed { /**默认延迟30分钟*/ private final static long DELAY = 30 * 60 * 1000L; /**数据id*/ private Long dataId; /**开始时间*/ private long startTime; /**到期时间*/ private long expire; /**创建时间*/ private Date now; /**泛型data*/ private T data; public ItemDelayed(Long dataId, long startTime, long secondsDelay) { super(); this.dataId = dataId; this.startTime = startTime; this.expire = startTime + (secondsDelay * 1000); this.now = new Date(); } public ItemDelayed(Long dataId, long startTime) { super(); this.dataId = dataId; this.startTime = startTime; this.expire = startTime + DELAY; this.now = new Date(); } @Override public int compareTo(Delayed o) { return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS)); } @Override public long getDelay(TimeUnit unit) { return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS); } }

public interface DelayOrder { /** * 添加延迟对象到延时队列 * * @param itemDelayed 延迟对象 * @return boolean */ boolean addToOrderDelayQueue(ItemDelayed itemDelayed); /** * 根据对象添加到指定延时队列 * * @param data 数据对象 * @return boolean */ boolean addToDelayQueue(T data); /** * 移除指定的延迟对象从延时队列中 * * @param data */ void removeToOrderDelayQueue(T data); }

@Slf4j @Lazy(false) @Component public class DelayOwnOrderImpl implements DelayOrder { @Autowired private OrderService orderService; @Autowired private ExecutorService delayOrderExecutor; private final static DelayQueue> DELAY_QUEUE = new DelayQueue<>(); /** * 初始化时加载数据库中需处理超时的订单 * 系统启动:扫描数据库中未支付(要在更新时:加上已支付就不用更新了),未过期的的订单 */ @PostConstruct public void init() { log.info("系统启动:扫描数据库中未支付,未过期的的订单"); List orderList = orderService.selectFutureOverTimeOrder(); for (Order order : orderList) { ItemDelayed orderDelayed = new ItemDelayed<>(order.getId(), order.getCreateDate().getTime()); this.addToOrderDelayQueue(orderDelayed); } log.info("系统启动:扫描数据库中未支付的订单,总共扫描了" + orderList.size() + "个订单,推入检查队列,准备到期检查..."); /*启动一个线程，去取延迟订单*/ delayOrderExecutor.execute(() -> { log.info("启动处理的订单线程:" + Thread.currentThread().getName()); ItemDelayed orderDelayed; while (true) { try { orderDelayed = DELAY_QUEUE.take(); //处理超时订单 orderService.updateCloseOverTimeOrder(orderDelayed.getDataId()); } catch (Exception e) { log.error("执行自营超时订单的_延迟队列_异常:" + e); } } }); } /** * 加入延迟消息队列 **/ @Override public boolean addToOrderDelayQueue(ItemDelayed orderDelayed) { return DELAY_QUEUE.add(orderDelayed); } /** * 加入延迟消息队列 **/ @Override public boolean addToDelayQueue(Order order) { ItemDelayed orderDelayed = new ItemDelayed<>(order.getId(), order.getCreateDate().getTime()); return DELAY_QUEUE.add(orderDelayed); } /** * 从延迟队列中移除 主动取消就主动从队列中取出 **/ @Override public void removeToOrderDelayQueue(Order order) { if (order == null) { return; } for (Iterator> iterator = DELAY_QUEUE.iterator(); iterator.hasNext(); ) { ItemDelayed queue = iterator.next(); if (queue.getDataId().equals(order.getId())) { DELAY_QUEUE.remove(queue); } } } }