【延时队列的实现方式】

延时队列

延时队列是一种特殊类型的队列，它允许元素在特定时间间隔后才能被处理。这种队列在处理具有延迟需求的任务时非常有用，例如定时任务、事件驱动系统等
延时队列在项目中的应用还是比较多的，尤其像电商类平台：
1、订单成功后，在30分钟内没有支付，自动取消订单
2、外卖平台发送订餐通知，下单成功后60s给用户推送短信。
3、如果订单一直处于某一个未完结状态时，及时处理关单，并退还库存
4、淘宝新建商户一个月内还没上传商品信息，将冻结商铺等

JDK自带的延时队列实现

在Java中，DelayQueue 是一个无界阻塞队列，它存放实现了 Delayed 接口的对象，只有在延迟时间到了，对象才能从队列中取出来。DelayQueue 是基于优先队列实现的，队列中的元素按照延迟时间的先后顺序排序。
DelayQueue是一个BlockingQueue（无界阻塞）队列，它本质就是封装了一个PriorityQueue（优先队列），PriorityQueue内部使用完全二叉堆（不知道的自行了解哈）来实现队列元素排序，我们在向DelayQueue队列中添加元素时，会给元素一个Delay（延迟时间）作为排序条件，队列中最小的元素会优先放在队首。队列中的元素只有到了Delay时间才允许从队列中取出。队列中可以放基本数据类型或自定义实体类，在存放基本数据类型时，优先队列中元素默认升序排列，自定义实体类就需要我们根据类属性值比较计算了。

package com.schdule.util;

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

class MyDelayedTask implements Delayed {
    private long delayTime;

    public MyDelayedTask(long delayTime) {
        this.delayTime = delayTime;
    }

    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(delayTime - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    @Override
    public int compareTo(Delayed other) {
        if (this.delayTime < ((MyDelayedTask) other).delayTime) {
            return -1;
        } else if (this.delayTime > ((MyDelayedTask) other).delayTime) {
            return 1;
        } else {
            return 0;
        }
    }

    public static void main(String[] args) throws InterruptedException {
        DelayQueue<MyDelayedTask> queue = new DelayQueue<>();
        queue.put(new MyDelayedTask(System.currentTimeMillis() + 5000)); // 5秒后执行

        while (true) {
            MyDelayedTask task = queue.take();
            System.out.println("Executing task at: " + System.currentTimeMillis());
        }
    }
}

DelayQueue的put方法是线程安全的，因为put方法内部使用了ReentrantLock锁进行线程同步。DelayQueue还提供了两种出队的方法 poll() 和 take() ， poll() 为非阻塞获取，没有到期的元素直接返回null；take() 阻塞方式获取，没有到期的元素线程将会等待。

Redis实现延迟队列

Redis的数据结构Zset，同样可以实现延迟队列的效果，主要利用它的score属性，redis通过score来为集合中的成员进行从小到大的排序。

Redis有序集合的一些常用命令包括：
ZADD: 向有序集合添加一个或多个成员，或者更新已存在成员的分数。
ZRANGE: 返回有序集合中指定区间内的成员列表。
ZREVRANGE: 返回有序集合中指定区间内的成员列表，按分数从高到低排序。
ZRANGEBYSCORE: 返回有序集合中指定分数区间内的成员列表。
ZREM: 移除有序集合中的一个或多个成员。
ZCARD: 获取有序集合的成员数。
ZSCORE: 获取有序集合中成员的分数。

    /**
     * 消费消息
     */
    public void pollOrderQueue() {

        while (true) {
            Set<Tuple> set = jedis.zrangeWithScores(DELAY_QUEUE, 0, 0);

            String value = ((Tuple) set.toArray()[0]).getElement();
            int score = (int) ((Tuple) set.toArray()[0]).getScore();
            
            Calendar cal = Calendar.getInstance();
            int nowSecond = (int) (cal.getTimeInMillis() / 1000);
            if (nowSecond >= score) {
                jedis.zrem(DELAY_QUEUE, value);
                System.out.println(sdf.format(new Date()) + " removed key:" + value);
            }

            if (jedis.zcard(DELAY_QUEUE) <= 0) {
                System.out.println(sdf.format(new Date()) + " zset empty ");
                return;
            }
            Thread.sleep(1000);
        }
    }

RabbitMQ 延时队列

利用 RabbitMQ 做延时队列是比较常见的一种方式，而实际上RabbitMQ 自身并没有直接支持提供延迟队列功能，而是通过 RabbitMQ 消息队列的 TTL和 DXL这两个属性间接实现的。
先来认识一下 TTL和 DXL两个概念：
Time To Live(TTL) ：
TTL 顾名思义：指的是消息的存活时间，RabbitMQ可以通过x-message-tt参数来设置指定Queue（队列）和 Message（消息）上消息的存活时间，它的值是一个非负整数，单位为微秒。
RabbitMQ 可以从两种维度设置消息过期时间，分别是队列和消息本身
设置队列过期时间，那么队列中所有消息都具有相同的过期时间。
设置消息过期时间，对队列中的某一条消息设置过期时间，每条消息TTL都可以不同。
如果同时设置队列和队列中消息的TTL，则TTL值以两者中较小的值为准。而队列中的消息存在队列中的时间，一旦超过TTL过期时间则成为Dead Letter（死信）。
Dead Letter Exchanges（DLX）
DLX即死信交换机，绑定在死信交换机上的即死信队列。RabbitMQ的 Queue（队列）可以配置两个参数x-dead-letter-exchange 和 x-dead-letter-routing-key（可选），一旦队列内出现了Dead Letter（死信），则按照这两个参数可以将消息重新路由到另一个Exchange（交换机），让消息重新被消费。
x-dead-letter-exchange：队列中出现Dead Letter后将Dead Letter重新路由转发到指定 exchange（交换机）。
x-dead-letter-routing-key：指定routing-key发送，一般为要指定转发的队列。
队列出现Dead Letter的情况有：
消息或者队列的TTL过期
队列达到最大长度
消息被消费端拒绝（basic.reject or basic.nack）

import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.BindingBuilder;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitMQConfig {
    @Bean
    public Queue myQueue() {
        Map<String, Object> args = new HashMap<>();
        args.put("x-message-ttl", 3000); // 设置队列的过期时间为 3 秒
        args.put("x-dead-letter-exchange", "myExchange"); // 设置死信交换机
        args.put("x-dead-letter-routing-key", "myRoutingKey"); // 设置死信路由键
        return new Queue("myQueue", false, false, false, args);
    }

    @Bean
    public DirectExchange myExchange() {
        return new DirectExchange("myExchange");
    }

    @Bean
    public Binding binding(Queue myQueue, DirectExchange myExchange) {
        return BindingBuilder.bind(myQueue).to(myExchange).with("myRoutingKey");
    }
}