利用redission延时队列，实现指定时长后，再执行其他操作

分手后的思念是犯贱 2022-12-21 06:12 186阅读 0赞

**目录**

1，核心原理

redission-delayQueue

监听器设计模式

PostConstruct

2，核心代码

1）RedissionConfig

2）RedisDeloyQueue

3）getQueueValue

3，执行示例

--------------------

# 1，核心原理 #

*  ## [redission-delayQueue][] ##
 *  ## 监听器设计模式 ##

即在类的内部，增加一个事件监听抽象类，类的内部有一个invoke方法。

在需要的地方，执行invoke方法。

如果需要执行监听，则继承该类，并实现这个方法。

好处，具体业务逻辑和队列可以剥离。

*  ## PostConstruct ##

这个注解是指当对象被spring管理后，在对象初始化后，会执行被这个注解注释的方法。类似配置方式中的init-method。

# 2，核心代码 #

## 1）RedissionConfig ##

package com.wayne.core.config;
    
    import io.netty.channel.nio.NioEventLoopGroup;
    import lombok.Data;
    import org.redisson.Redisson;
    import org.redisson.api.RedissonClient;
    import org.redisson.config.Config;
    import org.springframework.beans.factory.annotation.Value;
    import org.springframework.context.annotation.Bean;
    import org.springframework.context.annotation.Configuration;
    
    @Data
    @Configuration
    public class RessionConfig {
    
        @Value("${spring.redis.host}")
        private String host;
    
        @Value("${spring.redis.port}")
        private String port;
    
        @Value("${spring.redis.password}")
        private String password;
    
        @Value("${spring.redssion.threads}")
        private int threads; //当前处理核数量 * 2
    
        @Bean()
        public RedissonClient redissonClient(){
    
            Config config = new Config();
            config.useSingleServer()
                    .setAddress("redis://" + host + ":" + port)
                    .setPassword(password);
    
            config.setThreads(threads);
            config.setEventLoopGroup(new NioEventLoopGroup());
    
            //添加主从配置
    //        config.useMasterSlaveServers().setMasterAddress("").setPassword("").addSlaveAddress(new String[]{"",""});
    
            return Redisson.create(config);
        }
    }

## 2）RedisDeloyQueue ##

package com.wayne.core.utils;
    
    import lombok.extern.slf4j.Slf4j;
    import org.redisson.api.RBlockingQueue;
    import org.redisson.api.RDelayedQueue;
    import org.redisson.api.RedissonClient;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.util.concurrent.TimeUnit;
    
    @Slf4j
    @Component
    public class RedisDelayQueue {
    
        /**
         * 任务回调监听
         *
         * @param <T>
         */
        public abstract static class TaskEventListener<T> {
            /**
             * 执行方法
             *
             * @param t
             */
            public abstract void invoke(T t);
        }
    
        @Autowired
        private RedissonClient redissonClient;
    
        /**
         * 添加队列
         *
         * @param t        DTO传输类
         * @param delay    时间数量
         * @param timeUnit 时间单位
         * @param <T>      泛型
         */
        public <T> void addQueue(T t, long delay, TimeUnit timeUnit) {
            RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(t.getClass().getName());
            RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
            delayedQueue.offer(t, delay, timeUnit);
            //这里给destory了，造成队列在第一次应用启动时，队列没有效果。
            //delayedQueue.destroy();
        }
    
        /**
         * 获取队列
         *
         * @param zClass            DTO泛型
         * @param taskEventListener 任务回调监听
         * @param <T>               泛型
         * @return
         */
        public <T> void getQueue(Class zClass, TaskEventListener taskEventListener) {
            RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(zClass.getName());
            //由于此线程需要常驻，可以新建线程，不用交给线程池管理
            Thread thread = new Thread(() -> {
                while (true) {
                    try {
                        T t = blockingFairQueue.take();
                        log.info(Thread.currentThread().getName() + "取得数据后，开始处理");
                        taskEventListener.invoke(t);
                    } catch (InterruptedException e) {
                        log.error(e.getMessage());
                    }
                }
            });
            thread.start();
        }
    }

## 3）getQueueValue ##

@PostConstruct
        public void getQueueValue(){
            log.info("getQueueValue执行了");
            RedisDelayQueue.TaskEventListener<String> taskEventListener = new RedisDelayQueue.TaskEventListener<String>() {
                @Override
                public void invoke(String text) {
                    log.info("当前线程为："+Thread.currentThread().getName());
                    log.info("队列中的信息："+text);
                }
            };
            redisDelayQueue.getQueue(String.class,taskEventListener);
        }

# 3，执行示例 #

> 2020-11-15 13:33:21,950 INFO (DirectJDKLog.java:173)- Initializing Spring DispatcherServlet 'dispatcherServlet'  
> 2020-11-15 13:33:21,950 INFO (FrameworkServlet.java:525)- Initializing Servlet 'dispatcherServlet'  
> 2020-11-15 13:33:21,963 INFO (FrameworkServlet.java:547)- Completed initialization in 13 ms  
> 2020-11-15 13:33:22,005 INFO (RedisServiceImpl.java:22)- 插入数据==17  
> 2020-11-15 13:33:27,122 INFO (RedisDelayQueue.java:64)- Thread-56取得数据后，开始处理  
> 2020-11-15 13:33:27,122 INFO (RedisServiceImpl.java:32)- 当前线程为：Thread-56  
> 2020-11-15 13:33:27,122 INFO (RedisServiceImpl.java:33)- 队列中的信息：17

[redission-delayQueue]: https://github.com/redisson/redisson/wiki/7.-distributed-collections#715-delayed-queue