ThreadPoolTaskScheduler自定义定时任务框架

柔情只为你懂 2024-03-22 14:48 79阅读 0赞

### 一、背景 ###

之前用的@Sceduler定时框架，服务启动后就不能重新修改执行计划（或许有，我孤陋寡闻），后来换成这个，可以随时添加，删除，修改定时任务，配置修改发布，实时生效。主要内容包含：

### 二、说明 ###

*  `ScheduleConfig`，自定义配置，初始化设置调度线程池，以及任务缓存；
 *  `CustomizeTask`，自定义任务，任务的key和cron执行计划，以及任务执行入口（runable）；
 *  `DynamicScheduledTask`，动态创建、删除、修改任务
 *  `ExecuteScheduleJob`，任务具体执行入口，封装每个任务的具体业务逻辑
 *  `StartInitScheduler`，服务启动初始化任务
 *  `NacosListener`，我用的nacos配置中心，监听配置变化，cron变化，修改任务（这个和下面的`OrderDeliveryWarningController`二选一，因为业务上不太可能让任务既在页面能操作，又配置在配置文件中）
 *  `OrderDeliveryWarningController`，定时任务管理中心，在页面启动停止更新任务，调用这些任务（和`NacosListener`二选一，原因同上）

### 三、具体的代码 ###

*  自定义定时任务配置类

@Configuration
    public class ScheduleConfig {
        
    
        // 线程池大小，任务不多的话，按需设置
        private static final int POOL_SIZE = 5;
    
        // 用来存入线程执行情况, 方便于停止定时任务时使用
        public static ConcurrentHashMap<String, ScheduledFuture> cache= new ConcurrentHashMap<>();
    
        /**
         * 多几个线程。源码中ThreadPoolTaskScheduler默认大小是1
         * private volatile int poolSize = 1;
         * @return
         */
        @Bean
        public ThreadPoolTaskScheduler threadPoolTaskScheduler(){
        
            ThreadPoolTaskScheduler threadPoolTaskScheduler = new ThreadPoolTaskScheduler();
            threadPoolTaskScheduler.setPoolSize(POOL_SIZE);                        // 线程池大小，我们现在任务不多，可以小一点
            threadPoolTaskScheduler.setThreadNamePrefix("taskExecutor-");   // 线程名称
            threadPoolTaskScheduler.setAwaitTerminationSeconds(60);         // 等待时长
    //        threadPoolTaskScheduler.setWaitForTasksToCompleteOnShutdown(true);  // 调度器shutdown被调用时等待当前被调度的任务完成
            return threadPoolTaskScheduler;
        }
    }

*  自定义任务类

@Slf4j
    @Data
    @NoArgsConstructor
    public class CustomizeTask implements Runnable {
        
    
        private String name;    // 任务名字
        private String cron;    // 触发条件
    
        public CustomizeTask(String name, String cron) {
        
            this.name = name;
            this.cron = cron;
        }
    
        public CustomizeTask(String name) {
        
            this.name = name;
        }
    
        @Override
        public void run() {
        
            
            // 后续有其他类型的任务，在case中添加
            switch (name) {
        
                // 自定义，根据任务名称，进行不同的业务操作
                case Constant.SCHEDULER_TYPE_SYNC: {
        
                    // 业务代码执行
                    ExecuteScheduleJob.syncTenantUser();
                    break;
                }
                default: {
        
                    log.error("没有匹配到任务模块");
                    break;
                }
            }
        }
    }

*  动态新增、删除、修改定时任务类

@Component
    @Slf4j
    public class DynamicScheduledTask {
        
    
        private final ThreadPoolTaskScheduler threadPoolTaskScheduler;
    
    
        @Autowired
        public DynamicScheduledTask(ThreadPoolTaskScheduler threadPoolTaskScheduler) {
        
            this.threadPoolTaskScheduler = threadPoolTaskScheduler;
        }
    
        public List<String> getTask(String dbName) {
        
            if (StringUtils.isNotEmpty(dbName)) {
        
                return ScheduleConfig.cache.keySet().stream().filter(e -> e.contains(dbName + "###")).collect(Collectors.toList());
            } else {
        
                return new ArrayList<>(ScheduleConfig.cache.keySet());
            }
        }
    
        public void startTask(CustomizeTask customizeTask) {
        
            String taskKey = customizeTask.getName();
            if (ScheduleConfig.cache.get(taskKey) != null) {
        
                return;
            }
            log.info("启动定时任务：{}", JSON.toJSONString(customizeTask));
            // 开始执行调度
            ScheduledFuture scheduledFuture = threadPoolTaskScheduler.schedule(customizeTask, new CronTrigger(customizeTask.getCron()));
            ScheduleConfig.cache.put(taskKey, scheduledFuture);
            // 将 scheduledFuture 保存下来用于停止任务使用
        }
    
        public void endTask(CustomizeTask customizeTask) {
        
            String taskKey = customizeTask.getName();
            if (ScheduleConfig.cache.isEmpty() || ScheduleConfig.cache.get(taskKey) == null) {
        
                return;
            }
            log.info("结束定时任务：{}", JSON.toJSONString(customizeTask));
            ScheduledFuture scheduledFuture = ScheduleConfig.cache.get(taskKey);
            if (scheduledFuture != null) {
        
                scheduledFuture.cancel(true);   // 这里需要使用指定的 scheduledFuture 来停止当前的线程
                ScheduleConfig.cache.remove(taskKey);        // 移除缓存
            }
        }
    
        /**
         * 改变调度的时间，先停止定时器再启动新的定时器
         */
        public void changeTask(CustomizeTask customizeTask) {
        
            log.info("修改定时任务");
            // 停止定时器
            endTask(customizeTask);
            // 定义新的执行时间,并启动
            startTask(customizeTask);
        }
    }

*  业务代码处理

/**
     * 统一的任务执行入口，
     * 用于被CustomizeTask静态调用。
     */
    @Component
    @Slf4j
    public class ExecuteScheduleJob {
        
    
    
        private static IMesTenantProcessService iMesTenantProcessService;
    
        // 方便被自定义任务类直接调用，设置静态属性。
        @Autowired
        public void setStatic(IMesTenantProcessService iMesTenantProcessService) {
        
            ExecuteScheduleJob.iMesTenantProcessService = iMesTenantProcessService;
        }
    
        /**
         * 业务代码
         */
        public static void syncTenantUser() {
        
            log.info("开始定时同步" + new Date());
            // 业务代码
            iMesTenantProcessService.do(.......)
        }
    }

服务启动时注册定时任务

@Service
    @Slf4j
    public class StartInitScheduler implements ApplicationRunner {
        
    
        private final DynamicScheduledTask dynamicScheduledTask;
        private final MyConfig myConfig;
    
    
        @Autowired
        public StartInitScheduler(DynamicScheduledTask dynamicScheduledTask, MyConfig myConfig) {
        
            this.dynamicScheduledTask = dynamicScheduledTask;
            this.myConfig = myConfig;
        }
    
        @Override
        public void run(ApplicationArguments args) throws Exception {
        
            log.info("启动初始化定时任务");
            // 我是通过配置注册的定时任务信息，可以从数据库中查询等等方式
            dynamicScheduledTask.startTask(new CustomizeTask(Constant.SCHEDULER_TYPE_SYNC, myConfig.getSyncCron()));
        }
    }

*  业务上的任务管理中心，对每个任务启动，停止，更新任务频率

@RestController
    @RequestMapping("/orderDeliveryWarning")
    @Slf4j
    public class OrderDeliveryWarningController {
        
    
        private final DynamicScheduledTask dynamicScheduledTask;
    
        @Autowired
        public OrderDeliveryWarningController(DynamicScheduledTask dynamicScheduledTask) {
        
            this.dynamicScheduledTask = dynamicScheduledTask;
        }
    
    
        @ApiOperation(value = "开始定时任务", notes = "开始定时任务")
        @GetMapping(value = "/startDynamicScheduledTask")
        public WebResponse startDynamicScheduledTask(String cron) {
        
            try {
        
                dynamicScheduledTask.startTask(new CustomizeTask(Constant.ORDER_CODE, cron));
                return WebResponse.success();
            } catch (Exception e) {
        
                log.error("启动定时任务异常", e);
                return WebResponse.error(e.getMessage());
            }
        }
    
        @ApiOperation(value = "停止定时任务", notes = "停止定时任务")
        @GetMapping(value = "/endDynamicScheduledTask")
        public WebResponse endDynamicScheduledTask() {
        
            try {
        
                dynamicScheduledTask.endTask(new CustomizeTask(Constant.ORDER_CODE));
                return WebResponse.success();
            } catch (Exception e) {
        
                log.error("停止定时任务异常", e);
                return WebResponse.error(e.getMessage());
            }
        }
    
        @ApiOperation(value = "修改定时任务", notes = "修改止定时任务")
        @GetMapping(value = "/changeDynamicScheduledTask")
        public WebResponse changeDynamicScheduledTask(String cron) {
        
            try {
        
                dynamicScheduledTask.changeTask(new CustomizeTask(Constant.ORDER_CODE, cron));
                return WebResponse.success();
            } catch (Exception e) {
        
                log.error("改变定时任务异常", e);
                return WebResponse.error(e.getMessage());
            }
        }

*  nacos配置监听，cron变化修改任务

@Component
    @Slf4j
    public class NacosListener extends RefreshEventListener {
        
    
        private final MyConfig myConfig;
    
        private final DynamicScheduledTask dynamicScheduledTask;
    
        @Autowired
        public NacosListener(ContextRefresher refresh, MyConfig myConfig, DynamicScheduledTask dynamicScheduledTask) {
        
            super(refresh);
            this.myConfig = myConfig;
            this.dynamicScheduledTask = dynamicScheduledTask;
        }
    
        @Override
        public void handle(RefreshEvent event) {
        
            String syncCron = myConfig.getSyncCron();
            log.info("之前：{}", JSON.toJSONString(syncCron));
            super.handle(event);
            String afterSyncCron = myConfig.getSyncCron();
            log.info("之后：{}", JSON.toJSONString(afterSyncCron));
            if (!StringUtils.equals(syncCron, afterSyncCron)) {
        
                dynamicScheduledTask.changeTask(new CustomizeTask(Constant.SCHEDULER_TYPE_SYNC, myConfig.getSyncCron()));
            }
        }
    }