实战|我还是很建议你用DelayQueue搞定超时订单的

一、用三根鸡毛做引言

• 真的! 不骗你们的喔~ 相信大家都遇到类似于:订单30min后未支付自动取消的开发任务

• 那么 今日份 就来了解一下怎么用 延时队列 DelayQueue 搞定单机版的超时订单

  

二、延时队列使用场景

那么什么时候需要用延时队列呢？常见的延时任务场景 举栗子：

1. 订单在30分钟之内未支付则自动取消。
2. 重试机制实现,把调用失败的接口放入一个固定延时的队列,到期后再重试。
3. 新创建的店铺，如果在十天内都没有上传过商品，则自动发送消息提醒。
4. 用户发起退款，如果三天内没有得到处理则通知相关运营人员。
5. 预定会议后，需要在预定的时间点前十分钟通知各个与会人员参加会议。
6. 关闭空闲连接，服务器中，有很多客户端的连接，空闲一段时间之后需要关闭之。
7. 清理过期数据业务。比如缓存中的对象，超过了空闲时间，需要从缓存中移出。
8. 多考生考试,到期全部考生必须交卷,要求时间非常准确的场景。

三、解决办法多如鸡毛

1. 定期轮询（数据库等）
2. JDK DelayQueue
3. JDK Timer
4. ScheduledExecutorService 周期性线程池
5. 时间轮(kafka)
6. 时间轮(Netty的HashedWheelTimer)
7. Redis有序集合（zset）
8. zookeeper之curator
9. RabbitMQ
10. Quartz,xxljob等定时任务框架
11. Koala(考拉)
12. JCronTab(仿crontab的java调度器)
13. SchedulerX（阿里）
14. 有赞延迟队列
15. .....(鸡毛)

• 解决问题方法真是不胜枚举,正所谓一呼百应,一千个读者眼里有一千个哈姆雷特

:seedling: :seedling: :seedling: :seedling: :seedling: :seedling: :seedling: :seedling: :seedling: :seedling:

• 那我们第一篇先来实战JDK的 DelayQueue ,万祖归宗,万法同源,学会了最基础的 Queue ,就不愁其他的了
• 后续再写几篇使用Redis,Zk,MQ的一些机制,实战分布式情况下的使用

四、先认亲

延时队列，首先，它是一种 队列 ，队列意味着内部的元素是 有序 的，元素 出队 和 入队 是有方向性的，元素从一端进入，从另一端取出。

其次， 延时队列 ，最重要的特性就体现在它的 延时属性 上，跟普通的队列不一样的是， 普通队列 中的元素总是等着希望被早点取出处理，而 延时队列 中的元素则是希望被在 指定时间得到取出和处理 ，所以 延时队列 中的元素是都是 带时间属性 的，通常来说是需要被处理的消息或者任务。

一言以蔽之曰 : 延时队列就是用来存放需要在指定时间被处理的元素的队列。

1) DelayQueue 是谁,上族谱

这一代已经第五代传人了,

要知道 DelayQueue 自幼生在八戒家,长大就往外面拉,熊熊烈火它不怕,水是水来渣是渣。

不过它真的是文韬武略,有一把 ReentrantLock 就是它的九齿钉耙,抗的死死の捍卫着自己的 PriorityQueue .

有典故曰:

public class DelayQueue<E extends Delayed> extends AbstractQueue<E>
           implements BlockingQueue<E> {
// 用于控制并发的 可重入 全局 锁
private final transient ReentrantLock lock = new ReentrantLock();
// 根据Delay时间排序的 无界的 优先级队列
private final PriorityQueue<E> q = new PriorityQueue<E>();
// 用于优化阻塞通知的线程元素leader，标记当前是否有线程在排队（仅用于取元素时）
private Thread leader = null;
// 条件，用于阻塞和通知的Condition对象，表示现在是否有可取的元素
private final Condition available = lock.newCondition();

       /**
        * 省洛方法代码.....  你们懂我的省洛吗?
        */
复制代码

• 注释的已经很清楚他们的意思了,也具备了并发编程之艺术的 锁,队列,状态(条件)
• 他的几个方法也是通过 锁-->维护队列-->出队,入队-->根据 Condition 进行条件的判断-->进行线程之间的通信和唤起
• 以支持优先级无界队列的 PriorityQueue 作为一个容器，容器里面的元素都应该实现 Delayed 接口，在每次往优先级队列中添加元素时以元素的过期时间作为排序条件，最先过期的元素放在优先级最高。
• DelayQueue 是一个没有大小限制的队列，因此往队列中插入数据的操作（生产者）永远不会被阻塞，而只有获取数据的操作（消费者）才会被阻塞。

2) 优先级队列 PriorityQueue

因为我们的 DelayQueue 里面维护了一个优先级的队列 PriorityQueue 简单的看下:

//默认容量11
     private static final int DEFAULT_INITIAL_CAPACITY = 11;
    //存储元素的地方 数组
    transient Object[] queue; // non-private to simplify nested class access
    //元素个数
    private int size = 0;
    //比较器
    private final Comparator<? super E> comparator;
复制代码

1. 默认容量是11；
2. queue，元素存储在数组中，这跟我们之前说的堆一般使用数组来存储是一致的；
3. comparator，比较器，在优先级队列中，也有两种方式比较元素，一种是元素的自然顺序，一种是通过比较器来比较；
4. modCount，修改次数，有这个属性表示PriorityQueue也是fast-fail的；
5. PriorityQueue不是有序的，只有堆顶存储着最小的元素；
6. PriorityQueue 是非线程安全的；

3) DelayQueue的方法简介

• 入队方法 : 若添加的元素是队首(堆顶)元素，就把 leader 置为空，并唤醒等待在条件 available 上的线程；

public boolean add(E e) {    return offer(e);}
public void put(E e) {    offer(e);}
public boolean offer(E e, long timeout, TimeUnit unit) {    return offer(e);}
public boolean offer(E e) {    
    final ReentrantLock lock = this.lock;    
    lock.lock();   //加锁 因为优先队列线程不安全
    try {
        q.offer(e);  //判断优先级 进行入队      
    if (q.peek() == e) {    //-----[1]
        //leader记录了被阻塞在等待队列头生效的线程 新增一个元素到队列头，
        //表示等待原来队列头生效的阻塞的线程已经失去了阻塞的意义
        //，此时需要获取新的队列头进行返回了
        leader = null;      
        //获取队列头的线程被唤起,主要有两种场景:
        //1. 之前队列为空，导致被阻塞的线程
        //2. 之前队列非空，但是队列头没有生效(到期)导致被阻塞的线程
        available.signal();     
    }        
        return true; //因为是无界队列 所以添加元素肯定成功  直到OOM
    } finally {    
        lock.unlock();   //释放锁
    }
}
复制代码

offer() 方法,首先获取独占锁，然后添加元素到优先级队列，由于q是优先级队列，所以添加元素后，peek并不一定是当前添加的元素，如果[1]为true，说明当前元素e的优先级最小也就即将过期的，这时候激活avaliable变量条件队列里面的线程，通知他们队列里面有元素了。

• 出队方法 take()

请看我详细的注释,绝不是蜻蜓点水

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock; //获取锁 
    lock.lockInterruptibly();   //可中断锁 可以自行了解一下 嘻嘻嘻嘻...
    try {
        for (;;) {//会写死循环的都是高手
            E first = q.peek();//get队头元素
            if (first == null)
                // 队列头为空，则阻塞，直到新增一个入队为止(1)
                available.await();
            else {
                long delay = first.getDelay(NANOSECONDS);//获取剩余时间
                if (delay <= 0)
                    // 若队列头元素已生效，则直接返回(2)
                    return q.poll();
                first = null; // don't retain ref while waiting 等待的时候不能引用,表示释放当前引用的(3)
                if (leader != null)
                    // leader 非空时，表示有其他的一个线程在出队阻塞中 (4.1)
                    // 此时挂住当前线程，等待另一个线程出队完成
                    available.await();
                else {
                    //标识当前线程处于等待队列头生效的阻塞中 (4.2.1)
                    Thread thisThread = Thread.currentThread(); 
                    leader = thisThread;
                    try {
                        // 等待队列头元素生效(4.2.2)
                        available.awaitNanos(delay);
                    } finally {
                        //最终释放当前的线程 设置leader为null (4.2.3)
                        if (leader == thisThread)
                            leader = null;
                    }
                }
            }
        }     //(5)
    } finally {
        if (leader == null && q.peek() != null)
            // 当前线程出队完成，通知其他出队阻塞的线程继续执行(6)
            available.signal();
            lock.unlock();//解锁结束
    }
}
复制代码

那么，下面的结论肉眼可见:

1. 如果队列为空，则阻塞，直到有个线程(生产者投递数据)完成入队操作
2. 获取队列头，若队列头已生效，则直接返回
3. 未生效则释放当前引用
4. 当队列头部没有生效时候:
   1. 若有另一个线程已经处于等待队列头生效的阻塞过程中，则阻塞当前线程，直到另一个线程完成出队操作
   2. 若没有其他线程阻塞在出队过程中，即当前线程为第一个获取队列头的线程

      leader = thisThread
      leader=null
      

5. 再次进入循环，获取队列头并返回
6. 最后,当前线程出队完成，通知其他出队阻塞的线程继续执行

4)Delayed

public interface Delayed extends Comparable<Delayed> { 
    long getDelay(TimeUnit unit);
}
复制代码

据情报显示: Delayed 是一个继承自 Delayed 的接口，并且定义了一个 Delayed 方法，用于表示还有多少时间到期，到期了应返回小于等于0的数值。

很简答就是定义了一个,一个哈,一个表延迟的接口,就是个规范接口,目的就是骗我们去实现它的方法.哼~

五、再实战

说了那么多废话,让我想起了那句名言: 一切没有代码实操的讲解都是耍流氓 至今深深的烙在我心中,所以我一定要实战给你们看,显得我不是流氓...

• 实战以 订单下单后三十分钟内未支付则自动取消 为业务场景

• 该场景的代码逻辑分析如下:

  1. 下单后将订单直接放入未支付的延时队列中
  2. 如果超时未支付,则从队列中取出,进行修改为取消状态的订单
  3. 如果支付了,则不去进行取消,或者取消的时候做个状态筛选,即可避免更新
  4. 或者支付完成后,做个主动出队
  5. 还有就是用户主动取消订单,也做个主动出队

• 那么我们写代码一定要通用,先来写个通用的 Delayed 通用...嗯! 泛型的

import lombok.Getter;
import lombok.Setter;

import java.util.Date;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;

/**
 * @author LiJing
 * @ClassName: ItemDelayed
 * @Description: 数据延迟实现实例 用以包装具体的实例转型
 * @date 2019/9/16 15:53
 */

@Setter
@Getter
public class ItemDelayed<T> 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<T> {


    /**
     * 添加延迟对象到延时队列
     *
     * @param itemDelayed 延迟对象
     * @return boolean
     */
    boolean addToOrderDelayQueue(ItemDelayed<T> itemDelayed);

    /**
     * 根据对象添加到指定延时队列
     *
     * @param data 数据对象
     * @return boolean
     */
    boolean addToDelayQueue(T data);

    /**
     * 移除指定的延迟对象从延时队列中
     *
     * @param data
     */
    void removeToOrderDelayQueue(T data);
}
复制代码

• 来具体的任务,具体的逻辑具体实现

@Slf4j
@Lazy(false)
@Component
public class DelayOwnOrderImpl implements DelayOrder<Order> {

    @Autowired
    private OrderService orderService;

    @Autowired
    private ExecutorService delayOrderExecutor;

    private final static DelayQueue<ItemDelayed<Order>> DELAY_QUEUE = new DelayQueue<>();

    /**
     * 初始化时加载数据库中需处理超时的订单
     * 系统启动:扫描数据库中未支付(要在更新时:加上已支付就不用更新了),未过期的的订单
     */
    @PostConstruct
    public void init() {
        log.info("系统启动:扫描数据库中未支付,未过期的的订单");
        List<Order> orderList = orderService.selectFutureOverTimeOrder();
        for (Order order : orderList) {
            ItemDelayed<Order> orderDelayed = new ItemDelayed<>(order.getId(), order.getCreateDate().getTime());
            this.addToOrderDelayQueue(orderDelayed);
        }
        log.info("系统启动:扫描数据库中未支付的订单,总共扫描了" + orderList.size() + "个订单,推入检查队列,准备到期检查...");

        /*启动一个线程，去取延迟订单*/
        delayOrderExecutor.execute(() -> {
            log.info("启动处理的订单线程:" + Thread.currentThread().getName());
            ItemDelayed<Order> orderDelayed;
            while (true) {
                try {
                    orderDelayed = DELAY_QUEUE.take();
                    //处理超时订单
                    orderService.updateCloseOverTimeOrder(orderDelayed.getDataId());
                } catch (Exception e) {
                    log.error("执行自营超时订单的_延迟队列_异常:" + e);
                }
            }
        });
    }

    /**
     * 加入延迟消息队列
     **/
    @Override
    public boolean addToOrderDelayQueue(ItemDelayed<Order> orderDelayed) {
        return DELAY_QUEUE.add(orderDelayed);
    }

    /**
     * 加入延迟消息队列
     **/
    @Override
    public boolean addToDelayQueue(Order order) {
        ItemDelayed<Order> 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<ItemDelayed<Order>> iterator = DELAY_QUEUE.iterator(); iterator.hasNext(); ) {
            ItemDelayed<Order> queue = iterator.next();
            if (queue.getDataId().equals(order.getId())) {
                DELAY_QUEUE.remove(queue);
            }
        }
    }
}
复制代码

delayOrderExecutor
@PostConstruct
take()

六、后总结

id

那今日份的讲解就到此结束,具体的代码请移步我的 gitHub的mybot项目Master分支 查阅,fork体验一把,或者评论区留言探讨,写的不好,请多多指教~~