本文主要研究一下PowerJob的TimingStrategyHandler
tech/powerjob/server/core/scheduler/auxiliary/TimingStrategyHandler.java
public interface TimingStrategyHandler {
/**
* 校验表达式
*
* @param timeExpression 时间表达式
*/
void validate(String timeExpression);
/**
* 计算下次触发时间
*
* @param preTriggerTime 上次触发时间 (not null)
* @param timeExpression 时间表达式
* @param startTime 开始时间(include)
* @param endTime 结束时间(include)
* @return next trigger time
*/
Long calculateNextTriggerTime(Long preTriggerTime, String timeExpression, Long startTime, Long endTime);
/**
* 支持的定时策略
*
* @return TimeExpressionType
*/
TimeExpressionType supportType();
}
TimingStrategyHandler接口定义了validate、calculateNextTriggerTime、supportType方法
tech/powerjob/common/enums/TimeExpressionType.java
@Getter
@AllArgsConstructor
@ToString
public enum TimeExpressionType {
API(1),
CRON(2),
FIXED_RATE(3),
FIXED_DELAY(4),
WORKFLOW(5),
DAILY_TIME_INTERVAL(11);
private final int v;
public static final List<Integer> FREQUENT_TYPES = Collections.unmodifiableList(Lists.newArrayList(FIXED_RATE.v, FIXED_DELAY.v));
/**
* 首次计算触发时间时必须计算出一个有效值
*/
public static final List<Integer> INSPECT_TYPES = Collections.unmodifiableList(Lists.newArrayList(CRON.v, DAILY_TIME_INTERVAL.v));
public static TimeExpressionType of(int v) {
for (TimeExpressionType type : values()) {
if (type.v == v) {
return type;
}
}
throw new IllegalArgumentException("unknown TimeExpressionType of " + v);
}
}
TimeExpressionType枚举定义了API、CRON、FIXED_RATE、FIXED_DELAY、WORKFLOW、DAILY_TIME_INTERVAL几种类型
tech/powerjob/server/core/scheduler/auxiliary/AbstractTimingStrategyHandler.java
public abstract class AbstractTimingStrategyHandler implements TimingStrategyHandler {
@Override
public void validate(String timeExpression) {
// do nothing
}
@Override
public Long calculateNextTriggerTime(Long preTriggerTime, String timeExpression, Long startTime, Long endTime) {
// do nothing
return null;
}
}
AbstractTimingStrategyHandler实现了TimingStrategyHandler的validate、calculateNextTriggerTime方法
tech/powerjob/server/core/scheduler/auxiliary/impl/ApiTimingStrategyHandler.java
@Component
public class ApiTimingStrategyHandler extends AbstractTimingStrategyHandler {
@Override
public TimeExpressionType supportType() {
return TimeExpressionType.API;
}
}
ApiTimingStrategyHandler继承了AbstractTimingStrategyHandler，其supportType返回的是TimeExpressionType.API
tech/powerjob/server/core/scheduler/auxiliary/impl/FixedRateTimingStrategyHandler.java
@Component
public class FixedRateTimingStrategyHandler extends AbstractTimingStrategyHandler {
@Override
public void validate(String timeExpression) {
long delay;
try {
delay = Long.parseLong(timeExpression);
} catch (Exception e) {
throw new PowerJobException("invalid timeExpression!");
}
// 默认 120s ，超过这个限制应该使用考虑使用其他类型以减少资源占用
int maxInterval = Integer.parseInt(System.getProperty(PowerJobDKey.FREQUENCY_JOB_MAX_INTERVAL, "120000"));
if (delay > maxInterval) {
throw new PowerJobException("the rate must be less than " + maxInterval + "ms");
}
if (delay <= 0) {
throw new PowerJobException("the rate must be greater than 0 ms");
}
}
@Override
public Long calculateNextTriggerTime(Long preTriggerTime, String timeExpression, Long startTime, Long endTime) {
long r = startTime != null && startTime > preTriggerTime
? startTime : preTriggerTime + Long.parseLong(timeExpression);
return endTime != null && endTime < r ? null : r;
}
@Override
public TimeExpressionType supportType() {
return TimeExpressionType.FIXED_RATE;
}
}
FixedRateTimingStrategyHandler继承了AbstractTimingStrategyHandler，其validate方法校验interval参数，要求大于0而且不能大于120s；calculateNextTriggerTime方法先根据startTime、preTriggerTime、timeExpression计算再与endTime做比较；其supportType返回的是TimeExpressionType.FIXED_RATE
tech/powerjob/server/core/scheduler/auxiliary/impl/FixedDelayTimingStrategyHandler.java
@Component
public class FixedDelayTimingStrategyHandler extends AbstractTimingStrategyHandler {
@Override
public void validate(String timeExpression) {
long delay;
try {
delay = Long.parseLong(timeExpression);
} catch (Exception e) {
throw new PowerJobException("invalid timeExpression!");
}
// 默认 120s ，超过这个限制应该考虑使用其他类型以减少资源占用
int maxInterval = Integer.parseInt(System.getProperty(PowerJobDKey.FREQUENCY_JOB_MAX_INTERVAL, "120000"));
if (delay > maxInterval) {
throw new PowerJobException("the delay must be less than " + maxInterval + "ms");
}
if (delay <= 0) {
throw new PowerJobException("the delay must be greater than 0 ms");
}
}
@Override
public TimeExpressionType supportType() {
return TimeExpressionType.FIXED_DELAY;
}
}
FixedDelayTimingStrategyHandler继承了AbstractTimingStrategyHandler，其validate要求delay大于0且小于等于120s；其supportType返回的是TimeExpressionType.FIXED_DELAY
tech/powerjob/server/core/scheduler/auxiliary/impl/WorkflowTimingStrategyHandler.java
@Component
public class WorkflowTimingStrategyHandler extends AbstractTimingStrategyHandler {
@Override
public TimeExpressionType supportType() {
return TimeExpressionType.WORKFLOW;
}
}
WorkflowTimingStrategyHandler继承了AbstractTimingStrategyHandler，其supportType返回的是TimeExpressionType.WORKFLOW
tech/powerjob/server/core/scheduler/auxiliary/impl/CronTimingStrategyHandler.java
@Component
public class CronTimingStrategyHandler implements TimingStrategyHandler {
private final CronParser cronParser;
/**
* @see CronDefinitionBuilder#instanceDefinitionFor
* <p>
* Enhanced quartz cron，Support for specifying both a day-of-week and a day-of-month parameter.
* https://github.com/PowerJob/PowerJob/issues/382
*/
public CronTimingStrategyHandler() {
CronDefinition cronDefinition = CronDefinitionBuilder.defineCron()
.withSeconds().withValidRange(0, 59).and()
.withMinutes().withValidRange(0, 59).and()
.withHours().withValidRange(0, 23).and()
.withDayOfMonth().withValidRange(1, 31).supportsL().supportsW().supportsLW().supportsQuestionMark().and()
.withMonth().withValidRange(1, 12).and()
.withDayOfWeek().withValidRange(1, 7).withMondayDoWValue(2).supportsHash().supportsL().supportsQuestionMark().and()
.withYear().withValidRange(1970, 2099).withStrictRange().optional().and()
.instance();
this.cronParser = new CronParser(cronDefinition);
}
@Override
public void validate(String timeExpression) {
cronParser.parse(timeExpression);
}
@Override
public Long calculateNextTriggerTime(Long preTriggerTime, String timeExpression, Long startTime, Long endTime) {
Cron cron = cronParser.parse(timeExpression);
ExecutionTime executionTime = ExecutionTime.forCron(cron);
if (startTime != null && startTime > System.currentTimeMillis() && preTriggerTime < startTime) {
// 需要计算出离 startTime 最近的一次真正的触发时间
Optional<ZonedDateTime> zonedDateTime = executionTime.lastExecution(ZonedDateTime.ofInstant(Instant.ofEpochMilli(startTime), ZoneId.systemDefault()));
preTriggerTime = zonedDateTime.map(dateTime -> dateTime.toEpochSecond() * 1000).orElse(startTime);
}
Instant instant = Instant.ofEpochMilli(preTriggerTime);
ZonedDateTime preZonedDateTime = ZonedDateTime.ofInstant(instant, ZoneId.systemDefault());
Optional<ZonedDateTime> opt = executionTime.nextExecution(preZonedDateTime);
if (opt.isPresent()) {
long nextTriggerTime = opt.get().toEpochSecond() * 1000;
if (endTime != null && endTime < nextTriggerTime) {
return null;
}
return nextTriggerTime;
}
return null;
}
@Override
public TimeExpressionType supportType() {
return TimeExpressionType.CRON;
}
}
CronTimingStrategyHandler实现了TimingStrategyHandler接口，其构造器先创建了CronDefinition，再根据CronDefinition创建CronParser；其validate方法调用了cronParser.parse(timeExpression)；calculateNextTriggerTime方法先解析timeExpression，再解析为ExecutionTime，再根据startTime和preTriggerTime计算新的preTriggerTime；最后通过executionTime.nextExecution计算nextTriggerTime；其supportType返回的是TimeExpressionType.CRON
tech/powerjob/server/core/scheduler/auxiliary/impl/DailyTimeIntervalStrategyHandler.java
@Component
public class DailyTimeIntervalStrategyHandler implements TimingStrategyHandler {
/**
* 使用中国星期！！！
*/
private static final Set<Integer> ALL_DAY = Sets.newHashSet(1, 2, 3, 4, 5, 6, 7);
@Override
public TimeExpressionType supportType() {
return TimeExpressionType.DAILY_TIME_INTERVAL;
}
@Override
@SneakyThrows
public void validate(String timeExpression) {
DailyTimeIntervalExpress ep = JsonUtils.parseObject(timeExpression, DailyTimeIntervalExpress.class);
CommonUtils.requireNonNull(ep.interval, "interval can't be null or empty in DailyTimeIntervalExpress");
CommonUtils.requireNonNull(ep.startTimeOfDay, "startTimeOfDay can't be null or empty in DailyTimeIntervalExpress");
CommonUtils.requireNonNull(ep.endTimeOfDay, "endTimeOfDay can't be null or empty in DailyTimeIntervalExpress");
TimeOfDay startTime = TimeOfDay.from(ep.startTimeOfDay);
TimeOfDay endTime = TimeOfDay.from(ep.endTimeOfDay);
if (endTime.before(startTime)) {
throw new IllegalArgumentException("endTime should after startTime!");
}
if (StringUtils.isNotEmpty(ep.intervalUnit)) {
TimeUnit.valueOf(ep.intervalUnit);
}
}
@Override
@SneakyThrows
public Long calculateNextTriggerTime(Long preTriggerTime, String timeExpression, Long startTime, Long endTime) {
DailyTimeIntervalExpress ep = JsonUtils.parseObject(timeExpression, DailyTimeIntervalExpress.class);
// 未开始状态下，用起点算调度时间
if (startTime != null && startTime > System.currentTimeMillis() && preTriggerTime < startTime) {
return calculateInRangeTime(startTime, ep);
}
// 间隔时间
TimeUnit timeUnit = Optional.ofNullable(ep.intervalUnit).map(TimeUnit::valueOf).orElse(TimeUnit.SECONDS);
long interval = timeUnit.toMillis(ep.interval);
Long ret = calculateInRangeTime(preTriggerTime + interval, ep);
if (ret == null || ret <= Optional.ofNullable(endTime).orElse(Long.MAX_VALUE)) {
return ret;
}
return null;
}
/**
* 计算最近一次在范围中的时间
* @param time 当前时间基准，可能直接返回该时间作为结果
* @param ep 表达式
* @return 最近一次在范围中的时间
*/
static Long calculateInRangeTime(Long time, DailyTimeIntervalExpress ep) {
Calendar calendar = Calendar.getInstance();
calendar.setTime(new Date(time));
int year = calendar.get(Calendar.YEAR);
// 月份 + 1，转为熟悉的 1～12 月
int month = calendar.get(Calendar.MONTH) + 1;
int day = calendar.get(Calendar.DAY_OF_MONTH);
// 判断是否符合"日"的执行条件
int week = TimeUtils.calculateWeek(year, month, day);
Set<Integer> targetDays = CollectionUtils.isEmpty(ep.daysOfWeek) ? ALL_DAY : ep.daysOfWeek;
// 未包含情况下，将时间改写为符合条件日的 00:00 分，重新开始递归（这部分应该有性能更优的写法，不过这个调度模式应该很难触发瓶颈，先简单好用的实现）
if (!targetDays.contains(week)) {
simpleSetCalendar(calendar, 0, 0, 0);
Date tomorrowZero = DateUtils.addDays(calendar.getTime(), 1);
return calculateInRangeTime(tomorrowZero.getTime(), ep);
}
// 范围的开始时间
TimeOfDay rangeStartTime = TimeOfDay.from(ep.startTimeOfDay);
simpleSetCalendar(calendar, rangeStartTime.getHour(), rangeStartTime.getMinute(), rangeStartTime.getSecond());
long todayStartTs = calendar.getTimeInMillis();
// 未开始
if (time < todayStartTs) {
return todayStartTs;
}
TimeOfDay rangeEndTime = TimeOfDay.from(ep.endTimeOfDay);
simpleSetCalendar(calendar, rangeEndTime.getHour(), rangeEndTime.getMinute(), rangeEndTime.getSecond());
long todayEndTs = calendar.getTimeInMillis();
// 范围之间
if (time <= todayEndTs) {
return time;
}
// 已结束，重新计算第二天时间
simpleSetCalendar(calendar, 0, 0, 0);
return calculateInRangeTime(DateUtils.addDays(calendar.getTime(), 1).getTime(), ep);
}
//......
}
DailyTimeIntervalStrategyHandler实现了TimingStrategyHandler接口，其supportType返回的是TimeExpressionType.DAILY_TIME_INTERVAL；其validate方法先解析参数为DailyTimeIntervalExpress，然后校验其endTime不能比startTime小；其calculateNextTriggerTime方法主要是通过calculateInRangeTime来计算最近一次在范围中的时间
tech/powerjob/server/core/scheduler/TimingStrategyService.java
@Slf4j
@Service
public class TimingStrategyService {
private static final int NEXT_N_TIMES = 5;
private static final List<String> TIPS = Collections.singletonList("It is valid, but has not trigger time list!");
private final Map<TimeExpressionType, TimingStrategyHandler> strategyContainer;
public TimingStrategyService(List<TimingStrategyHandler> timingStrategyHandlers) {
// init
strategyContainer = new EnumMap<>(TimeExpressionType.class);
for (TimingStrategyHandler timingStrategyHandler : timingStrategyHandlers) {
strategyContainer.put(timingStrategyHandler.supportType(), timingStrategyHandler);
}
}
/**
* 计算接下来几次的调度时间
*
* @param timeExpressionType 定时表达式类型
* @param timeExpression 表达式
* @param startTime 起始时间(include)
* @param endTime 结束时间(include)
* @return 调度时间列表
*/
public List<String> calculateNextTriggerTimes(TimeExpressionType timeExpressionType, String timeExpression, Long startTime, Long endTime) {
TimingStrategyHandler timingStrategyHandler = getHandler(timeExpressionType);
List<Long> triggerTimeList = new ArrayList<>(NEXT_N_TIMES);
Long nextTriggerTime = System.currentTimeMillis();
do {
nextTriggerTime = timingStrategyHandler.calculateNextTriggerTime(nextTriggerTime, timeExpression, startTime, endTime);
if (nextTriggerTime == null) {
break;
}
triggerTimeList.add(nextTriggerTime);
} while (triggerTimeList.size() < NEXT_N_TIMES);
if (triggerTimeList.isEmpty()) {
return TIPS;
}
return triggerTimeList.stream().map(t -> DateFormatUtils.format(t, OmsConstant.TIME_PATTERN)).collect(Collectors.toList());
}
/**
* 计算下次的调度时间
*
* @param preTriggerTime 上次触发时间(nullable)
* @param timeExpressionType 定时表达式类型
* @param timeExpression 表达式
* @param startTime 起始时间(include)
* @param endTime 结束时间(include)
* @return 下次的调度时间
*/
public Long calculateNextTriggerTime(Long preTriggerTime, TimeExpressionType timeExpressionType, String timeExpression, Long startTime, Long endTime) {
if (preTriggerTime == null || preTriggerTime < System.currentTimeMillis()) {
preTriggerTime = System.currentTimeMillis();
}
return getHandler(timeExpressionType).calculateNextTriggerTime(preTriggerTime, timeExpression, startTime, endTime);
}
/**
* 计算下次的调度时间并检查校验规则
*
* @param timeExpressionType 定时表达式类型
* @param timeExpression 表达式
* @param startTime 起始时间(include)
* @param endTime 结束时间(include)
* @return 下次的调度时间
*/
public Long calculateNextTriggerTimeWithInspection( TimeExpressionType timeExpressionType, String timeExpression, Long startTime, Long endTime) {
Long nextTriggerTime = calculateNextTriggerTime(null, timeExpressionType, timeExpression, startTime, endTime);
if (TimeExpressionType.INSPECT_TYPES.contains(timeExpressionType.getV()) && nextTriggerTime == null) {
throw new PowerJobException("time expression is out of date: " + timeExpression);
}
return nextTriggerTime;
}
public void validate(TimeExpressionType timeExpressionType, String timeExpression, Long startTime, Long endTime) {
if (endTime != null) {
if (endTime <= System.currentTimeMillis()) {
throw new PowerJobException("lifecycle is out of date!");
}
if (startTime != null && startTime > endTime) {
throw new PowerJobException("lifecycle is invalid! start time must earlier then end time.");
}
}
getHandler(timeExpressionType).validate(timeExpression);
}
private TimingStrategyHandler getHandler(TimeExpressionType timeExpressionType) {
TimingStrategyHandler timingStrategyHandler = strategyContainer.get(timeExpressionType);
if (timingStrategyHandler == null) {
throw new PowerJobException("No matching TimingStrategyHandler for this TimeExpressionType:" + timeExpressionType);
}
return timingStrategyHandler;
}
}
TimingStrategyService的构造器遍历timingStrategyHandlers，然后根据其supportType构建
Map<TimeExpressionType, TimingStrategyHandler>；其calculateNextTriggerTimes先根据timeExpressionType获取到对应的TimingStrategyHandler，再循环调用TimingStrategyHandler.calculateNextTriggerTime方法来计算nextTriggerTime，最后返回最近5次的调度时间；calculateNextTriggerTimeWithInspection方法会计算nextTriggerTime并针对CRON及DAILY_TIME_INTERVAL类型的要求其不能为null；validate方法调用的是对应TimingStrategyHandler的validate方法
TimingStrategyHandler接口定义了validate、calculateNextTriggerTime、supportType方法；其支持的TimeExpressionType枚举定义了API、CRON、FIXED_RATE、FIXED_DELAY、WORKFLOW、DAILY_TIME_INTERVAL几种类型，分别对应了ApiTimingStrategyHandler、CronTimingStrategyHandler、FixedRateTimingStrategyHandler、FixedDelayTimingStrategyHandler、WorkflowTimingStrategyHandler、DailyTimeIntervalStrategyHandler；TimingStrategyService则聚合了这些TimingStrategyHandler，对外提供了calculateNextTriggerTimes、calculateNextTriggerTime、calculateNextTriggerTimeWithInspection、validate方法。