redis工具集之基于zset设计的延迟队列方案

站长

2024年03月28日 12:04 · 阅读数 6

代码仓库

gitee: gitee.com/listen_w/re… github: github.com/jettwangcj/…

背景

但是有同学使用过程中发现这个延迟队列组件有BUG，具体是消息丢失。

所以，需要有其他延迟队列方案来替换，目前调研之后有以下两个思路：

接入新的支持延迟队列消息的消息中间件，例如: Rocketmq。
自研一个延迟队列组件，例如：基于 ZSET 实现。

使用方式

引入依赖

<dependency>
    <groupId>cn.org.wangchangjiu</groupId>
    <artifactId>redis-util-spring-boot-starter</artifactId>
    <version>1.0.1-SNAPSHOT</version>
</dependency>

开启配置
发送延迟消息
接收延迟消息
1. 自动ACK，方法上增加 @RedisDelayMessageListener 注解，topic 和发送的topic对应，值得注意的事，自动ACK只能是一个参数，参数就是所需要接收的消息实体。
2. 手动ACK，方法上增加 @RedisDelayMessageListener 注解，topic 和发送的topic对应，值得注意的事，手动ACK必须是两个参数，第一个参数是所需要接收的消息实体，第二个参数固定是 Acknowledgment，当执行完方法是或者有需要时通过 acknowledgment.acknowledge() 手动提交ACK，如果未提交ACK，那么过一段时间（时间可配置），该消息会被重新投递到延迟队列中。

设计方案流程图

redis工具集之基于zset设计的延迟队列方案

重点流程说明

服务启动后会启动三个线程组：boss 线程池、worker线程池、back线程池；

Boss 线程每隔500毫秒（可配置）周期性发布获取延迟消息的任务；

worker线程根据不同的topic去获取延迟消息，如果获取到延迟消息，那么反射调用目标方法执行，执行成功自动/手动ACK消息；

back线程每隔5秒（可配置）周期性检查是否有未提交ACK的消息，如果有，则重新投递到延迟队列中；
服务启动后，boss 线程会发布批量获取延迟消息的任务，worker线程执行任务，批量获取延迟消息，这么做的原因是，怕服务宕机或者重启时，大量延迟消息未消费，所以一次性批量消费。
发送延迟消息过程：
- 添加消息元数据，元数据存储在 hash数据结构中，key : task_meta_data:{service} (service为服务名称)
- 将消息ID添加到 ZSET数据结构中，value为MessageID，score为当前时间戳+过期时间

实现细节

整体概论

redis工具集之基于zset设计的延迟队列方案

核心Bean 简介

DelayQueueAutoConfiguration：延迟队列自动装配Bean。

DelayQueueMessageProducer：延迟消息发送器。

DelayQueueMessageConsumer：延迟消息消费者，该Bean实现了 BeanPostProcessor 接口，主要作用是，获取被注解 @RedisMessageListener 修饰的方法，把信息封装在 RedisDelayMessageConsumerContainer对象里，方便后面反射调用。

DelayQueue：延迟队列，发送消息/获取消息/提交 ACK 核心处理类，该类封装了 redis + lua的脚本，通过调用lua脚本处理redis延迟消息；

服务启动以及发送消息调用过程

redis工具集之基于zset设计的延迟队列方案

以上流程介绍

容器启动时主要操作如下：
- 检查哪些Bean的方法上有注解@RedisDelayMessageListener标注的，封装目标方法信息，用于反射调用；如下图：会将以下方法信息封装到 RedisDelayMessageConsumerContainer 对象里。实现代码：
- 启动三个线程组，boss 线程池、worker线程池、back线程池
  - Boss 线程每隔500毫秒（可配置）周期性发布获取延迟消息的任务；
  - worker线程根据不同的topic去获取延迟消息，如果获取到延迟消息，那么反射调用目标方法执行，执行成功自动/手动ACK消息；
  - back线程每隔5秒（可配置）周期性检查是否有未提交ACK的消息，如果有，则重新投递到延迟队列中；
发送消息

生成消息ID（UUID），调用lua脚本添加消息
获取消息