用户自定义定时任务、分布式延时队列

叁歲伎倆 2023-10-02 20:23 11阅读 0赞

#### 文章目录 ####

*  前言
 *  一、常见的延时队列对比
 *  二、基于redisson实现一个分布式延时队列
 *   *  1.引入库
     *  2.封装接口
     *  3.核心实现
     *  4.封装推送消息的方法
     *  5.编写延时任务监听
     *  6.配置yml
     *  7.测试推送消息
 *  总结

--------------------

## 前言 ##

很多同学经常遇到用户指定时间的定时任务、自定义延迟时间任务的需求。指定时间定时任务可使用循环方式或者延时任务解决，但是循环执行消耗我们服务器资源，大多时间是无用功。延时任务常用的解决方案有很多，本文主要讲基于redisson的分布式延时队列实现。  
[代码地址][Link 1]

--------------------

## 一、常见的延时队列对比 ##

<table> 
 <thead> 
  <tr> 
   <th>延时队列</th> 
   <th>支持分布式</th> 
   <th>支持自定义时间</th> 
   <th>支持删除延时任务</th> 
   <th>是否高消耗资源</th> 
   <th>延迟时间是否稳定</th> 
   <th>是否支持集群</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>jdk自带delayed</td> 
   <td>否</td> 
   <td>是</td> 
   <td>是</td> 
   <td>否</td> 
   <td>是</td> 
   <td>否</td> 
  </tr> 
  <tr> 
   <td>rabbitmq的死信队列</td> 
   <td>是</td> 
   <td>是</td> 
   <td>否</td> 
   <td>否</td> 
   <td>否</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>rabbitmq的delayed-queue插件</td> 
   <td>是</td> 
   <td>不完全（延迟时间类型是Integer，最多延迟24天）</td> 
   <td>否</td> 
   <td>否</td> 
   <td>是</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>rocketmq</td> 
   <td>是</td> 
   <td>否</td> 
   <td>否</td> 
   <td>否</td> 
   <td>是</td> 
   <td>是</td> 
  </tr> 
  <tr> 
   <td>redisson的RdelayedQueue</td> 
   <td>是</td> 
   <td>是</td> 
   <td>是</td> 
   <td>否</td> 
   <td>是</td> 
   <td>是</td> 
  </tr> 
 </tbody> 
</table>

## 二、基于redisson实现一个分布式延时队列 ##

### 1.引入库 ###

<?xml version="1.0" encoding="UTF-8"?>
    <project xmlns="http://maven.apache.org/POM/4.0.0"
             xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
             xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
        <parent>
            <artifactId>timed-task</artifactId>
            <groupId>org.hjrpc</groupId>
            <version>1.0-SNAPSHOT</version>
        </parent>
        <modelVersion>4.0.0</modelVersion>
    
        <artifactId>distributed-timed-task-starter</artifactId>
    
        <properties>
            <maven.compiler.source>8</maven.compiler.source>
            <maven.compiler.target>8</maven.compiler.target>
        </properties>
        <dependencies>
            <dependency>
                <groupId>org.springframework.boot</groupId>
                <artifactId>spring-boot-starter-web</artifactId>
                <version>2.2.5.RELEASE</version>
                <scope>compile</scope>
            </dependency>
            <dependency>
                <groupId>org.redisson</groupId>
                <artifactId>redisson-spring-boot-starter</artifactId>
                <version>3.16.3</version>
                <scope>compile</scope>
            </dependency>
            <dependency>
                <groupId>org.projectlombok</groupId>
                <artifactId>lombok</artifactId>
                <version>1.18.12</version>
                <scope>compile</scope>
            </dependency>
        </dependencies>
    </project>

### 2.封装接口 ###

/**
     * 分布式演示队列任务监听接口
     * 底层已经使用异步执行
     * @param <T>
     */
    public interface DistributedDelayedQueueListener<T> {
        
        /**
         * 重启后初始化待监听的数据到分布式延时队列，需要支持幂等
         */
        void init();
    
        /**
         * 监听到消息执行调用的方法     * @param t 泛型参数
         */
        void invoke(T t);
    }

### 3.核心实现 ###

package com.hjrpc.timedtask;
    
    import lombok.extern.slf4j.Slf4j;
    import org.redisson.api.RBlockingQueue;
    import org.redisson.api.RLock;
    import org.redisson.api.RedissonClient;
    import org.springframework.boot.context.event.ApplicationStartedEvent;
    import org.springframework.context.ApplicationListener;
    import org.springframework.data.redis.core.RedisTemplate;
    import org.springframework.util.CollectionUtils;
    import reactor.util.annotation.NonNull;
    
    import javax.annotation.PreDestroy;
    import java.util.ArrayList;
    import java.util.List;
    import java.util.concurrent.ExecutorService;
    import java.util.concurrent.Executors;
    import java.util.concurrent.TimeUnit;
    
    @Slf4j
    public class DistributedDelayedQueueInit implements ApplicationListener<ApplicationStartedEvent> {
        
        private final List<DistributedDelayedQueueListener<?>> distributedDelayedQueueListenerList;
        private final RedissonClient redissonClient;
        private final RedisTemplate<String, Object> redisTemplate;
        private final ExecutorService executor = Executors.newFixedThreadPool(10, r -> {
        
            Thread thread = new Thread(r);
            thread.setName("delay-base");
            return thread;
        });
        private final List<Thread> businessThread = new ArrayList<>();
        public static final String LOCK_PREFIX = "DISTRIBUTED_DELAYED_QUEUE_INIT_LOCK";
        public static final String FLAG_PREFIX = "DISTRIBUTED_DELAYED_QUEUE_INIT_FLAG";
    
        public DistributedDelayedQueueInit(List<DistributedDelayedQueueListener<?>> distributedDelayedQueueListenerList
                , RedissonClient redissonClient, RedisTemplate<String, Object> redisTemplate) {
        
            this.distributedDelayedQueueListenerList = distributedDelayedQueueListenerList;
            this.redissonClient = redissonClient;
            this.redisTemplate = redisTemplate;
        }
    
        @PreDestroy
        public void destroy() {
        
            log.info("开始销毁延时队列任务监听和业务线程...");
            executor.shutdown();
            businessThread.forEach(Thread::interrupt);
            log.info("结束销毁延时队列任务监听和业务线程...");
        }
    
        /**
         * 启动监听任务线程
         *
         * @param queueName 队列名称
         * @param listener  监听器
         * @param <T>       泛型
         */
        private <T> void startThread(String queueName, DistributedDelayedQueueListener<T> listener) {
        
            Thread thread = new Thread(() -> {
        
                while (!Thread.currentThread().isInterrupted()) {
        
                    RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
                    redissonClient.getDelayedQueue(blockingFairQueue);
                    try {
        
                        T t = blockingFairQueue.take();
                        log.info("监听队列线程{},获取到值:{}", queueName, t);
                        executor.execute(() -> {
        
                            try {
        
                                listener.invoke(t);
                            } catch (Exception e) {
        
                                log.error("延迟队列{}发送任务异常：", queueName, e);
                            }
                        });
                    } catch (InterruptedException e) {
        
                        Thread.currentThread().interrupt();
                    }
                }
            });
            thread.setName("delay-business");
            thread.start();
            // 缓存线程，方便停止的时候释放
            businessThread.add(thread);
            log.info("=====>启动监听任务{}成功...", queueName);
        }
    
        @Override
        public void onApplicationEvent(@NonNull ApplicationStartedEvent applicationStartedEvent) {
        
            if (CollectionUtils.isEmpty(distributedDelayedQueueListenerList)) {
        
                return;
            }
            log.info("===> 开始初始化分布式延时队列 ...");
            // 读取所有监听器（实现DistributedDelayedQueueListener接口，且注入到spring的bean）
            for (DistributedDelayedQueueListener<?> listener : distributedDelayedQueueListenerList) {
        
                // 获取监听器的class名称作为一个延时队列的名称（这里获取的是子类名）,如果是走代理的话，这里获取特殊处理一下
                String className = handlerListenerClassName(listener);
                // 给不同的延时队列名创建对应的延时队列
                startThread(className, listener);
                // 上分布式锁
                RLock lock = redissonClient.getLock(LOCK_PREFIX + ":" + className);
                if (lock.tryLock()) {
        
                    try {
        
                        String listenerFlag = FLAG_PREFIX + ":" + className;
                        if (Boolean.TRUE.equals(redisTemplate.opsForValue().setIfAbsent(listenerFlag, "已初始化"
                                , 5, TimeUnit.MINUTES))) {
        
                            // 初始化需要监听的数据
                            listener.init();
                        }
                    } finally {
        
                        lock.unlock();
                    }
                }
            }
            log.info("===> 延时队列初始化完成 ...");
        }
    
        private String handlerListenerClassName(DistributedDelayedQueueListener<?> listener) {
        
            String name = listener.getClass().getName();
            int index = name.indexOf("$");
            if (index >= 0) {
        
                name = name.substring(0, index);
            }
            return name;
        }
    }

### 4.封装推送消息的方法 ###

import lombok.extern.slf4j.Slf4j;
    import org.redisson.api.RBlockingQueue;
    import org.redisson.api.RDelayedQueue;
    import org.redisson.api.RedissonClient;
    
    import java.util.Collection;
    import java.util.Date;
    import java.util.concurrent.TimeUnit;
    
    @Slf4j
    @Component
    public class DistributedDelayedQueue {
        
        private final RedissonClient redissonClient;
    
        public DistributedDelayedQueue(RedissonClient redissonClient) {
        
            this.redissonClient = redissonClient;
        }
    
        /**
         * 添加延时队列任务
         *
         * @param t         演示队列DTO对象
         * @param delay     延时时间
         * @param timeUnit  延时时间的单位
         * @param queueName 队列名称
         * @param <T>       泛型
         */
        public <T> void addDelayedTask(T t, long delay, TimeUnit timeUnit, String queueName) {
        
            if (log.isDebugEnabled()) {
        
                log.debug("添加队列:{},delay:{},timeUnit:{},data:{}", queueName, delay, timeUnit, t);
            }
            // 兼容延迟时间负数问题
            if (delay < 0) {
        
                delay = 0;
            }
            // 默认为公平队列
            RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(queueName);
            RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
            delayedQueue.offer(t, delay, timeUnit);
        }
    
        /**
         * @param t        演示队列DTO对象
         * @param delay    延时时间
         * @param timeUnit 延时时间的单位
         * @param clazz    取对应监听器的class
         * @param <T>      泛型
         */
        public <T> void addDelayedTask(T t, long delay, TimeUnit timeUnit, Class<?> clazz) {
        
            addDelayedTask(t, delay, timeUnit, clazz.getName());
        }
    
        /**
         * @param t     演示队列DTO对象
         * @param dest  指定延迟到的日期
         * @param clazz 取对应监听器的class
         * @param <T>   泛型
         */
        public <T> void addDelayedTask(T t, Date dest, Class<?> clazz) {
        
            long delay = dest.getTime() - System.currentTimeMillis();
            addDelayedTask(t, delay, TimeUnit.MILLISECONDS, clazz);
        }
    
        /**
         * 删除延时任务，注意对象必须是同一个，redis采用默认序列化吗，没有采用json序列化。
         * 在业务量不是很大时，推荐在任务弹出时判断任务状态是否发生变化
         *
         * @param t     演示队列DTO对象
         * @param clazz 取对应监听器的class
         * @param <T>   泛型
         */
        public <T> void removeDelayedTask(T t, Class<?> clazz) {
        
            if (log.isDebugEnabled()) {
        
                log.debug("移除队列:{},data:{}", clazz.getName(), t);
            }
            // 默认为公平队列
            RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(clazz.getName());
            RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
            try {
        
                delayedQueue.remove(t);
            } catch (Exception e) {
        
                log.error("删除任务[{}]异常:", t, e);
            }
        }
    
        /**
         * 删除延时任务，注意对象必须是同一个，redis采用默认序列化吗，没有采用json序列化。
         * * 在业务量不是很大时，推荐在任务弹出时判断任务状态是否发生变化 * *
         *
         * @param collections 被批量删除的对象集合
         * @param clazz       取对应监听器的class
         * @param <T>         泛型
         */
        public <T> void removeAllDelayedTask(Collection<T> collections, Class<?> clazz) {
        
            if (log.isDebugEnabled()) {
        
                log.debug("移除队列:{},data:{}", clazz.getName(), collections);
            }
            // 默认为公平队列
            RBlockingQueue<T> blockingFairQueue = redissonClient.getBlockingQueue(clazz.getName());
            RDelayedQueue<T> delayedQueue = redissonClient.getDelayedQueue(blockingFairQueue);
            try {
        
                delayedQueue.removeAll(collections);
            } catch (Exception e) {
        
                log.error("删除任务[{}]异常:", collections, e);
            }
        }
    }

### 5.编写延时任务监听 ###

package com.hjrpc.demo.listener;
    
    import com.hjrpc.timedtask.DistributedDelayedQueue;
    import com.hjrpc.timedtask.DistributedDelayedQueueListener;
    import lombok.extern.slf4j.Slf4j;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    
    import java.text.SimpleDateFormat;
    import java.util.Date;
    import java.util.HashSet;
    import java.util.Set;
    
    @Slf4j
    @Component
    public class MyListener implements DistributedDelayedQueueListener<Integer> {
        
        @Autowired
        private DistributedDelayedQueue distributedDelayedQueue;
    
        @Override
        public void init() {
        
            // 项目启动时初始化历史数据到延时队列
            log.info("项目启动时初始化历史数据");
            Set<Integer> todoTaskIds = new HashSet<>();
            distributedDelayedQueue.removeAllDelayedTask(todoTaskIds, this.getClass());
        }
    
        @Override
        public void invoke(Integer integer) {
        
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            log.info("{}监听到定时任务，{}", format.format(new Date()), integer);
        }
    }

### 6.配置yml ###

spring:
      redis:
        host: 192.168.**.**
        port: 6379

### 7.测试推送消息 ###

@RestController
    @Slf4j
    @Api("测试定时任务")
    public class TestController {
        
        @Autowired
        private DistributedDelayedQueue distributedDelayedQueue;
    
    
        @PostMapping("/pushTask")
        @ApiOperation("推送定时任务")
        public void pushTask(@RequestBody TaskDTO taskDTO) {
        
            SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            Date destDate = taskDTO.getDestDate();
            log.info("推送定时任务{}，在{}执行", taskDTO.getTaskNum(), format.format(destDate));
            distributedDelayedQueue.addDelayedTask(taskDTO.getTaskNum(), destDate, MyListener.class);
        }
    }

--------------------

## 总结 ##

如果您仔细阅读代码会发现，当延时任务的时间过长时，我们的延时任务会一直保存在redis中，浪费redis空间，可考虑优化成定时任务每天初始化一次来实现。

[Link 1]: https://gitee.com/HuJianMonn/timed-task.git