devbox
weixin_40725706
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

使用redis实现延时队列的一个简单方案（延时队列的设计方案，源码分享）

作者：weixin_40725706 | 2024-05-26 12:55:57

赞

踩

redis实现延时队列

一、背景，开发场景


        之前负责开发过一个会议室系统，这个系统整体不难，唯一有一个技术难点，就是需要延时任务，会议的开始、结束需要系统自动更新状态（进行中、已结束）以及开始前多分钟、快结束后多少分钟需要发送邮件提醒、短信提醒，以及超时待审批的会议室申请自动审核不通过。这些操作对我们系统来说，不是固定时刻的定时任务轮询就可以的了，因为每一时刻（粒度可能会小到秒）都有可能有执行的。这种开发场景，就需要使用到延时任务了。

        不使用延时任务不可以吗？使用定时任务每隔一段很短的时间轮询去数据库查询数据？当然可以，但是如果数据很大，数据库的处理能力却十分有限，导致系统有性能问题。
        使用延时队列，可以事先每个一段时间，把接下来一段时间内要执行的任务查询出来，放到延时队列。然后每次只需查询延时队列即可，不需要再查询数据库，可以减缓数据库的压力

二、分析方案

1、使用redis的zset数据结构开发一个延时队列

        ZSet数据结构类似于Set结构，只是ZSet结构中，每个元素都会有一个分值，然后所有元素按照分值的大小进行排列，相当于是一个进行了排序的链表。Redis中的ZSet是一个有序的Set，内部使用HashMap和跳表(SkipList)来保证数据的存储和有序，HashMap里放的是成员到score的映射，而跳跃表里存放的是所有的成员，排序依据是HashMap里存的score,使用跳跃表的结构可以获得比较高的查找效率，并且在实现上比较简单。

2、JDK ScheduledExecutorService

3、时间轮

4、redis的key过期回调

        由上分析，使用ScheduledExecutorService虽然简单，但是不支持分布式的部署，没有高可用，最终采取了 使用redis的zset数据结构开发一个延时队列，开发一个支持分布式、轻量简单、低延时、消息可靠的延时队列

三、 代码解析


这个使用redis的延时队列的设计方案分为四部分

  • 延时队列（使用redis的zset结构开发延时队列）
  • 延时任务的具体策略类（使用策略模式，不同延时任务有不同的类获取和执行延时任务）
  • 消费延时任务的监听器，消费到期的消息任务（也是定时器）
  • 增加延时任务的定时器，每隔一段时间执行，加入接下来一段时间内要执行的定时任务

在这里插入图片描述

延时队列

封装延时任务的bean

  1. /**
  2.  * @author zhangxinlin
  3.  * @Description: 消息 封装
  4.  * @date 2020/4/13 23:37
  5.  */
  6. @Data
  7. public class DelayMessage {
  8.     /**
  9.      * 申请单id
  10.      */
  11.     private Integer applyId;
  12.     /**
  13.      * 执行的时间
  14.      */
  15.     private Date executeTime;
  16.  
  17.     /**
  18.      * 任务类型
  19.      */
  20.     private String jobType;
  21.  
  22.     /**
  23.      * 等级，默认0
  24.      * 当超时没有执行成功后重新加入执行的队列时会+1，直到超出最大重试次数则丢弃这个消息
  25.      */
  26.     private int level;
  27.  
  28.     public DelayMessage(){
  29.  
  30.     }
  31.     public DelayMessage(Integer applyId, Date executeTime, String jobType) {
  32.         this.applyId = applyId;
  33.         this.executeTime = executeTime;
  34.         this.jobType = StringUtils.uncapitalize(jobType);
  35.         this.level = 0;
  36.     }
  37. }
  38.

 根据redis的zset结构封装一个延时队列的类
（这个延时队列主要由增加任务到延时队列，从延时队列删除任务）
核心方法是take()：从延时队列获取到期需要执行的消息

  1. /**
  2.  * @author zhangxinlin
  3.  * @Description: redis延时队列
  4.  * @date 2020/4/13 23:48
  5.  */
  6. @Component
  7. @Slf4j
  8. public class DelayQueue {
  9.  
  10.     /**
  11.      * 延时队列的key
  12.      */
  13.     @Value("${queue.key}")
  14.     private String MEETING_QUEUE_KEY;
  15.  
  16.     /**
  17.      * 延时队列任务丢失后的最大重试次数,-1表示无限次
  18.      */
  19.     @Value("${queue.job.max.retires}")
  20.     private int jobMaxRetires;
  21.  
  22.     /**
  23.      * 延时队列任务最大的消费时间,超过时间未完成任务即为丢失，单位：秒
  24.      */
  25.     @Value("${queue.job.max.ms}")
  26.     private int jobMaxMs;
  27.  
  28.     @Autowired
  29.     private RedisTemplate redisTemplate;
  30.     @Autowired
  31.     private RedisLock redisLock;
  32.  
  33.     /**
  34.      * 增加任务到延时队列
  35.      * @param message
  36.      */
  37.     public void add(DelayMessage message){
  38.         //序列化
  39.         String value =JSONObject.toJSONString(message);
  40.         //加入redis
  41.         redisTemplate.opsForZSet().add(MEETING_QUEUE_KEY,value,message.getExecuteTime().getTime());
  42.     }
  43.  
  44.     /**
  45.      * 增加任务到延时队列
  46.      * @param applyId
  47.      * @param executeTime
  48.      * @param jobType
  49.      */
  50.     public void add(Integer applyId, Date executeTime,String jobType){
  51.         DelayMessage message = new DelayMessage(applyId,executeTime,jobType);
  52.         this.add(message);
  53.     }
  54.  
  55.     public void addAll(List<DelayMessage> messages){
  56.         if(!CollectionUtils.isEmpty(messages)){
  57.             messages.stream().forEach(message->{
  58.                 this.add(message);
  59.             });
  60.         }
  61.     }
  62.  
  63.     /**
  64.      * 从延时队列中移除任务
  65.      * @param message
  66.      */
  67.     public void remove(DelayMessage message){
  68.         //序列化
  69.         String value =JSONObject.toJSONString(message);
  70.         redisTemplate.opsForZSet().remove(MEETING_QUEUE_KEY,value);
  71.     }
  72.  
  73.     public void remove(Integer applyId,Date executeTime,String jobType){
  74.         DelayMessage message = new DelayMessage(applyId,executeTime,jobType);
  75.         this.remove(message);
  76.     }
  77.  
  78. //    public void removeRange(List<String> delValues){
  79. //        if(!CollectionUtils.isEmpty(delValues)){
  80. //            redisTemplate.opsForZSet().remove(MEETING_QUEUE_KEY,delValues.toArray());
  81. //        }
  82. //    }
  83.  
  84.     /**
  85.      * 删除处理中队列的消息
  86.      * @param message
  87.      */
  88.     public void removeHandingMessage(DelayMessage message){
  89.         //序列化
  90.         String value =JSONObject.toJSONString(message);
  91.         redisTemplate.opsForZSet().remove(MEETING_QUEUE_KEY+"_HANDING",value);
  92.     }
  93.  
  94.     /**
  95.      * 从队列获取到期需要执行的消息
  96.      * @return
  97.      */
  98.     public List<DelayMessage> take(){
  99.  
  100.         List<DelayMessage> delayMessageList = Collections.emptyList();
  101.         try{
  102.             //加锁，一次只能由一个线程读消息队列，避免同一个消息被重复读
  103.             boolean success = this.redisLock.lock(MEETING_QUEUE_KEY);
  104.             if (!success) {
  105.                 return delayMessageList;
  106.             }
  107.             //获取待处理的任务消息
  108.             List<String> toDoMessageList = this.getMessageByQueue(MEETING_QUEUE_KEY);
  109.             //获取处理中超时的任务消息
  110.             List<String> overTimeMessageList = this.getMessageByQueue(MEETING_QUEUE_KEY+"_HANDING");
  111.  
  112.             //待处理的任务处理，更新待处理队列和进行中队列
  113.             this.handToDoMessage(toDoMessageList);
  114.             //超时的任务处理，更新待处理队列和进行中队列
  115.             this.handOvertimeMessage(overTimeMessageList);
  116.  
  117.             if(!CollectionUtils.isEmpty(toDoMessageList)){
  118.                 delayMessageList = toDoMessageList.stream().map(value->{
  119.                     DelayMessage message = JSONObject.parseObject(value,DelayMessage.class);
  120.                     return message;
  121.                 }).collect(Collectors.toList());
  122.             }
  123.  
  124.         }finally {
  125.             this.redisLock.delete(MEETING_QUEUE_KEY);
  126.         }
  127.  
  128.         return delayMessageList;
  129.     }
  130.  
  131.     /**
  132.      * 根据key获取需要执行的任务消息
  133.      * @param queueKey
  134.      * @return
  135.      */
  136.     private List<String> getMessageByQueue(String queueKey){
  137.         Set<ZSetOperations.TypedTuple<Object>> tuples = redisTemplate.opsForZSet().rangeWithScores(queueKey,0,-1);
  138.         if(queueKey.equals(MEETING_QUEUE_KEY)){
  139.             log.info("当前延时队列 待处理任务数量：{}",tuples.size());
  140.         }
  141.         Iterator<ZSetOperations.TypedTuple<Object>> iterator = tuples.iterator();
  142.         List<String> messageList = null;
  143.         while (iterator.hasNext())
  144.         {
  145.  
  146.             ZSetOperations.TypedTuple<Object> typedTuple = iterator.next();
  147.             long nowScore = System.currentTimeMillis();
  148.             double score = typedTuple.getScore();
  149.             if(nowScore > score){
  150.                 if(messageList == null){
  151.                     messageList = new ArrayList<>();
  152.                 }
  153.                 String value = (String)typedTuple.getValue();
  154.                 messageList.add(value);
  155.             }else{
  156.                 break;
  157.             }
  158.  
  159.         }
  160.         return messageList;
  161.     }
  162.  
  163.     /**
  164.      * 待处理的任务处理，更新待处理队列和进行中队列
  165.      * @param toDoMessageList
  166.      */
  167.     private void handToDoMessage(List<String> toDoMessageList){
  168.         if(!CollectionUtils.isEmpty(toDoMessageList)){
  169.  
  170.             List<String> delValues = toDoMessageList;
  171.             //redis事物
  172.             redisTemplate.execute(new SessionCallback<Object>(){
  173.                 @Override
  174.                 public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
  175.                     //开启事务
  176.                     operations.multi();
  177.                     //从待处理的延迟队列中移除
  178.                     redisTemplate.opsForZSet().remove(MEETING_QUEUE_KEY,delValues.toArray());
  179.                     //加入进行中的延时队列
  180.                     for(String value:delValues){
  181.                         long score = DateUtils.addSecond(new Date(),jobMaxMs).getTime();
  182.                         redisTemplate.opsForZSet().add(MEETING_QUEUE_KEY+"_HANDING",value,score);
  183.                     }
  184.                     //执行事务
  185.                     operations.exec();
  186.                     return null;
  187.                 }
  188.             });
  189.         }
  190.     }
  191.  
  192.     /**
  193.      * 超时的任务处理，更新待处理队列和进行中队列
  194.      * @param overTimeMessageList
  195.      */
  196.     private void handOvertimeMessage(List<String> overTimeMessageList){
  197.         //判断进行中的队列中是否有超时的，如果有则重新加入待处理队列中等待重新消费
  198.         if(!CollectionUtils.isEmpty(overTimeMessageList)){
  199.             List<String> delValues = overTimeMessageList;
  200.             //redis事物
  201.             redisTemplate.execute(new SessionCallback<Object>(){
  202.                 @Override
  203.                 public <K, V> Object execute(RedisOperations<K, V> operations) throws DataAccessException {
  204.                     //开启事务
  205.                     operations.multi();
  206.                     //从处理中的队列移除
  207.                     redisTemplate.opsForZSet().remove(MEETING_QUEUE_KEY+"_HANDING",delValues.toArray());
  208.                     //加入待处理的延时队列
  209.                     for(String value:delValues){
  210.                         DelayMessage message = JSONObject.parseObject(value,DelayMessage.class);
  211.                         //小于最大重试次数加入待处理队列，否则丢掉这个任务
  212.                         if(jobMaxRetires == -1 || message.getLevel() < jobMaxRetires){
  213.                             message.setLevel(message.getLevel()+1);
  214.                             String nowValue = JSONObject.toJSONString(message);
  215.                             redisTemplate.opsForZSet().add(MEETING_QUEUE_KEY,nowValue,message.getExecuteTime().getTime());
  216.                         }
  217.                     }
  218.                     //执行事务
  219.                     operations.exec();
  220.                     return null;
  221.                 }
  222.             });
  223.         }
  224.     }
  225. }

延时任务
上面有了延时队列了，但是延时任务要怎么加入延时队列呢？以及延时任务要如何获取？
不同延时任务有不同的延时任务的获取和执行任务的具体实现是不一样，例如修改会议为进行中、已结束和发送邮件提醒等任务，这些任务获取逻辑是不一样的，我们使用策略模式，把不同的延时任务封装为不同的策略，一个策略类对应一个延时任务的获取和执行。

延时任务的策略类的接口
 

  1. public interface DelayJob {
  2.  
  3.     /**
  4.      * 获取任务名称
  5.      * @return
  6.      */
  7.     String getJobName();
  8.  
  9.     /**
  10.      * 获取当前时间的一定范围时间内需要执行的延时消息
  11.      * @param curDate 当前时间（可指定，不指定时默认当前系统时间）
  12.      * @return
  13.      */
  14.     List<DelayMessage> getJob(Date curDate);
  15.  
  16.     /**
  17.      * 延时任务到期后的执行任务
  18.      * @param message
  19.      * @throws Exception
  20.      */
  21.     void executeJob(DelayMessage message) throws Exception;
  22. }

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/weixin_40725706/article/detail/626758
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号