Quartz多个调度器+线程池模式分别调度任务

配置，配置多个调度器及其线程池

import com.wipinfo.central.engine.constant.JobConstant;
import com.wipinfo.central.engine.job.MyJobFactory;
import lombok.extern.slf4j.Slf4j;
import org.quartz.Scheduler;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;

import java.util.concurrent.ThreadPoolExecutor;

/** * @author echoo华地 * @description 定时任务配置 * @date 2022/5/31 9:06 */
@Configuration
@Slf4j
public class QuartzConfig {
    

    /** * 数据清理任务调度器 */
    @Bean(JobConstant.JOB_EXECUTOR_CLEAN)
    public Scheduler cleanScheduler(@Qualifier("CleanSchedulerFactoryBean") SchedulerFactoryBean timingSchedulerFactoryBean) {
    
        return timingSchedulerFactoryBean.getScheduler();
    }


    /** * 虚拟机定时开关机任务调度器 */
    @Bean(JobConstant.JOB_EXECUTOR_TIMING)
    public Scheduler timingScheduler(@Qualifier("TimingSchedulerFactoryBean") SchedulerFactoryBean timingSchedulerFactoryBean) {
    
        return timingSchedulerFactoryBean.getScheduler();
    }

    /** * 数据上传任务调度器 */
    @Bean(JobConstant.JOB_EXECUTOR_UPLOAD)
    public Scheduler uploadScheduler(@Qualifier("UploadSchedulerFactoryBean") SchedulerFactoryBean timingSchedulerFactoryBean) {
    
        return timingSchedulerFactoryBean.getScheduler();
    }

    /** * 数据统计任务调度器 */
    @Bean(JobConstant.JOB_EXECUTOR_CALCULATE)
    public Scheduler calScheduler(@Qualifier("CalSchedulerFactoryBean") SchedulerFactoryBean timingSchedulerFactoryBean) {
    
        return timingSchedulerFactoryBean.getScheduler();
    }


    /** * 虚拟机定时开关机调度器工厂 */
    @Bean("TimingSchedulerFactoryBean")
    public SchedulerFactoryBean timingSchedulerFactoryBean() {
    
        ThreadPoolTaskExecutor threadPoolExecutor = createThreadPoolExecutor(1, 2, 4, 60);
        return createSchedulerFactoryBean(JobConstant.JOB_EXECUTOR_TIMING, 5, false, threadPoolExecutor, jobFactory());
    }

    /** * 虚拟机定时开关机调度器工厂 */
    @Bean("UploadSchedulerFactoryBean")
    public SchedulerFactoryBean uploadSchedulerFactoryBean() {
    
        ThreadPoolTaskExecutor threadPoolExecutor = createThreadPoolExecutor(1, 2, 4, 60);
        return createSchedulerFactoryBean(JobConstant.JOB_EXECUTOR_UPLOAD, 5, false, threadPoolExecutor, jobFactory());
    }

    /** * 虚拟机定时开关机调度器工厂 */
    @Bean("CalSchedulerFactoryBean")
    public SchedulerFactoryBean calSchedulerFactoryBean() {
    
        ThreadPoolTaskExecutor threadPoolExecutor = createThreadPoolExecutor(1, 2, 4, 60);
        return createSchedulerFactoryBean(JobConstant.JOB_EXECUTOR_CALCULATE, 5, false, threadPoolExecutor, jobFactory());
    }

    /** * 虚拟机定时开关机调度器工厂 */
    @Bean("CleanSchedulerFactoryBean")
    public SchedulerFactoryBean cleanSchedulerFactoryBean() {
    
        ThreadPoolTaskExecutor threadPoolExecutor = createThreadPoolExecutor(1, 2, 4, 60);
        return createSchedulerFactoryBean(JobConstant.JOB_EXECUTOR_CLEAN, 5, false, threadPoolExecutor, jobFactory());
    }

    /** * @param name 调度器名称 * @param delay 启动延时 * @param waitForJob 等待任务执行完毕再关闭程序 * @param threadPoolExecutor 调度器线程池 * @return org.springframework.scheduling.quartz.SchedulerFactoryBean * @description 生成调度器工厂 * @author echoo华地 * @date 2022/5/31 9:04 */
    private SchedulerFactoryBean createSchedulerFactoryBean(String name, Integer delay, boolean waitForJob, ThreadPoolTaskExecutor threadPoolExecutor, MyJobFactory jobFactory) {
    
        SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
        System.out.println("当前 = " + schedulerFactoryBean);
        schedulerFactoryBean.setSchedulerName(name); //
        schedulerFactoryBean.setStartupDelay(delay); // 延时启动
        schedulerFactoryBean.setWaitForJobsToCompleteOnShutdown(waitForJob); // 等待任务执行完毕再关闭程序
        // schedulerFactoryBean.setDataSource(); // 因为 Quartz 默认数据库为 Mysql，因此做自定义磁盘化
        schedulerFactoryBean.setJobFactory(jobFactory);
        schedulerFactoryBean.setTaskExecutor(threadPoolExecutor);
        return schedulerFactoryBean;
    }


    /** * @param corePoolSize 核心线程 * @param maxPoolSize 最大线程 * @param queueCapacity 队列容量 * @param keepAliveSeconds 闲置时间 * @return org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor * @description 生成线程池 * @author echoo华地 * @date 2022/5/31 8:54 */
    private static ThreadPoolTaskExecutor createThreadPoolExecutor(Integer corePoolSize, Integer maxPoolSize, Integer queueCapacity, Integer keepAliveSeconds) {
    
        ThreadPoolTaskExecutor threadPoolTaskExecutor = new ThreadPoolTaskExecutor();
        threadPoolTaskExecutor.setCorePoolSize(corePoolSize);
        threadPoolTaskExecutor.setMaxPoolSize(maxPoolSize);
        threadPoolTaskExecutor.setQueueCapacity(queueCapacity);
        threadPoolTaskExecutor.setKeepAliveSeconds(keepAliveSeconds); // Default is 60
        threadPoolTaskExecutor.setRejectedExecutionHandler(new ThreadPoolExecutor.AbortPolicy()); // AbortPolicy : 默认拒绝策略，抛出异常 DiscardPolicy：不会抛出异常，会丢掉任务
        threadPoolTaskExecutor.initialize(); // 手动初始化
        return threadPoolTaskExecutor;
    }

    @Bean
    public MyJobFactory jobFactory() {
    
        return new MyJobFactory();
    }
}

调度管理器，定义自己需要的功能

import com.wipinfo.central.engine.constant.JobConstant;
import com.wipinfo.central.engine.entity.QuartzTask;
import com.wipinfo.central.engine.enums.JobStatusEnums;
import com.wipinfo.central.engine.enums.JobTypeEnums;
import com.wipinfo.central.engine.service.IQuartzService;
import lombok.extern.slf4j.Slf4j;
import org.quartz.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Collections;
import java.util.List;

/** * @author echoo华地 * @description 调度管理器 * @date 2022/5/30 13:47 */
@Slf4j
@Component
public class SchedulerManager {
    

    @Resource(name = JobConstant.JOB_EXECUTOR_TIMING)
    private Scheduler timingScheduler;
    @Resource(name = JobConstant.JOB_EXECUTOR_UPLOAD)
    private Scheduler uploadScheduler;
    @Resource(name = JobConstant.JOB_EXECUTOR_CALCULATE)
    private Scheduler calculateScheduler;
    @Autowired
    @Qualifier(JobConstant.JOB_EXECUTOR_CLEAN)
    private Scheduler cleanScheduler;
    @Autowired
    private IQuartzService quartzService;

    /** * 调度所有任务 */
    public void schedulerAllJobs() {
    
        scheduleTimingJobs();
        scheduleCalculateJobs();
        scheduleUploadJobs();
// scheduleCleanJobs();
    }

    /** * 调度数据上传任务 */
    public void scheduleUploadJobs() {
    
        log.info("启动数据上传任务调度器");
        scheduleUploadJobFromDb(uploadScheduler);
    }

    /** * 调度数据统计任务 */
    public void scheduleCalculateJobs() {
    
        log.info("启动数据统计任务调度器");
        scheduleCalculateJobFromDb(calculateScheduler);
    }

    /** * 调度数据清理任务 */
    public void scheduleCleanJobs() {
    
        log.info("启动数据清理任务调度器");
        scheduleCleanJobFromDb(cleanScheduler);
    }

    /** * 调度定时开关机任务 */
    public void scheduleTimingJobs() {
    
        log.info("启动开关机定时任务调度器");
        scheduleTimingJobFromDb(timingScheduler);
    }

    /** * 调度数据库数据上传任务 */
    private void scheduleUploadJobFromDb(Scheduler scheduler) {
    
        List<QuartzTask> tasks = quartzService.selectJobsByGroup(JobConstant.JOB_GROUP_UPLOAD);
        log.info("当前数据上传任务数：{}", tasks.size());
        tasks.stream().filter(task ->
                task.getStatus() == 0
        ).forEach(task -> {
    
            try {
    
                schedulerJob(task, scheduler);
            } catch (Exception e) {
    
                log.error("数据上传任务[{}]调度异常;错误:[{}]", task.getJobName(), e.getMessage());
            }
        });
    }

    /** * 调度数据库数据统计任务 */
    private void scheduleCalculateJobFromDb(Scheduler scheduler) {
    
        List<QuartzTask> tasks = quartzService.selectJobsByGroup(JobConstant.JOB_GROUP_CALCULATE);
        log.info("当前数据统计任务数：{}", tasks.size());
        tasks.stream().filter(task ->
                task.getStatus() == 0
        ).forEach(task -> {
    
            try {
    
                schedulerJob(task, scheduler);
            } catch (Exception e) {
    
                log.error("数据统计任务[{}]调度异常;错误:[{}]", task.getJobName(), e.getMessage());
            }
        });
    }

    /** * 调度数据库数据清理任务 */
    private void scheduleCleanJobFromDb(Scheduler scheduler) {
    
        List<QuartzTask> tasks = quartzService.selectJobsByGroup(JobConstant.JOB_GROUP_CLEAN);
        log.info("当前数据清理任务数：{}", tasks.size());
        tasks.stream().filter(task ->
                task.getStatus() == 0
        ).forEach(task -> {
    
            try {
    
                schedulerJob(task, scheduler);
            } catch (Exception e) {
    
                log.error("数据清理任务[{}]调度异常;错误:[{}]", task.getJobName(), e.getMessage());
            }
        });
    }

    /** * 调度数据库定时开关机任务 */
    private void scheduleTimingJobFromDb(Scheduler scheduler) {
    
        List<QuartzTask> tasks = Collections.emptyList();
        // 查询 定时开关机组 任务
        tasks = quartzService.selectJobsByGroup(JobConstant.JOB_GROUP_TIMING);
        log.info("当前定时开关机任务数: {}", tasks.size());
        // 把所有定时任务注入调度器中
        tasks.forEach(task -> {
    
            try {
    
                schedulerJob(task, scheduler);
            } catch (Exception e) {
    
                if (e.getMessage().contains("will never fire")) {
    
                    // 修改任务状态为已过期
                    quartzService.updateStatus(task.getId(), JobStatusEnums.ALREADY_EXECUTE.getStatus());
                    log.warn("定时任务[{}]已过期失效", task.getJobName());
                } else {
    
                    log.error("定时任务[{}]调度异常;错误:[{}]", task.getJobName(), e.getMessage());
                }
            }
        });
    }

    /** * @param schedulerName 需要重载的调度器名称 * @description 重载指定调度器任务 * @author echoo华地 * @date 2022/5/31 10:00 */
    public void rescheduleJobs(String schedulerName) {
    
        log.info("正在重载[{}]调度器任务", schedulerName);
        try {
    
            switch (schedulerName) {
    
                case JobConstant.JOB_EXECUTOR_TIMING:
                    if (clearSchedulerJobs(timingScheduler))
                        scheduleTimingJobFromDb(timingScheduler);
                    break;
                case JobConstant.JOB_EXECUTOR_UPLOAD:
                    if (clearSchedulerJobs(uploadScheduler))
                        scheduleUploadJobFromDb(uploadScheduler);
                    break;
                case JobConstant.JOB_EXECUTOR_CALCULATE:
                    if (clearSchedulerJobs(calculateScheduler))
                        scheduleCalculateJobFromDb(calculateScheduler);
                    break;
                case JobConstant.JOB_EXECUTOR_CLEAN:
                    if (clearSchedulerJobs(calculateScheduler))
                        scheduleCleanJobFromDb(calculateScheduler);
                    break;
                default:
                    log.error("重载调度器任务失败;错误[{}调度器不存在]", schedulerName);
            }
        } catch (Exception e) {
    
            log.error("重载定时任务失败；错误:[{}]", e.getMessage());
        }
    }


    /** * 清空定时开关机任务 */
    private boolean clearSchedulerJobs(Scheduler scheduler) {
    
        try {
    
            log.info("清空[{}]调度器所有任务", scheduler.getSchedulerName());
            scheduler.clear();
            return true;
        } catch (SchedulerException e) {
    
            log.error("清空调度器任务异常;错误[{}]", e.getMessage());
            return false;
        }
    }

    /** * @param task 单个任务 * @param scheduler 调度器 * @description 调度器添加调度任务 * @author echoo华地 * @date 2022/5/30 17:12 */
    private void schedulerJob(QuartzTask task, Scheduler scheduler) throws SchedulerException {
    
        JobKey jobKey = new JobKey(task.getJobName(), task.getJobGroup()); // 任务唯一标识
        Integer type = task.getType();
        Class<? extends Job> jobClass = getJobClass(type);
        if ((jobClass == null) || (!Job.class.isAssignableFrom(jobClass))) {
    
            log.error("调度失败，任务类型错误！");
        }
        JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(jobKey) // 任务类型,任务数据
                .usingJobData(JobConstant.JOB_DATA_JOB_ID, task.getId())
                .usingJobData(JobConstant.JOB_DATA_JOB_TYPE, task.getType())
                .usingJobData(JobConstant.JOB_DATA_JOB_NAME, task.getJobName()).build();
        CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(task.getCron());
        // 以任务名为名称，并根据cron表达式创建触发器
        CronTrigger cronTrigger = TriggerBuilder.newTrigger().withIdentity(task.getJobName(),
                task.getJobGroup()).withSchedule(scheduleBuilder).build();
        if (scheduler.checkExists(jobKey)) {
    
            // 任务已在调度器内，先删除
            scheduler.deleteJob(jobKey);
        }
        // 加入调度
        scheduler.scheduleJob(jobDetail, cronTrigger);
    }

    private Class<? extends Job> getJobClass(Integer jobType) {
    
        Class<?> jobClass = JobTypeEnums.getJobClassByType(jobType);
        if (Job.class.isAssignableFrom(jobClass)) {
    
            return (Class<Job>) jobClass;
        } else {
    
            return null;
        }
    }
}