Quartz定时任务 1.QuartzDemo1 QuartzDemo2: 2.Cron表达式 3.Cron表达式栗子 4.表达式栗子 5.工具类Demo2:Quartz实现定时任务 6.分布式定时任务实现 7.动态配置定时任务触发时间 8.重启服务器，启动所有定时任务
Quartz定时任务
添加依赖：
<dependency>
<groupId>org.quartz-scheduler</groupId>
<artifactId>quartz</artifactId>
<version>2.3.0</version>
</dependency>
1.1Quartz工具类
import static org.quartz.JobBuilder.newJob;
import static org.quartz.JobKey.jobKey;
import static org.quartz.TriggerBuilder.newTrigger;
import static org.quartz.CronScheduleBuilder.cronSchedule;
import static org.quartz.TriggerKey.triggerKey;
import java.util.Date;
import org.quartz.CronTrigger;
import org.quartz.JobDetail;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
public class QuartzUtils {
private static String JOB_NAME = "EXTJWEB_NAME";
private static String JOB_GROUP_NAME = "EXTJWEB_JOBGROUP_NAME";
private static String TRIGGER_NAME = "EXTJWEB_NAME";
private static String TRIGGER_GROUP_NAME = "EXTJWEB_TRIGGERGROUP_NAME";
/**
* @Description: 添加一个定时任务，使用默认的任务组名，触发器名，触发器组名
* @param sched:调度器
* @param jobClass:任务
* @param time:时间设置，CronExpression表达式
*/
public static void addJob(Scheduler sched, @SuppressWarnings("rawtypes") Class jobClass, String time) {
addJob(sched, jobClass,time,JOB_NAME,JOB_GROUP_NAME,TRIGGER_NAME,TRIGGER_GROUP_NAME);
}
/**
* @Description: 添加一个定时任务
* @param sched:调度器
* @param jobClass:任务
* @param time:时间设置，CronExpression表达式
* @param jobName:任务名
* @param jobGroupName:任务组名
* @param triggerName:触发器名
* @param triggerGroupName:触发器组名
*/
public static void addJob(Scheduler sched, @SuppressWarnings("rawtypes") Class jobClass, String time,
String jobName, String jobGroupName, String triggerName, String triggerGroupName) {
JobDetail job = newJob(jobClass).withIdentity(jobName, jobGroupName).build();
CronTrigger trigger = newTrigger().withIdentity(triggerName, triggerGroupName)
.withSchedule(cronSchedule(time)).build();
try {
// 返回为 null 添加失败
Date ft = sched.scheduleJob(job, trigger);
} catch (SchedulerException e) {
e.printStackTrace();
}
}
/**
* @Description: 定义一个任务之后进行触发设定(使用默认的任务组名，触发器名，触发器组名)
* @param sched:调度器
* @param time
*/
@SuppressWarnings("rawtypes")
public static void addJObLaterUse(Scheduler sched, @SuppressWarnings("rawtypes") Class jobClass, String time) {
addJObLaterUse(sched,jobClass,time,JOB_NAME,JOB_GROUP_NAME);
}
/**
* @Description: 定义一个任务之后进行触发设定
* @param sched:调度器
* @param time
* @param jobName:任务名
* @param jobGroupName:任务组名
*/
@SuppressWarnings("rawtypes")
public static void addJObLaterUse(Scheduler sched, @SuppressWarnings("rawtypes") Class jobClass, String time,
String jobName,String jobGroupName) {
JobDetail job = newJob(jobClass).withIdentity(jobName, jobGroupName).storeDurably().build();
try {
sched.addJob(job, false);
} catch (SchedulerException e) {
e.printStackTrace();
}
}
/**
* @Description: 对已存储的任务进行scheduling(使用默认的任务组名，触发器名，触发器组名)
* @param sched:调度器
* @param time
* @param jobName:任务名
* @param jobGroupName:任务组名
*/
@SuppressWarnings("rawtypes")
public static void schedulingStoredJOb(Scheduler sched, @SuppressWarnings("rawtypes") Class jobClass, String time) {
schedulingStoredJOb(sched,jobClass,time,JOB_NAME,JOB_GROUP_NAME,TRIGGER_NAME,TRIGGER_GROUP_NAME);
}
/**
* @Description: 对已存储的任务进行scheduling
* @param sched:调度器
* @param time
* @param jobName:任务名
* @param jobGroupName:任务组名
*/
@SuppressWarnings("rawtypes")
public static void schedulingStoredJOb(Scheduler sched, @SuppressWarnings("rawtypes") Class jobClass, String time,
String jobName,String jobGroupName,String triggerName, String triggerGroupName) {
Trigger trigger = newTrigger().withIdentity(triggerName, triggerGroupName).startNow()
.forJob(jobKey(jobName,jobGroupName))
.build();
try {
sched.scheduleJob(trigger);
} catch (SchedulerException e) {
e.printStackTrace();
}
}
/**
* @Description: 修改一个任务的触发时间(使用默认的任务组名，触发器名，触发器组名)
* @param sched:调度器
* @param time
*/
@SuppressWarnings("rawtypes")
public static void modifyJobTime(Scheduler sched, String time) {
modifyJobTime(sched, TRIGGER_NAME, TRIGGER_GROUP_NAME, time);
}
/**
* @Description: 修改一个任务的触发时间
* @param sched:调度器
* @param triggerName
* @param triggerGroupName
* @param time
*/
public static void modifyJobTime(Scheduler sched, String triggerName, String triggerGroupName, String time) {
Trigger trigger = newTrigger().withIdentity(triggerName, triggerGroupName).withSchedule(cronSchedule(time)).startNow().build();
try {
sched.rescheduleJob(triggerKey(triggerName, triggerGroupName), trigger);
} catch (SchedulerException e) {
e.printStackTrace();
}
}
/**
* @Description: 修改一个任务(使用默认的任务组名，任务名)
* @param sched:调度器
*/
@SuppressWarnings("rawtypes")
public static void modifyJob(Scheduler sched, @SuppressWarnings("rawtypes") Class jobClass) {
modifyJob(sched,jobClass,JOB_NAME,JOB_GROUP_NAME);
}
/**
* @Description: 修改一个任务
* @param sched:调度器
* @param jobName:任务名
* @param jobGroupName:任务组名
*/
public static void modifyJob(Scheduler sched, @SuppressWarnings("rawtypes") Class jobClass, String jobName,String jobGroupName) {
JobDetail job1 = newJob(jobClass).withIdentity(jobName,jobGroupName).build();
try {
sched.addJob(job1, true);
} catch (SchedulerException e) {
e.printStackTrace();
}
}
/**
* @Description: 删除一个任务的的trigger
* @param sched:调度器
* @param triggerName
* @param triggerGroupName
*/
public static void unschedulingJob(Scheduler sched,String triggerName, String triggerGroupName) {
try {
sched.unscheduleJob(triggerKey(triggerName,triggerGroupName));
} catch (SchedulerException e) {
e.printStackTrace();
}
}
/**
* @Description: 移除一个任务，以及任务的所有trigger
* @param sched:调度器
* @param jobName
*/
public static void removeJob(Scheduler sched, String jobName,String jobGroupName) {
try {
sched.deleteJob(jobKey(jobName,jobGroupName));
} catch (SchedulerException e) {
e.printStackTrace();
}
}
/**
* @Description:启动所有定时任务
* @param sched:调度器
*/
public static void startJobs(Scheduler sched) {
try {
sched.start();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
/**
* @Description:关闭所有定时任务
* @param sched:调度器
*/
public static void shutdownJobs(Scheduler sched) {
try {
if (!sched.isShutdown()) {
//未传参或false：不等待执行完成便结束；true：等待任务执行完才结束
sched.shutdown();
}
} catch (Exception e) {
throw new RuntimeException(e);
}
}
}
1.2新建一个Job
public class QueryVehicleToRedis implements Job {
@Override
public void execute(JobExecutionContext jobExecutionContext) {
System.out.println("定时任务开始吧！");
}
}
1.3添加一个定时任务或调用工具类的启动定时任务方法，都可启动定时任务
Scala版本：
import java.time.LocalDateTime
import com.xuhai.util.QuartzUtil
object StartOrShutDownQuartz {
def startJobs(): Unit = {
println("开启定时任务：车辆数据推送至Redis!" + LocalDateTime.now())
QuartzUtil.startJobs()
}
def main(args: Array[String]): Unit = {
//添加一个定时任务，每周天1点，添加时会启动定时任务
QuartzUtil.addJob("CalDetrial", classOf[QueryVehicleToRedis], "0 0 1 ? * SUN")
}
def shutDownJobs(): Unit = {
println("关闭所有定时任务!" + LocalDateTime.now())
QuartzUtil.shutdownJobs()
}
}
QuartzDemo2:
新建一个能够打印任意内容的Job：
public class PrintWordsJob implements Job{
@Override
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
String printTime = new SimpleDateFormat("yy-MM-dd HH-mm-ss").format(new Date());
System.out.println("PrintWordsJob start at:" + printTime + ", prints: Hello Job-" + new Random().nextInt(100));
}
}
创建Schedule，执行任务
public class MyScheduler {
public static void main(String[] args) throws SchedulerException, InterruptedException {
// 1、创建调度器Scheduler
SchedulerFactory schedulerFactory = new StdSchedulerFactory();
Scheduler scheduler = schedulerFactory.getScheduler();
// 2、创建JobDetail实例，并与PrintWordsJob类绑定(Job执行内容)
JobDetail jobDetail = JobBuilder.newJob(PrintWordsJob.class)
.withIdentity("job1", "group1").build();
// 3、构建Trigger实例,每隔1s执行一次
Trigger trigger = TriggerBuilder.newTrigger().withIdentity("trigger1", "triggerGroup1")
.startNow()//立即生效
.withSchedule(SimpleScheduleBuilder.simpleSchedule()
.withIntervalInSeconds(1)//每隔1s执行一次
.repeatForever()).build();//一直执行
//4、执行
scheduler.scheduleJob(jobDetail, trigger);
System.out.println("--------scheduler start ! ------------");
scheduler.start();
//睡眠
TimeUnit.MINUTES.sleep(1);
scheduler.shutdown();
System.out.println("--------scheduler shutdown ! ------------");
}
}
运行程序，可以看到程序每隔1s会打印出内容，且在一分钟后结束：
每周一到周五上午10:30执行定时任务
public class MyScheduler2 {
public static void main(String[] args) throws SchedulerException, InterruptedException {
// 1、创建调度器Scheduler
SchedulerFactory schedulerFactory = new StdSchedulerFactory();
Scheduler scheduler = schedulerFactory.getScheduler();
// 2、创建JobDetail实例，并与PrintWordsJob类绑定(Job执行内容)
JobDetail jobDetail = JobBuilder.newJob(PrintWordsJob.class)
.usingJobData("jobDetail1", "这个Job用来测试的")
.withIdentity("job1", "group1").build();
// 3、构建Trigger实例,每隔1s执行一次
Date startDate = new Date();
startDate.setTime(startDate.getTime() + 5000);
Date endDate = new Date();
endDate.setTime(startDate.getTime() + 5000);
CronTrigger cronTrigger = TriggerBuilder.newTrigger().withIdentity("trigger1", "triggerGroup1")
.usingJobData("trigger1", "这是jobDetail1的trigger")
.startNow()//立即生效
.startAt(startDate)
.endAt(endDate)
.withSchedule(CronScheduleBuilder.cronSchedule("* 30 10 ? * 1/5 2018"))
.build();
//4、执行
scheduler.scheduleJob(jobDetail, cronTrigger);
System.out.println("--------scheduler start ! ------------");
scheduler.start();
System.out.println("--------scheduler shutdown ! ------------");
}
}
2.Cron表达式
晓梦_知行：Quartz定时任务的表达式总结
| 名称 | 是否必须 | 允许值 | 特殊字符 |
|---|---|---|---|
| 秒 | 是 | 0-59 | , - * / |
| 分 | 是 | 0-59 | , - * / |
| 时 | 是 | 0-23 | , - * / |
| 日 | 是 | 1-31 | , - * ? / L W C |
| 月 | 是 | 1-12或JAN-DEC | , - * / |
| 周 | 是 | 1-7或SUN-SAT | , - * ? / L C # |
| 年 | 否 | 空或1970-2099 | z, - * / |
注意事项：
1、月份和星期的名称是不区分大小写的。FRI和fri是一样的。
2、域之间有空格分隔。
2.1特殊字符含义
2.1 *星号
使用星号(*)表示在这个域上包含所有合法的值。
表达式样例：
0 * 17 * * ?
意义：每天从下午5点到下午5:59中的每分钟激发一次trigger。它停在下午5:59是因为值17在小时域上，在下午6点时，小时变为18了，也就不再理会这个trigger，直到下一天的下午5点。
如果你希望trigger在该域的所有有效值上被激发时使用*字符。
2.2 ? 问号
? 号只能用在日和周域上，但是不能在这两个域上同时使用。你可以认为?字符是"我并不关心在该域上是什么值。"这不同于星号，星号是指示着该域上的每一个值。?是说不为该域指定值。
为了表达式语义明确，不要在日和周上同时指定值，记住，假如你为这两个域中的其中一个域指定了值，那就必须在另一个域上放一个?。
表达式样例：
0 10,44 14 ? 3 WEB
意义：在3月份，每个星期三的下午2:10和下午2:44被触发。
2.3 , 逗号
逗号(,)是用来在给某个域上指定一个值列表的。例如，使用值0,15,30,45在秒域上意味着每15秒触发一个trigger。
表达式样例：
0 0,15,30,45 * * * ?
意义：每刻钟触发一次 trigger。
2.4 /斜杠
斜杠(/)是用于时间表的递增。我们刚刚用了逗号来表示每15分钟的递增，但是我们也能写成这样0/15。
表达式样例：
0/15 0/30 * * * ?
意义：在整点和半点时每15秒触发trigger。
2.5 - 中划线
中划线(-)用于指定一个范围。例如，在小时域上的3-8意味着"3,4,5,6,7和8点。" ,域的值不允许回卷，所以像50-10这样的值是不允许的。
表达式样例：
0 45 3-8 ? * *
意义：在上午的3点至上午的8点的45分时触发trigger。
2.6L 字母
L说明了某域上允许的最后一个值。它仅被日和周域支持。当用在日域上，表示的是在月域上指定的月份的最后一天。例如，当月域上指定了JAN时，在日域上的L会促使trigger在1月31号被触发。假如月域上是SEP，那么L会预示着在9月30号触发。换句话说，就是不管指定了哪个月，都是在相应月份的时最后一天触发trigger。
表达式样例：
0 0 8 L * ?
意义：在每个月最后一天的上午 8:00 触发 trigger。
当 L 字母用于周域上，指示着周的最后一天，就是星期六(或者数字7)。所以如果你需要在每个月的最后一个星期六下午的11:59触发trigger，你可以用这样的表达式0 59 23 ? * L。
当使用于周域上，你可以用一个数字与 L 连起来表示月份的最后一个星期 X。例如，表达式0 0 12 ? * 2L说的是在每个月的最后一个星期一的12:00触发trigger。
不要让范围和列表值与 L 连用，虽然你能用星期数(1-7)与L连用，但是不允许你用一个范围值和列表值与L连用。这会产生不可预知的结果。
2.7W 字母
W 字符代表着平日 (Mon-Fri)，并且仅能用于日域中。它用来指定离指定日的最近的一个平日。大部分的商业处理都是基于工作周的，所以W字符可能是非常重要的。例如，日域中的15W意味着"离该月15号的最近一个平日。"假如15号是星期六，那么trigger会在14号(星期五)触发，因为星期五比星期一（这个例子中是17号）离15号更近。W只能用在指定的日域为单天，不能是范围或列表值。
2.8 # 井号
井号字符仅能用于周域中。它用于指定月份中的第几周的哪一天。例如，如果你指定周域的值为6#3，它意思是某月的第三个周五(6=星期五，#3意味着月份中的第三周)。另一个例子2#1意思是某月的第一个星期一(2=星期一，#1意味着月份中的第一周)。注意，假如你指定#5，然而月份中没有第5周，那么该月不会触发。
3.Cron表达式栗子
3.1分钟Cron表达式
| 用法 | 表达式 |
|---|---|
| 每天的从 5:00 PM 至5:59 PM中的每分钟触发 | 0 * 17 * * ? |
| 每天的从 11:00 PM 至11:55 PM中的每五分钟触发 | 0 0/5 23 * * ? |
| 每天的从 3:00 至3:55 PM和6:00 PM至6:55 PM之中的每五分钟触发 | 0 0/5 15,18 * * ? |
| 每天的从 5:00 AM 至5:05 AM中的每分钟触发 | 0 0-5 5 * * ? |
3.2天（日）Cron表达式
| 用法 | 表达式 |
|---|---|
| 每天的 3:00 AM | 0 0 3 * * ? |
| 每天的 3:00 AM (另一种写法) | 0 0 3 ? * * |
| 每天的 12:00 PM (中午) | 0 0 12 * * ? |
| 在 2019 中每天的10:15 AM | 0 15 10 * * ? 2019 |
3.3周和月的Cron表达式
| 用法 | 表达式 |
|---|---|
| 在每个周一,二,三,四和周五的10:15 AM | 0 15 10 ? * MON-FRI |
| 每月15号的 10:15 AM | 0 15 10 15 * ? |
| 每月最后一天的 10:15 AM | 0 15 10 L * ? |
| 每月最后一个周五的 10:15 AM | 0 15 10 ? * 6L |
| 在 2002, 2003, 2004, 和2005年中的每月最后一个周五的10:15 AM | 0 15 10 ? * 6L 2002-2005 |
| 每月第三个周五的 10:15 AM | 0 15 10 ? * 6#3 |
| 每月从第一天算起每五天的 12:00 PM (中午) | 0 0 12 1/5 * ? |
| 每一个 11 月11号的11:11 AM | 0 11 11 11 11 ? |
| 三月份每个周三的 2:10 PM 和2:44 PM | 0 10,44 14 ? 3 WED |
4.表达式栗子
高达一号:Quartz_quartz定时任务 cron表达式详解
0 * * * * ? 每1分钟触发一次
0 0 * * * ? 每天每1小时触发一次
0 0 10 * * ? 每天10点触发一次
0 * 14 * * ? 在每天下午2点到下午2:59期间的每1分钟触发
0 30 9 1 * ? 每月1号上午9点半
0 15 10 15 * ? 每月15日上午10:15触发
*/5 * * * * ? 每隔5秒执行一次
0 */1 * * * ? 每隔1分钟执行一次
0 0 5-15 * * ? 每天5-15点整点触发
0 0/3 * * * ? 每三分钟触发一次
0 0-5 14 * * ? 在每天下午2点到下午2:05期间的每1分钟触发
0 0/5 14 * * ? 在每天下午2点到下午2:55期间的每5分钟触发
0 0/5 14,18 * * ? 在每天下午2点到2:55期间和下午6点到6:55期间的每5分钟触发
0 0/30 9-17 * * ? 朝九晚五工作时间内每半小时
0 0 10,14,16 * * ? 每天上午10点，下午2点，4点
0 0 12 ? * WED 表示每个星期三中午12点
0 0 17 ? * TUES,THUR,SAT 每周二、四、六下午五点
0 10,44 14 ? 3 WED 每年三月的星期三的下午2:10和2:44触发
0 15 10 ? * MON-FRI 周一至周五的上午10:15触发
0 0 23 L * ? 每月最后一天23点执行一次
0 15 10 L * ? 每月最后一日的上午10:15触发
0 15 10 ? * 6L 每月的最后一个星期五上午10:15触发
0 15 10 * * ? 2005 2005年的每天上午10:15触发
0 15 10 ? * 6L 2002-2005 2002年至2005年的每月的最后一个星期五上午10:15触发
0 15 10 ? * 6#3 每月的第三个星期五上午10:15触发
5.工具类Demo2:Quartz实现定时任务
以下博文来源于：BarryWang
import java.text.SimpleDateFormat
import java.util.Date
import org.quartz.{CronScheduleBuilder, CronTrigger, JobBuilder, JobKey, TriggerBuilder, TriggerKey, _}
import org.quartz.impl.StdSchedulerFactory
import scala.collection.JavaConverters._
/**
* 定时任务管理类
*
* @author BarryWang create at 2018/5/11 14:22
* @version 0.0.1
*/
object QuartzManager {
private val stdSchedulerFactory = new StdSchedulerFactory
private val JOB_GROUP_NAME = "JOB_GROUP_NAME"
private val TRIGGER_GROUP_NAME = "TRIGGER_NAME"
/**
* 根据指定格式(yyyy-MM-dd HH:mm:ss)时间字符串添加定时任务，使用默认的任务组名，触发器名，触发器组名
* @param jobName 任务名
* @param time 时间设置，参考quartz说明文档
* @param jobClass 任务类名
*/
def addJobByTime(jobName: String, time: String, jobClass: Class[_ <: Job]) : Unit = {
QuartzManager.addJobByTime(jobName, time, jobClass, Map("1"->"otherData"))
}
/**
* 根据指定时间(java.util.Date)添加定时任务，使用默认的任务组名，触发器名，触发器组名
*
* @param jobName 任务名
* @param date 日期
* @param jobClass 任务类名
*/
def addJobByDate(jobName: String, date: Date, jobClass: Class[_ <: Job]): Unit = {
QuartzManager.addJobByDate(jobName, date, jobClass, Map("1"->"otherData"))
}
/**
* 根据指定cron表达式添加定时任务，使用默认的任务组名，触发器名，触发器组名
*
* @param jobName 任务名
* @param jobClass 任务类名
* @param cron cron表达式
*/
def addJobByCron(jobName: String, cron : String, jobClass: Class[_ <: Job]): Unit = {
QuartzManager.addJobByCron(jobName, cron, jobClass, Map("1"->"otherData"))
}
/**
* 函数描述: 添加一个定时任务，使用默认的任务组名，触发器名，触发器组名
* @param jobName 任务名
* @param time 时间字符串, 格式为(yyyy-MM-dd HH:mm:ss)
* @param jobClass 任务类名
* @param paramsMap 定时器需要额外数据
*/
def addJobByTime(jobName: String, time: String, jobClass: Class[_ <: Job], paramsMap: Map[_ <: String, _ <: AnyRef]): Unit = {
addJobByTime(jobName, time, "yyyy-MM-dd HH:mm:ss", jobClass, paramsMap)
}
/**
* 函数描述: 添加一个定时任务，使用默认的任务组名，触发器名，触发器组名
* @param jobName 任务名
* @param time 时间设置，参考quartz说明文档
* @param jobClass 任务类名
* @param paramsMap 定时器需要额外数据
*/
def addJobByTime(jobName: String, time: String, timePattern: String, jobClass: Class[_ <: Job], paramsMap: Map[_ <: String, _ <: AnyRef]): Unit = {
val df = new SimpleDateFormat(timePattern)
val cron = getCron(df.parse(time))
addJobByCron(jobName, cron, jobClass, paramsMap)
}
/**
* Description: 添加一个定时任务，使用默认的任务组名，触发器名，触发器组名
*
* @param jobName 任务名
* @param date 日期
* @param cls 任务
* @param paramsMap 定时器需要额外数据
*/
def addJobByDate(jobName: String, date: Date, cls: Class[_ <: Job], paramsMap: Map[_ <: String, _ <: AnyRef]): Unit = {
val cron = getCron(date)
addJobByCron(jobName, cron, cls, paramsMap)
}
/**
* 函数描述: 根据cron表达式添加定时任务(默认触发器组名及任务组名)
* @param jobId 任务ID
* @param cron 时间设置 表达式，参考quartz说明文档
* @param jobClass 任务的类
* @param paramsMap 可变参数需要进行传参的值
*/
def addJobByCron(jobId: String, cron: String, jobClass: Class[_ <: Job], paramsMap: Map[_ <: String, _ <: AnyRef]): Unit = {
addJob(jobId, cron, jobClass, paramsMap, JOB_GROUP_NAME, TRIGGER_GROUP_NAME)
}
/**
* 函数描述: 根据cron表达式添加定时任务
* @param jobId 任务ID
* @param cron 时间设置 表达式，参考quartz说明文档
* @param jobClass 任务的类类型 eg:TimedMassJob.class
* @param paramsMap 可变参数需要进行传参的值
* @param jobGroupName 任务组名
* @param triggerGroupName 触发器组名
*/
def addJob(jobId: String, cron: String, jobClass: Class[_ <: Job], paramsMap: Map[_ <: String, _ <: AnyRef],
jobGroupName: String, triggerGroupName: String): Unit = {
val scheduler = stdSchedulerFactory.getScheduler
// 任务名，任务组，任务执行类
val jobDetail = JobBuilder.newJob(jobClass).withIdentity(jobId, jobGroupName).build
//设置参数
jobDetail.getJobDataMap.putAll(paramsMap.asJava)
val triggerBuilder = TriggerBuilder.newTrigger
// 触发器名,触发器组
//默认设置触发器名与任务ID相同
val triggerName = jobId
triggerBuilder.withIdentity(triggerName, triggerGroupName)
triggerBuilder.startNow
// 触发器时间设定
triggerBuilder.withSchedule(CronScheduleBuilder.cronSchedule(cron))
// 创建Trigger对象
val trigger = triggerBuilder.build.asInstanceOf[CronTrigger]
// 调度容器设置JobDetail和Trigger
scheduler.scheduleJob(jobDetail, trigger)
// 启动
if (!scheduler.isShutdown) scheduler.start()
}
/**
* 函数描述: 修改一个任务的触发时间(使用默认的任务组名，触发器名，触发器组名)
* @param cron 时间字符串
*/
def modifyJobTime(jobId: String, cron: String, jobClass: Class[_ <: Job]): Unit = {
modifyJobTime(jobId, cron, jobClass, Map("1"->"otherData"), JOB_GROUP_NAME, TRIGGER_GROUP_NAME)
}
/**
* 函数描述: 修改一个任务的触发时间(使用默认的任务组名，触发器名，触发器组名)
* @param cron 时间字符串
*/
def modifyJobTime(jobId: String, cron: String, jobClass: Class[_ <: Job], paramsMap: Map[_ <: String, _ <: AnyRef]): Unit = {
modifyJobTime(jobId, cron, jobClass, paramsMap, JOB_GROUP_NAME, TRIGGER_GROUP_NAME)
}
/**
* 函数描述: 修改一个任务的触发时间
* @param jobId 任务ID
* @param cron cron表达式
* @param jobClass 任务类名
* @param paramsMap 其他参数
* @param jobGroupName 任务组名
* @param triggerGroupName 触发器组
*/
def modifyJobTime(jobId: String, cron: String, jobClass: Class[_ <: Job], paramsMap: Map[_ <: String, _ <: AnyRef],
jobGroupName: String, triggerGroupName: String): Unit = {
val scheduler = stdSchedulerFactory.getScheduler()
//默认设置触发器名与任务ID相同
val triggerName = jobId
val triggerKey = TriggerKey.triggerKey(triggerName, triggerGroupName)
var trigger = scheduler.getTrigger(triggerKey).asInstanceOf[CronTrigger]
if (trigger != null) {
removeJob(jobId)
}
addJob(jobId, cron, jobClass, paramsMap, jobGroupName, triggerGroupName)
}
/**
* 函数描述: 移除一个任务(使用默认的任务组名，触发器名，触发器组名)
* @param jobId 任务名称
*/
def removeJob(jobId: String): Unit = {
val scheduler = stdSchedulerFactory.getScheduler
//默认设置触发器名与任务ID相同
val triggerName = jobId
val triggerKey = TriggerKey.triggerKey(triggerName, TRIGGER_GROUP_NAME)
// 停止触发器
scheduler.pauseTrigger(triggerKey)
// 移除触发器
scheduler.unscheduleJob(triggerKey)
// 删除任务
scheduler.deleteJob(JobKey.jobKey(jobId , JOB_GROUP_NAME))
}
/**
* 函数描述: 移除一个任务
* @param jobId 任务ID
* @param jobGroupName 任务组
* @param triggerName 触发器名称
* @param triggerGroupName 触发器组名
*/
def removeJob(jobId: String, jobGroupName: String, triggerName: String, triggerGroupName: String): Unit = {
val scheduler = stdSchedulerFactory.getScheduler
val triggerKey = TriggerKey.triggerKey(triggerName, triggerGroupName)
// 停止触发器
scheduler.pauseTrigger(triggerKey)
// 移除触发器
scheduler.unscheduleJob(triggerKey)
// 删除任务
scheduler.deleteJob(JobKey.jobKey(jobId , jobGroupName))
}
/**
* 函数描述:启动所有定时任务
*/
def startJobs(): Unit = {
stdSchedulerFactory.getScheduler.start()
}
/**
* 函数描述:关闭所有定时任务
*
*/
def shutdownJobs(): Unit = {
val sched = stdSchedulerFactory.getScheduler
if (!sched.isShutdown) sched.shutdown()
}
/**
* 根据时间获取Cron表达式
* @param date 日期
* @return
*/
def getCron(date: Date): String = {
val dateFormat = "ss mm HH dd MM ? yyyy"
formatDateByPattern(date, dateFormat)
}
/**
* 日期格式转换
* @param date 日期
* @param dateFormat 格式
* @return
*/
def formatDateByPattern(date : Date, dateFormat : String): String = {
val sdf = new SimpleDateFormat(dateFormat)
sdf.format(date)
}
}
6.分布式定时任务实现
分布式定时任务：即在分布式服务的环境下，启动定时任务。分布式定时任务需要解决的问题是：同一定时任务，同一时间点，只能在一台服务器上启动。
为了解决分布式定时任务的问题，我们需要框架层面解决定时任务触发时，选举一台服务器作为master节点。
实现思路如下：
1.服务启动注册服务时，为服务编号；
2.从注册服务中随机选中一台服务器作为master节点；
3.服务挂掉或添加时重新选举。
import java.util.HashMap;
import java.util.Map;
/**
* Created by tangliu on 2016/7/13.
*/
public class MasterHelper {
public static Map<String, Boolean> isMaster = new HashMap<>();
/**
* 根据serviceName, versionName，判断当前服务是否集群中的master
* todo 服务版本号是否作为master判断的依据??
* @param servieName
* @param versionName
* @return
*/
public static boolean isMaster(String servieName, String versionName) {
String key = generateKey(servieName, versionName);
if (!isMaster.containsKey(key))
return false;
else
return isMaster.get(key);
}
public static String generateKey(String serviceName, String versionName) {
return serviceName + ":" + versionName;
}
}
竞选master:
/**
* 监听服务节点下面的子节点（临时节点，实例信息）变化
*/
public void watchInstanceChange(RegisterContext context) {
String watchPath = context.getServicePath();
try {
List<String> children = zk.getChildren(watchPath, event -> {
LOGGER.warn("ServerZk::watchInstanceChange zkEvent:" + event);
//Children发生变化，则重新获取最新的services列表
if (event.getType() == Watcher.Event.EventType.NodeChildrenChanged) {
LOGGER.info("容器状态:{}, {}子节点发生变化，重新获取子节点...", ContainerFactory.getContainer().status(), event.getPath());
if (ContainerFactory.getContainer().status() == Container.STATUS_SHUTTING
|| ContainerFactory.getContainer().status() == Container.STATUS_DOWN) {
LOGGER.warn("Container is shutting down");
return;
}
watchInstanceChange(context);
}
});
boolean _isMaster = false;
if (children.size() > 0) {
_isMaster = checkIsMaster(children, MasterHelper.generateKey(context.getService(), context.getVersion()), context.getInstanceInfo());
}
//masterChange响应
LifecycleProcessorFactory.getLifecycleProcessor().onLifecycleEvent(
new LifeCycleEvent(LifeCycleEvent.LifeCycleEventEnum.MASTER_CHANGE,
context.getService(), _isMaster));
} catch (KeeperException | InterruptedException e) {
LOGGER.error(e.getMessage(), e);
create(context.getServicePath() + "/" + context.getInstanceInfo(), context, true);
}
}
//-----竞选master---
private static Map<String, Boolean> isMaster = MasterHelper.isMaster;
/**
* @param children 当前方法下的实例列表， eg 127.0.0.1:9081:1.0.0,192.168.1.12:9081:1.0.0
* @param serviceKey 当前服务信息 eg com.github.user.UserService:1.0.0
* @param instanceInfo 当前服务节点实例信息 eg 192.168.10.17:9081:1.0.0
*/
public boolean checkIsMaster(List<String> children, String serviceKey, String instanceInfo) {
if (children.size() <= 0) {
return false;
}
boolean _isMaster = false;
/**
* 排序规则
* a: 192.168.100.1:9081:1.0.0:0000000022
* b: 192.168.100.1:9081:1.0.0:0000000014
* 根据 lastIndexOf : 之后的数字进行排序，由小到大，每次取zk临时有序节点中的序列最小的节点作为master
*/
try {
Collections.sort(children, (o1, o2) -> {
Integer int1 = Integer.valueOf(o1.substring(o1.lastIndexOf(":") + 1));
Integer int2 = Integer.valueOf(o2.substring(o2.lastIndexOf(":") + 1));
return int1 - int2;
});
String firstNode = children.get(0);
LOGGER.info("serviceInfo firstNode {}", firstNode);
String firstInfo = firstNode.replace(firstNode.substring(firstNode.lastIndexOf(":")), "");
if (firstInfo.equals(instanceInfo)) {
isMaster.put(serviceKey, true);
_isMaster = true;
LOGGER.info("({})竞选master成功, master({})", serviceKey, CURRENT_CONTAINER_ADDR);
} else {
isMaster.put(serviceKey, false);
_isMaster = false;
LOGGER.info("({})竞选master失败，当前节点为({})", serviceKey);
}
} catch (NumberFormatException e) {
LOGGER.error("临时节点格式不正确,请使用新版，正确格式为 etc. 192.168.100.1:9081:1.0.0:0000000022");
}
return _isMaster;
}
分布式环境判断是否是master：
if (!MasterHelper.isMaster("com.today.api.financetask.service.FinanceScheduledService", "1.0.0")) {
//excute the task
}
7.动态配置定时任务触发时间
实现定时任务可动态配置，需要通过数据库表保存最新一次修改的cron表达式来实现：
建表如下：
CREATE TABLE t_scheduled_task (
job_name varchar(50) NOT NULL COMMENT 'job名称'
, job_id varchar(40) NOT NULL COMMENT 'job ID'
, job_cron varchar(50) NOT NULL COMMENT 'Job cron表达式'
, job_type int COMMENT 'Job类型'
, is_start tinyint(2) DEFAULT 1 NOT NULL COMMENT '是否已启动,0:否(no);1:是(yes)'
, remark varchar(256) COMMENT '备注'
, updated_at timestamp DEFAULT CURRENT_TIMESTAMP NOT NULL COMMENT '更新时间'
, created_at timestamp DEFAULT CURRENT_TIMESTAMP NOT NULL COMMENT '创建时间'
, has_deleted tinyint(2) DEFAULT 0 NOT NULL COMMENT '是否已删除,0:否(no);1:是(yes)'
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='定时任务配置表';
CREATE UNIQUE INDEX uk_t_scheduled_task ON t_scheduled_task ( job_id);
建表后需要页面调用接口实现先停掉上次的定时任务，再根据最新修改的触发时间新建一个新的定时任务：
如下：
//页面设置每天触发的时间，格式：HH:mm
val cron = convertHourMinuteToCron(processTime)
//修改定时任务时间, 保存入库
ScheduledTaskQuerySql.isExists(jobId) match {
case true => ScheduledTaskActionSql.updateTaskCron(jobId, cron)
case false => ScheduledTaskActionSql.insertTScheduledTask(
TScheduledTask( report.name,
jobId,
cron,
None,
TScheduledTaskIsStartEnum.YES.id,
None,
null,
null,
TScheduledTaskHasDeletedEnum.NO.id))
}
//关掉老的定时任务，添加新的定时任务
QuartzManager.modifyJobTime(jobId, cron, classOf[DailyGenIncomeDetailJob])
/**
* 每天定时触发-转换时分格式(hh:mm)为cron表达式
* @param hourMibuteStr
* @return
*/
def convertHourMinuteToCron(hourMibuteStr : String) : String = {
val splitHm = hourMibuteStr.split(":")
s"0 ${splitHm(1).trim.toInt} ${splitHm(0).trim.toInt} * * ?"
}
/**
* 每天定时触发-转换时分格式(hh:hh:mm)为cron表达式
* @param dayStr
* @return
*/
def convertDayToCron(dayStr : String) : String = {
val splitHm = dayStr.split(":")
s"0 ${splitHm(1).trim.toInt} ${splitHm(0).trim.toInt} * * ?"
}
定义Job及父类:
Job定义
import java.util.Calendar
import com.today.api.checkaccount.scala.enums.FlatFormTypeEnum
import com.today.api.checkaccount.scala.request.ReconciliationRequest
import com.today.api.checkaccount.scala.CheckAccountServiceClient
import com.today.service.financetask.job.define.{AbstractJob, JobEnum}
import org.quartz.JobExecutionContext
class CheckAccountJob extends AbstractJob{
/**
* get the api information
*
* @return (interface name, interface version, JobEnum)
*/
override def getJobAndApiInfo(context: JobExecutionContext): (String, String, JobEnum) = {
("com.today.api.financetask.service.CloseAccountScheduleService", "1.0.0", JobEnum.CHECK_ACCOUNT_PROCESS)
}
/**
* start up the scheduled task
*
* @param context JobExecutionContext
*/
override def run(context: JobExecutionContext): Unit = {
val cal = Calendar.getInstance
cal.add(Calendar.DATE, -1)
new CheckAccountServiceClient().appReconciliation(new ReconciliationRequest(FlatFormTypeEnum.TODAY_APP,None))
}
}
公共父类:
import java.io.{PrintWriter, StringWriter}
import com.github.dapeng.core.helper.MasterHelper
import com.today.api.financetask.scala.enums.TScheduledTaskLogEnum
import com.today.service.financetask.action.sql.ScheduledTaskLogSql
import com.today.service.financetask.util.{AppContextHolder, Debug}
import org.quartz.{Job, JobExecutionContext}
import org.slf4j.LoggerFactory
import org.springframework.transaction.TransactionStatus
import org.springframework.transaction.support.TransactionTemplate
import scala.util.{Failure, Success, Try}
/**
* the abstract class for job
*/
trait AbstractJob extends Job{
/** 日志 */
val logger = LoggerFactory.getLogger(getClass)
/**
* execute the job
* @param context
*/
override def execute(context: JobExecutionContext): Unit = {
val jobAndApiInfo = getJobAndApiInfo(context)
if (!MasterHelper.isMaster(jobAndApiInfo._1, jobAndApiInfo._2)) {
logger.info(s"Can't select master to run the job ${jobAndApiInfo._3.jobId}: ${jobAndApiInfo._3.jobName}")
return
}
val logId = ScheduledTaskLogSql.insertScheduledTaskLog(jobAndApiInfo._3)
context.put("logId", logId)
logger.info(s"Starting the job ${jobAndApiInfo._3.jobId}: ${jobAndApiInfo._3.jobName} ...")
/**
* 事物处理
*/
val transactionTemplate: TransactionTemplate = AppContextHolder.getBean("transactionTemplate")
transactionTemplate.execute((status: TransactionStatus) =>{
Debug.reset()
Try(Debug.trace(s"${jobAndApiInfo._3.jobId}:${jobAndApiInfo._3.jobName}")(run(context))) match
{
case Success(x) => {
logger.info(s"Successfully execute the job ${jobAndApiInfo._3.jobId}: ${jobAndApiInfo._3.jobName}")
successLog(logId)
}
case Failure(e) => {
logger.error(s"Failure execute the job ${jobAndApiInfo._3.jobId}: ${jobAndApiInfo._3.jobName}", e)
failureLog(logId, status, e)
}
}
Debug.info()
})
}
/**
* get the api information
* @return (interface name, interface version, JobEnum)
*/
def getJobAndApiInfo(context: JobExecutionContext) : (String, String, JobEnum)
/**
* start up the scheduled task
* @param context JobExecutionContext
*/
def run(context: JobExecutionContext)
/**
* 成功日志记录
* @param logId
*/
def successLog(logId: Long): Unit ={
ScheduledTaskLogSql.updateExportReportRecord(logId, TScheduledTaskLogEnum.SUCCESS, "Success")
}
/**
* 失败日志记录
* @param logId
*/
def failureLog(logId: Long, status: TransactionStatus, e: Throwable): Unit ={
status.setRollbackOnly()
ScheduledTaskLogSql.updateExportReportRecord(logId, TScheduledTaskLogEnum.FAILURE, getExceptionStack(e))
}
/**
*
* 功能说明:在日志文件中 ，打印异常堆栈
* @param e : Throwable
* @return : String
*/
def getExceptionStack(e: Throwable): String = {
val errorsWriter = new StringWriter
e.printStackTrace(new PrintWriter(errorsWriter))
errorsWriter.toString
}
}
8.重启服务器，启动所有定时任务
重启定时任务，需要重启所有定时任务
这个过程需要将所有定时任务及触发周期保存到数据库，重启后，读取数据库启动恢复所有定时任务
代码如下(Spring框架下启动服务自动会启动ApplicationListener.onApplicationEvent(event: ContextRefreshedEvent))：
import com.today.api.financetask.scala.enums.{TReportTypeEnum, TScheduledTaskIsStartEnum}
import com.today.api.financetask.scala.request.QueryAutoConfigRequest
import com.today.service.financetask.job._
import com.today.service.financetask.job.define.JobEnum
import com.today.service.financetask.query.sql.{AutoConfigQuerySql, ScheduledTaskQuerySql}
import com.today.service.financetask.util.QuartzManager
import org.slf4j.LoggerFactory
import org.springframework.context.ApplicationListener
import org.springframework.context.event.ContextRefreshedEvent
import org.springframework.stereotype.Service
/**
* 类功能描述: 定时器监听器, 服务启动时启动定时器
*
* @author BarryWang create at 2018/5/11 12:04
* @version 0.0.1
*/
@Service
class ScheduleStartListener extends ApplicationListener[ContextRefreshedEvent] {
/** 日志 */
val logger = LoggerFactory.getLogger(getClass)
/**
* 启动加载执行定时任务
*/
override def onApplicationEvent(event: ContextRefreshedEvent): Unit = {
logger.info("=======服务器重启定时任务启动start=======")
//1. 恢复日次处理定时任务
recoveryDayTimeProcessJob()
//2. 恢复每天营收明细报表生成定时任务
recoveryImcomeDetailGenJob()
logger.info("=======服务器重启定时任务启动end=======")
}
/**
* 恢复日次处理定时任务
*/
private def recoveryDayTimeProcessJob(): Unit ={
try {
ScheduledTaskQuerySql.queryByJobId(JobEnum.DAY_TIME_PROCESS.jobId) match {
case Some(x) =>
if(x.isStart == TScheduledTaskIsStartEnum.YES.id)
QuartzManager.addJobByCron(JobEnum.DAY_TIME_PROCESS.jobId, x.jobCron, classOf[DayTimeProcessJob])
else
logger.info("定时任务：" + JobEnum.DAY_TIME_PROCESS.jobName + "is_start标志为0，不启动")
case None =>
QuartzManager.addJobByCron(JobEnum.DAY_TIME_PROCESS.jobId, "0 30 2 * * ?", classOf[DayTimeProcessJob])
}
} catch {
case e : Exception => logger.error(JobEnum.DAY_TIME_PROCESS.jobName + "启动失败， 失败原因：", e)
}
}
/**
* 恢复营收明细报表生成定时任务
*/
private def recoveryImcomeDetailGenJob(): Unit = {
val jobName = TReportTypeEnum.INCOMEDETAIL_REPORT.name
try {
val jobId = TReportTypeEnum.INCOMEDETAIL_REPORT.id.toString
ScheduledTaskQuerySql.queryByJobId(jobId) match {
case Some(x) =>
if (x.isStart == TScheduledTaskIsStartEnum.YES.id)
QuartzManager.addJobByCron(jobId, x.jobCron, classOf[DailyGenIncomeDetailJob])
else
logger.info("定时任务：" + jobName + "is_start标志为0，不启动")
case None =>
QuartzManager.addJobByCron(jobId, "0 10 0 * * ?", classOf[DailyGenIncomeDetailJob])
}
}catch {
case e : Exception => logger.error(jobName + "启动失败， 失败原因：", e)
}
}
}
使用Scala枚举定义定时信息如下：
case class JobEnum(val jobId: String, val jobName: String) extends Enumeration
/**
* 所有Job 枚举定义在此类, 不能重复
* jobId不能重复
* @author BarryWang create at 2018/5/12 10:45
* @version 0.0.1
*/
object JobEnum {
val DAY_TIME_PROCESS = JobEnum("DAY_TIME_PROCESS", "日次处理定时任务")
val INVOICE_SYNC_PROCESS = JobEnum("INVOICE_SYNC_PROCESS", "采购系统同步单据数据定时任务")
val RETIREMENT_SYNC_PROCESS = JobEnum("RETIREMENT_SYNC_PROCESS", "采购系统同步报废单据数据定时任务")
val CLOSE_ACCOUNT_STATE_PROCESS = JobEnum("CLOSE_ACCOUNT_STATE_PROCESS","更新关账状态定时任务")
val PURCHASE_ORDER_2_SYNC_PROCESS = JobEnum("PURCHASE_ORDER_2_SYNC_PROCESS","采购系统同步PO2数据定时任务")
val SEND_EMAIL_PROCESS = JobEnum("SEND_EMAIL_PROCESS","计划付款通知和已付款通知定时任务")
val CLOSE_ACCOUNT_BASE_DATA_PROCESS = JobEnum("CLOSE_ACCOUNT_BASE_DATA_PROCESS","关账基础数据同步定时任务")
}