2036 字

10 分钟

... 次访问

Redisson 延迟队列

2024-02-25

技术分享

Redis

/

Redisson

/

消息队列

/

SpringBoot

Redisson 延迟队列#

使用场景#

1、下单成功，30分钟未支付。支付超时，自动取消订单

2、订单签收，签收后7天未进行评价。订单超时未评价，系统默认好评

3、下单成功，商家5分钟未接单，订单取消

4、配送超时，推送短信提醒

5、三天会员试用期，三天到期后准时准点通知用户，试用产品到期了

6、用户进店后按照顺序播放音频

…

对于延时比较长的场景、实时性不高的场景，我们可以采用任务调度的方式定时轮询处理。如：xxl-job。

今天我们讲解延迟队列的实现方式，而延迟队列有很多种实现方式，普遍会采用如下等方式，如：

1.如基于RabbitMQ的队列ttl+死信路由策略：通过设置一个队列的超时未消费时间，配合死信路由策略，到达时间未消费后，回会将此消息路由到指定队列

（暴力开门和卡门）
2.基于RabbitMQ延迟队列插件（rabbitmq-delayed-message-exchange）：发送消息时通过在请求头添加延时参数（headers.put(“x-delay”, 5000)）即可达到延迟队列的效果。
3.使用redis的zset有序性，轮询zset中的每个元素

Redis的数据结构Zset，可以实现延迟队列的效果，主要利用它的score属性，redis通过score来为集合中的成员进行从小到大的排序。
4.使用redis的key的过期通知策略，设置一个key的过期时间为延迟时间，过期后通知客户端(此方式依赖redis过期检查机制key多后延迟会比较严重；Redis的pubsub不会被持久化，服务器宕机就会被丢弃)。
Redisson 延时队列

一、先整合实践一下#

1、引入 Redisson 依赖并配置redis#

1
        <dependency>
2
            <groupId>org.redisson</groupId>
3
            <artifactId>redisson-spring-boot-starter</artifactId>
4
            <version>3.10.5</version>
5
        </dependency>

1
  redis:
2
    # 地址
3
    host: 127.0.0.1
4
    # 端口，默认为6379
5
    port: 6379
6
    # 数据库索引
7
    database: 0
8
    # 密码
9
    password: 123456
10
    # 连接超时时间
11
    timeout: 10s

2、创建 RedissonConfig 配置#

1
/**
2
 * Redission配置类
3
 */
4
@Slf4j
5
@Configuration
6
public class RedissionConfig {
7
    private final String REDISSON_PREFIX = "redis://";
8
    private final RedisProperties redisProperties;
9

10
    public RedissionConfig(RedisProperties redisProperties) {
11
        this.redisProperties = redisProperties;
12
    }
13

14
    @Bean
15
    public RedissonClient redissonClient() {
16
        Config config = new Config();
17
        String url = REDISSON_PREFIX + redisProperties.getHost() + ":" + redisProperties.getPort();
18
        // 这里以单台redis服务器为例
19
        config.useSingleServer()
20
            .setAddress(url)
21
            .setPassword(redisProperties.getPassword())
22
            .setDatabase(redisProperties.getDatabase())
23
            .setPingConnectionInterval(2000);
24
        config.setLockWatchdogTimeout(10000L);
25
        try {
26
            return Redisson.create(config);
27
        } catch (Exception e) {
28
            log.error("RedissonClient init redis url:[{}], Exception:", url, e);
29
            return null;
30
        }
31
    }
32
}

3、封装 Redis 延迟队列工具类#

1
/**
2
 * 分布式延时队列工具类
3
 */
4
@Slf4j
5
@Component
6
//@ConditionalOnBean({RedissonClient.class})
7
public class RedisDelayQueueUtil {
8

9
    @Resource
10
    private RedissonClient redissonClient;
11

12
    /**
13
     * 添加延迟队列
14
     *
15
     * @param value     队列值
16
     * @param delay     延迟时间
17
     * @param timeUnit  时间单位
18
     * @param queueCode 队列键
19
     * @param <T>
20
     */
21
    public <T> boolean addDelayQueue(@NonNull T value, @NonNull long delay, @NonNull TimeUnit timeUnit, @NonNull String queueCode) {
22
        if (StringUtils.isBlank(queueCode) || Objects.isNull(value)) {
23
            return false;
24
        }
25
        try {
26
            RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode);
27
            RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
28
            delayedQueue.offer(value, delay, timeUnit);
29
            //delayedQueue.destroy();
30
            log.info("(添加延时队列成功) 队列键：{}，队列值：{}，延迟时间：{}", queueCode, value, timeUnit.toSeconds(delay) + "秒");
31
        } catch (Exception e) {
32
            log.error("(添加延时队列失败) {}", e.getMessage());
33
            throw new RuntimeException("(添加延时队列失败)");
34
        }
35
        return true;
36
    }
37

38
    /**
39
     * 获取延迟队列
40
     *
41
     * @param queueCode
42
     * @param <T>
43
     */
44
    public <T> T getDelayQueue(@NonNull String queueCode) throws InterruptedException {
45
        if (StringUtils.isBlank(queueCode)) {
46
            return null;
47
        }
48
        RBlockingDeque<Map> blockingDeque = redissonClient.getBlockingDeque(queueCode);
49
        RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque);
50
        T value = (T) blockingDeque.poll();
51
        return value;
52
    }
53
    /**
54
     * 删除指定队列中的消息
55
     *
56
     * @param o 指定删除的消息对象队列值(同队列需保证唯一性)
57
     * @param queueCode 指定队列键
58
     */
59
     public boolean removeDelayedQueue(@NonNull Object o, @NonNull String queueCode) {
60
        if (StringUtils.isBlank(queueCode) || Objects.isNull(o)) {
61
            return false;
62
        }
63
        RBlockingDeque<Object> blockingDeque = redissonClient.getBlockingDeque(queueCode); RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingDeque); boolean flag = delayedQueue.remove(o);
64
        //delayedQueue.destroy();
65
        return flag;
66
    }
67
}

4、创建延迟队列业务枚举#

1
/**
2
 * 延迟队列业务枚举
3
 */
4
@Getter
5
@NoArgsConstructor
6
@AllArgsConstructor
7
public enum RedisDelayQueueEnum {
8

9
    ORDER_PAYMENT_TIMEOUT("ORDER_PAYMENT_TIMEOUT","订单支付超时，自动取消订单", "orderPaymentTimeout"),
10
    ORDER_TIMEOUT_NOT_EVALUATED("ORDER_TIMEOUT_NOT_EVALUATED", "订单超时未评价，系统默认好评", "orderTimeoutNotEvaluated");
11

12
    /**
13
     * 延迟队列 Redis Key
14
     */
15
    private String code;
16

17
    /**
18
     * 中文描述
19
     */
20
    private String name;
21

22
    /**
23
     * 延迟队列具体业务实现的 Bean
24
     * 可通过 Spring 的上下文获取
25
     */
26
    private String beanId;
27

28
}

5、定义延迟队列执行器#

1
/**
2
 * 延迟队列执行器
3
 */
4
public interface RedisDelayQueueHandle<T> {
5

6
    void execute(T t);
7

8
}

6、创建枚举中定义的Bean,实现延迟队列执行器#

OrderPaymentTimeout：订单支付超时延迟队列处理类

1
/**
2
 * 订单支付超时处理类
3
 */
4
@Component
5
@Slf4j
6
public class OrderPaymentTimeout implements RedisDelayQueueHandle<Map> {
7
    @Override
8
    public void execute(Map map) {
9
        log.info("(收到订单支付超时延迟消息) {}", map);
10
        // TODO 订单支付超时，自动取消订单处理业务...
11

12
    }
13
}

OrderTimeoutNotEvaluated：订单超时未评价延迟队列处理类

1
/**
2
 * 订单超时未评价处理类
3
 */
4
@Component
5
@Slf4j
6
public class OrderTimeoutNotEvaluated implements RedisDelayQueueHandle<Map> {
7
    @Override
8
    public void execute(Map map) {
9
        log.info("(收到订单超时未评价延迟消息) {}", map);
10
        // TODO 订单超时未评价，系统默认好评处理业务...
11

12
    }
13
}

7、创建延迟队列消费线程，项目启动完成后开启#

1
/**
2
 * 启动延迟队列监测扫描
3
 */
4
@Slf4j
5
@Component
6
public class RedisDelayQueueRunner implements CommandLineRunner {
7
    @Autowired
8
    private RedisDelayQueueUtil redisDelayQueueUtil;
9
    @Autowired
10
    private ApplicationContext context;
11
    @Autowired
12
    private ThreadPoolTaskExecutor ptask;
13

14
    ThreadPoolExecutor executorService = new ThreadPoolExecutor(3, 5, 30, TimeUnit.SECONDS,
15
        new LinkedBlockingQueue<Runnable>(1000),new ThreadFactoryBuilder().setNameFormat("order-delay-%d").build());
16

17
    @Override
18
    public void run(String... args) throws Exception {
19
        ptask.execute(() -> {
20
            // 无限循环监测延迟队列
21
            while (true){
22
                try {
23
                     // 获取所有延迟队列的枚举值
24
                    RedisDelayQueueEnum[] queueEnums = RedisDelayQueueEnum.values();
25
                    // 遍历枚举值
26
                    for (RedisDelayQueueEnum queueEnum : queueEnums) {
27
                        // 获取延迟队列中的值
28
                        Object value = redisDelayQueueUtil.getDelayQueue(queueEnum.getCode());
29
                        // 如果值不为空，则执行处理操作
30
                        if (value != null) {
31
                            // 获取处理器对象并执行
32
                            RedisDelayQueueHandle<Object> redisDelayQueueHandle = (RedisDelayQueueHandle<Object>)context.getBean(queueEnum.getBeanId());
33
                            // 在线程池中执行处理操作
34
                            executorService.execute(() -> {redisDelayQueueHandle.execute(value);});
35
                        }
36
                    }
37
                    // 控制循环速度为每次 500 毫秒
38
                    TimeUnit.MILLISECONDS.sleep(500);
39
                } catch (InterruptedException e) {
40
                    log.error("(Redission延迟队列监测异常中断) {}", e.getMessage());
41
                }
42
            }
43
        });
44
        log.info("(Redission延迟队列监测启动成功)");
45
    }
46
}

8、创建一个测试接口，模拟添加延迟队列#

1
/**
2
 * 延迟队列测试
3
 */
4
@RestController
5
public class RedisDelayQueueController {
6

7
    @Autowired
8
    private RedisDelayQueueUtil redisDelayQueueUtil;
9

10
    @PostMapping("/addQueue")
11
    public void addQueue() {
12
        Map<String, String> map1 = new HashMap<>();
13
        map1.put("orderId", "100");
14
        map1.put("remark", "订单支付超时，自动取消订单");
15

16
        Map<String, String> map2 = new HashMap<>();
17
        map2.put("orderId", "200");
18
        map2.put("remark", "订单超时未评价，系统默认好评");
19

20
        // 添加订单支付超时，自动取消订单延迟队列。为了测试效果，延迟10秒钟
21
        redisDelayQueueUtil.addDelayQueue(map1, 10, TimeUnit.SECONDS, RedisDelayQueueEnum.ORDER_PAYMENT_TIMEOUT.getCode());
22

23
        // 订单超时未评价，系统默认好评。为了测试效果，延迟20秒钟
24
        redisDelayQueueUtil.addDelayQueue(map2, 20, TimeUnit.SECONDS, RedisDelayQueueEnum.ORDER_TIMEOUT_NOT_EVALUATED.getCode());
25
    }
26

27
}

启动 SpringBoot 项目，用 PostMan 调用接口添加延迟队列通过 Redis 客户端可看到两个延迟队列已添加成功

查看 IDEA 控制台日志可看到延迟队列已消费成功

小结一下#

这两个队列中，带有timeout关键字的那条是一个ZSet集合，另一个是普通的List。

客户端启动，redisson先订阅一个key，同时 BLPOP key 0 无限监听一个阻塞队列（等里面有数据了就返回）。
当有数据put时，redisson先把数据放到一个zset集合（按延时到期时间的时间戳为分数排序），同时发布上面订阅的key，发布内容为数据到期的timeout，此时客户端进程开启一个延时任务，延时时间为发布的timeout。
客户端进程的延时任务到了时间执行，从zset分页取出过了当前时间的数据，然后将数据rpush到第一步的阻塞队列里。然后将当前数据从zset移除，取完之后，又执行 BLPOP key 0 无限监听一个阻塞队列。这一部分的逻辑，客户端会发送一个lua脚本给到服务端去操作：org.redisson.RedissonDelayedQueue源码里面的pushTaskAsync函数有lua脚本内容。
上一步客户端监听的阻塞队列返回取到数据，回调到 RBlockingQueue 的 take方法。于是，我们就收到了数据。

这种设计利用了 Redis 的发布订阅机制和有序集合的特性，结合延时任务和阻塞队列的处理方式，实现了高效的延迟队列功能。这种设计可以确保任务按照延时时间顺序被处理，同时保证了任务的可靠性和高性能。

参考笔记：