基于springboot和quartz的任务调度系统

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
package com.littlehow.job.base.config;
import com.littlehow.job.base.BaseTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Configurable;
import org.springframework.context.annotation.Lazy;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.*;
import org.springframework.scheduling.support.CronTrigger;
import org.springframework.stereotype.Component;
import java.util.Date;
@Lazy(false)
@Configurable
@Component
@EnableScheduling
public class ScheduleTaskConfig implements SchedulingConfigurer {
private Logger log = LoggerFactory.getLogger(ScheduleTaskConfig.class);
private final static String TASK_NOT_EXISTS = "not exists";
private final static String TASK_EXISTS = "exists";
private final static String FAILURE = "failure";
private final static String SUCCESS = "success";
private ScheduledTaskRegistrar scheduledTaskRegistrar;
@Override
public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
this.scheduledTaskRegistrar = scheduledTaskRegistrar;
initTask();
}
/**
* 添加任务
* @param task
* @return
*/
public String addTask(BaseTask task) {
if (scheduledTaskRegistrar == null || task == null) {
return FAILURE;
}
if (TaskConfig.containsTask(task.getId())) {
return TASK_EXISTS;
}
try {
addTask0(task);
TaskConfig.addTask(task);
return SUCCESS;
} catch (Exception e) {
log.error("新增定时任务失败:" + task, e);
throw e;
}
}
/**
* 改变任务执行频率
* @param taskId
* @param expression
* @return
*/
public String changeTask(String taskId, String expression) {
BaseTask baseTask = TaskConfig.getTask(taskId);
if (baseTask == null || TaskType.TRIGGER != baseTask.taskType || expression == null) {
return TASK_NOT_EXISTS;
}
log.info("change trigger expression:(id=" + taskId + ",expression=" + expression+")");
baseTask.setExpression(expression);
return SUCCESS;
}
/**
* 取消定时任务
* @param taskId
* @return
*/
public String cancelTask(String taskId) {
if (!TaskConfig.containsTask(taskId)) {
return TASK_NOT_EXISTS;
}
try {
log.info("cancel task:" + taskId);
TaskConfig.removeTask(taskId).getScheduledTask().cancel();
} catch (Exception e) {
log.error("取消任务失败:" + taskId, e);
throw e;
}
return SUCCESS;
}
/**
* 初始化已配置任务
*/
private void initTask() {
TaskConfig.getTasks().forEach(task -> addTask0(task));
}
private void addTask0(BaseTask task) {
log.info("add task:" + task);
switch (task.taskType) {
case TRIGGER: task.setScheduledTask(addTriggerTask(task));
break;
case CRON: task.setScheduledTask(addCronTask(task, task.getExpression()));
break;
case FIXED_RATE: task.setScheduledTask(addFixedRateTask(task, task.interval()));
break;
case FIXED_DELAY: task.setScheduledTask(addFixedDelayTask(task, task.interval(), task.delay()));
break;
default:
}
}
/**
* 添加不可改变时间表的定时任务
* @param task
*/
private ScheduledTask addCronTask(Runnable task, String expression) {
return scheduledTaskRegistrar.scheduleCronTask(new CronTask(task, expression));
}
/**
* 添加可变时间task
* @param task
* @return
*/
private ScheduledTask addTriggerTask(BaseTask task) {
return scheduledTaskRegistrar.scheduleTriggerTask(new TriggerTask(task, triggerContext -> {
CronTrigger trigger = new CronTrigger(task.getExpression());
Date nextExec = trigger.nextExecutionTime(triggerContext);
return nextExec;
}));
}
/**
* 设置固定频率的定时任务
* @param task
* @param interval
*/
private ScheduledTask addFixedRateTask(Runnable task, long interval) {
return scheduledTaskRegistrar.scheduleFixedRateTask(new IntervalTask(task, interval, 0L));
}
/**
* 设置延迟以固定频率执行的定时任务
* @param task
* @param interval
* @param delay
*/
private ScheduledTask addFixedDelayTask(Runnable task, long interval, long delay) {
return scheduledTaskRegistrar.scheduleFixedDelayTask(new IntervalTask(task, interval, delay));
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
package com.littlehow.job.base;
import com.littlehow.job.base.config.TaskConfig;
import com.littlehow.job.base.config.TaskType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.scheduling.config.ScheduledTask;
/**
* 基本的任务配置类
*/
public abstract class BaseTask implements Runnable,InitializingBean {
protected final Logger log = LoggerFactory.getLogger(this.getClass());
public final TaskType taskType;
private ScheduledTask scheduledTask;
private final String id;
public BaseTask(TaskType taskType, String id) {
this.taskType = taskType;
this.id = id;
}
/**
*
获取任务表达式如：0 0 0/1 * * *? (每个整点执行)
* @return
*/
public abstract String getExpression();
/**
* 固定频率执行的时间间隔
* @return
*/
public abstract long interval();
/**
* 固定频率执行的延迟时间
* @return
*/
public abstract long delay();
/**
* 设置任务表达式
* @param expression
*/
public abstract void setExpression(String expression);
/**
* 获取任务唯一标识
* @return
*/
public String getId() {
return id;
}
public final ScheduledTask getScheduledTask() {
return scheduledTask;
}
public final void setScheduledTask(ScheduledTask scheduledTask) {
this.scheduledTask = scheduledTask;
}
@Override
public void afterPropertiesSet() {
TaskConfig.addTask(this);
}
public String toString() {
return this.getClass().getSimpleName() + "(id:" + id + ",expression:" + getExpression()
+ ",type:" + taskType + ",interval:" + interval()
+ ", delay:" + delay() + ")";
}
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
package com.littlehow.job.service;
import com.littlehow.job.base.BaseCronTask;
import com.littlehow.job.base.BaseFixedTask;
import com.littlehow.job.base.ScheduleBaseService;
import com.littlehow.job.manager.pojo.TaskDto;
import com.littlehow.job.manager.pojo.TaskType;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.StringUtils;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@Service
public class TaskExecuteService {
@Autowired
private ScheduleBaseService scheduleBaseService;
@Autowired
private CallbackExecuteService executeService;
private static final Pattern fixed = Pattern.compile("fixed=(\d+)\s*(,\s*delay=(\d+))?");
/**
* 新增任务
* 因表达式解析错误可能会抛出异常
* @param taskDto
*/
public void addTask(TaskDto taskDto) {
if (taskDto == null) {
return;
}
String taskId = getTaskId(taskDto);
CallbackExecuteService.addOrUpdateTask(taskId, taskDto.getCallbackUrl());
TaskType taskType = taskDto.getTaskType();
switch (taskType) {
case FIXED:
case FIXED_DELAY:
long[] params = getFixedParameters(taskDto.getExpression());
BaseFixedTask baseFixedTask = new BaseFixedTask(taskId,
params[0], params[1]) {
@Override
public void run() {
executeService.execute(this.getId());
}
};
scheduleBaseService.addTask(baseFixedTask);
break;
case CRON:
case TRIGGER:
BaseCronTask baseCronTask = new BaseCronTask(taskType == TaskType.CRON ?
com.littlehow.job.base.config.TaskType.CRON :
com.littlehow.job.base.config.TaskType.TRIGGER, taskId) {
@Override
public void run() {
executeService.execute(this.getId());
}
};
baseCronTask.setExpression(taskDto.getExpression());
scheduleBaseService.addTask(baseCronTask);
}
}
/**
* 修改任务
* @param taskDto
*/
public void updateTask(TaskDto taskDto) {
if (!StringUtils.isEmpty(taskDto.getExpression())) {
//修改表达式
scheduleBaseService.changeTask(getTaskId(taskDto), taskDto.getExpression());
}
if (!StringUtils.isEmpty(taskDto.getCallbackUrl())) {
//修改回调地址
CallbackExecuteService.addOrUpdateTask(getTaskId(taskDto), taskDto.getCallbackUrl());
}
}
/**
* 删除任务
* @param taskDto
*/
public void removeTask(TaskDto taskDto) {
String taskId = getTaskId(taskDto);
scheduleBaseService.removeTask(taskId);
CallbackExecuteService.removeTask(taskId);
}
/**
* 获取fixed类型的实际参数信息
* @param expression
* @return
*/
private static long[] getFixedParameters(String expression) {
Matcher matcher = fixed.matcher(expression);
if (matcher.find()) {
long[] fixedParam = {0L, 0L};
fixedParam[0] = Long.parseLong(matcher.group(1));
String delay = matcher.group(3);
if (delay != null) {
fixedParam[1] = Long.parseLong(delay);
}
return fixedParam;
}
throw new IllegalArgumentException("invalid expression:" +expression);
}
/**
* 获取任务编号
* @param task
* @return
*/
private String getTaskId(TaskDto task) {
return task.getBusinessId() + "-" + task.getTaskId();
}
}