定时任务总会遇到任务重叠执行的情况，比如一个任务1分钟执行一次，而任务的执行时间超过了1分钟，这样就会有两个相同任务并发执行了。有时候我们是允许这种情况的发生的，比如任务执行的代码是幂等的，而有时候我们可能考虑到一些情况是不允许这种事情发生的。
在实际场景中，我们定时任务调度使用quartz来实现触发的，定时任务的业务代码分布在各个应用，用soa调用。
对于quartz来说，官方文档上明确对这种需求有指定的解决办法，就是使用注解@DisallowConcurrentExecution；
意思是：禁止并发执行多个相同定义的JobDetail，就是我们想要的。
下面一个实现的例子：可以对比两个job：AllowConcurrentExecutionTestJob，DisallowConcurrentExecutionTestJob
public class AllowConcurrentExecutionTestJob implements Job {
public AllowConcurrentExecutionTestJob() {
}
public void execute(JobExecutionContext context) throws JobExecutionException {
try {
List<JobExecutionContext> list = context.getScheduler().getCurrentlyExecutingJobs();
for(JobExecutionContext jobExecutionContext : list){
// job内部获取容器内变量
System.out.println(jobExecutionContext.getJobDetail().getKey().getName());
}
Thread.sleep(4000);
} catch (SchedulerException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Hello World! AllowConcurrentExecutionTestJob is executing.");
}
}
@DisallowConcurrentExecution
public class DisallowConcurrentExecutionTestJob implements org.quartz.Job {
public DisallowConcurrentExecutionTestJob() {
} public void execute(JobExecutionContext context) throws JobExecutionException { try {
List<JobExecutionContext> list = context.getScheduler().getCurrentlyExecutingJobs();
for(JobExecutionContext jobExecutionContext : list){
// job内部获取容器内变量
System.out.println(jobExecutionContext.getJobDetail().getKey().getName());
}
Thread.sleep(4000);
} catch (SchedulerException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Hello World! DisallowConcurrentExecutionTestJob is executing.");
}
}
测试代码：
public class QuartzTest {
public static void main(String[] args) throws InterruptedException {
try {
// Grab the Scheduler instance from the Factory
Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
// and start it off
scheduler.start();
// define the job and tie it to our HelloJob class
JobDetail job = JobBuilder.newJob(DisallowConcurrentExecutionTestJob.class)
.withIdentity("job1", "group1")
.build();
// Trigger the job to run now, and then repeat every 40 seconds
Trigger trigger = TriggerBuilder.newTrigger()
.withIdentity("trigger1", "group1")
.startNow()
.withSchedule(SimpleScheduleBuilder.simpleSchedule()
.withIntervalInSeconds(1)
.repeatForever())
.build();
// define the job and tie it to our HelloJob class
JobDetail job2 = JobBuilder.newJob(AllowConcurrentExecutionTestJob.class)
.withIdentity("job2", "group1")
.build();
// Trigger the job to run now, and then repeat every 40 seconds
Trigger trigger2 = TriggerBuilder.newTrigger()
.withIdentity("trigger2", "group1")
.startNow()
.withSchedule(SimpleScheduleBuilder.simpleSchedule()
.withIntervalInSeconds(1)
.repeatForever())
.build();
// Tell quartz to schedule the job using our trigger
scheduler.scheduleJob(job2, trigger2);
// scheduler.scheduleJob(job2, trigger2);
// wait trigger
Thread.sleep(20000);
scheduler.shutdown();
} catch (SchedulerException se) {
se.printStackTrace();
}
}
}
我们还发现在job的execute里传参是JobExecutionContext，它可以让我们拿到正在执行的job的信息。所以我想在job里直接判断一下就可以知道有没有已经在执行的相同job。
public class SelfDisAllowConExeTestJob implements org.quartz.Job{
public void execute(JobExecutionContext context) throws JobExecutionException {
try {
List<JobExecutionContext> list = context.getScheduler().getCurrentlyExecutingJobs();
Set<String> jobs = new HashSet<String>();
int i=0;
for (JobExecutionContext jobExecutionContext : list){
if(context.getJobDetail().getKey().getName().equals(jobExecutionContext.getJobDetail().getKey().getName())){
i++;
}
}
if(i>1){
System.out.printf("self disallow ");
return;
}
Thread.sleep(4000);
} catch (SchedulerException e) {
e.printStackTrace();
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println("Hello World! SelfDisAllowConExeTestJob is executing.");
}
}
测试代码：
public class OuartzSelfMapTest {
public static void main(String[] args) throws InterruptedException {
try {
// Grab the Scheduler instance from the Factory
Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
// and start it off
scheduler.start();
// define the job and tie it to our HelloJob class
JobDetail job = JobBuilder.newJob(SelfDisAllowConExeTestJob.class)
.withIdentity("job1", "group1")
.build();
// Trigger the job to run now, and then repeat every 40 seconds
Trigger trigger = TriggerBuilder.newTrigger()
.withIdentity("trigger1", "group1")
.startNow()
.withSchedule(SimpleScheduleBuilder.simpleSchedule()
.withIntervalInSeconds(1)
.repeatForever())
.build();
// Tell quartz to schedule the job using our trigger
scheduler.scheduleJob(job, trigger);
// wait trigger
Thread.sleep(20000);
scheduler.shutdown();
} catch (SchedulerException se) {
se.printStackTrace();
}
}
}
我们在实际代码中经常会结合spring，特地去看了一下MethodInvokingJobDetailFactoryBean的concurrent属性来控制是否限制并发执行的实现：
Class<?> jobClass = (this.concurrent ? MethodInvokingJob.class : StatefulMethodInvokingJob.class);
/**
* Extension of the MethodInvokingJob, implementing the StatefulJob interface.
* Quartz checks whether or not jobs are stateful and if so,
* won't let jobs interfere with each other.
*/
@PersistJobDataAfterExecution
@DisallowConcurrentExecution
public static class StatefulMethodInvokingJob extends MethodInvokingJob { // No implementation, just an addition of the tag interface StatefulJob
// in order to allow stateful method invoking jobs.
}
当然，在quartz里有一个StatefulJob，方便直接继承就实现了concurrent=false的事情了。
那么啰嗦了这么多，其实就是想表达，quartz里并没有一个可以设置说是否并发的接口，而是需要自定义的job自行继承，或使用注解来实现的。
另外，还有一个相关的注解：@PersistJobDataAfterExecution
意思是：放在JobDetail 里的JobDataMap是共享的，也就是相同任务之间执行时可以传输信息。很容易想到既然是共享的，那么就会有并发的问题，就如开头说的这个场景就会导致并发问题。所以官方文档也特别解释这个注解最好和@DisallowConcurrentExecution一起使用。
以下是例子：
@PersistJobDataAfterExecution
public class PersistJob implements Job {
public void execute(JobExecutionContext context) throws JobExecutionException {
JobDataMap data = context.getJobDetail().getJobDataMap();
int i = data.getInt("P");
System.out.printf("PersistJob=>"+i);
i++;
data.put("P", i);
}
}
测试代码：
public class PersistJobDataQuartzTest {
public static void main(String[] args) throws InterruptedException {
try {
// Grab the Scheduler instance from the Factory
Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
// and start it off
scheduler.start();
JobDataMap jobDataMap = new JobDataMap();
jobDataMap.put("P",1);
// define the job and tie it to our HelloJob class
JobDetail job = JobBuilder.newJob(PersistJob.class)
.withIdentity("job1", "group1").usingJobData(jobDataMap)
.build();
// Trigger the job to run now, and then repeat every 40 seconds
Trigger trigger = TriggerBuilder.newTrigger()
.withIdentity("trigger1", "group1")
.startNow()
.withSchedule(SimpleScheduleBuilder.simpleSchedule()
.withIntervalInSeconds(1)
.repeatForever())
.build();
// Tell quartz to schedule the job using our trigger
scheduler.scheduleJob(job, trigger);
// wait trigger
Thread.sleep(20000);
scheduler.shutdown();
} catch (SchedulerException se) {
se.printStackTrace();
}
}
}
参考文档：