2种方式用redis实现延时队列

谁践踏了优雅 2023-02-15 10:04 4阅读 0赞

## 第一种：采用redisson ##

**依赖：**

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

**配置：**

spring.redis.host=47.98.182.201
    spring.redis.port=7007
    spring.redis.password=124
    spring.redis.database=2

**添加队列：**

package com.xxxx.faceadd.redis;
    
    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;
    
    @Component
    @Slf4j
    public class RedisDelayedQueue { 
    
        @Autowired
        RedissonClient redissonClient;
    
        /** * 添加队列 * * @param t DTO传输类 * @param delay 时间数量 * @param timeUnit 时间单位 * @param <T> 泛型 */
        public <T> void addQueue(T t, long delay, TimeUnit timeUnit, String queueName) { 
            log.info("添加队列{},delay:{},timeUnit:{}" + queueName, delay, timeUnit);
            RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
            RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
            delayedQueue.offer(t, delay, timeUnit);
            delayedQueue.destroy();
        }
    
    }

**监听队列：**

package com.xxxx.faceadd.redis;
    
    import com.alibaba.fastjson.JSON;
    import lombok.extern.slf4j.Slf4j;
    import org.redisson.api.RBlockingQueue;
    import org.redisson.api.RedissonClient;
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    import org.springframework.beans.BeansException;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.ApplicationContextAware;
    import org.springframework.stereotype.Component;
    
    import java.util.Map;
    
    /** * 初始化队列监听 */
    @Component
    @Slf4j
    public class RedisDelayedQueueInit implements ApplicationContextAware { 
    
        @Autowired
        RedissonClient redissonClient;
    
        /** * 获取应用上下文并获取相应的接口实现类 * * @param applicationContext * @throws BeansException */
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException { 
            Map<String, RedisDelayedQueueListener> map = applicationContext.getBeansOfType(RedisDelayedQueueListener.class);
            map.entrySet().stream().forEach(taskEventListenerEntry->{ 
                String listenerName = taskEventListenerEntry.getValue().getClass().getName();
                startThread(listenerName, taskEventListenerEntry.getValue());
            });
        }
    
        /** * 启动线程获取队列* * * @param queueName queueName * @param redisDelayedQueueListener 任务回调监听 * @param <T> 泛型 * @return */
        private <T> void startThread(String queueName, RedisDelayedQueueListener redisDelayedQueueListener) { 
            RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
            //由于此线程需要常驻，可以新建线程，不用交给线程池管理
            Thread thread = new Thread(() -> { 
                log.info("启动监听队列线程" + queueName);
                while (true) { 
                    try { 
                        T t = blockingFairQueue.take();
                        log.info("监听队列线程{},获取到值:{}", queueName, JSON.toJSONString(t));
                        new Thread(() -> redisDelayedQueueListener.invoke(t)).start();
                    } catch (Exception e) { 
                        log.info("监听队列线程错误,", e);
                        try { 
                            Thread.sleep(10000);
                        } catch (InterruptedException ex) { 
                            log.error("线程出现异常,", ex);
                        }
                    }
                }
            });
            thread.setName(queueName);
            thread.start();
        }
    
    }

**队列事件监听接口，需要实现这个方法：**

package com.xxxx.faceadd.redis;
    
    /** * 队列事件监听接口，需要实现这个方法 * * @param <T> */
    public interface RedisDelayedQueueListener<T> { 
        /** * 执行方法 * * @param t */
        void invoke(T t);
    }

**参数：**

package com.xxxx.faceadd.redis;
    
    import lombok.Data;
    
    import java.io.Serializable;
    
    /** * @author nick */
    @Data
    public class TaskBodyDTO implements Serializable { 
    
        private String name;
    
        private String body;
    
    }

**延时执行代码：**

package com.xxxx.faceadd.redis;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.stereotype.Component;
    
    /** * 监听器 */
    @Component
    @Slf4j
    public class TestListener implements RedisDelayedQueueListener<TaskBodyDTO> { 
    
        @Override
        public void invoke(TaskBodyDTO taskBodyDTO) { 
            //这里调用你延迟之后的代码
            log.info("执行...." + taskBodyDTO.getBody() + "===" + taskBodyDTO.getName());
        }
    }

**测试接口：**

package com.xxxx.faceadd.redis;
    
    
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import java.util.concurrent.TimeUnit;
    
    /** * @author nick */
    @RestController
    @RequestMapping(value = "/test")
    public class RedisTestController { 
    
        @Autowired
        private RedisDelayedQueue redisDelayedQueue;
    
        @GetMapping("/redis")
        private void test(){ 
            TaskBodyDTO taskBody = new TaskBodyDTO();
            taskBody.setBody("测试DTO,3秒之后执行");
            taskBody.setName("测试DTO,3秒之后执行");
            //添加队列3秒之后执行
            redisDelayedQueue.addQueue(taskBody, 10, TimeUnit.SECONDS, TestListener.class.getName());
            taskBody.setBody("测试DTO,10秒之后执行");
            taskBody.setName("测试DTO,10秒之后执行");
            //添加队列10秒之后执行
            redisDelayedQueue.addQueue(taskBody, 20, TimeUnit.SECONDS, TestListener.class.getName());
            taskBody.setBody("测试DTO,20秒之后执行");
            taskBody.setName("测试DTO,20秒之后执行");
            //添加队列30秒之后执行
            redisDelayedQueue.addQueue(taskBody, 30, TimeUnit.SECONDS, TestListener.class.getName());
        }
    }

**第二种方式：采用redis实现**

**依赖：**

<dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-data-redis</artifactId>
                <version>2.3.0.RELEASE</version>
            </dependency>

**配置：**

spring.redis.host=47.98.182.201
    spring.redis.port=7007
    spring.redis.password=124
    spring.redis.database=2

**核心代码：**

package com.xxxx.faceadd.redis;
    
    import com.alibaba.fastjson.JSON;
    import org.apache.commons.collections.CollectionUtils;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.data.redis.core.ZSetOperations;
    
    import java.util.Set;
    import java.util.concurrent.ThreadLocalRandom;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    import java.util.function.Consumer;
    
    /** * @author nick */
    public class RedisDelayedQueueT<T> { 
    
        private static final AtomicInteger COUNTER = new AtomicInteger(1);
    
        private volatile boolean started = false;
    
        private String queueKey;
    
        private Consumer<T> handler;
    
        private Class<T> classOfT;
    
        private StringRedisTemplate stringRedisTemplate;
    
        /** * @param queueKey 队列键值 * @param classOfT 元素的类型 * @param handler 处理器，如果元素到了时间，需要做的处理 */
        public RedisDelayedQueueT(String queueKey, Class<T> classOfT, Consumer<T> handler) { 
            this.queueKey = queueKey;
            this.handler = handler;
            this.classOfT = classOfT;
        }
    
        /** * 往该延时队列中放入数据,达到指定时间时处理 * * @param value 数据 * @param deadLine 截止时间戳，单位是毫秒 * @return 是否执行成功 */
        public boolean putForDeadLine(T value, long deadLine) { 
            if (value == null) { 
                return false;
            }
            long current = System.currentTimeMillis();
            if (deadLine < current) { 
                throw new IllegalArgumentException(String.format("deadline: %d 小于当前时间: %d !", deadLine, current));
            }
            if (stringRedisTemplate == null) { 
                throw new IllegalStateException("请设置stringRedisTemplate!");
            }
            String json = JSON.toJSONString(value);
            Boolean flag = stringRedisTemplate.opsForZSet().add(queueKey, json, deadLine);
            return Boolean.TRUE.equals(flag);
        }
    
        /** * 往该延时队列中放入数据,指定时间后执行 * * @param value 数据 * @param delayedTime 需要延长的时间，单位是毫秒 * @return 是否执行成功 */
        public boolean putForDelayedTime(T value, long delayedTime) { 
            return putForDeadLine(value, System.currentTimeMillis() + delayedTime);
        }
    
        /** * 清除队列中的数据 */
        public void clear() { 
            stringRedisTemplate.opsForZSet().removeRangeByScore(queueKey, Double.MIN_VALUE, Double.MAX_VALUE);
        }
    
        /** * 验证队列是否存在 true 存在 false 不存在 */
        public Boolean verify() { 
            Long value = stringRedisTemplate.opsForZSet().zCard(queueKey);
            return value != null ? value > 0 : false;
        }
    
    
        public void setStringRedisTemplate(StringRedisTemplate stringRedisTemplate) { 
            if (this.stringRedisTemplate == null && !started) { 
                this.stringRedisTemplate = stringRedisTemplate;
                Worker worker = new Worker();
                worker.setName("delayed-queue-task-" + queueKey + "-" + COUNTER.getAndIncrement());
                worker.start();
                started = true;
            }
        }
    
        class Worker extends Thread { 
    
            @Override
            public void run() { 
                while (!Thread.currentThread().isInterrupted()) { 
                    long current = System.currentTimeMillis();
                    Set<ZSetOperations.TypedTuple<String>> typedTuples = stringRedisTemplate.opsForZSet()
                            .rangeByScoreWithScores(queueKey, 0, current, 0, 1);
                    if (CollectionUtils.isNotEmpty(typedTuples)) { 
                        ZSetOperations.TypedTuple<String> next = typedTuples.iterator().next();
                        if (next.getScore() != null && next.getScore() < current) { 
                            Long removedCount = stringRedisTemplate.opsForZSet().remove(queueKey, next.getValue());
                            // 只有一个线程可以删除成功，代表拿到了这个需要处理的数据
                            if (removedCount != null && removedCount > 0) { 
                                handler.accept(JSON.parseObject(next.getValue(), classOfT));
                            }
                        }
                    }
                    try { 
                        TimeUnit.MILLISECONDS.sleep(10L + ThreadLocalRandom.current().nextInt(10));
                    } catch (InterruptedException e) { 
                        Thread.currentThread().interrupt();
                    }
                }
                started = false;
            }
        }
    
    }

**测试接口：**

package com.xxxx.faceadd.redis;
    
    import com.alibaba.fastjson.JSON;
    import org.springframework.data.redis.core.StringRedisTemplate;
    import org.springframework.scheduling.annotation.Async;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    
    import javax.annotation.Resource;
    
    @RestController
    @RequestMapping(value = "/test")
    public class RedisDelayedQueueTController { 
    
        @Resource
        private StringRedisTemplate stringRedisTemplate;
    
    
        private RedisDelayedQueueT<String> redisDelayedQueue = new RedisDelayedQueueT<>("test-queue", String.class, this::head);
        /** * 异步执行方法 * @param t * @param <T> */
        @Async
        public <T> void head(T t) { 
            System.out.println("执行方法"+ JSON.toJSONString(t));
        }
    
    
        @GetMapping("/redisT")
        public void addTest(){ 
            redisDelayedQueue.setStringRedisTemplate(stringRedisTemplate);
            redisDelayedQueue.putForDelayedTime("测试延时定时任务",5000);
        }
    }

参考：[叶飞][Link 1]  
参考：[https://www.wncode.cn/detail/24.html][https_www.wncode.cn_detail_24.html]

[Link 1]: http://www.%E5%8F%B6%E9%A3%9E.com/
[https_www.wncode.cn_detail_24.html]: https://www.wncode.cn/detail/24.html