SpringBoot2.0 整合 QuartJob ,實現定時器實時管理

本文源碼：GitHub·點這里 || GitEE·點這里

一、QuartJob簡介

1、一句話描述

Quartz是一個完全由java編寫的開源作業調度框架，形式簡易，功能強大。

2、核心API

(1)、Scheduler
代表一個 Quartz 的獨立運行容器，Scheduler 將 Trigger 綁定到特定 JobDetail， 這樣當 Trigger 觸發時， 對應的 Job 就會被調度。
(2)、Trigger
描述 Job 執行的時間觸發規則。主要有 SimpleTrigger 和 CronTrigger 兩個子類，通過一個 TriggerKey 唯一標識。
(3)、Job
定義一個任務，規定了任務是執行時的行為。JobExecutionContext 提供了調度器的上下文信息，Job 的數據可從 JobDataMap 中獲取。
(4)、JobDetail
Quartz 在每次執行 Job 時，都重新創建一個 Job 實例，所以它不直接接受一個 Job 的實例，相反它接收一個 Job 實現類。描述 Job 的實現類及其它相關的靜態信息，如 Job 名字、描述等。

二、與SpringBoot2.0 整合

1、項目結構

版本描述

spring-boot：2.1.3.RELEASE
quart-job：2.3.0

2、定時器配置

import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;
import javax.sql.DataSource;
import java.util.Properties;
@Configuration
public class ScheduleConfig {
    @Bean
    public SchedulerFactoryBean schedulerFactoryBean(DataSource dataSource) {
        // Quartz參數配置
        Properties prop = new Properties();
        // Schedule調度器的實體名字
        prop.put("org.quartz.scheduler.instanceName", "HuskScheduler");
        // 設置為AUTO時使用，默認的實現org.quartz.scheduler.SimpleInstanceGenerator是基于主機名稱和時間戳生成。
        prop.put("org.quartz.scheduler.instanceId", "AUTO");
        // 線程池配置
        prop.put("org.quartz.threadPool.class", "org.quartz.simpl.SimpleThreadPool");
        prop.put("org.quartz.threadPool.threadCount", "20");
        prop.put("org.quartz.threadPool.threadPriority", "5");
        // JobStore配置:Scheduler在運行時用來存儲相關的信息
        // JDBCJobStore和JobStoreTX都使用關系數據庫來存儲Schedule相關的信息。
        // JobStoreTX在每次執行任務后都使用commit或者rollback來提交更改。
        prop.put("org.quartz.jobStore.class", "org.quartz.impl.jdbcjobstore.JobStoreTX");
        // 集群配置：如果有多個調度器實體的話則必須設置為true
        prop.put("org.quartz.jobStore.isClustered", "true");
        // 集群配置：檢查集群下的其他調度器實體的時間間隔
        prop.put("org.quartz.jobStore.clusterCheckinInterval", "15000");
        // 設置一個頻度(毫秒)，用于實例報告給集群中的其他實例
        prop.put("org.quartz.jobStore.maxMisfiresToHandleAtATime", "1");
        // 觸發器觸發失敗后再次觸犯的時間間隔
        prop.put("org.quartz.jobStore.misfireThreshold", "12000");
        // 數據庫表前綴
        prop.put("org.quartz.jobStore.tablePrefix", "qrtz_");
        // 從 LOCKS 表查詢一行并對這行記錄加鎖的 SQL 語句
        prop.put("org.quartz.jobStore.selectWithLockSQL", "SELECT * FROM {0}LOCKS UPDLOCK WHERE LOCK_NAME = ?");

        // 定時器工廠配置
        SchedulerFactoryBean factory = new SchedulerFactoryBean();
        factory.setDataSource(dataSource);
        factory.setQuartzProperties(prop);
        factory.setSchedulerName("HusScheduler");
        factory.setStartupDelay(30);
        factory.setApplicationContextSchedulerContextKey("applicationContextKey");
        // 可選，QuartzScheduler 啟動時更新己存在的Job
        factory.setOverwriteExistingJobs(true);
        // 設置自動啟動，默認為true
        factory.setAutoStartup(true);
        return factory;
    }
}

3、定時器管理工具

import com.quart.job.entity.ScheduleJobBean;
import org.quartz.*;
/**
 * 定時器工具類
 */
public class ScheduleUtil {
    private ScheduleUtil (){}
    private static final String SCHEDULE_NAME = "HUS_" ;
    /**
     * 觸發器 KEY
     */
    public static TriggerKey getTriggerKey(Long jobId){
        return TriggerKey.triggerKey(SCHEDULE_NAME+jobId) ;
    }
    /**
     * 定時器 Key
     */
    public static JobKey getJobKey (Long jobId){
        return JobKey.jobKey(SCHEDULE_NAME+jobId) ;
    }
    /**
     * 表達式觸發器
     */
    public static CronTrigger getCronTrigger (Scheduler scheduler,Long jobId){
        try {
            return (CronTrigger)scheduler.getTrigger(getTriggerKey(jobId)) ;
        } catch (SchedulerException e){
            throw new RuntimeException("getCronTrigger Fail",e) ;
        }
    }
    /**
     * 創建定時器
     */
    public static void createJob (Scheduler scheduler, ScheduleJobBean scheduleJob){
        try {
            // 構建定時器
            JobDetail jobDetail = JobBuilder.newJob(TaskJobLog.class).withIdentity(getJobKey(scheduleJob.getJobId())).build() ;
            CronScheduleBuilder scheduleBuilder = CronScheduleBuilder
                    .cronSchedule(scheduleJob.getCronExpression())
                    .withMisfireHandlingInstructionDoNothing() ;
            CronTrigger trigger = TriggerBuilder.newTrigger()
                    .withIdentity(getTriggerKey(scheduleJob.getJobId()))
                    .withSchedule(scheduleBuilder).build() ;
            jobDetail.getJobDataMap().put(ScheduleJobBean.JOB_PARAM_KEY,scheduleJob);
            scheduler.scheduleJob(jobDetail,trigger) ;
            // 如果該定時器處于暫停狀態
            if (scheduleJob.getStatus() == 1){
                pauseJob(scheduler,scheduleJob.getJobId()) ;
            }
        } catch (SchedulerException e){
            throw new RuntimeException("createJob Fail",e) ;
        }
    }
    /**
     * 更新定時任務
     */
    public static void updateJob(Scheduler scheduler, ScheduleJobBean scheduleJob) {
        try {
            // 構建定時器
            TriggerKey triggerKey = getTriggerKey(scheduleJob.getJobId());
            CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(scheduleJob.getCronExpression())
                    .withMisfireHandlingInstructionDoNothing();
            CronTrigger trigger = getCronTrigger(scheduler, scheduleJob.getJobId());
            trigger = trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(scheduleBuilder).build();
            trigger.getJobDataMap().put(ScheduleJobBean.JOB_PARAM_KEY, scheduleJob);
            scheduler.rescheduleJob(triggerKey, trigger);
            // 如果該定時器處于暫停狀態
            if(scheduleJob.getStatus() == 1){
                pauseJob(scheduler, scheduleJob.getJobId());
            }
        } catch (SchedulerException e) {
            throw new RuntimeException("updateJob Fail",e) ;
        }
    }
    /**
     * 停止定時器
     */
    public static void pauseJob (Scheduler scheduler,Long jobId){
        try {
            scheduler.pauseJob(getJobKey(jobId));
        } catch (SchedulerException e){
            throw new RuntimeException("pauseJob Fail",e) ;
        }
    }
    /**
     * 恢復定時器
     */
    public static void resumeJob (Scheduler scheduler,Long jobId){
        try {
            scheduler.resumeJob(getJobKey(jobId));
        } catch (SchedulerException e){
            throw new RuntimeException("resumeJob Fail",e) ;
        }
    }
    /**
     * 刪除定時器
     */
    public static void deleteJob (Scheduler scheduler,Long jobId){
        try {
            scheduler.deleteJob(getJobKey(jobId));
        } catch (SchedulerException e){
            throw new RuntimeException("deleteJob Fail",e) ;
        }
    }
    /**
     * 執行定時器
     */
    public static void run (Scheduler scheduler, ScheduleJobBean scheduleJob){
        try {
            JobDataMap dataMap = new JobDataMap() ;
            dataMap.put(ScheduleJobBean.JOB_PARAM_KEY,scheduleJob);
            scheduler.triggerJob(getJobKey(scheduleJob.getJobId()),dataMap);
        } catch (SchedulerException e){
            throw new RuntimeException("run Fail",e) ;
        }
    }
}

4、定時器執行和日志

import com.quart.job.entity.ScheduleJobBean;
import com.quart.job.entity.ScheduleJobLogBean;
import com.quart.job.service.ScheduleJobLogService;
import org.quartz.JobExecutionContext;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.quartz.QuartzJobBean;
import java.lang.reflect.Method;
import java.util.Date;
/**
 * 定時器執行日志記錄
 */
public class TaskJobLog extends QuartzJobBean {

    private static final Logger LOG = LoggerFactory.getLogger(TaskJobLog.class) ;

    @Override
    protected void executeInternal(JobExecutionContext context) {
        ScheduleJobBean jobBean = (ScheduleJobBean)context.getMergedJobDataMap().get(ScheduleJobBean.JOB_PARAM_KEY) ;
        ScheduleJobLogService scheduleJobLogService = (ScheduleJobLogService)SpringContextUtil.getBean("scheduleJobLogService") ;
        // 定時器日志記錄
        ScheduleJobLogBean logBean = new ScheduleJobLogBean () ;
        logBean.setJobId(jobBean.getJobId());
        logBean.setBeanName(jobBean.getBeanName());
        logBean.setParams(jobBean.getParams());
        logBean.setCreateTime(new Date());
        long beginTime = System.currentTimeMillis() ;
        try {
            // 加載并執行定時器的 run 方法
            Object target = SpringContextUtil.getBean(jobBean.getBeanName());
            Method method = target.getClass().getDeclaredMethod("run", String.class);
            method.invoke(target, jobBean.getParams());
            long executeTime = System.currentTimeMillis() - beginTime;
            logBean.setTimes((int)executeTime);
            logBean.setStatus(0);
            LOG.info("定時器 === >> "+jobBean.getJobId()+"執行成功,耗時 === >> " + executeTime);
        } catch (Exception e){
            // 異常信息
            long executeTime = System.currentTimeMillis() - beginTime;
            logBean.setTimes((int)executeTime);
            logBean.setStatus(1);
            logBean.setError(e.getMessage());
        } finally {
            scheduleJobLogService.insert(logBean) ;
        }
    }
}

三、定時器服務封裝

1、定時器初始化

@Service
public class ScheduleJobServiceImpl implements ScheduleJobService {

    @Resource
    private Scheduler scheduler ;
    @Resource
    private ScheduleJobMapper scheduleJobMapper ;

    /**
     * 定時器初始化
     */
    @PostConstruct
    public void init (){
        ScheduleJobExample example = new ScheduleJobExample() ;
        List<ScheduleJobBean> scheduleJobBeanList = scheduleJobMapper.selectByExample(example) ;
        for (ScheduleJobBean scheduleJobBean : scheduleJobBeanList) {
            CronTrigger cronTrigger = ScheduleUtil.getCronTrigger(scheduler,scheduleJobBean.getJobId()) ;
            if (cronTrigger == null){
                ScheduleUtil.createJob(scheduler,scheduleJobBean);
            } else {
                ScheduleUtil.updateJob(scheduler,scheduleJobBean);
            }
        }
    }
}

2、添加定時器

@Override
@Transactional(rollbackFor = Exception.class)
public int insert(ScheduleJobBean record) {
    ScheduleUtil.createJob(scheduler,record);
    return scheduleJobMapper.insert(record);
}

3、立即執行一次定時器

@Override
@Transactional(rollbackFor = Exception.class)
public void run(Long jobId) {
    ScheduleJobBean scheduleJobBean = scheduleJobMapper.selectByPrimaryKey(jobId) ;
    ScheduleUtil.run(scheduler,scheduleJobBean);
}

4、更新定時器

@Override
@Transactional(rollbackFor = Exception.class)
public int updateByPrimaryKeySelective(ScheduleJobBean record) {
    ScheduleUtil.updateJob(scheduler,record);
    return scheduleJobMapper.updateByPrimaryKeySelective(record);
}

5、停止定時器

@Override
@Transactional(rollbackFor = Exception.class)
public void pauseJob(Long jobId) {
    ScheduleJobBean scheduleJobBean = scheduleJobMapper.selectByPrimaryKey(jobId) ;
    ScheduleUtil.pauseJob(scheduler,jobId);
    scheduleJobBean.setStatus(1);
    scheduleJobMapper.updateByPrimaryKeySelective(scheduleJobBean) ;
}

6、恢復定時器

@Override
@Transactional(rollbackFor = Exception.class)
public void resumeJob(Long jobId) {
    ScheduleJobBean scheduleJobBean = scheduleJobMapper.selectByPrimaryKey(jobId) ;
    ScheduleUtil.resumeJob(scheduler,jobId);
    scheduleJobBean.setStatus(0);
    scheduleJobMapper.updateByPrimaryKeySelective(scheduleJobBean) ;
}

7、刪除定時器

@Override
@Transactional(rollbackFor = Exception.class)
public void delete(Long jobId) {
    ScheduleUtil.deleteJob(scheduler, jobId);
    scheduleJobMapper.deleteByPrimaryKey(jobId) ;
}

四、配置一個測試的定時器

1、定時接口封裝

public interface TaskService {
    void run(String params);
}

2、測試定時器

@Component("getTimeTask")
public class GetTimeTask implements TaskService {
    private static final Logger LOG = LoggerFactory.getLogger(GetTimeTask.class.getName()) ;
    private static final SimpleDateFormat format =
            new SimpleDateFormat("yyyy-MM-dd HH:mm:ss") ;
    @Override
    public void run(String params) {
        LOG.info("Params === >> " + params);
        LOG.info("當前時間::::"+format.format(new Date()));
    }
}

五、源代碼

GitHub·地址
https://github.com/cicadasmile/middle-ware-parent
GitEE·地址
https://gitee.com/cicadasmile/middle-ware-parent