基于Redisson实现延迟队列

Redisson实现延迟队列

1.场景介绍

假设有这样一个场景，我们有一个订单，或者工单等等。需要在超时30分钟后进行关闭。这个时候我们最先想到的应该是采用定时任务去进行轮训判断，但是呢，每个订单的创建时间是不一样的，这个时间怎么确定才好呢，5分钟。。1分钟。。执行一次吗。这样就会非常影响性能。且时间误差很大。基于以上业务需要我们想到了有以下解决方案。

• JDK延迟队列,但是数据都在内存中，重启后什么都没了。
• MQ中的延迟队列，比如RocketMQ。
• 基于Redisson的延迟队列

2.JDK延迟队列

我们首先来回顾下JDK的延迟队列

基于延迟队列要实现接口Delayed，并且实现getDelay方法和compareTo方法

• getDelay主要是计算返回剩余时间，单位时间戳（毫秒）延迟任务是否到时就是按照这个方法判断如果返回的是负数则说明到期否则还没到期
• compareTo主要是自定义实现比较方法返回 1 0 -1三个参数

@ToString
public class MyDelayed<T> implements Delayed {
    /**
     * 延迟时间
     */
    Long delayTime;

    /**
     * 过期时间
     */
    Long expire;
    /**
     * 数据
     */
    T t;

    public MyDelayed(long delayTime, T t) {
        this.delayTime = delayTime;
        // 过期时间 = 当前时间 + 延迟时间
        this.expire = System.currentTimeMillis() + delayTime;
        this.t = t;
    }

    /**
     * 剩余时间 = 到期时间 - 当前时间
     */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.expire - System.currentTimeMillis(), TimeUnit.MILLISECONDS);
    }

    /**
     * 优先级规则：两个任务比较，时间短的优先执行
     */
    @Override
    public int compareTo(Delayed o) {
        long f = this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS);
        return (int) f;
    }

订单的实体,为了简单就定义基础几个字段。

@Data
public class OrderInfo implements Serializable {
    private static final long serialVersionUID = -2837036864073566484L;
    /**
     * 订单id
     */
    private Long id;
    
    /**
     * 订单金额
     */
    private Double salary;

    /**
     * 订单创建时间   对于java8LocalDateTime 以下注解序列化反序列化用到
     */
    @JsonDeserialize(using = LocalDateTimeDeserializer.class)
    @JsonSerialize(using = LocalDateTimeSerializer.class)
    private LocalDateTime createTime;

}

为了简单我们暂且定义延迟时间为10s

public static void main(String[] args) throws InterruptedException {
        OrderInfo orderInfo = new OrderInfo();
        orderInfo.setCreateTime(LocalDateTimeUtil.parse("2022-07-01 15:00:00", "yyyy-MM-dd HH:mm:ss"));
        MyDelayed<OrderInfo> myDelayed = new MyDelayed<>(10000L,orderInfo);
        DelayQueue<MyDelayed<OrderInfo>> queue = new DelayQueue<>();
        queue.add(myDelayed);
       
        System.out.println(queue.take().getT().getCreateTime());
        System.out.println("当前时间：" + LocalDateTime.now());
    }

输出结果

2022-07-01T15:00
当前时间：2022-07-01T15:10:37.375

3.基于Redisson的延迟队列

当然今天的主角是它了，我们主要围绕着基于Redisson的延迟队列来说。

其实Redisson延迟队列内部也是基于redis来实现的，我们先来进行整合使用看看效果。基于springboot

1.依赖：

 <dependency>
      <groupId>org.redisson</groupId>
      <artifactId>redisson-spring-boot-starter</artifactId>
      <version>3.16.7</version>
 </dependency>

2.创建redisson.yml

# 单节点配置
singleServerConfig:
  # 连接空闲超时，单位：毫秒
  idleConnectionTimeout: 10000
  # 连接超时，单位：毫秒
  connectTimeout: 10000
  # 命令等待超时，单位：毫秒
  timeout: 3000
  # 命令失败重试次数,如果尝试达到 retryAttempts（命令失败重试次数） 仍然不能将命令发送至某个指定的节点时，将抛出错误。
  # 如果尝试在此限制之内发送成功，则开始启用 timeout（命令等待超时） 计时。
  retryAttempts: 3
  # 命令重试发送时间间隔，单位：毫秒
  retryInterval: 1500
  # 密码
  password:
  # 单个连接最大订阅数量
  subscriptionsPerConnection: 5
  # 客户端名称
  clientName: null
  # 节点地址
  address: redis://127.0.0.1:6379
  # 发布和订阅连接的最小空闲连接数
  subscriptionConnectionMinimumIdleSize: 1
  # 发布和订阅连接池大小
  subscriptionConnectionPoolSize: 50
  # 最小空闲连接数
  connectionMinimumIdleSize: 32
  # 连接池大小
  connectionPoolSize: 64
  # 数据库编号
  database: 0
  # DNS监测时间间隔，单位：毫秒
  dnsMonitoringInterval: 5000
# 线程池数量,默认值: 当前处理核数量 * 2
#threads: 0
# Netty线程池数量,默认值: 当前处理核数量 * 2
#nettyThreads: 0
# 编码
codec: !<org.redisson.codec.JsonJacksonCodec> {}
# 传输模式
transportMode : "NIO"

3.创建配置类RedissonConfig,这里是为了读取我们刚刚创建在配置文件中的yml

@Configuration
public class RedissonConfig {
    @Bean
    public RedissonClient redissonClient() throws IOException {
        Config config = Config.fromYAML(RedissonConfig.class.getClassLoader().getResource("redisson.yml"));;
        return Redisson.create(config);
    }
}

4.测试

        // redisson  延迟队列
        // Redisson的延时队列是对另一个队列的再包装，使用时要先将延时消息添加到延时队列中，
        // 当延时队列中的消息达到设定的延时时间后，该延时消息才会进行进入到被包装队列中，因此，我们只需要对被包装队列进行监听即可。
        RBlockingQueue<OrderInfo> blockingFairQueue = redissonClient.getBlockingQueue("my-test");

        RDelayedQueue<OrderInfo> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);

        OrderInfo orderInfo = new OrderInfo();
        // 订单生成时间
        orderInfo.setCreateTime(LocalDateTime.now());
        // 10秒钟以后将消息发送到指定队列
        delayedQueue.offer(orderInfo, 10, TimeUnit.SECONDS);
        RBlockingQueue<OrderInfo> outQueue = redissonClient.getBlockingQueue("my-test");

        OrderInfo orderInfo2 = outQueue.take();
        System.out.println("订单生成时间" + orderInfo2.getCreateTime());
        System.out.println("订单关闭时间" + LocalDateTime.now());

        // 在该对象不再需要的情况下，应该主动销毁。仅在相关的Redisson对象也需要关闭的时候可以不用主动销毁
        delayedQueue.destroy();

控制台输出：

订单生成时间2022-07-01T15:22:10.304
订单关闭时间2022-07-01T15:22:20.414

解决项目重新启动并不会消费之前队列里的消息的问题,增加如下代码

 redissonClient.getDelayedQueue(deque);

4.深入探究Redisson的延迟队列实现原理

我们首先来了解两个API

• RBlockingQueue 就是目标队列

• RDelayedQueue 就是中转队列

那么为什么会涉及到两个队列呢，这两个队列到底有什么用呢？

首先我们实际操作的是RBlockingQueue阻塞队列，并不是RDelayedQueue队列，RDelayedQueue对接主要是提供中间转发的一个队列，类似中间商的意思

画个小图理解下

这里不难看出我们都是基于RBlockingQueue 目标队列在进行消费，而RDelayedQueue就是会把过期的消息放入到我们的目标队列中

我们只要从RBlockingQueue队列中取数据即可。

好像还是不够深入，我们接着看。我们知道Redisson是基于redis来实现的那么我们看看里面到底做了什么事

打开redis客户端，执行monitor命令，看下在执行上面订单操作时redis到底执行了哪些命令

monitor命令可以看到操作redis时执行了什么命令

// 这里订阅了一个固定的队列 redisson_delay_queue_channel:{my-test}，为了开启进程里面的延时任务
"SUBSCRIBE" "redisson_delay_queue_channel:{my-test}"

// Redis Zrangebyscore 返回有序集合中指定分数区间的成员列表。有序集成员按分数值递增(从小到大)次序排列。
// redisson_delay_queue_channel:{my-test} 是一个zset，当有延时数据存入Redisson队列时，就会在此队列中插入 数据，排序分数为延时的时间戳(毫秒 以下同理)。
"zrangebyscore" "redisson_delay_queue_timeout:{my-test}" "0" "1656404479385" "limit" "0" "100"

// 取出第一个数，也就是判断上面执行的操作是否有下一页。（因为刚刚开始总是0的）除非是之前的操作（zrangebyscore）没有取完
"zrange" "redisson_delay_queue_timeout:{my-test}" "0" "0" "WITHSCORES"

// 往zset里面设置 数据过期的时间戳（当前执行的时间戳+延时的时间毫秒值）内容就是订单数据
"zadd" "redisson_delay_queue_timeout:{my-test}" "1656404489400" "b\x99M9\x9b\x0c\xd3\xc3\\x00\x00\x00{\"@class\":\"com.example.mytest.domain.OrderInfo\",\"createTime\":[2022,6,28,16,21,19,400000000]}"

// 同步一份数据到list队列
"rpush" "redisson_delay_queue:{my-test}" "b\x99M9\x9b\x0c\xd3\xc3\\x00\x00\x00{\"@class\":\"com.example.mytest.domain.OrderInfo\",\"createTime\":[2022,6,28,16,21,19,400000000]}"

// 取出排序好的第一个数据，也就是最临近要触发的数据，然后发送通知
"zrange" "redisson_delay_queue_timeout:{my-test}" "0" "0"

// 发送通知  之前第一步 SUBSCRIBE 订阅了 客户端收到通知后，就在自己进程里面开启延时任务（HashedWheelTimer），到时间后就可以从redis取数据发送
"publish" "redisson_delay_queue_channel:{my-test}" "1656404489400"

// 这里就是取数据环节了
"BLPOP" "my-test" "0"

// 在范围 0-过期时间  取出100条数据
"zrangebyscore" "redisson_delay_queue_timeout:{my-test}" "0" "1656404489444" "limit" "0" "100"

// 将上面取到的数据push到阻塞队列 很显然能看到 com.example.mytest.domain.OrderInfo 是我们的订单数据
"rpush" "my-test" "{\"@class\":\"com.example.mytest.domain.OrderInfo\",\"createTime\":[2022,6,28,16,21,19,400000000]}"

// 删除数据
"lrem" "redisson_delay_queue:{my-test}" "1" "b\x99M9\x9b\x0c\xd3\xc3\\x00\x00\x00{\"@class\":\"com.example.mytest.domain.OrderInfo\",\"createTime\":[2022,6,28,16,21,19,400000000]}"
"zrem" "redisson_delay_queue_timeout:{my-test}" "b\x99M9\x9b\x0c\xd3\xc3\\x00\x00\x00{\"@class\":\"com.example.mytest.domain.OrderInfo\",\"createTime\":[2022,6,28,16,21,19,400000000]}"

// 取zset第一个数据，有的话继续上面逻辑取数据
"zrange" "redisson_delay_queue_timeout:{my-test}" "0" "0" "WITHSCORES"

// 退订
"UNSUBSCRIBE" "redisson_delay_queue_channel:{my-test}"

这里参考：https://zhuanlan.zhihu.com/p/343811173

我们知道Zset是按照分数升序的也就是最小的分数在最前面，基于这个特点，大致明白，利用过期时间的时间戳作为分数放入到Zset中，那么即将过期的就在最上面。

直接上个图解