基于Rabbitmq和Redis的延迟消息实现

1 基于Rabbitmq延迟消息实现

支付时间设置为30，未支付的消息会积压在mq中，给mq带来巨大压力。我们可以利用Rabbitmq的延迟队列插件实现消息前一分钟尽快处理

1.1定义延迟消息实体

由于我们要多次发送延迟消息，因此需要先定义一个记录消息延迟时间的消息体

@Data
public class MultiDelayMessage {
    /**
     * 消息体
     */
    private T data;
    /**
     * 记录延迟时间的集合
     */
    private List delayMillis;
    public MultiDelayMessage(T data, List delayMillis) {
        this.data = data;
        this.delayMillis = delayMillis;
    }
    public static  MultiDelayMessage of(T data, Long ... delayMillis){
        return new MultiDelayMessage<>(data, CollUtils.newArrayList(delayMillis));
    }
    /**
     * 获取并移除下一个延迟时间
     * @return 队列中的第一个延迟时间
     */
    public Long removeNextDelay(){
        return delayMillis.remove(0);
    }
    /**
     * 是否还有下一个延迟时间
     */
    public boolean hasNextDelay(){
        return !delayMillis.isEmpty();
    }
}

1.2 定义常量，用于记录交换机、队列、RoutingKey等常量

package com.hmall.trade.constants;
public interface MqConstants {
    String DELAY_EXCHANGE = "trade.delay.topic";
    String DELAY_ORDER_QUEUE = "trade.order.delay.queue";
    String DELAY_ORDER_ROUTING_KEY = "order.query";
}

1.3 抽取mq配置到nacos中

spring:
  rabbitmq:
    host: ${hm.mq.host:192.168.150.101} # 主机名
    port: ${hm.mq.port:5672} # 端口
    virtual-host: ${hm.mq.vhost:/hmall} # 虚拟主机
    username: ${hm.mq.un:hmall} # 用户名
    password: ${hm.mq.pw:123} # 密码
    listener:
      simple:
        prefetch: 1 # 每次只能获取一条消息，处理完成才能获取下一个消息

1.4 定义消息处理器

使用延迟消息处理器发送消息

1.5 消息监听与延迟消息再次发送

2 延迟消息实现

DelayQueue:基于JVM，保存在内存中，会出现消息丢失

Rabbitmq的延迟任务：基于TTL和死信交换机

2.1 redis的延迟任务：基于zset的去重和排序功能

1.为什么任务需要存储在数据库中?

延迟任务是一个通用的服务，任何有延迟需求的任务都可以调用该服务，内存数据库的存储是有限的，需要考虑数据持久化的问题，存储数据库中是一种数据安全的考虑

2.为什么使用redis中的两种数据类型，list和zset?

• 原因一: list存储立即执行的任务，zset存储未来的数据
• 原因二:任务量过大以后，zset的性能会下降

  时间复杂度:执行时间(次数) 随着数据规模增长的变化趋势

  • 操作redis中的list命令LPUSH: 时间复杂度: O(1)
  • 操作redis中的zset命令zadd: 时间复杂度: (Mlog(n))

    2.2 设计mybatis映射实体类：

    /**
         * 版本号,用乐观锁
         */
        @Version
        private Integer version;
    乐观锁支持：
    /**
         * mybatis-plus乐观锁支持
         * @return
         */
    @Bean
    public MybatisPlusInterceptor optimisticLockerInterceptor(){
        MybatisPlusInterceptor interceptor = new MybatisPlusInterceptor();
        interceptor.addInnerInterceptor(new OptimisticLockerInnerInterceptor());
        return interceptor;
    }
    

    2.3 创建task类，用于接收添加任务的参数

    @Data
    public class Task implements Serializable {
        /**
         * 任务id
         */
        private Long taskId;
        /**
         * 类型
         */
        private Integer taskType;
        /**
         * 优先级
         */
        private Integer priority;
        /**
         * 执行id
         */
        private long executeTime;
        /**
         * task参数
         */
        private byte[] parameters;
        
    }
    

    2.4 添加任务

    2.4.1 添加任务到数据库中

    addTaskToDb(task);修改任务表和日志表

    @Autowired
        private TaskinfoMapper taskinfoMapper;
        @Autowired
        private TaskinfoLogsMapper taskinfoLogsMapper;
        /**
         * 添加任务到数据库中
         *
         * @param task
         * @return
         */
        private boolean addTaskToDb(Task task) {
            boolean flag = false;
            try {
                //保存任务表
                Taskinfo taskinfo = new Taskinfo();
                BeanUtils.copyProperties(task, taskinfo);
                taskinfo.setExecuteTime(new Date(task.getExecuteTime()));
                taskinfoMapper.insert(taskinfo);
                //设置taskID
                task.setTaskId(taskinfo.getTaskId());
                //保存任务日志数据
                TaskinfoLogs taskinfoLogs = new TaskinfoLogs();
                BeanUtils.copyProperties(taskinfo, taskinfoLogs);
                taskinfoLogs.setVersion(1);
                taskinfoLogs.setStatus(ScheduleConstants.SCHEDULED);
                taskinfoLogsMapper.insert(taskinfoLogs);
                flag = true;
            } catch (Exception e) {
                e.printStackTrace();
            }
            return flag;
        }
    

    2.4.2 添加任务到redis

    addTaskToCache(task);判断任务执行之间是否在现在还是未来五分钟内

    @Autowired
        private CacheService cacheService;
        /**
         * 把任务添加到redis中
         *
         * @param task
         */
        private void addTaskToCache(Task task) {
            String key = task.getTaskType() + "_" + task.getPriority();
            //获取5分钟之后的时间  毫秒值
            Calendar calendar = Calendar.getInstance();
            calendar.add(Calendar.MINUTE, 5);
            long nextScheduleTime = calendar.getTimeInMillis();
            //2.1 如果任务的执行时间小于等于当前时间，存入list
            if (task.getExecuteTime() <= System.currentTimeMillis()) {
                cacheService.lLeftPush(ScheduleConstants.TOPIC + key, JSON.toJSONString(task));
            } else if (task.getExecuteTime() <= nextScheduleTime) {
                //2.2 如果任务的执行时间大于当前时间 && 小于等于预设时间（未来5分钟） 存入zset中
                cacheService.zAdd(ScheduleConstants.FUTURE + key, JSON.toJSONString(task), task.getExecuteTime());
            }
        }
    

    2.5 删除任务

    1、删除数据库任务表，更改日志表任务状态

    2、删除list或者zset中的任务

    在TaskService中添加方法

    /**
         * 取消任务
         * @param taskId        任务id
         * @return              取消结果
         */
    public boolean cancelTask(long taskId);
    

    /**
         * 取消任务
         * @param taskId
         * @return
         */
    @Override
    public boolean cancelTask(long taskId) {
        boolean flag = false;
        //删除任务，更新日志
        Task task = updateDb(taskId,ScheduleConstants.EXECUTED);
        //删除redis的数据
        if(task != null){
            removeTaskFromCache(task);
            flag = true;
        }
        return false;
    }
    /**
         * 删除redis中的任务数据
         * @param task
         */
    private void removeTaskFromCache(Task task) {
        String key = task.getTaskType()+"_"+task.getPriority();
        if(task.getExecuteTime()<=System.currentTimeMillis()){
            cacheService.lRemove(ScheduleConstants.TOPIC+key,0,JSON.toJSONString(task));
        }else {
            cacheService.zRemove(ScheduleConstants.FUTURE+key, JSON.toJSONString(task));
        }
    }
    /**
         * 删除任务，更新任务日志状态
         * @param taskId
         * @param status
         * @return
         */
    private Task updateDb(long taskId, int status) {
        Task task = null;
        try {
            //删除任务
            taskinfoMapper.deleteById(taskId);
            TaskinfoLogs taskinfoLogs = taskinfoLogsMapper.selectById(taskId);
            taskinfoLogs.setStatus(status);
            taskinfoLogsMapper.updateById(taskinfoLogs);
            task = new Task();
            BeanUtils.copyProperties(taskinfoLogs,task);
            task.setExecuteTime(taskinfoLogs.getExecuteTime().getTime());
        }catch (Exception e){
            log.error("task cancel exception taskid={}",taskId);
        }
        return task;
    }
    

    2.6 消费任务

    1、删除list中的数据

    2、使用updateDB删除任务表、跟新日志表

    在TaskService中添加方法

    /**
     * 按照类型和优先级来拉取任务
     * @param type
     * @param priority
     * @return
     */
    public Task poll(int type,int priority);
    

    实现

    /**
         * 按照类型和优先级拉取任务
         * @return
         */
    @Override
    public Task poll(int type,int priority) {
        Task task = null;
        try {
            String key = type+"_"+priority;
            String task_json = cacheService.lRightPop(ScheduleConstants.TOPIC + key);
            if(StringUtils.isNotBlank(task_json)){
                task = JSON.parseObject(task_json, Task.class);
                //更新数据库信息
                updateDb(task.getTaskId(),ScheduleConstants.EXECUTED);
            }
        }catch (Exception e){
            e.printStackTrace();
            log.error("poll task exception");
        }
        return task;
    }
    

    2.7 未来定时任务更新-reids管道

    减少与redis的交互次数

    1、在引导类中添加开启任务调度注解：@EnableScheduling

    2、在service中添加定时任务 @Scheduled(cron = “0 */1 * * * ?”)，每分钟一次

    @Scheduled(cron = "0 */1 * * * ?")
    public void refresh() {
        System.out.println(System.currentTimeMillis() / 1000 + "执行了定时任务");
        // 获取所有未来数据集合的key值
        Set futureKeys = cacheService.scan(ScheduleConstants.FUTURE + "*");// future_*
        for (String futureKey : futureKeys) { // future_250_250
            String topicKey = ScheduleConstants.TOPIC + futureKey.split(ScheduleConstants.FUTURE)[1];
            //获取该组key下当前需要消费的任务数据
            Set tasks = cacheService.zRangeByScore(futureKey, 0, System.currentTimeMillis());
            if (!tasks.isEmpty()) {
                //将这些任务数据添加到消费者队列中
                cacheService.refreshWithPipeline(futureKey, topicKey, tasks);
                System.out.println("成功的将" + futureKey + "下的当前需要执行的任务数据刷新到" + topicKey + "下");
            }
        }
    }
    

    public List