Quartz实现数据库动态配置定时任务

或许实现的方式跟之前的代码有点不一样的

1.定时任务的配置信息

&＃64;Configuration
public class ScheduleConfigration {&＃64;Autowiredprivate ScheduleInfoAction scheduleInfoAction;&＃64;Autowiredprivate ChannelSyncTask ChannelSyncTask;/*** 用于5分钟轮训检查cron修改&＃xff08;基本不需要修改&＃xff09;* &＃64;return*/&＃64;Bean(name &＃61; "jobDetail")public MethodInvokingJobDetailFactoryBean jobDetail(){MethodInvokingJobDetailFactoryBean methodInvokingJobDetailFactoryBean &＃61; new MethodInvokingJobDetailFactoryBean();methodInvokingJobDetailFactoryBean.setConcurrent(false);methodInvokingJobDetailFactoryBean.setTargetObject(scheduleInfoAction);methodInvokingJobDetailFactoryBean.setTargetMethod("reScheduleJob");return methodInvokingJobDetailFactoryBean;}/*** 用于5分钟轮训检查cron修改&＃xff08;基本不需要修改&＃xff09;* &＃64;return*/&＃64;Bean(name &＃61; "cronTrigger")public CronTriggerFactoryBean cronTrigger(){CronTriggerFactoryBean cronTriggerFactoryBean &＃61; new CronTriggerFactoryBean();cronTriggerFactoryBean.setJobDetail(jobDetail().getObject());// 设置默认刷新croncronTriggerFactoryBean.setCronExpression(Properties.getString("refresh.default.cron"));return cronTriggerFactoryBean;}/*** dycChannel 任务&＃xff0c;需要添加新的定时任务&＃xff0c;需要重复配置JobDetail,CronTrigger* &＃64;return*/&＃64;Bean(name &＃61; "channelSyncJobCronJobDetail")public MethodInvokingJobDetailFactoryBean channelSyncJobCronJobDetail(){MethodInvokingJobDetailFactoryBean methodInvokingJobDetailFactoryBean &＃61; new MethodInvokingJobDetailFactoryBean();methodInvokingJobDetailFactoryBean.setConcurrent(false);methodInvokingJobDetailFactoryBean.setTargetObject(channelSyncTask);methodInvokingJobDetailFactoryBean.setTargetMethod("doTask");return methodInvokingJobDetailFactoryBean;}&＃64;Bean(name &＃61; "channelSyncJobCronTrigger")public CronTriggerFactoryBean channelSyncJobCronTrigger(){CronTriggerFactoryBean cronTriggerFactoryBean &＃61; new CronTriggerFactoryBean();cronTriggerFactoryBean.setJobDetail(channelSyncJobCronJobDetail().getObject());cronTriggerFactoryBean.setCronExpression("0 0 1 * * ?");return cronTriggerFactoryBean;}/*** repeat code JobDetail and CronTrigger ...**//*** 调度工厂* &＃64;return*/&＃64;Bean(name &＃61; "schedulerFactory")public SchedulerFactoryBean schedulerFactoryBean(){SchedulerFactoryBean schedulerFactoryBean &＃61; new SchedulerFactoryBean();//在添加新的触发器时&＃xff0c;增加相应的触发器schedulerFactoryBean.setTriggers(cronTrigger().getObject(),channelSyncJobCronTrigger().getObject());return schedulerFactoryBean;}
}


2.刷新的定时任务

&＃64;Component
public class ScheduleInfoAction{private Logger logger &＃61; LoggerFactory.getLogger(ScheduleInfoAction.class);private Scheduler scheduler;public void reScheduleJob() throws SchedulerException {scheduler &＃61; (Scheduler) SpringContextUtils.getBean("schedulerFactory");// 数据库取任务列表List jobs &＃61; yunyingDao.getJobCodesFromDB(null);if (jobs !&＃61; null && !jobs.isEmpty()) {for (String job : jobs) {TriggerKey triggerKey &＃61; new TriggerKey(job &＃43; "CronTrigger", Scheduler.DEFAULT_GROUP);CronTriggerImpl trigger &＃61; (CronTriggerImpl) scheduler.getTrigger(triggerKey);String dbCronExpression &＃61; getCronExpressionFromDB(job);String originConExpression;if (StringUtils.isNotBlank(dbCronExpression)){if (trigger !&＃61; null){originConExpression &＃61; trigger.getCronExpression();if(!dbCronExpression.equalsIgnoreCase(originConExpression)){try{trigger.setCronExpression(dbCronExpression);scheduler.rescheduleJob(triggerKey, trigger);logger.info("jobCode:{}, dbCron:{}, originCron:{}",job,dbCronExpression,originConExpression);} catch (Exception e) {logger.error("jobCode:{}, dbCron:{}, originCron:{}, refresh Cron Error:{}" , job,dbCronExpression, originConExpression, e);}}}else {trigger &＃61; (CronTriggerImpl)SpringContextUtils.getBean(job &＃43; "CronTrigger");if (trigger !&＃61; null){scheduler.scheduleJob(trigger);logger.info("scheduleJob:{}", job);}}}else {if (trigger !&＃61; null){scheduler.unscheduleJob(triggerKey);logger.info("unscheduleJob:{}", job);}}}}}// 数据库取最新的Cron值private String getCronExpressionFromDB(String jobName){return yunyingDao.getCronByCode(jobName);}
}

3.设置定时任务

    1&＃xff09;基础抽象类的实现&＃xff0c;&＃xff08;由于可能处于分布式环境中&＃xff0c;需要使用zookeeper的分布式锁&＃xff09;

public abstract class BaseTask {private Logger logger &＃61; LoggerFactory.getLogger(BaseTask.class);public abstract void execute();public void doTask(){String clazzName &＃61; this.getClass().getSimpleName();String className &＃61; clazzName.substring(0,1).toLowerCase() &＃43; clazzName.substring(1);String lockPath &＃61; "/oper/schedule/" &＃43; className &＃43; "/lock";ZooKeeperClient zkClient &＃61; (ZooKeeperClient)SpringContextUtils.getBean("zkClient");CronTriggerImpl cronTrigger &＃61; (CronTriggerImpl)SpringContextUtils.getBean(className.replace("Task","Job") &＃43; "CronTrigger");logger.info("lock path:{},zkClient:{},trigger cron:{}", lockPath, zkClient, cronTrigger.getCronExpression());Boolean isGetLocker &＃61; false;InterProcessMutex lock &＃61; new InterProcessMutex(zkClient.getClient(), lockPath);try {if (lock.acquire(2000, TimeUnit.MILLISECONDS)){isGetLocker &＃61; true;execute();}} catch (Exception e) {logger.error("BaseTask 执行锁任务时抛错&＃xff1a;", e);}finally {try {if(isGetLocker){lock.release();}} catch (Exception e) {logger.error("释放锁出错",e);}}}
}

需要执行的定时任务的代码实现继承BaseTask&＃xff0c;并实现业务代码。

不过&＃xff0c;这里我们中间又使用了Groovy代码&＃xff0c;由于Groovy代码在这里可以直接对数据库进行操作&＃xff0c;所以这里继承BaseTask 的定时任务&＃xff0c;我们是调用Groovy Class 的方法。

    2&＃xff09;定时任务的实现

&＃64;Component
public class DycChannelSyncTask extends BaseTask{private Logger logger &＃61; LoggerFactory.getLogger(DycChannelSyncTask.class);public void execute() {logger.info("渠道信息开始执行任务:" &＃43; TimeUtils.formatTime(new Date(), TimeUtils.FORMAT_YYYYMMDDHHMMSS));//进行数据抽取存储&＃xff0c;这里调的是groovy的方法channelSyncJob.doJob();logger.info("渠道信息任务执行完成:" &＃43; TimeUtils.formatTime(new Date(), TimeUtils.FORMAT_YYYYMMDDHHMMSS));}
}

4.需要引入的Maven

org.quartz-schedulerquartz2.2.1org.codehaus.groovygroovy-all2.4.6mysqlmysql-connector-java5.1.31org.apache.zookeeperzookeeperorg.slf4jslf4j-log4j12log4jlog4jorg.apache.curatorcurator-frameworkorg.apache.curatorcurator-recipes


5.数据库中定时任务的表结构

CREATE TABLE &＃96;tb_job_config&＃96; (&＃96;id&＃96; bigint(20) NOT NULL AUTO_INCREMENT COMMENT &＃39;自增ID&＃39;,&＃96;jobCode&＃96; varchar(32) DEFAULT NULL COMMENT &＃39;商户&＃39;,&＃96;jobName&＃96; varchar(32) DEFAULT NULL COMMENT &＃39;渠道编号&＃39;,&＃96;cron&＃96; varchar(128) DEFAULT NULL,&＃96;jobDetail&＃96; longtext COMMENT &＃39;渠道名称&＃39;,&＃96;status&＃96; smallint(1) DEFAULT NULL COMMENT &＃39;1、有效 2 无效&＃39;,PRIMARY KEY (&＃96;id&＃96;),UNIQUE KEY &＃96;merchantNo_channelCode_unique&＃96; (&＃96;jobCode&＃96;,&＃96;jobName&＃96;) USING BTREE
) ENGINE&＃61;InnoDB AUTO_INCREMENT&＃61;164 DEFAULT CHARSET&＃61;utf8 COMMENT&＃61;&＃39;任务调度配置表&＃39;;
INSERT INTO &＃96;tb_job_config&＃96; (&＃96;id&＃96;, &＃96;jobCode&＃96;, &＃96;jobName&＃96;, &＃96;cron&＃96;, &＃96;jobDetail&＃96;, &＃96;status&＃96;)
VALUES (&＃39;139&＃39;, &＃39;channelUserSyncJob&＃39;, &＃39;渠道用户同步&＃39;, &＃39;0 0 0/1 * * ? &＃39;, &＃39;{\"intervalHours\":\"2\"}&＃39;, &＃39;1&＃39;);// 定时任务执行完&＃xff0c;记录一下任务执行情况&＃xff0c;有助于排查问题等等
CREATE TABLE &＃96;tb_job_execute_log&＃96; (&＃96;id&＃96; bigint(64) NOT NULL AUTO_INCREMENT,&＃96;jobCode&＃96; varchar(255) DEFAULT NULL COMMENT &＃39;任务编号&＃39;,&＃96;totalNum&＃96; bigint(64) DEFAULT NULL COMMENT &＃39;记录总数&＃39;,&＃96;successNum&＃96; bigint(64) DEFAULT NULL COMMENT &＃39;记录成功数&＃39;,&＃96;errorNum&＃96; bigint(64) DEFAULT NULL COMMENT &＃39;记录错误数&＃39;,&＃96;recordTime&＃96; datetime DEFAULT CURRENT_TIMESTAMP COMMENT &＃39;记录时间&＃39;,PRIMARY KEY (&＃96;id&＃96;),KEY &＃96;Index_jobCode&＃96; (&＃96;jobCode&＃96;) USING BTREE
) ENGINE&＃61;InnoDB AUTO_INCREMENT&＃61;52810 DEFAULT CHARSET&＃61;utf8 COMMENT&＃61;&＃39;任务执行情况记录表&＃39;;

另外注意的是数据库中命名的Job&＃xff0c;要与实际代码中的命名规则保持一致。

我觉得使用之前的Job&＃xff0c;获取我们这边的代码可以简化很多。