Java 延时实现（订单超时等）

野性酷女 2023-10-01 10:56 23阅读 0赞

#### Java 延时任务实现 ####

*  一、Java自带延时队列
 *   *  1.定义需要进行延时处理的对象类型，实现延时接口：Delayed
     *  2.定义对象生成者：
     *  3.定义对象消费者：
     *  4.定义测试类：
     *  5.测试误差 延时触发时间与定义的延迟时间存在毫秒级误差
 *  二、通过Scheduled定时器实现
 *   *  1.定义SpringUtil用于获取定时器任务线程池
     *  2.定义ScheduledDelay配置类
     *  3.调用定时器配置
     *  4.Pom配置
 *  三、Java 时间轮算法（内容过长，见链接）
 *   *  1.Dubbo时间轮解析：\[HashedWheelTimer\](https://blog.csdn.net/weixin\_42176639/article/details/122354714)
 *  四、Redis 过期回调
 *   *  1.Redis轮询
     *   *  1.配置RedisTemplate 使其支持redisTemplate.keys("\*")
         *  2.轮询ZSET集合，并进行超时处理和数据移除（缺点：耗内存、延迟与性能成反比）
         *  3.添加任务接口
         *  4.Pom配置
     *  2.Redis回调
     *   *  1.容器启动命令：
         *  1.开启远程访问：注释掉bing 127.0.0.1
         *  2.修改配置文件：notify-keyspace-events "Ex"
         *  3.监听配置：
         *  5.过期执行方法：
         *  6.调用方法：
         *  7.结果：
 *  五、通过RocketMQ延迟队列实现
 *   *  1.RocketMQ环境搭建：\[RocketMQ\](https://blog.csdn.net/weixin\_42176639/article/details/106630656)
     *  2.延时数据生产者：
     *  3.延时数据消费者：
     *  4.延时数据调用接口：
     *  5.Pom配置：
     *  6.YML配置：
     *  7.调试效果：
     *  8.延时等级调整：

## 一、Java自带延时队列 ##

### 1.定义需要进行延时处理的对象类型，实现延时接口：Delayed ###

package com.demo.entity;
    
    import java.util.concurrent.Delayed;
    import java.util.concurrent.TimeUnit;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /**
     * Description: 延迟数据对象
     * 需要实现延时接口：Delayed 才能将数据存入延时队列 DelayQueue
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2021-12-28 0:13
     * @version: V1.0.0
     */
    public class DelayMessageData implements Delayed {
        
    
        private static AtomicInteger num = new AtomicInteger(0);
    
        /**
         * 数据 ID
         */
        private long orderId;
        /**
         * 生成时间
         */
        private long currentTime;
        /**
         * 延迟周期
         */
        private long expirationTime;
    
        /**
         * 延迟后，失效时间点（创建时用当前时间+失效周期 方便处理）
         */
        private long timeOut;
    
        /**
         * 有参构造
         * @param currentTime
         * @param expirationTime
         */
        public DelayMessageData(long currentTime,long expirationTime){
        
            this.orderId = num.getAndIncrement();
            this.currentTime = currentTime;
            this.expirationTime = expirationTime;
            this.timeOut = currentTime + expirationTime;
        }
    
        /**
         * 获取剩余失效时间
         * @param unit
         * @return
         */
        @Override
        public long getDelay(TimeUnit unit) {
        
            return unit.convert(this.timeOut-System.currentTimeMillis(),TimeUnit.MILLISECONDS);
        }
    
        /**
         * 比较两个对象顺序 一致 0 先失效 -1 后失效 1
         * 1.如果是对象本身 则直接返回 0
         * 2.比较失效时间：先失效返回 -1；后失效返回 1
         * 3.如果失效时间一致： 序号小的返回 -1；否则返回 1
         * 4.数据类型不一致时：直接比较剩余时间（一般不会出现不同类型数据在同一队列）
         *
         * @param o
         * @return
         */
        @Override
        public int compareTo(Delayed o) {
        
    
            if (o.equals(this)){
        
                return 0;
            }
    
            if (o instanceof DelayMessageData){
        
    
                DelayMessageData data = (DelayMessageData) o;
    
                long delay = this.timeOut - data.timeOut;
    
                if (delay < 0){
        
                    return -1;
                }
                else if (delay > 0){
        
                    return 1;
                }
                else if (this.orderId < data.orderId){
        
                    return -1;
                }
                else {
        
                    return 1;
                }
            }
    
            /**
             * 数据类型不一致时
             */
            long delay = this.getDelay(TimeUnit.NANOSECONDS) - o.getDelay(TimeUnit.NANOSECONDS);
    
            if (delay < 0){
        
                return -1;
            }
            else {
        
                return 1;
            }
        }
    
        @Override
        public String toString(){
        
            long nowTime = System.currentTimeMillis();
            return "orderId:" + orderId + ",currentTime:" + currentTime + ",expirationTime:" + expirationTime + ",timeOut:" + timeOut + " Now:" + nowTime + " 误差（MS）:" + (nowTime - timeOut);
        }
    }

### 2.定义对象生成者： ###

package com.demo.config;
    
    import com.demo.entity.DelayMessageData;
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.concurrent.DelayQueue;
    import java.util.concurrent.ThreadLocalRandom;
    
    /**
     * Description: 生产者
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2021-12-28 0:59
     * @version: V1.0.0
     */
    @Slf4j
    public record Producer(DelayQueue<DelayMessageData> queue) implements Runnable {
        
    
        @Override
        public void run() {
        
    
            while (true) {
        
                /**
                 * 通过系统当前时间 + 延迟时间 生成失效时间点
                 * 每 10 秒生成一条数据
                 */
                long currentTime = System.currentTimeMillis();
                long expirationTime = ThreadLocalRandom.current().nextLong(1000, 5000);
                queue.put(new DelayMessageData(currentTime,expirationTime));
                try {
        
                    Thread.sleep(1000*10);
                } catch (Exception e) {
        
                    log.error("消息生产异常：", e);
                }
            }
        }
    }

### 3.定义对象消费者： ###

package com.demo.config;
    
    import com.demo.entity.DelayMessageData;
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.concurrent.DelayQueue;
    
    /**
     * Description: 消费者
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2021-12-28 0:59
     * @version: V1.0.0
     */
    @Slf4j
    public record Consumer(DelayQueue<DelayMessageData> queue) implements Runnable{
        
    
        @Override
        public void run() {
        
            while (true){
        
                try {
        
                    DelayMessageData delayMessageData = queue.take();
                    System.out.println("线程名称：" + Thread.currentThread().getName() + " 数据：" + delayMessageData.toString());
                    Thread.yield();
                }catch (Exception e){
        
                    log.error("消费者异常：",e);
                }
            }
        }
    }

### 4.定义测试类： ###

package com.demo.config;
    
    import com.demo.entity.DelayMessageData;
    
    import java.util.concurrent.DelayQueue;
    
    /**
     * Description: 测试类
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2021-12-28 1:16
     * @version: V1.0.0
     */
    public class Test {
        
    
        public static void main(String[] args) {
        
            DelayQueue<DelayMessageData> queue = new DelayQueue<>();
            Thread t1 = new Thread(new Consumer(queue),"consumer");
            Thread t2 = new Thread(new Producer(queue),"producer");
            t1.start();
            t2.start();
        }
    }

### 5.测试误差 延时触发时间与定义的延迟时间存在毫秒级误差 ###

![在这里插入图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16]

## 二、通过Scheduled定时器实现 ##

### 1.定义SpringUtil用于获取定时器任务线程池 ###

package com.demo.utils;
    
    import org.springframework.beans.BeansException;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.ApplicationContextAware;
    import org.springframework.stereotype.Component;
    
    /**
     * Description: spring 工具类
     *
     * @Date: 2022-01-03 0:06
     * @version: V1.0.0
     */
    @Component
    public class SpringUtil implements ApplicationContextAware {
        
        /**
         * Spring应用上下文环境
         */
        private static ApplicationContext applicationContext;
    
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        
            SpringUtil.applicationContext = applicationContext;
        }
    
        public static ApplicationContext getApplicationContext() {
        
            return applicationContext;
        }
    
        @SuppressWarnings("unchecked")
        public static <T> T getBean(String name) throws BeansException {
        
            return (T) applicationContext.getBean(name);
        }
    
        @SuppressWarnings("unchecked")
        public static <T> T getBean(Class<?> clz) throws BeansException {
        
            return (T) applicationContext.getBean(clz);
        }
    }

### 2.定义ScheduledDelay配置类 ###

package com.demo.config.scheduled;
    
    import com.demo.utils.SpringUtil;
    import io.netty.util.internal.StringUtil;
    import org.springframework.scheduling.TaskScheduler;
    import org.springframework.scheduling.support.CronTrigger;
    import java.util.Date;
    import java.util.Map;
    import java.util.Objects;
    import java.util.concurrent.*;
    
    /**
     * Description: 通过定时器实现延时
     * 动态延时设置
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-01-02 20:59
     * @version: V1.0.0
     */
    
    public class ScheduledDelay{
        
    
        /**
         * Future handles, to cancel the running jobs
         */
        public static final Map<String, ScheduledFuture> FUTURE_MAP = new ConcurrentHashMap<>();
        public static String DEFAULT_CORN="0/1 * * * * *";
        public static Runnable DEFAULT_RUNNABLE=() -> {
        };
    
        /**
         * 动态定时参数
         */
        public String corn;
    
        /**
         * 动态任务
         */
        public Runnable task;
    
        /**
         * TaskScheduler 定时器线程池
         */
        private TaskScheduler taskScheduler;
    
        /**
         * 构造方法
         * @param corn
         * @param task
         */
        public ScheduledDelay(String corn,Runnable task){
        
            if (StringUtil.isNullOrEmpty(corn)){
        
                this.corn = DEFAULT_CORN;
            }else{
        
                this.corn = corn;
            }
            if (Objects.isNull(task)){
        
                this.task = DEFAULT_RUNNABLE;
            }else{
        
                this.task = task;
            }
    
            this.taskScheduler = SpringUtil.getBean("taskScheduler");
        }
    
        /**
         * 启动定时器 并存入HashMap 用于删除
         * @param uuid
         */
        public void start(String uuid){
        
            FUTURE_MAP.put(uuid,taskScheduler.schedule(task,triggerContext -> {
        
                //任务触发，可修改任务的执行周期
                CronTrigger cronTrigger = new CronTrigger(corn);
                Date date = cronTrigger.nextExecutionTime(triggerContext);
                return date;
            }));
        }
    
    }

### 3.调用定时器配置 ###

package com.demo.controller;
    
    import com.demo.config.scheduled.ScheduledDelay;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.web.bind.annotation.GetMapping;
    import org.springframework.web.bind.annotation.RequestMapping;
    import org.springframework.web.bind.annotation.RestController;
    import java.util.UUID;
    
    /**
     * Description: 定时器延时
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-01-02 21:53
     * @version: V1.0.0
     */
    @RestController
    @RequestMapping("/delay")
    @Slf4j
    public class ScheduledController {
        
    
        @GetMapping("/scheduled")
        public void scheduled(){
        
            log.info("scheduled");
            String uuid = UUID.randomUUID().toString();
            ScheduledDelay delay = new ScheduledDelay("0/30 * * * * *",()->{
        
    
                log.info("定时任务开始执行");
    
                //TODO 待执行的方法 可根据需要决定是否停止任务
                
                log.info("定时任务结束执行");
    
                ScheduledDelay.FUTURE_MAP.get(uuid).cancel(true);
            });
            delay.start(uuid);
        }
    }

### 4.Pom配置 ###

<dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-quartz</artifactId>
            <version>2.3.1.RELEASE</version>
        </dependency>

## 三、Java 时间轮算法（内容过长，见链接） ##

### 1.Dubbo时间轮解析：[HashedWheelTimer][] ###

## 四、Redis 过期回调 ##

### 1.Redis轮询 ###

#### 1.配置RedisTemplate 使其支持redisTemplate.keys("\*") ####

#### 2.轮询ZSET集合，并进行超时处理和数据移除（缺点：耗内存、延迟与性能成反比） ####

package com.demo.config.redis;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.boot.ApplicationArguments;
    import org.springframework.boot.ApplicationRunner;
    import org.springframework.data.redis.core.DefaultTypedTuple;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.stereotype.Component;
    import org.springframework.util.CollectionUtils;
    import javax.annotation.Resource;
    import java.util.*;
    
    /**
     * Description: Redis 延迟回调
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-01-03 1:19
     * @version: V1.0.0
     */
    @Slf4j
    @Component
    public class RedisDelay implements ApplicationRunner {
        
    
        @Resource
        private RedisTemplate redisTemplate;
    
        @Override
        public void run(ApplicationArguments args) throws Exception {
        
            while (true){
        
    
                long time = System.currentTimeMillis();
    
                Set<String> key = redisTemplate.opsForZSet().getOperations().keys("*");
    
                for (String k:key){
        
                    /**
                     * 获取超时的数据 开始时间直接设为0即可
                     */
                    Set<DefaultTypedTuple> result = redisTemplate.opsForZSet().rangeByScoreWithScores(k,0,time);
    
                    //TODO 对Result进行超时处理 处理完移除
                    if (!CollectionUtils.isEmpty(result)){
        
    
                        DefaultTypedTuple typedTuple = result.stream().findFirst().get();
    
                        log.info("当前为：{} 结果 {}超时:{}",time, typedTuple.getValue(),typedTuple.getScore());
    
    
                        redisTemplate.opsForZSet().removeRangeByScore(k,0,time);
                    }
                }
    
                /**
                 * 延迟500毫秒 则延迟误差为500毫秒
                 */
                Thread.sleep(500);
            }
        }
    }

#### 3.添加任务接口 ####

package com.demo.controller;
    
    import org.springframework.data.redis.core.RedisTemplate;
    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;
    import java.util.UUID;
    
    /**
     * Description: Redis 延迟时间
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-01-03 2:40
     * @version: V1.0.0
     */
    @RestController
    @RequestMapping("/delay")
    public class RedisController {
        
    
        @Resource
        private RedisTemplate redisTemplate;
    
        private static Integer delayTime = 1000 * 3;
        private static String product = "product_0";
    
        @GetMapping("/zset")
        public void z_set(){
        
    
            redisTemplate.opsForZSet().add(product, UUID.randomUUID().toString(), System.currentTimeMillis() + delayTime);
    
        }
    }

#### 4.Pom配置 ####

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

### 2.Redis回调 ###

#### 1.容器启动命令： ####

docker run -itd -p 6379:6379 --privileged=true --name redis -v /mnt/mount/redis/redis.conf:/usr/local/etc/redis.conf redis:6.2.6 redis-server /usr/local/etc/redis.conf

#### 1.开启远程访问：注释掉bing 127.0.0.1 ####

![在这里插入图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16 1]

#### 2.修改配置文件：notify-keyspace-events “Ex” ####

![在这里插入图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16 2]

#### 3.监听配置： ####

package com.demo.config.redis;
    
    import org.springframework.context.annotation.Bean;
    import org.springframework.data.redis.connection.RedisConnectionFactory;
    import org.springframework.data.redis.listener.ChannelTopic;
    import org.springframework.data.redis.listener.RedisMessageListenerContainer;
    import org.springframework.stereotype.Component;
    
    /**
     * Description: Redis 回调
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-01-03 22:13
     * @version: V1.0.0
     */
    @Component
    public class RedisConfiguration {
        
    
        private RedisConnectionFactory redisConnectionFactory;
        private RedisKeyExpirationListener redisKeyExpirationListener;
    
        public RedisConfiguration(RedisConnectionFactory redisConnectionFactory,RedisKeyExpirationListener redisKeyExpirationListener){
        
            this.redisConnectionFactory = redisConnectionFactory;
            this.redisKeyExpirationListener = redisKeyExpirationListener;
        }
    
        public ChannelTopic expiredTopic(){
        
            /**
             * 选择数据库：1
             */
            return new ChannelTopic("__keyevent@1__:expired");
        }
    
        @Bean
        public RedisMessageListenerContainer redisMessageListenerContainer(){
        
            RedisMessageListenerContainer redisMessageListenerContainer = new RedisMessageListenerContainer();
            redisMessageListenerContainer.setConnectionFactory(redisConnectionFactory);
            redisMessageListenerContainer.addMessageListener(redisKeyExpirationListener,expiredTopic());
            return redisMessageListenerContainer;
        }
    }

#### 5.过期执行方法： ####

package com.demo.config.redis;
    
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.data.redis.connection.Message;
    import org.springframework.data.redis.connection.MessageListener;
    import org.springframework.stereotype.Component;
    
    /**
     * Description: Redis 过期监听
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-01-03 22:21
     * @version: V1.0.0
     */
    @Component
    @Slf4j
    public class RedisKeyExpirationListener implements MessageListener {
        
    
        @Override
        public void onMessage(Message message, byte[] pattern) {
        
            log.info("失效键：{}",message);
            //TODO 失效键需要执行的操作
    
        }
    }

#### 6.调用方法： ####

package com.demo.controller;
    
    import org.springframework.data.redis.core.StringRedisTemplate;
    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;
    import java.util.UUID;
    import java.util.concurrent.TimeUnit;
    
    /**
     * Description: Redis Key失效监听
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-01-03 2:40
     * @version: V1.0.0
     */
    @RestController
    @RequestMapping("/delay")
    public class RedisController {
        
    
        @Resource
        private StringRedisTemplate stringRedisTemplate;
    
        private static String product = "product_0";
    
        @GetMapping("/expiration")
        public void expiration(){
        
            stringRedisTemplate.opsForValue().set(product, UUID.randomUUID().toString(), 5, TimeUnit.SECONDS);
        }
    }

#### 7.结果： ####

![在这里插入图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16 3]

## 五、通过RocketMQ延迟队列实现 ##

### 1.RocketMQ环境搭建：[RocketMQ][] ###

### 2.延时数据生产者： ###

package com.demo.config.rocketmq;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.client.producer.SendResult;
    import org.apache.rocketmq.spring.core.RocketMQTemplate;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.messaging.support.MessageBuilder;
    import org.springframework.stereotype.Component;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * Description: RocketMQ 延时数据生产者
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-01-10 15:29
     * @version: V1.0.0
     */
    @Component
    @Slf4j
    public class DelayProduce {
        
    
        private String messageDelayLevel = "";
    
        /**
         * 延时等级：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
         */
        private enum DELAY_LEVEL {
        
            ONE_L,
            TOW_L,
            THREE_L,
            FOUR_L,
            FIVE_L,
            SIX_L,
            SEVEN_L,
            EIGHT_L,
            NINE_L,
            TEN_L
        }
    
        private final RocketMQTemplate rocketMQTemplate;
    
        @Autowired
        public DelayProduce(RocketMQTemplate rocketMQTemplate) {
        
            this.rocketMQTemplate = rocketMQTemplate;
        }
    
        /**
         * 生成延时数据
         * @param topic
         * @param message
         * @param delayLevel
         */
        public void sendDelayMessage(String topic,String message,int delayLevel){
        
            SendResult sendResult = rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis())).build(), 2000, delayLevel);
            log.info("Send Result is {}",sendResult);
        }
    }

### 3.延时数据消费者： ###

package com.demo.config.rocketmq;
    
    import lombok.extern.slf4j.Slf4j;
    import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
    import org.apache.rocketmq.spring.core.RocketMQListener;
    import org.springframework.stereotype.Component;
    
    import java.util.concurrent.TimeUnit;
    
    /**
     * Description: RocketMQ 延时数据消费者
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-01-10 15:42
     * @version: V1.0.0
     */
    @Component
    @Slf4j
    @RocketMQMessageListener(topic = "delay-topic",consumerGroup = "delay-group")
    public class DelayConsumer implements RocketMQListener<String> {
        
    
    
        /**
         * 消费数据
         * @param s
         */
        @Override
        public void onMessage(String s) {
        
            log.info("消费时间：{} 开始时间：{}", TimeUnit.MILLISECONDS.toSeconds(System.currentTimeMillis()),s);
        }
    }

### 4.延时数据调用接口： ###

package com.demo.controller;
    
    import com.demo.config.rocketmq.DelayProduce;
    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;
    
    /**
     * Description: RocketMQ 延时数据生产
     *
     * @Author: zhx & moon hongxu_1234@163.com
     * @Date: 2022-01-10 15:48
     * @version: V1.0.0
     */
    @RestController
    @RequestMapping("delay")
    public class RocketController {
        
    
        private final DelayProduce delayProduce;
    
        @Autowired
        public RocketController(DelayProduce delayProduce) {
        
            this.delayProduce = delayProduce;
        }
    
        @GetMapping("/rocket")
        public void rocket(int delay){
        
            delayProduce.sendDelayMessage("delay-topic","",delay);
        }
    }

### 5.Pom配置： ###

<dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-spring-boot-starter</artifactId>
        <version>2.2.1</version>
    </dependency>

### 6.YML配置： ###

rocketmq:
      name-server: 106.12.148.211:9876
      producer:
        group: delay-group

### 7.调试效果： ###

RocketMQ 默认延时等级：1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    http://127.0.0.1:8899/delay/rocket?delay=4
    http://127.0.0.1:8899/delay/rocket?delay=3

1月 10, 2022 6:05:19 下午 org.apache.catalina.core.ApplicationContext log
    信息: Initializing Spring DispatcherServlet 'dispatcherServlet'
    2022-01-10 18:05:19.364  INFO 2428 --- [nio-8899-exec-1] o.s.w.s.DispatcherServlet                : Initializing Servlet 'dispatcherServlet'
    2022-01-10 18:05:19.364  INFO 2428 --- [nio-8899-exec-1] o.s.w.s.DispatcherServlet                : Completed initialization in 0 ms
    2022-01-10 18:05:19.415  INFO 2428 --- [nio-8899-exec-1] c.d.c.r.DelayProduce                     : Send Result is SendResult [sendStatus=SEND_OK, msgId=7F000001097C63947C6B323AE0AA0000, offsetMsgId=6A0C94D300002A9F0000000000000135, messageQueue=MessageQueue [topic=delay-topic, brokerName=broker-a, queueId=3], queueOffset=0]
    2022-01-10 18:05:25.832  INFO 2428 --- [nio-8899-exec-3] c.d.c.r.DelayProduce                     : Send Result is SendResult [sendStatus=SEND_OK, msgId=7F000001097C63947C6B323AF9BC0001, offsetMsgId=6A0C94D300002A9F0000000000000269, messageQueue=MessageQueue [topic=delay-topic, brokerName=broker-a, queueId=0], queueOffset=0]
    2022-01-10 18:05:35.843  INFO 2428 --- [MessageThread_1] c.d.c.r.DelayConsumer                    : 消费时间：1641809135 开始时间：1641809125
    2022-01-10 18:05:49.420  INFO 2428 --- [MessageThread_2] c.d.c.r.DelayConsumer                    : 消费时间：1641809149 开始时间：1641809119

### 8.延时等级调整： ###

启动时RocketMQ Broker配置文件映射位置: -v /mnt/bd/rmq/broker/conf/broker.conf:/opt/rocketmq-4.3.2/conf/broker.conf
    默认值18个级别：messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
    修改第一级别时间，在增加一个级别
    messageDelayLevel=20s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h 8s
    修改broker.conf：vim /mnt/bd/rmq/broker/conf/broker.conf
    重启broker：docker restart rmqbroker

![在这里插入图片描述][watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16 4]

调用：
    http://127.0.0.1:8899/delay/rocket?delay=1
    http://127.0.0.1:8899/delay/rocket?delay=19
    
    2022-01-10 18:38:55.915  INFO 2428 --- [nio-8899-exec-5] c.d.c.r.DelayProduce                     : Send Result is SendResult [sendStatus=SEND_OK, msgId=7F000001097C63947C6B3259A5930002, offsetMsgId=6A0C94D300002A9F00000000000005F5, messageQueue=MessageQueue [topic=delay-topic, brokerName=broker-a, queueId=0], queueOffset=0]
    2022-01-10 18:39:15.910  INFO 2428 --- [MessageThread_3] c.d.c.r.DelayConsumer                    : 消费时间：1641811155 开始时间：1641811135
    2022-01-10 18:39:34.989  INFO 2428 --- [nio-8899-exec-7] c.d.c.r.DelayProduce                     : Send Result is SendResult [sendStatus=SEND_OK, msgId=7F000001097C63947C6B325A3E400003, offsetMsgId=6A0C94D300002A9F0000000000000855, messageQueue=MessageQueue [topic=delay-topic, brokerName=broker-a, queueId=3], queueOffset=0]
    2022-01-10 18:39:42.992  INFO 2428 --- [MessageThread_4] c.d.c.r.DelayConsumer                    : 消费时间：1641811182 开始时间：1641811174

[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16]: https://img-blog.csdnimg.cn/bd2d83a801a84c6dbe6c320fbbdb9bc3.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA54yq5oKf6YGT,size_20,color_FFFFFF,t_70,g_se,x_16
[HashedWheelTimer]: https://blog.csdn.net/weixin_42176639/article/details/122354714
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16 1]: https://img-blog.csdnimg.cn/335ce66961cb4390bb50ec67c836cfb6.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA54yq5oKf6YGT,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16 2]: https://img-blog.csdnimg.cn/1ddf91bca2924b78950ea89b5fcd4197.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA54yq5oKf6YGT,size_20,color_FFFFFF,t_70,g_se,x_16
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16 3]: https://img-blog.csdnimg.cn/b17de4b7ea484d57abb17350ec0309f3.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA54yq5oKf6YGT,size_20,color_FFFFFF,t_70,g_se,x_16
[RocketMQ]: https://blog.csdn.net/weixin_42176639/article/details/106630656
[watermark_type_d3F5LXplbmhlaQ_shadow_50_text_Q1NETiBA54yq5oKf6YGT_size_20_color_FFFFFF_t_70_g_se_x_16 4]: https://img-blog.csdnimg.cn/69cb5be0b8b94eb5b0eb04bf2e760729.png?x-oss-process=image/watermark,type_d3F5LXplbmhlaQ,shadow_50,text_Q1NETiBA54yq5oKf6YGT,size_20,color_FFFFFF,t_70,g_se,x_16