Spring Boot + Redis 实现延时队列，写得太好了！

忘是亡心i 2022-09-14 00:22 338阅读 0赞

来源：blog.csdn.net/qq330983778/article/details/99341671

## 业务流程 ##

首先我们分析下这个流程

1.  [用户提交任务。首先将任务推送至延迟队列中。][Link 1]
2.  [延迟队列接收到任务后，首先将任务推送至job pool中，然后计算其执行时间。][Link 1]
3.  [然后生成延迟任务（仅仅包含任务id）放入某个桶中][Link 1]
4.  [时间组件时刻轮询各个桶，当时间到达的时候从job pool中获得任务元信息。][Link 1]
5.  [监测任务的合法性如果已经删除则pass。继续轮询。如果任务合法则再次计算时间][Link 1]
6.  [如果合法则计算时间，如果时间合法：根据topic将任务放入对应的ready queue，然后从bucket中移除。如果时间不合法，则重新计算时间再次放入bucket，并移除之前的bucket中的内容][Link 1]
7.  [消费端轮询对应topic的ready queue。获取job后做自己的业务逻辑。与此同时，服务端将已经被消费端获取的job按照其设定的TTR，重新计算执行时间，并将其放入bucket。][Link 1]
8.  [完成消费后，发送finish消息，服务端根据job id删除对应信息。][Link 1]

![65a1d509935a43500c5501465fa574f1.png][]

## 对象 ##

我们现在可以了解到中间存在的几个组件

*  延迟队列，为Redis延迟队列。实现消息传递
 *  Job pool 任务池保存job元信息。根据文章描述使用K/V的数据结构，key为ID，value为job
 *  Delay Bucket 用来保存业务的延迟任务。文章中描述使用轮询方式放入某一个Bucket可以知道其并没有使用topic来区分，个人这里默认使用顺序插入
 *  Timer 时间组件，负责扫描各个Bucket。根据文章描述存在多个Timer，但是同一个Timer同一时间只能扫描一个Bucket
 *  Ready Queue 负责存放需要被完成的任务，但是根据描述根据Topic的不同存在多个Ready Queue。

其中Timer负责轮询，Job pool、Delay Bucket、Ready Queue都是不同职责的集合。

## 任务状态 ##

*  ready：可执行状态，
 *  delay：不可执行状态，等待时钟周期。
 *  reserved：已被消费者读取，但没有完成消费。
 *  deleted：已被消费完成或者已被删除。

## 对外提供的接口 ##

![2e124edd4b5b13831781ca7a31d8ac9d.png][]

## 额外的内容 ##

1.  首先根据状态状态描述，finish和delete操作都是将任务设置成deleted状态。
2.  根据文章描述的操作，在执行finish或者delete的操作的时候任务已经从元数据中移除，此时deleted状态可能只存在极短时间，所以实际实现中就直接删除了。
3.  文章中并没有说明响应超时后如何处理，所以个人现在将其重新投入了待处理队列。
4.  文章中因为使用了集群，所以使用redis的setnx锁来保证多个时间循环处理多个桶的时候不会出现重复循环。这里因为是简单的实现，所以就很简单的每个桶设置一个时间队列处理。也是为了方便简单处理。关于分布式锁可以看我之前的文章里面有描述。

## 实现 ##

现在我们根据设计内容完成设计。这一块设计我们分四步完成

### 任务及相关对象 ###

[目前需要两个对象，一个是任务对象（job）一个负责保存任务引用的对象（delay job）][Link 1]

[任务对象][Link 1]

@Data
    @AllArgsConstructor
    @NoArgsConstructor
    public class Job implements Serializable {
    
        /**
         * 延迟任务的唯一标识，用于检索任务
         */
        @JsonSerialize(using = ToStringSerializer.class)
        private Long id;
    
        /**
         * 任务类型（具体业务类型）
         */
        private String topic;
    
        /**
         * 任务的延迟时间
         */
        private long delayTime;
    
        /**
         * 任务的执行超时时间
         */
        private long ttrTime;
    
        /**
         * 任务具体的消息内容，用于处理具体业务逻辑用
         */
        private String message;
    
        /**
         * 重试次数
         */
        private int retryCount;
        /**
         * 任务状态
         */
        private JobStatus status;
    }

[任务引用对象][Link 1]

@Data
    @AllArgsConstructor
    public class DelayJob implements Serializable {
    
    
        /**
         * 延迟任务的唯一标识
         */
        private long jodId;
        
        /**
         * 任务的执行时间
         */
        private long delayDate;
    
        /**
         * 任务类型（具体业务类型）
         */
        private String topic;
    
    
        public DelayJob(Job job) {
            this.jodId = job.getId();
            this.delayDate = System.currentTimeMillis() + job.getDelayTime();
            this.topic = job.getTopic();
        }
    
        public DelayJob(Object value, Double score) {
            this.jodId = Long.parseLong(String.valueOf(value));
            this.delayDate = System.currentTimeMillis() + score.longValue();
        }
    }

### 容器 ###

目前我们需要完成三个容器的创建，Job任务池、延迟任务容器、待完成任务容器。[Spring Boot 学习笔记][Spring Boot]

job任务池，为普通的K/V结构，提供基础的操作。Spring Boot 基础就不介绍了，推荐下这个实战教程：https://github.com/javastacks/spring-boot-best-practice

@Component
    @Slf4j
    public class JobPool {
        
        @Autowired
        private RedisTemplate redisTemplate;
    
        private String NAME = "job.pool";
        
        private BoundHashOperations getPool () {
            BoundHashOperations ops = redisTemplate.boundHashOps(NAME);
            return ops;
        }
    
        /**
         * 添加任务
         * @param job
         */
        public void addJob (Job job) {
            log.info("任务池添加任务：{}", JSON.toJSONString(job));
            getPool().put(job.getId(),job);
            return ;
        }
    
        /**
         * 获得任务
         * @param jobId
         * @return
         */
        public Job getJob(Long jobId) {
            Object o = getPool().get(jobId);
            if (o instanceof Job) {
                return (Job) o;
            }
            return null;
        }
    
        /**
         * 移除任务
         * @param jobId
         */
        public void removeDelayJob (Long jobId) {
            log.info("任务池移除任务：{}",jobId);
            // 移除任务
            getPool().delete(jobId);
        }
    }

[延迟任务，使用可排序的ZSet保存数据，提供取出最小值等操作][Link 1]

@Slf4j
    @Component
    public class DelayBucket {
    
        @Autowired
        private RedisTemplate redisTemplate;
    
        private static AtomicInteger index = new AtomicInteger(0);
    
        @Value("${thread.size}")
        private int bucketsSize;
    
        private List <String> bucketNames = new ArrayList <>();
    
        @Bean
        public List <String> createBuckets() {
            for (int i = 0; i < bucketsSize; i++) {
                bucketNames.add("bucket" + i);
            }
            return bucketNames;
        }
    
        /**
         * 获得桶的名称
         * @return
         */
        private String getThisBucketName() {
            int thisIndex = index.addAndGet(1);
            int i1 = thisIndex % bucketsSize;
            return bucketNames.get(i1);
        }
    
        /**
         * 获得桶集合
         * @param bucketName
         * @return
         */
        private BoundZSetOperations getBucket(String bucketName) {
            return redisTemplate.boundZSetOps(bucketName);
        }
    
        /**
         * 放入延时任务
         * @param job
         */
        public void addDelayJob(DelayJob job) {
            log.info("添加延迟任务:{}", JSON.toJSONString(job));
            String thisBucketName = getThisBucketName();
            BoundZSetOperations bucket = getBucket(thisBucketName);
            bucket.add(job,job.getDelayDate());
        }
    
        /**
         * 获得最新的延期任务
         * @return
         */
        public DelayJob getFirstDelayTime(Integer index) {
            String name = bucketNames.get(index);
            BoundZSetOperations bucket = getBucket(name);
            Set<ZSetOperations.TypedTuple> set = bucket.rangeWithScores(0, 1);
            if (CollectionUtils.isEmpty(set)) {
                return null;
            }
            ZSetOperations.TypedTuple typedTuple = (ZSetOperations.TypedTuple) set.toArray()[0];
            Object value = typedTuple.getValue();
            if (value instanceof DelayJob) {
                return (DelayJob) value;
            }
            return null;
        }
    
        /**
         * 移除延时任务
         * @param index
         * @param delayJob
         */
        public void removeDelayTime(Integer index,DelayJob delayJob) {
            String name = bucketNames.get(index);
            BoundZSetOperations bucket = getBucket(name);
            bucket.remove(delayJob);
        }
    
    }

[待完成任务，内部使用topic进行细分，每个topic对应一个list集合][Link 1]、

@Component
    @Slf4j
    public class ReadyQueue {
    
        @Autowired
        private RedisTemplate redisTemplate;
    
        private String NAME = "process.queue";
    
        private String getKey(String topic) {
            return NAME + topic;
        }
    
        /**
         * 获得队列
         * @param topic
         * @return
         */
        private BoundListOperations getQueue (String topic) {
            BoundListOperations ops = redisTemplate.boundListOps(getKey(topic));
            return ops;
        }
    
        /**
         * 设置任务
         * @param delayJob
         */
        public void pushJob(DelayJob delayJob) {
            log.info("执行队列添加任务:{}",delayJob);
            BoundListOperations listOperations = getQueue(delayJob.getTopic());
            listOperations.leftPush(delayJob);
        }
    
        /**
         * 移除并获得任务
         * @param topic
         * @return
         */
        public DelayJob popJob(String topic) {
            BoundListOperations listOperations = getQueue(topic);
            Object o = listOperations.leftPop();
            if (o instanceof DelayJob) {
                log.info("执行队列取出任务:{}", JSON.toJSONString((DelayJob) o));
                return (DelayJob) o;
            }
            return null;
        }
        
    }

### 轮询处理 ###

设置了线程池为每个bucket设置一个轮询操作

@Component
    public class DelayTimer implements ApplicationListener <ContextRefreshedEvent> {
    
        @Autowired
        private DelayBucket delayBucket;
        @Autowired
        private JobPool     jobPool;
        @Autowired
        private ReadyQueue  readyQueue;
        
        @Value("${thread.size}")
        private int length;
        
        @Override 
        public void onApplicationEvent(ContextRefreshedEvent contextRefreshedEvent) {
            ExecutorService executorService = new ThreadPoolExecutor(
                    length, 
                    length,
                    0L, TimeUnit.MILLISECONDS,
                    new LinkedBlockingQueue <Runnable>());
    
            for (int i = 0; i < length; i++) {
                executorService.execute(
                        new DelayJobHandler(
                                delayBucket,
                                jobPool,
                                readyQueue,
                                i));
            }
            
        }
    }

### 测试请求 ###

/**
     * 测试用请求
     * @author daify
     **/
    @RestController
    @RequestMapping("delay")
    public class DelayController {
        
        @Autowired
        private JobService jobService;
        /**
         * 添加
         * @param request
         * @return
         */
        @RequestMapping(value = "add",method = RequestMethod.POST)
        public String addDefJob(Job request) {
            DelayJob delayJob = jobService.addDefJob(request);
            return JSON.toJSONString(delayJob);
        }
    
        /**
         * 获取
         * @return
         */
        @RequestMapping(value = "pop",method = RequestMethod.GET)
        public String getProcessJob(String topic) {
            Job process = jobService.getProcessJob(topic);
            return JSON.toJSONString(process);
        }
    
        /**
         * 完成一个执行的任务
         * @param jobId
         * @return
         */
        @RequestMapping(value = "finish",method = RequestMethod.DELETE)
        public String finishJob(Long jobId) {
            jobService.finishJob(jobId);
            return "success";
        }
    
        @RequestMapping(value = "delete",method = RequestMethod.DELETE)
        public String deleteJob(Long jobId) {
            jobService.deleteJob(jobId);
            return "success";
        }
        
    }

## 测试 ##

### 添加延迟任务 ###

通过postman请求：localhost:8000/delay/add

![01604c63dc6723623d95ba3a907fabaa.png][]

此时这条延时任务被添加进了线程池中

2019-08-12 21:21:36.589  INFO 21444 --- [nio-8000-exec-6] d.samples.redis.delay.container.JobPool  : 任务池添加任务：{"delayTime":10000,"id":3,"message":"tag:testid:3","retryCount":0,"status":"DELAY","topic":"test","ttrTime":10000}
    2019-08-12 21:21:36.609  INFO 21444 --- [nio-8000-exec-6] d.s.redis.delay.container.DelayBucket    : 添加延迟任务:{"delayDate":1565616106609,"jodId":3,"topic":"test"}

根据设置10秒钟之后任务会被添加至ReadyQueue中

2019-08-12 21:21:46.744  INFO 21444 --- [pool-1-thread-4] d.s.redis.delay.container.ReadyQueue     : 执行队列添加任务:DelayJob(jodId=3, delayDate=1565616106609, topic=test)

### 获得任务 ###

这时候我们请求localhost:8000/delay/pop

这个时候任务被响应，修改状态的同时设置其超时时间，然后放置在DelayBucket中

2019-08-09 19:36:02.342  INFO 58456 --- [nio-8000-exec-3] d.s.redis.delay.container.ReadyQueue     : 执行队列取出任务:{"delayDate":1565321728704,"jodId":1,"topic":"测试"}
    2019-08-09 19:36:02.364  INFO 58456 --- [nio-8000-exec-3] d.samples.redis.delay.container.JobPool  : 任务池添加任务：{"delayTime":10000,"id":1,"message":"延迟10秒，超时30秒","retryCount":0,"status":"RESERVED","topic":"测试","ttrTime":30000}
    2019-08-09 19:36:02.384  INFO 58456 --- [nio-8000-exec-3] d.s.redis.delay.container.DelayBucket    : 添加延迟任务:{"delayDate":1565321792364,"jodId":1,"topic":"测试"}

按照设计在30秒后，任务假如没有被消费将会重新放置在ReadyQueue中

2019-08-12 21:21:48.239  INFO 21444 --- [nio-8000-exec-7] d.s.redis.delay.container.ReadyQueue     : 执行队列取出任务:{"delayDate":1565616106609,"jodId":3,"topic":"test"}
    2019-08-12 21:21:48.261  INFO 21444 --- [nio-8000-exec-7] d.samples.redis.delay.container.JobPool  : 任务池添加任务：{"delayTime":10000,"id":3,"message":"tag:testid:3","retryCount":0,"status":"RESERVED","topic":"test","ttrTime":10000}

### 任务的删除/消费 ###

现在我们请求：localhost:8000/delay/delete

![4edb935bdfa8c3a58f59d933f0c68eab.png][]

此时在Job pool中此任务将会被移除，此时元数据已经不存在，但任务还在DelayBucket中循环，然而在循环中当检测到元数据已经不存的话此延时任务会被移除。

2019-08-12 21:21:54.880  INFO 21444 --- [nio-8000-exec-8] d.samples.redis.delay.container.JobPool  : 任务池移除任务：3
    2019-08-12 21:21:59.104  INFO 21444 --- [pool-1-thread-5] d.s.redis.delay.handler.DelayJobHandler  : 移除不存在任务:{"delayDate":1565616118261,"jodId":3,"topic":"test"}

（完）

——长按关注——

[Link 1]: http://mp.weixin.qq.com/s?__biz=MzUyNDc0NjM0Nw==&mid=2247492574&idx=2&sn=f27a39ad8bf4540785d08d7d4be889df&chksm=fa2a08dacd5d81cc3b043fcf01b6b0d9f12e0ed43f02a97c0941c5d325d989c6af5fb0276dc7&scene=21#wechat_redirect
[65a1d509935a43500c5501465fa574f1.png]: /images/20220828/d62f347d5cfd468b836c1bc3f7f62a43.png
[2e124edd4b5b13831781ca7a31d8ac9d.png]: /images/20220828/38412f88b9774e5fb1e4441f5a531d3b.png
[Spring Boot]: https://blog.csdn.net/superbeyone/article/details/82356271
[01604c63dc6723623d95ba3a907fabaa.png]: /images/20220828/3a88782475504fd8927188cc724436a6.png
[4edb935bdfa8c3a58f59d933f0c68eab.png]: /images/20220828/1cbfe092eda946299d5c749884aa5c70.png