基于Redis实现DelayQueue延迟队列设计方案

冷不防 2023-10-03 15:55 31阅读 0赞

实现说明：[基于Redis实现DelayQueue延迟队列设计方案][Redis_DelayQueue]

源码地址:[RedisDelayQueue][]

## RedisDelayQueue接入说明 ##

### 一、引入pom ###

将项目中的 redis-delay-queue-core 模块打包 推送到自己公司的中央仓库,然后引入pom依赖

<dependency>
        <artifactId>redis-delay-queue-core</artifactId>
        <groupId>com.shirc</groupId>
        <version>0.0.1-SNAPSHOT</version>
    </dependency>

### 二、将RedisDelayQueue被Spring管理 ###

/**
     * @Description 引入 redisdelayqueue
     * @Author shirenchuang
     * @Date 2019/8/6 5:58 PM
     **/
    @Component
    public class DelayConfig {
        
     
     
        @Autowired
        private RedisTemplate redisTemplate;
     
     
        @Bean
        public RedisDelayQueueContext getRdctx(){
        
            /**传入redisTemplate实例, 第二个参数为项目名 projectName ;不同项目需要设置不一样**/
            RedisDelayQueueContext context =  new RedisDelayQueueContext(redisTemplate,"dq_demo");
            return context;
        }
     
        /**加了这个  可以在其他地方直接   @Autowire RedisDelayQueue 使用了**/
        @Bean
        public RedisDelayQueue getRedisOperation(RedisDelayQueueContext context){
        
            return context.getRedisDelayQueue();
        }
    }

### 三、注册Topic任务 ###

/**
     * @Description 注册延迟队列 Demo
     * @Author shirenchuang
     * @Date 2019/8/8 10:07 AM
     **/
    @Service
    public class DelayQueueDemoJob extends AbstractTopicRegister<DemoArgs> {
        
     
     
        @Override
        public String getTopic() {
        
            return DelayJobTopicEnums.DEMO_TEST.getTopic();
        }
     
        @Override
        public void execute(DemoArgs demoArgs) {
        
            // 延迟任务回调接口
             
            //id : 这个Topic下的唯一值
            String id = demoArgs.getId();
            //重试次数;  如果回调接口超时失败, 调用失败,会自动重试2次; 这个代表重试次数
            int retryCount = demoArgs.getRetryCount();
             
            //DemoArgs 是要继承 Args的;  如果没有自己需要定义的回调参数; 泛型那里直接写  Args就行了
            System.out.println(demoArgs.getTest());
        }
     
        /******下面方法可选重写  根据消费情况可以自行调节 超时时间,线程池大小等等******/
     
        /**
         * 重试2次仍然失败; 通知接口; 可以在这个接口写自己的通知逻辑; 比如发送邮件或者钉钉消息
         * @param demoArgs
         */
        @Override
        public void retryOutTimes(DemoArgs demoArgs) {
        
            super.retryOutTimes(demoArgs);
        }
     
        /**
         * 设置核心线程池数量  默认2
         * @return
         */
        @Override
        public int getCorePoolSize() {
        
            return super.getCorePoolSize();
        }
     
        /**
         * 设置线程池最大线程数量 默认
         * @return
         */
        @Override
        public int getMaxPoolSize() {
        
            return super.getMaxPoolSize();
        }
     
        /**
         * 可以重写 回调方法的超时时间 默认回调接口超时时间  6秒
         * @return
         */
        @Override
        public int getMethodTimeout() {
        
            return super.getMethodTimeout();
        }
    }

上面的DemoArgs

/**
     * @Description 回调参数Demo
     * @Author shirenchuang
     * @Date 2019/8/8 10:07 AM
     **/
    public class DemoArgs  extends Args {
        
     
        private String test;
     
        public String getTest() {
        
            return test;
        }
     
        public void setTest(String test) {
        
            this.test = test;
        }
    }

如果没有自己定义的回调属性，或者只需要一个id ,那么泛型那里传 Args就行了,这个会返回id的;

下面就是已经定义好的Args

public class Args implements Serializable {
        
     
        private static final long serialVersionUID = 66666l;
     
        /**唯一键 不能为空**/
        private String id;
     
        /**
         * 已经重试的次数:
         * 重试机制: 默认重试2次; 总共最多执行3次
         * 添加任务的时候可以设置为<0 的值;则表示不希望重试;
         * 回调接口自己做好幂等
         ***/
        private int retryCount;
     
     
        /**
         * 重入次数:
         * 这里标记的是当前Job某些异常情况导致并没有真正消费到,然后重新放入待消费池的次数;
         * 比如: BLPOP出来了之后,在去获取Job的时候redis超时了,导致没有正常消费掉;
         * 重入次数最大 3次; 避免某些不可控因素出现,超过3次则丢弃
         */
        private int reentry;
     
        public Args() {
        
        }
     
        public Args(String id) {
        
            this.id = id;
        }
     
        public Args(String id, int retryCount) {
        
            this.id = id;
            this.retryCount = retryCount;
        }
     
        public String getId() {
        
            return id;
        }
     
        public void setId(String id) {
        
            this.id = id;
        }
     
        public int getRetryCount() {
        
            return retryCount;
        }
     
        public void setRetryCount(int retryCount) {
        
            this.retryCount = retryCount;
        }
     
        public int getReentry() {
        
            return reentry;
        }
     
        public void setReentry(int reentry) {
        
            this.reentry = reentry;
        }
     
        @Override
        public String toString() {
        
            return "Args{" +
                    "id='" + id + '\'' +
                    ", retryCount=" + retryCount +
                    ", reentry=" + reentry +
                    '}';
        }

### 四、建议新建一个Topic的枚举类 ###

因为 新增任务的Topic 和 注册地方的Topic,还有删除Topic 要一致,建议用枚举

/**
     * @Description 所有延迟任务的Topic
     * @Author shirenchuang
     **/
    public enum DelayJobTopicEnums {
        
     
         DEMO_TEST("DEMO_TEST","测试"),
        ;
     
        private String topic;
     
        private String desc;
     
        DelayJobTopicEnums(String topic, String desc) {
        
            this.topic = topic;
            this.desc = desc;
        }
     
        public String getTopic() {
        
            return topic;
        }
     
        public String getDesc() {
        
            return desc;
        }
     
    }

### 五、如何新增一个延迟任务 、删除一个延迟任务 ###

/**
     * @Description 新增删除延迟任务
     * @Author shirenchuang
     * @Date 2019/8/8 10:29 AM
     **/
    @Component
    public class DelayQueueUseDemo {
        
     
        @Autowired
        RedisDelayQueue redisDelayQueue;
     
     
        private void addDelayQueue(){
        
     
            //do something
     
            //新增一个延迟任务
            DemoArgs demoArgs = new DemoArgs();
            demoArgs.setId(UUID.randomUUID().toString());
            //设置-1 表示我不想要重试
            demoArgs.setRetryCount(-1);
            demoArgs.setTest("我是个Test");
            //异步新增一个  一分钟之后执行的延时任务
            redisDelayQueue.addAsync(demoArgs,DelayJobTopicEnums.DEMO_TEST.getTopic(),60000);
     
            //也可以同步新增延迟任务
            redisDelayQueue.add(demoArgs,60000,DelayJobTopicEnums.DEMO_TEST.getTopic(),RunTypeEnum.SYNC);
     
            //也可以指定某个时间点执行
            redisDelayQueue.add(demoArgs,DelayJobTopicEnums.DEMO_TEST.getTopic(),System.currentTimeMillis()+60000,RunTypeEnum.ASYNC);
     
            /**PS:如果同一个ID添加了多次,以最新添加的为准,会覆盖之前的**/
     
     
            /**如果自己没有需要回调的参数 直接用Args**/
            redisDelayQueue.addAsync(new Args(id),DelayJobTopicEnums.DEMO_TEST.getTopic(),60000);
        }
         
         
        //删除一个之前添加的延迟任务
        private void  delDelayQueue(String id){
        
            //异步删除
            redisDelayQueue.deleteAsync(DelayJobTopicEnums.DEMO_TEST.getTopic(),id);
            //同步删除
            redisDelayQueue.delete(DelayJobTopicEnums.DEMO_TEST.getTopic(),id,RunTypeEnum.SYNC);
        }
    }

### 六、Redis任务查看 ###

![在这里插入图片描述][watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ZOS5LiN5q2755qE5bCP5bCP5by6_size_20_color_FFFFFF_t_70_g_se_x_16]

[Redis_DelayQueue]: https://blog.csdn.net/u010634066/article/details/98864764
[RedisDelayQueue]: https://gitee.com/songwaimai/redis-delay-queue
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA5ZOS5LiN5q2755qE5bCP5bCP5by6_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/57c63e019f414d10a472ee7f0cd0a1ee.png?x-oss-process=image/watermark,type_ZHJvaWRzYW5zZmFsbGJhY2s,shadow_50,text_Q1NETiBA5ZOS5LiN5q2755qE5bCP5bCP5by6,size_20,color_FFFFFF,t_70,g_se,x_16