延时任务设计

超时30分钟自动关闭订单

数据库定时任务quartz-scheduler 支持集群；对服务器内存消耗大，数据库压力大，最坏延时为定时时间

//org.quartz-scheduler
public class MyJob implements Job {
    public void execute(JobExecutionContext context) throws JobExecutionException {
    }
    public static void main(String[] args) throws Exception {
        JobDetail jobDetail = JobBuilder.newJob(MyJob.class)
                .withIdentity("job1", "group1").build();
        //触发器 每3秒执行一次
        Trigger trigger = TriggerBuilder
                .newTrigger()
                .withIdentity("trigger1", "group3")
                .withSchedule(
                        SimpleScheduleBuilder
                                .simpleSchedule()
                                .withIntervalInSeconds(3).
                                repeatForever())
                .build();
        Scheduler scheduler = new StdSchedulerFactory().getScheduler();
        // 将任务及其触发器放入调度器
        scheduler.scheduleJob(jobDetail, trigger);
        // 调度器开始调度任务
        scheduler.start();
    }
}

JDK的DelayQueue无界阻塞延时队列，延迟期满时才能获取元素。延迟低；重启后数据丢失，集群扩展难

Poll()：获取并移除队列的超时元素，没有则返回空
take()：获取并移除队列的超时元素，如果没有则wait当前线程，直到有元素返回结果

public class OrderDelay implements Delayed {
    private String orderId;
    private long timeout;
    OrderDelay(String orderId, long timeout) {
        this.orderId = orderId;
        this.timeout = timeout + System.nanoTime();
    }

    public int compareTo(Delayed other) {
        if (other == this) {
            return 0;
        }
        OrderDelay t = (OrderDelay) other;
        long d = (getDelay(TimeUnit.NANOSECONDS) - t.getDelay(TimeUnit.NANOSECONDS));
        return (d == 0) ? 0 : ((d < 0) ? -1 : 1);
    }

    // 返回距离你自定义的超时时间还有多少
    public long getDelay(TimeUnit unit) {
        return unit.convert(timeout - System.nanoTime(), TimeUnit.NANOSECONDS);
    }

    void print() {
        System.out.println(orderId + "编号的订单要删除啦。。。。");
    }

    public static void main(String[] args) {
        DelayQueue<OrderDelayed> delayQueue = new DelayQueue<>();
        delayQueue.put(new OrderDelayed("220101001", 8, TimeUnit.SECONDS));
        delayQueue.put(new OrderDelayed("220101002", 4, TimeUnit.SECONDS));  
        while (true) {
            // 取队列头部元素是否过期
            OrderDelayed task = delayQueue.poll();
            if (task != null) {
                System.out.println("订单 ---> " + task.orderNo + " 已过期准备取消");
            }
        }
    }
}

Redis Sorted Set zadd key score毫秒 member score member 且long int = zrem key >0且大于当前时间

public class AppTest {
    private static JedisPool jedisPool = new JedisPool("127.0.0.1", 6379);
    public static Jedis getJedis() {
        return jedisPool.getResource();
    }

    //生产者,生成5个订单放进去
    public void productionDelayMessage() {
        for (int i = 0; i < 5; i++) {
            //延迟3秒
            Calendar cal1 = Calendar.getInstance();
            cal1.add(Calendar.SECOND, 3);
            int second3later = (int) (cal1.getTimeInMillis() / 1000);
            AppTest.getJedis().zadd("OrderId", second3later, "OID0000001" + i);
            System.out.println(System.currentTimeMillis() + "ms:redis生成了一个订单任务：订单ID为" + "OID0000001" + i);
        }
    }

    //消费者，取订单
    public void consumerDelayMessage() {
        Jedis jedis = AppTest.getJedis();
        while (true) {
            Set<Tuple> items = jedis.zrangeWithScores("OrderId", 0, 1);
            if (items == null || items.isEmpty()) {
                System.out.println("当前没有等待的任务");
                try {
                    Thread.sleep(500);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                continue;
            }
            int score = (int) ((Tuple) items.toArray()[0]).getScore();
            Calendar cal = Calendar.getInstance();
            int nowSecond = (int) (cal.getTimeInMillis() / 1000);
            if (nowSecond >= score) {
                String orderId = ((Tuple) items.toArray()[0]).getElement();
                Long num = jedis.zrem("OrderId", orderId);
                if( num != null && num>0){
                    System.out.println(System.currentTimeMillis() + "ms:redis消费了一个任务：消费的订单OrderId为" + orderId);
                }
            }
        }
    }

    public static void main(String[] args) {
        AppTest appTest = new AppTest();
        appTest.productionDelayMessage();
        appTest.consumerDelayMessage();
    }

}

基于时间轮算法实现延迟队列，如Kafka。Netty的HashedWheelTimer延迟时间比delayQueue低，重启后数据丢失，集群扩展难

netty-all
public class HashedWheelTimerTest {
    static class MyTimerTask implements TimerTask {
        boolean flag;
        public MyTimerTask(boolean flag) {
            this.flag = flag;
        }
        public void run(Timeout timeout) throws Exception {
            System.out.println("要去数据库删除订单了。。。。");
            this.flag = false;
        }
    }

    public static void main(String[] argv) {
        MyTimerTask timerTask = new MyTimerTask(true);
        Timer timer = new HashedWheelTimer();
        timer.newTimeout(timerTask, 5, TimeUnit.SECONDS);
        int i = 1;
        while (timerTask.flag) {
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            System.out.println(i + "秒过去了");
            i++;
        }
    }
}

基于RocketMQ延迟消息，发送方通过设置延迟时间Message#setDelayTimeLevel指定18个延迟等级，11s 5s 10s 30s 1-10m 20m 30m 1h 2h，消费方消费失败默认将在发送方的延迟等级基础上加1。如果消费方可以使用ConsumeConcurrentlyContext#setDelayLevelwhenNextConsume指定其他的延迟等级
简单，不支持自定义延迟时间
RabbitMQ死信队列：Queue和Message设置x-message-tt控制消息的生存时间，如果超时则消息变为dead letter。Queue可配置x-dead-letter-exchange 和x-dead-letter-routing-key（可选）控制队列内deadletter按照这两个参数重新路由。高效，横向扩展容易，消息支持持久化。运维复杂度和成本变高