SpringBoot整合Quartz

2020-11-19

注：配置文件中配置测试任务每10秒执行一次

1. 引入依赖

<dependency>
    <groupId>mysql</groupId>
    <artifactId>mysql-connector-java</artifactId>
    <version>8.0.15</version>
</dependency>
<!--为了使用HikariDataSource数据源-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-jdbc</artifactId>
</dependency>
<!--Spring Quartz依赖-->
<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-quartz</artifactId>
</dependency>

2. 配置文件

数据库连接配置
数据源配置
Quartz配置
Job配置(定时任务配置) 这里配置的测试任务每10秒执行一次

####################### 数据源配置 ##############################
spring.datasource.type=com.zaxxer.hikari.HikariDataSource
spring.datasource.hikari.minimum-idle=5
spring.datasource.hikari.maximum-pool-size=15
spring.datasource.hikari.auto-commit=true
spring.datasource.hikari.idle-timeout=30000
spring.datasource.hikari.pool-name=DatebookHikariCP
spring.datasource.hikari.max-lifetime=1800000
spring.datasource.hikari.connection-timeout=30000
spring.datasource.hikari.connection-test-query=SELECT 1
##########################  Quartz配置   ##########################
#######相关属性配置
spring.quartz.properties.org.quartz.scheduler.instanceName=clusteredScheduler
spring.quartz.properties.org.quartz.scheduler.instanceId=AUTO
#######数据库方式（内存方式与数据库存储方式）
spring.quartz.job-store-type=jdbc
spring.quartz.properties.org.quartz.jobStore.class=org.quartz.impl.jdbcjobstore.JobStoreTX
spring.quartz.properties.org.quartz.jobStore.driverDelegateClass=org.quartz.impl.jdbcjobstore.StdJDBCDelegate
spring.quartz.properties.org.quartz.jobStore.tablePrefix=QRTZ_
#spring.quartz.properties.org.quartz.jobStore.useProperties=false
########线程池配置
#线程池的实现类
spring.quartz.properties.org.quartz.threadPool.class=org.quartz.simpl.SimpleThreadPool
#指定线程数
spring.quartz.properties.org.quartz.threadPool.threadCount=10
#设置线程的优先级
spring.quartz.properties.org.quartz.threadPool.threadPriority=5
spring.quartz.properties.org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread=true
#######检查quartz是否有版本更新true是不检查
spring.quartz.properties.org.terracotta.quartz.skipUpdateCheck=true
#初始化表结构
#spring.quartz.jdbc.initialize-schema=never
##########################  Job配置(定时任务配置)  ##########################
#1-cron;2-simple
quartz.jobType=simple
quartz.jobName=Test
quartz.jobGroupName=myTest
quartz.triggerName=Test
quartz.triggerGroupName=myTest
#jobType为simple时time为正整数，为cron时time为cron表达式
#simple例子每10秒一次
quartz.time=10
#cron例子每天1点执行一次 0 0 1 * * ?
#quartz.time="0 0 1 * * ?"

3. 定时任务配置

@Configuration
public class ScheduledTaskConfig implements SchedulerFactoryBeanCustomizer{

    @Override
    public void customize(SchedulerFactoryBean schedulerFactoryBean) {
        schedulerFactoryBean.setStartupDelay(2);
        schedulerFactoryBean.setAutoStartup(true);
        schedulerFactoryBean.setOverwriteExistingJobs(true);
    }
}

4. schedule任务管理类

动态实现任务的开始，暂停，停止等操作

@Component
public class ScheduleManager {
    private final static Logger LOGGER = LoggerFactory.getLogger(ScheduleManager.class);
    @Autowired
    private Scheduler scheduler;

    /**
     * 使用SimpleScheduleBuilder创建一个定时任务
     * @param jobName job名称
     * @param jobGroupName job分组名称
     * @param triggerName 触发器名称
     * @param triggerGroupName 触发器分组名称
     * @param jobClass job的class名
     * @param durability 是否持久化
     * @param interval 间隔时间，单位秒
     * @param delay 延迟启动时间，单位秒
     * @return
     */
    public Boolean addJob(String jobName, String jobGroupName, String triggerName, String triggerGroupName, Class jobClass,
                          boolean durability, int interval, int delay) {
        /*
         * 1.取到任务调度器Scheduler
         * 2.定义jobDetail;
         * 3.定义trigger;
         * 4.使用Scheduler添加任务;
         */
        try {
            JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(jobName, jobGroupName).storeDurably(durability).usingJobData("test","test").build();

            SimpleScheduleBuilder scheduleBuilder = SimpleScheduleBuilder.simpleSchedule()
                    .withIntervalInSeconds(interval).repeatForever();

            Trigger trigger = TriggerBuilder.newTrigger()
                    .withIdentity(triggerName, triggerGroupName)
                    .withSchedule(scheduleBuilder)
                    .usingJobData("test1","test1")
                    .startAt(new Date(new Date().getTime() + delay * 1000))
                    .build();

            scheduler.scheduleJob(jobDetail, trigger);
            return true;
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 使用CronScheduleBuilder创建一个定时任务
     *
     * @param jobName job名称
     * @param jobGroupName job分组名称
     * @param triggerName 触发器名称
     * @param triggerGroupName 触发器分组名称
     * @param jobClass job的class名
     * @param durability 是否持久化
     * @param cronExpression 定时表达式
     * @param delay 延迟启动时间，单位秒
     * @return
     */
    public Boolean addJob(String jobName, String jobGroupName, String triggerName, String triggerGroupName, Class jobClass,
                          boolean durability, String cronExpression, int delay) {
        /*
         * 1.取到任务调度器Scheduler
         * 2.定义jobDetail;
         * 3.定义trigger;
         * 4.使用Scheduler添加任务;
         */
        try {
            JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(jobName, jobGroupName).storeDurably(durability).build();

            Trigger trigger = TriggerBuilder.newTrigger()
                    .withIdentity(triggerName, triggerGroupName)
                    .withSchedule(CronScheduleBuilder.cronSchedule(cronExpression))
                    .startAt(new Date(new Date().getTime() + delay * 1000))
                    .build();

            scheduler.scheduleJob(jobDetail, trigger);
            return true;
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
        return false;
    }
    /**
     * 动态停止job任务
     * @param jobName job名称
     * @param jobGroupName job分组名称
     * @return
     */
    public Boolean pause(String jobName, String jobGroupName) {
        try {
            JobKey key = new JobKey(jobName, jobGroupName);
            scheduler.pauseJob(key);
            return true;
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 检查job任务是否存在
     * @param jobName job名称
     * @param jobGroupName job分组名称
     * @return
     */
    public Boolean checkJob(String jobName, String jobGroupName) {
        try {
            JobKey key = new JobKey(jobName, jobGroupName);
            return scheduler.checkExists(key);

        } catch (SchedulerException e) {
            e.printStackTrace();
        }
        return false;
    }
    /**
     * 检查trigger是否存在
     * @param triggerName trigger名称
     * @param triggerGroupName trigger分组名称
     * @return
     */
    public Boolean checkTrigger(String triggerName, String triggerGroupName) {
        try {
            TriggerKey key = new TriggerKey(triggerName, triggerGroupName);
            return scheduler.checkExists(key);

        } catch (SchedulerException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 动态开始job任务
     *
     * @param jobName job名称
     * @param jobGroupName job分组名称
     * @return
     */
    public Boolean start(String jobName, String jobGroupName) {
        try {
            JobKey key = new JobKey(jobName, jobGroupName);
            scheduler.resumeJob(key);
            return true;
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 移除定时任务
     *
     * @param jobName job名称
     * @param jobGroupName job分组名称
     * @param triggerName 触发器名称
     * @param triggerGroupName 触发器分组名称
     * @return
     */
    public Boolean delete(String jobName, String jobGroupName, String triggerName, String triggerGroupName) {
        try {
            Integer status =getJobStatus(triggerName, triggerGroupName);
            if (status == 0) {
                //NONE - 0,该job不存在
                LOGGER.warn("NONE - 0,该job不存在");
                return null;
            }

            TriggerKey triggerKey = TriggerKey.triggerKey(triggerName, triggerGroupName);//通过触发器名和组名获取TriggerKey
            JobKey jobKey = new JobKey(jobName, jobGroupName); //通过任务名和组名获取JobKey
            scheduler.pauseTrigger(triggerKey);    //停止触发器
            scheduler.unscheduleJob(triggerKey);//移除触发器
            scheduler.deleteJob(jobKey);// 删除任务
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 动态修改任务执行的时间
     *
     * @param type 创建类型，1-cron,2-simple
     * @param jobName job名称
     * @param jobGroupName job分组名称
     * @param triggerName 触发器名称
     * @param triggerGroupName 触发器分组名称
     * @param time type为1时-填cron表达式，为2-整数
     * @return
     */
    public Boolean modify(String type,String jobName, String jobGroupName, String triggerName, String triggerGroupName, String time) {

        try {
            Integer status =getJobStatus(triggerName, triggerGroupName);
            if (status == 0) {
                //NONE - 0,该job不存在
                LOGGER.warn("NONE - 0,该job不存在");
                return false;
            }

            //获取任务
            JobKey key = new JobKey(jobName, jobGroupName);
            //获取jobDetail
            JobDetail jobDetail = scheduler.getJobDetail(key);
            //生成trigger
            if (Constants.jobTypeCron.equals(type)) {
                // 1.CronSchedule
                Trigger trigger = TriggerBuilder
                        .newTrigger()
                        .withIdentity(triggerName, triggerGroupName)
                        .withSchedule(CronScheduleBuilder.cronSchedule(time))
                        .build();
                //删除旧的任务，否则报错
                //scheduler.deleteJob(key);
                delete(jobName, jobGroupName, triggerName, triggerGroupName);
                //重新启动任务
                scheduler.scheduleJob(jobDetail, trigger);
                return true;
            }

            if (Constants.jobTypeSimple.equals(type)) {
                // 2.SimpleSchedule
                Trigger trigger = TriggerBuilder
                        .newTrigger()
                        .withIdentity(triggerName, triggerGroupName)
                        .withSchedule(SimpleScheduleBuilder.repeatSecondlyForever(Integer.valueOf(time)))
                        .build();
                //删除旧的任务，否则报错
                //scheduler.deleteJob(key);
                delete(jobName, jobGroupName, triggerName, triggerGroupName);
                //重新启动任务
                scheduler.scheduleJob(jobDetail, trigger);
                return true;
            }
        } catch (SchedulerException e) {
            e.printStackTrace();
            return false;
        }
        return false;
    }
    /**
     * 动态修改任务执行的时间
     *
     * @param type 创建类型，1-cron,2-simple
     * @param triggerName 触发器名称
     * @param triggerGroupName 触发器分组名称
     * @param time type为1时-填cron表达式，为2-整数
     * @return
     */
    public Boolean modifyTrigger(String type, String triggerName, String triggerGroupName, String time) {

        try {
            Integer status =getJobStatus(triggerName, triggerGroupName);
            if (status == 0) {
                //NONE - 0,该job不存在
                LOGGER.warn("NONE - 0,该job不存在");
                return false;
            }

            if (Constants.jobTypeCron.equals(type)) {
                // 1.CronSchedule
                Trigger trigger = TriggerBuilder
                        .newTrigger()
                        .withIdentity(triggerName, triggerGroupName)
                        .withSchedule(CronScheduleBuilder.cronSchedule(time))
                        .build();

                //重新启动任务
                scheduler.rescheduleJob(new TriggerKey(triggerName,triggerGroupName),trigger);
                return true;
            }

            if (Constants.jobTypeSimple.equals(type)) {
                // 2.SimpleSchedule
                Trigger trigger = TriggerBuilder
                        .newTrigger()
                        .withIdentity(triggerName, triggerGroupName)
                        .withSchedule(SimpleScheduleBuilder.repeatSecondlyForever(Integer.valueOf(time)))
                        .build();

                //重新启动任务
                scheduler.rescheduleJob(new TriggerKey(triggerName,triggerGroupName),trigger);
                return true;
            }
        } catch (SchedulerException e) {
            e.printStackTrace();
            return false;
        }
        return false;
    }
    /**
     * 查看job运行状态
     * @param triggerName 触发器名称
     * @param triggerGroupName 触发器分组名称
     * @return TriggerState枚举值： NONE - 0；NORMAL - 1；PAUSED - 2；COMPLETE - 3；ERROR - 4；BLOCKED - 5
     */
    public Integer getJobStatus(String triggerName, String triggerGroupName) {

        /*
            TriggerState枚举值
            NONE - 0,
            NORMAL -1,
            PAUSED - 2,
            COMPLETE - 3,
            ERROR - 4,
            BLOCKED - 5;
        */
        try {
            TriggerKey triggerKey = TriggerKey.triggerKey(triggerName, triggerGroupName);
            return scheduler.getTriggerState(triggerKey).ordinal();
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 启动所有定时任务
     */
    public Boolean startAll() {
        try {
            scheduler.start();
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 关闭所有定时任务
     */
    public Boolean shutdownAll() {
        try {
            if (!scheduler.isShutdown()) {
                scheduler.shutdown();
            }
            return true;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 清理数据库中的定时任务
     */
    public Boolean taskClear() {
        try {
            scheduler.clear();
            return true;
        } catch (SchedulerException e) {
            e.printStackTrace();
        }
        return false;
    }
}

5. 测试用的定时任务

@DisallowConcurrentExecution    //设置任务能否并行执行，加注解则不能
public class TestJob extends QuartzJobBean {
    @Override
    protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        System.out.println("*********计划任务开始执行**********");
        JobDataMap dataMap = jobExecutionContext.getJobDetail().getJobDataMap();
        String param = dataMap.getString("test");//获取参数
        System.out.println(param);
        JobDataMap tdataMap = jobExecutionContext.getTrigger().getJobDataMap();
        String param1 = tdataMap.getString("test1");
        System.out.println(param1);
        System.out.println("*********计划任务结束执行**********");
        System.out.println(new Date().getTime());
    }
}

6. 在应用启动时启动定时任务

实现了ApplicationRunner

@Component
public class MyApplicationRunner implements ApplicationRunner {
    private final static Logger LOGGER = LoggerFactory.getLogger(MyApplicationRunner.class);

    @Autowired
    private ScheduleManager scheduleManager;

    @Value("${quartz.jobType}")
    private String jobType;
    @Value("${quartz.jobName}")
    private String jobName;
    @Value("${quartz.jobGroupName}")
    private String jobGroupName;
    @Value("${quartz.triggerName}")
    private String triggerName;
    @Value("${quartz.triggerGroupName}")
    private String triggerGroupName;
    @Value("${quartz.time}")
    private String time;
    @Value("${spring.quartz.job-store-type}")
    private String jobStoreType;

    @Override
    public void run(ApplicationArguments applicationArguments) {
        LOGGER.info("--MyApplicationRunner--");

        boolean durability = false;
        Class jobClass = TestJob.class;

        // 存储类型 jdbc|memory
        if (Constants.jobStoreTypeJdbc.equals(jobStoreType)) {
            durability = true;
            LOGGER.info("job存储方式："+Constants.jobStoreTypeJdbc);
        } else if (Constants.jobStoreTypeMemory.equals(jobStoreType)) {
            durability = false;
            LOGGER.info("job存储方式："+Constants.jobStoreTypeMemory);
        } else {
            LOGGER.warn("不存在存储类型："+jobStoreType+"，默认为："+Constants.jobStoreTypeMemory+"请选择:"+Constants.jobStoreTypeJdbc+"或者"+Constants.jobStoreTypeMemory);
        }

        // 任务类型 simple|cron
        if (Constants.jobTypeSimple.equals(jobType)) {
            int interval = Constants.time;
            // 配置文件中配置的quartz.time是否符合要求。符合就采用配置文件中的，否则使用默认配置Constants.time
            if (StringUtils.testIsInt(time)) {
                interval = Integer.parseInt(time);
            }
            if (scheduleManager.checkJob(jobName,jobGroupName) && scheduleManager.checkTrigger(triggerName,triggerGroupName)) {
                LOGGER.info("Job与Trigger已存在不新增");
            } else {
                scheduleManager.addJob(jobName,jobGroupName,triggerName,triggerGroupName,
                        jobClass,durability,interval,5);
            }
        } else if (Constants.jobTypeCron.equals(jobType)) {
            // 配置文件中配置的quartz.time是否是cron表达式。是就采用配置文件中的，否则使用默认配置 Constants.cron
            if (!CronExpression.isValidExpression(time)) {
                time = Constants.cron;
            }
            if (scheduleManager.checkJob(jobName,jobGroupName) && scheduleManager.checkTrigger(triggerName,triggerGroupName)) {
                LOGGER.info("Job与Trigger已存在不新增");
            } else {
                scheduleManager.addJob(jobName, jobGroupName, triggerName, triggerGroupName,
                        jobClass, durability, time, 5);
            }
        } else {
            LOGGER.warn("--MyApplicationRunner--不存在jobType："+jobType+",请选择："+Constants.jobTypeSimple+"或者"+Constants.jobTypeCron);
        }
    }
}

7. 运行

每10秒执行一次TestJob

参考

https://www.cnblogs.com/yunliu0603/p/10112791.html