定时/延时任务-万字解析Spring定时任务原理

1. 概要

上一篇文章：定时/延时任务-Spring定时任务的两种实现方式。这篇文章就来看下 Spring 中 @Scheduled 和接口方式的定时任务是如何实现的。

2. @EnableScheduling 注解

Spring 中如果需要使用定时任务，就需要引入 @EnableScheduling，我们看下这个注解是怎么定义的。

@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Import(SchedulingConfiguration.class)
@Documented
public @interface EnableScheduling {}

在这个注解中关键就是：@Import(SchedulingConfiguration.class)，这里面通过 Import 引入了 SchedulingConfiguration 配置类。Import 是 Spring 中定义的一种引入配置类的方式，通过 Import 注解可以把对应的类交给 Spring 管理，达到动态开关配置的目的。然后我们再来看下引入的这个注解配置类。

@Configuration(proxyBeanMethods = false)
@Role(BeanDefinition.ROLE_INFRASTRUCTURE)
public class SchedulingConfiguration {@Bean(name = TaskManagementConfigUtils.SCHEDULED_ANNOTATION_PROCESSOR_BEAN_NAME)@Role(BeanDefinition.ROLE_INFRASTRUCTURE)public ScheduledAnnotationBeanPostProcessor scheduledAnnotationProcessor() {return new ScheduledAnnotationBeanPostProcessor();}}

可以看到这个配置类代码比较简单，就是注册了一个 ScheduledAnnotationBeanPostProcessor 后置处理器。后置处理器是 Spring 中用于在 bean 初始化之后调用来处理 bean 的方法。对于 @Scheduled 和 @Schedules 注解解析的核心逻辑就在 postProcessAfterInitialization 中。但是在这之前，我们看下 @Scheduled 注解的属性。

3. @Scheduled 注解

@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Repeatable(Schedules.class)
public @interface Scheduled {/*** 是否禁用 cron 表达式，如果设置成 '-' 就表示不使用 cron 表达式*/String CRON_DISABLED = ScheduledTaskRegistrar.CRON_DISABLED;/*** cron 表达式* @return*/String cron() default "";/*** 时区* @return*/String zone() default "";/*** 固定延时任务的延时时间* @return*/long fixedDelay() default -1;/*** 固定延时任务的延时时间字符串，可动态配置* @return*/String fixedDelayString() default "";/*** 固定速率任务的延时时间* @return*/long fixedRate() default -1;/*** 固定速率任务的延时时间字符串，可动态配置* @return*/String fixedRateString() default "";/*** 第一次执行任务之前延迟多少秒* @return*/long initialDelay() default -1;/*** 第一次执行任务之前延迟多少秒，字符串，可动态配置* @return*/String initialDelayString() default "";/*** 时间单位，默认是毫秒* @return*/TimeUnit timeUnit() default TimeUnit.MILLISECONDS;}

@Scheduled 定义了几种定时任务的实现方式

1. cron 表达式任务
2. 固定延时任务 fixedDelay
3. 固定速率任务 fixedRate

然后我们再看下 postProcessAfterInitialization 是如何处理上面这几种方法的。

4. postProcessAfterInitialization 解析

/*** 初始化之后回回调后置处理器处理定时任务* @param bean the new bean instance* @param beanName the name of the bean* @return*/
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) {if (bean instanceof AopInfrastructureBean || bean instanceof TaskScheduler ||bean instanceof ScheduledExecutorService) {// 注册的 bean 是上面类型的就不处理// Ignore AOP infrastructure such as scoped proxies.return bean;}// 获取实际要处理的 bean 的类型Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);// 1.如果这个目标类还没有被处理过// 2.这个类能不能被 Scheduled 或者 Schedules 注解处理（如果这个类是以 java. 开头或者是 Ordered 类，就不可以）if (!this.nonAnnotatedClasses.contains(targetClass) &&AnnotationUtils.isCandidateClass(targetClass, Arrays.asList(Scheduled.class, Schedules.class))) {// 从目标类中获取有 Scheduled 或者 Schedules 注解的方法Map<Method, Set<Scheduled>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,(MethodIntrospector.MetadataLookup<Set<Scheduled>>) method -> {Set<Scheduled> scheduledAnnotations = AnnotatedElementUtils.getMergedRepeatableAnnotations(method, Scheduled.class, Schedules.class);return (!scheduledAnnotations.isEmpty() ? scheduledAnnotations : null);});// 如果找不到方法if (annotatedMethods.isEmpty()) {// 加入 nonAnnotatedClasses 集合中this.nonAnnotatedClasses.add(targetClass);if (logger.isTraceEnabled()) {logger.trace("No @Scheduled annotations found on bean class: " + targetClass);}}else {// 找到了包含特定注解的方法annotatedMethods.forEach((method, scheduledAnnotations) ->// 遍历来处理所有的方法scheduledAnnotations.forEach(scheduled -> processScheduled(scheduled, method, bean)));if (logger.isTraceEnabled()) {logger.trace(annotatedMethods.size() + " @Scheduled methods processed on bean '" + beanName +"': " + annotatedMethods);}}}return bean;
}

当 ScheduledAnnotationBeanPostProcessor 初始化完成之后，调用 postProcessAfterInitialization 来处理相关的注解，下面来看下具体逻辑。

if (bean instanceof AopInfrastructureBean || bean instanceof TaskScheduler ||bean instanceof ScheduledExecutorService) {// 注册的 bean 是上面类型的就不处理// Ignore AOP infrastructure such as scoped proxies.return bean;
}

如果 bean 的类型是 上面三个类型的，就不处理。

// 获取实际要处理的 bean 的类型
Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);
// 1.如果这个目标类还没有被处理过
// 2.这个类能不能被 Scheduled 或者 Schedules 注解处理（如果这个类是以 java. 开头或者是 Ordered 类，就不可以）
if (!this.nonAnnotatedClasses.contains(targetClass) &&AnnotationUtils.isCandidateClass(targetClass, Arrays.asList(Scheduled.class, Schedules.class))) {// 从目标类中获取有 Scheduled 或者 Schedules 注解的方法Map<Method, Set<Scheduled>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,(MethodIntrospector.MetadataLookup<Set<Scheduled>>) method -> {Set<Scheduled> scheduledAnnotations = AnnotatedElementUtils.getMergedRepeatableAnnotations(method, Scheduled.class, Schedules.class);return (!scheduledAnnotations.isEmpty() ? scheduledAnnotations : null);});// 如果找不到方法if (annotatedMethods.isEmpty()) {// 加入 nonAnnotatedClasses 集合中this.nonAnnotatedClasses.add(targetClass);if (logger.isTraceEnabled()) {logger.trace("No @Scheduled annotations found on bean class: " + targetClass);}}else {// 找到了包含特定注解的方法annotatedMethods.forEach((method, scheduledAnnotations) ->// 遍历来处理所有的方法scheduledAnnotations.forEach(scheduled -> processScheduled(scheduled, method, bean)));if (logger.isTraceEnabled()) {logger.trace(annotatedMethods.size() + " @Scheduled methods processed on bean '" + beanName +"': " + annotatedMethods);}}
}
return bean;

然后获取下 bean 的类型，并且判断下这个类型能不能被 Scheduled 或者 Schedules 注解处理，如果这个类是以 java. 开头或者是 Ordered 类，就不可以。最后遍历这些标记了上面两个注解的方法，一个一个处理，处理的逻辑是 processScheduled，看下 processScheduled 的逻辑

protected void processScheduled(Scheduled scheduled, Method method, Object bean) {// 根据调用的对象和调用的方法创建一个任务Runnable runnable = createRunnable(bean, method);boolean processedSchedule = false;String errorMessage ="Exactly one of the 'cron', 'fixedDelay(String)', or 'fixedRate(String)' attributes is required";Set<ScheduledTask> tasks = new LinkedHashSet<>(4);...
}

这个方法中首先会创建一个任务，然后设置几个属性。

// 把 @Scheduled 注解的 initialDelay 属性转化成毫秒，initialDelay 是指延时多少秒进行第一次执行
long initialDelay = convertToMillis(scheduled.initialDelay(), scheduled.timeUnit());
// @Scheduled 注解的 initialDelayString 属性，作用和上面的 initialDelay 作用一样
String initialDelayString = scheduled.initialDelayString();
if (StringUtils.hasText(initialDelayString)) {// 如果指定了 initialDelayString，那么就不能指定 initialDelay 了// 同时指定 initialDelay 会报错Assert.isTrue(initialDelay < 0, "Specify 'initialDelay' or 'initialDelayString', not both");if (this.embeddedValueResolver != null) {// 因为 @Scheduled 注解的几个 String 类型的值都可以通过配置文件引入，也就是 ${} 的方式// 这个方法就是去解析动态配置值的initialDelayString = this.embeddedValueResolver.resolveStringValue(initialDelayString);}// 如果配置了if (StringUtils.hasLength(initialDelayString)) {try {// 转换成毫秒单位initialDelay = convertToMillis(initialDelayString, scheduled.timeUnit());}catch (RuntimeException ex) {throw new IllegalArgumentException("Invalid initialDelayString value \"" + initialDelayString + "\" - cannot parse into long");}}
}

上面就是解析 @Scheduled 中 initialDelay 和 initialDelayString 的逻辑。

1. 首先把 initialDelay 属性转化成毫秒
2. 然后再解析 initialDelayString，需要注意的是 initialDelay 和 initialDelayString 只能选一个，如果两个都填就会报错
3. this.embeddedValueResolver.resolveStringValue 是解析动态配置的逻辑，因为 initialDelayString 可以使用 ${} 动态配置
4. 最后都转化成 initialDelay 毫秒

上面解析的 initialDelay 和 initialDelayString 表示延时多少 ms 再执行第一次任务。下面再来看下 cron 表达式的解析，这是第一种定时任务的配置方式。

// 获取 cron 表达式
String cron = scheduled.cron();
// 如果设置了 cron 表达式
if (StringUtils.hasText(cron)) {// 获取指定的时区String zone = scheduled.zone();// 同时也是去解析 cron 表达式和时区的动态值if (this.embeddedValueResolver != null) {cron = this.embeddedValueResolver.resolveStringValue(cron);zone = this.embeddedValueResolver.resolveStringValue(zone);}// 如果设置了 cron 表达式if (StringUtils.hasLength(cron)) {// 如果设置了 cron 表达式，那么就不能设置 initialDelay 了Assert.isTrue(initialDelay == -1, "'initialDelay' not supported for cron triggers");// 设置下标记，表示已经设置 cron 表达式了processedSchedule = true;// 如果 cron 表达式没有被设置成 '-' 了，就代表用 cron 去触发if (!Scheduled.CRON_DISABLED.equals(cron)) {// 创建 cron 触发器CronTrigger trigger;if (StringUtils.hasText(zone)) {// 设置时区和 cron 表达式trigger = new CronTrigger(cron, StringUtils.parseTimeZoneString(zone));}else {// 不需要设置时区，设置 cron 表达式trigger = new CronTrigger(cron);}// 将创建的 ScheduledTask 加入任务集合中// 使用 ScheduledTaskRegistrar 创建一个 ScheduledTask，同时需要传入触发器，这个触发器里面需要传入 cron 表达式和时区tasks.add(this.registrar.scheduleCronTask(new CronTask(runnable, trigger)));}}
}

1. 解析 cron 表达式，先获取下时区
2. 然后解析 cron 表达式和时区的动态值
3. 如果设置了 cron 表达式，那么就不能设置 initialDelay 了，需要依靠 cron 表达式来执行定时任务
4. processedSchedule 标记位是代表有没有设置了定时任务的调度方式，这里如果使用 cron 表达式来调度任务，processedSchedule 就会设置为 true
5. 检查下 cron 有没有被设置成 -，这个标记代表禁用 cron 表达式，如果没有设置为 -，就通过 cron 和时区创建一个 CronTrigger，这是 cron 触发器，在这个触发器里面可以获取表达式和时区等信息，同时也可以获取 cron 下一次执行的时间
6. 新建一个 ScheduledTask，把这个任务添加到任务集合中，这个是任务的统一包装，里面可以对 CronTask、FixedDelayTask、FixedRateTask 进行包装

上面就是解析 cron 表达式的逻辑，下面继续看解析@Scheduled 注解的 fixedDelay 字段逻辑，这个字段就是固定延时任务。

// 下面检查 @Scheduled 注解的 fixedDelay 字段，这个字段就是固定延时任务
long fixedDelay = convertToMillis(scheduled.fixedDelay(), scheduled.timeUnit());
if (fixedDelay >= 0) {// 如果前面 cron 已经设置了，那么这里就不能设置 fixedDelay 了Assert.isTrue(!processedSchedule, errorMessage);// 设置标记processedSchedule = true;// 同样添加任务，不过这里是创建一个 FixedDelayTask，表示固定延时的任务tasks.add(this.registrar.scheduleFixedDelayTask(new FixedDelayTask(runnable, fixedDelay, initialDelay)));
}
// 看下有没有设置 fixedDelayString
String fixedDelayString = scheduled.fixedDelayString();
// 如果设置 fixedDelayString 了
if (StringUtils.hasText(fixedDelayString)) {if (this.embeddedValueResolver != null) {// 解析动态字符串fixedDelayString = this.embeddedValueResolver.resolveStringValue(fixedDelayString);}if (StringUtils.hasLength(fixedDelayString)) {// 如果前面没有设置 cron 和 fixDelayAssert.isTrue(!processedSchedule, errorMessage);// 设置标记，表示已解析processedSchedule = true;try {// 转化成毫秒fixedDelay = convertToMillis(fixedDelayString, scheduled.timeUnit());}catch (RuntimeException ex) {throw new IllegalArgumentException("Invalid fixedDelayString value \"" + fixedDelayString + "\" - cannot parse into long");}// 同样添加任务，不过这里是创建一个 FixedDelayTask，表示固定延时的任务tasks.add(this.registrar.scheduleFixedDelayTask(new FixedDelayTask(runnable, fixedDelay, initialDelay)));}
}

1. 首先检测 @Scheduled 注解的 fixedDelay，转化成毫秒
   • 判断下 processedSchedule 字段，如果前面已经设置 cron了，那么这里就会抛出异常，也就是说如果设置了 cron 表达式的调度方式，这里就不能设置 fixedDelay 了
   • 创建一个 ScheduledTask 添加到集合里面，这个任务包装类包装的是 FixedDelayTask，里面设置了初始延时时间和固定速率
2. 然后继续检测 @Scheduled 注解的 fixedDelayString
   • 解析动态字符串
   • 判断下 processedSchedule 字段，如果前面已经设置 cron 或者设置了 fixedDelay 了，那么这里就会抛出异常，也就是说如果设置了 cron 表达式的调度方式或者设置了 fixedDelay 字段，这里就不能设置 fixedDelayString 了
   • 创建一个 ScheduledTask 添加到集合里面，这个任务包装类包装的是 FixedDelayTask，里面设置了初始延时时间和固定速率

上面就是解析固定速率的逻辑，下面继续看解析@Scheduled 注解的 fixedRate 字段逻辑，这个字段就是固定速率任务。

// 下面检查 @Scheduled 注解的 fixedRate 字段，这个字段就是固定速率任务
long fixedRate = convertToMillis(scheduled.fixedRate(), scheduled.timeUnit());
if (fixedRate >= 0) {// 如果设置了 fixedRate，同时前面几种方式都没有设置Assert.isTrue(!processedSchedule, errorMessage);// 设置标记processedSchedule = true;// 同样添加任务，不过这里是创建一个 FixedRateTask，表示固定延时的任务tasks.add(this.registrar.scheduleFixedRateTask(new FixedRateTask(runnable, fixedRate, initialDelay)));
}
// 继续检测 fixedRateString
String fixedRateString = scheduled.fixedRateString();
if (StringUtils.hasText(fixedRateString)) {if (this.embeddedValueResolver != null) {// 解析动态表达式fixedRateString = this.embeddedValueResolver.resolveStringValue(fixedRateString);}if (StringUtils.hasLength(fixedRateString)) {// 如果前面几种方式已经设置过了，这里就抛出异常Assert.isTrue(!processedSchedule, errorMessage);// 设置标记位processedSchedule = true;try {// 把 fixedRateString 转化成毫秒fixedRate = convertToMillis(fixedRateString, scheduled.timeUnit());}catch (RuntimeException ex) {throw new IllegalArgumentException("Invalid fixedRateString value \"" + fixedRateString + "\" - cannot parse into long");}// 同样添加任务，不过这里是创建一个 FixedRateTask，表示固定延时的任务tasks.add(this.registrar.scheduleFixedRateTask(new FixedRateTask(runnable, fixedRate, initialDelay)));}
}

1. 首先检测 @Scheduled 注解的 fixedRate，转化成毫秒
   • 判断下 processedSchedule 字段，如果前面已经设置 cron了，或者已经设置固定延时的方式了，那么这里就会抛出异常
   • 创建一个 ScheduledTask 添加到集合里面，这个任务包装类包装的是 FixedRateTask，里面设置了初始延时时间和固定速率
2. 然后继续检测 @Scheduled 注解的 fixedRateString
   • 解析动态字符串
   • 判断下 processedSchedule 字段，如果前面已经设置 cron 或者设置了固定延时或者 fixedRate 字段了，那么这里就会抛出异常fixedRate 字段，这里就不能设置 fixedRateString了
   • 创建一个 ScheduledTask 添加到集合里面，这个任务包装类包装的是 FixedRateTask，里面设置了初始延时时间和固定速率

最后把定时任务注册到 scheduledTasks 中，这个属性是一个 map 任务集合，表示 bean -> Set<Task> 的映射。

private final Map<Object, Set<ScheduledTask>> scheduledTasks = new IdentityHashMap<>(16);

到这里任务就已经被解析完并添加到 scheduledTasks 集合中了，当 bean 注销的时候会从这个集合里面拿出任务来一个一个取消，这个后面再看。

4.1 createRunnable

这个方法一开始就创建了一个任务，我们这里就简单看下创建任务的逻辑。

/*** 根据传入的对象和方法来创建一个任务* @param target* @param method* @return*/
protected Runnable createRunnable(Object target, Method method) {// 如果方法设置了参数，就抛出异常Assert.isTrue(method.getParameterCount() == 0, "Only no-arg methods may be annotated with @Scheduled");// 选择注解可调用的的同名方法，这个方法不能是 private 或者是 static，又或者是代理类型的Method invocableMethod = AopUtils.selectInvocableMethod(method, target.getClass());// 创建 ScheduledMethodRunnablereturn new ScheduledMethodRunnable(target, invocableMethod);
}

其实逻辑不难，就是从对象里面获取同名方法，这些方法不能是 private 或者是 static，同时这个对象的类不能是代理类。

5. 任务 Task 和子类

上面就是 postProcessAfterInitialization 的逻辑，但是有一些方法我们还没有细看，比如 this.registrar.scheduleCronTask，而这个方法只是把任务添加到集合中，那么这些方法是在哪被调用的呢？既然不是在 ScheduledAnnotationBeanPostProcessor 调度的，是在哪调度的呢？同时在上一篇文章通过接口动态设置 cron 表达式的时候我们不是通过 addTriggerTask 添加了定时任务吗，那么这个方法是在哪里设置任务的， 带着这些问题，下面我们先看下要调度的任务的结构。
不管是上面添加的 CronTask、FixedDelayTask 和 FixedRateTask，都是子类的实现，但是这些类的顶层结构式怎么样的呢？

1. 首先顶层结构是 Task
2. 下面实现类是 TriggerTask 和 IntervalTask
3. 最后就是 CronTask、FixedDelayTask 和 FixedRateTask，这三个就是我们设置的 cron、固定延时和固定速率任务

下面来看下里面的源码结构，首先是 Task：

public class Task {private final Runnable runnable;public Task(Runnable runnable) {Assert.notNull(runnable, "Runnable must not be null");this.runnable = runnable;}public Runnable getRunnable() {return this.runnable;}@Overridepublic String toString() {return this.runnable.toString();}}

顶层 Task 里面封装了具体的任务 Runnable

public class TriggerTask extends Task {private final Trigger trigger;public TriggerTask(Runnable runnable, Trigger trigger) {super(runnable);Assert.notNull(trigger, "Trigger must not be null");this.trigger = trigger;}public Trigger getTrigger() {return this.trigger;}}public class IntervalTask extends Task {private final long interval;private final long initialDelay;public IntervalTask(Runnable runnable, long interval, long initialDelay) {super(runnable);this.interval = interval;this.initialDelay = initialDelay;}public IntervalTask(Runnable runnable, long interval) {this(runnable, interval, 0);}public long getInterval() {return this.interval;}public long getInitialDelay() {return this.initialDelay;}}

• TriggerTask 触发器任务是对 cron 任务的封装，里面创建了一个 Trigger 变量，这个 trigger 里面可以获取 cron 的下一次执行的时间。
• IntervalTask 周期任务是固定速率任务和固定延时任务的父类，这个类里面需要设置 initialDelay（初始延时） 和 interval（周期）。

最后来看下底层的三个实现类：

public class CronTask extends TriggerTask {// cron 表达式private final String expression;public CronTask(Runnable runnable, String expression) {this(runnable, new CronTrigger(expression));}public CronTask(Runnable runnable, CronTrigger cronTrigger) {super(runnable, cronTrigger);this.expression = cronTrigger.getExpression();}/*** Return the cron expression defining when the task should be executed.*/public String getExpression() {return this.expression;}}public class FixedRateTask extends IntervalTask {public FixedRateTask(Runnable runnable, long interval, long initialDelay) {super(runnable, interval, initialDelay);}}public class FixedDelayTask extends IntervalTask {public FixedDelayTask(Runnable runnable, long interval, long initialDelay) {super(runnable, interval, initialDelay);}}

其实类里面的结构不复杂，直接看代码就能看明白了。

6. ScheduledTaskRegistrar

6.1 添加任务的逻辑

在 ScheduledAnnotationBeanPostProcessor 初始化完之后，被 @Scheduled 注解标注的方法会被封装成 Task 存到集合中，那么这些方法是什么时候被调度的，就在这个类里面会解答。在看里面的方法之前，还是先把目光 ScheduledAnnotationBeanPostProcessor 里面，processScheduled 方法中，不管是注册 cron、fixRate、fixDelay 类型的任务，都会通过 registrar 里面的方法去创建。



这个 registrar 其实就是 ScheduledTaskRegistrar，那么我们就跟进这几个方法去看看是怎么处理任务的。

/**
* 调度 cron 表达式任务的逻辑* @param task* @return*/
@Nullable
public ScheduledTask scheduleCronTask(CronTask task) {// 从未解析任务集合中移除任务ScheduledTask scheduledTask = this.unresolvedTasks.remove(task);boolean newTask = false;// 如果不存在 ScheduledTask，就创建一个if (scheduledTask == null) {// ScheduledTask 是对 Task 任务的包装scheduledTask = new ScheduledTask(task);// 表示新建了一个 ScheduledTask 任务newTask = true;}// 如果调度器不为空if (this.taskScheduler != null) {// 调度器调度任务scheduledTask.future = this.taskScheduler.schedule(task.getRunnable(), task.getTrigger());}else {// 此时调度器还没有初始化，先把任务存起来// 1.添加到 cronTasks 中// 2.添加到 unresolvedTasks 中addCronTask(task);this.unresolvedTasks.put(task, scheduledTask);}return (newTask ? scheduledTask : null);
}public void addCronTask(CronTask task) {if (this.cronTasks == null) {this.cronTasks = new ArrayList<>();}this.cronTasks.add(task);
}

首先是 scheduleCronTask 方法，这个方法会传入一个 cron 表达式任务，流程是这样的：

1. 先从 unresolvedTasks 集合中获取这个任务对应的 ScheduledTask，Spring 里面的定时任务都是统一用 ScheduledTask 来调度的，ScheduledTask 里面会封装 CronTask、FixedRateTask、FixedDelayTask
2. 如果不存在对应的 scheduleCronTask，就创建一个
3. 如果还没有初始化调度器，说明定时任务还没能调度任务，这时候先把任务加入到 unresolvedTasks 和 cronTasks 集合中
4. 如果初始化调度器了，就通过调度器去调度 cron 任务
5. 最后返回任务，如果是新建的就返回新建的任务

这个就是调度 CronTask 的流程，可以看到在里面如果调度器没有初始化的时候是不会调度任务的，只会把任务添加到集合中，等调度器 taskScheduler 初始化之后再调度任务。再看下其他两个调度方法。

/*** 调度固定延时任务* @param task* @return*/
@Nullable
public ScheduledTask scheduleFixedDelayTask(FixedDelayTask task) {// 从未解析任务集合中移除任务ScheduledTask scheduledTask = this.unresolvedTasks.remove(task);boolean newTask = false;if (scheduledTask == null) {// 如果不存在就创建一个scheduledTask = new ScheduledTask(task);newTask = true;}// 如果设置调度器了if (this.taskScheduler != null) {if (task.getInitialDelay() > 0) {// 启动时间是当前时间 + 延时时间Date startTime = new Date(this.taskScheduler.getClock().millis() + task.getInitialDelay());// 通过调度器去调度固定延时任务scheduledTask.future =this.taskScheduler.scheduleWithFixedDelay(task.getRunnable(), startTime, task.getInterval());}else {// 如果没有设置初始延时时间，直接执行固定延时任务scheduledTask.future =this.taskScheduler.scheduleWithFixedDelay(task.getRunnable(), task.getInterval());}}else {// 如果调度器还没有初始化，就先把任务存起来addFixedDelayTask(task);this.unresolvedTasks.put(task, scheduledTask);}return (newTask ? scheduledTask : null);
}/*** Schedule the specified fixed-rate task, either right away if possible* or on initialization of the scheduler.* @return a handle to the scheduled task, allowing to cancel it* (or {@code null} if processing a previously registered task)* @since 5.0.2*/
@Nullable
public ScheduledTask scheduleFixedRateTask(FixedRateTask task) {// 从未调度任务中获取需要调度的固定速率任务ScheduledTask scheduledTask = this.unresolvedTasks.remove(task);boolean newTask = false;if (scheduledTask == null) {// 如果获取不到就创建一个scheduledTask = new ScheduledTask(task);newTask = true;}// 使用调度器去调度任务if (this.taskScheduler != null) {if (task.getInitialDelay() > 0) {// 第一次执行的时间 = 当前时间 + @Scheduled 的 initialDelayDate startTime = new Date(this.taskScheduler.getClock().millis() + task.getInitialDelay());scheduledTask.future =// 进行调度this.taskScheduler.scheduleAtFixedRate(task.getRunnable(), startTime, task.getInterval());}else {scheduledTask.future =this.taskScheduler.scheduleAtFixedRate(task.getRunnable(), task.getInterval());}}else {// 调度器还没有设置，先存到没有调度集合中addFixedRateTask(task);this.unresolvedTasks.put(task, scheduledTask);}return (newTask ? scheduledTask : null);
}

这两个方法跟上面的 scheduleCronTask 逻辑是一样的。

6.2 调度器初始化

上面就是在 ScheduledTaskRegistrar 中的调度任务的逻辑，只是由于调度器还没有初始化，所以任务还不能被调度。那么调度器是在哪被初始化的呢？

public class ScheduledTaskRegistrar implements ScheduledTaskHolder, InitializingBean, DisposableBean {...
}

可以看到 ScheduledTaskRegistrar 实现了 InitializingBean，在初始化之后会调用 afterPropertiesSet

/*** Calls {@link #scheduleTasks()} at bean construction time.* 初始化之后调用*/
@Override
public void afterPropertiesSet() {scheduleTasks();
}/**
* Schedule all registered tasks against the underlying* {@linkplain #setTaskScheduler(TaskScheduler) task scheduler}.*/
@SuppressWarnings("deprecation")
protected void scheduleTasks() {// 如果我们没有自己设置 taskSchedulerif (this.taskScheduler == null) {// 单线程的线程池this.localExecutor = Executors.newSingleThreadScheduledExecutor();// 这里默认就创建一个 ConcurrentTaskScheduler，使用单线程的线程池，其实这个线程池类型就是 ScheduledExecutorServicethis.taskScheduler = new ConcurrentTaskScheduler(this.localExecutor);}// 把需要调度的任务添加到 scheduleTasks 中if (this.triggerTasks != null) {for (TriggerTask task : this.triggerTasks) {addScheduledTask(scheduleTriggerTask(task));}}// 把需要调度的 cron 任务添加到 scheduleTasks 中if (this.cronTasks != null) {for (CronTask task : this.cronTasks) {addScheduledTask(scheduleCronTask(task));}}// 把需要调度的固定速率任务添加到 scheduleTasks 中if (this.fixedRateTasks != null) {for (IntervalTask task : this.fixedRateTasks) {addScheduledTask(scheduleFixedRateTask(task));}}// 把需要调度的固定延时任务添加到 scheduleTasks 中if (this.fixedDelayTasks != null) {for (IntervalTask task : this.fixedDelayTasks) {addScheduledTask(scheduleFixedDelayTask(task));}}
}

这个方法会调度所有的任务，可以看到，如果我们没有设置调度器 TaskScheduler，就设置一个单线程的 ScheduledExecutorService 作为任务执行的线程池。所以 @Scheduled 的核心调度是通过 ScheduledExecutorService 来调度的。
还记得上面 processScheduled 方法吗，在这个方法中

1. CronTask 被添加到 ScheduledTaskRegistrar 的 cronTasks 集合中
2. FixedDelayTask 被添加到 ScheduledTaskRegistrar 的 fixedDelayTasks 集合中
3. FixedRateTask 被添加到 ScheduledTaskRegistrar 的 fixedRateTasks 集合中

在上面的 scheduleTasks 方法中，调度的任务顺序是：

1. triggerTasks
2. cronTasks
3. fixedRateTasks
4. fixedDelayTasks

triggerTasks 是接口实现类的扩展点，上一篇文章中我们通过接口实现动态配置定时任务的时候，就是通过 addTriggerTask 方法把任务添加到 triggerTasks 集合里面。

public abstract class ScheduledConfig implements SchedulingConfigurer {// 定时任务周期表达式private String cron;@Overridepublic void configureTasks(ScheduledTaskRegistrar taskRegistrar) {// 设置线程池,可开启多线程taskRegistrar.setScheduler(taskExecutor());taskRegistrar.addTriggerTask(// 执行定时任务() -> {taskService();},triggerContext -> {// 这里就是动态获取任务队列的逻辑cron = getCron();if(cron == null){throw new RuntimeException("cron not exist");}// 重新获取cron表达式CronTrigger trigger = new CronTrigger(cron);return trigger.nextExecutionTime(triggerContext);});}private Executor taskExecutor() {return BeanUtils.getBean(ThreadPoolTaskScheduler.class);}public abstract void taskService();public abstract String getCron();public abstract int getOpen();
}

所以我们通过 addTriggerTask 添加的任务会在这里被调度，先看调度的逻辑。

@Nullable
public ScheduledTask scheduleTriggerTask(TriggerTask task) {// 从未解析任务集合中获取 ScheduledTaskScheduledTask scheduledTask = this.unresolvedTasks.remove(task);boolean newTask = false;if (scheduledTask == null) {// 如果不存在就创建一个scheduledTask = new ScheduledTask(task);newTask = true;}// 这里调度器已经初始化了if (this.taskScheduler != null) {// 通过调度器去调度任务scheduledTask.future = this.taskScheduler.schedule(task.getRunnable(), task.getTrigger());}else {// 否则就把任务添加到集合中addTriggerTask(task);this.unresolvedTasks.put(task, scheduledTask);}return (newTask ? scheduledTask : null);
}

这个方法我们上面已经看过了，基本是一样的逻辑，只不过在这里 taskScheduler 已经被初始化了，所以这里会通过 ScheduledExecutorService 去调度任务。scheduleCronTask、scheduleFixedRateTask、scheduleFixedDelayTask 的逻辑上面已经说了，这里就不多说了。

6.3 调用时机

上面我们说了 ScheduledTaskRegistrar 的初始化逻辑，那么你有没有想过，ScheduledAnnotationBeanPostProcessor 是怎么和 ScheduledTaskRegistrar 联系起来的，要知道如果 ScheduledTaskRegistrar 是初始化之后调用的 afterPropertiesSet，那跟 ScheduledAnnotationBeanPostProcessor 的初始化就控制不了先后顺序，因为肯定要 ScheduledAnnotationBeanPostProcessor 先初始化解析任务，然后再通过 ScheduledTaskRegistrar 去调度的。并且 ScheduledAnnotationBeanPostProcessor 里面并没有通过 spring 的方式引入 ScheduledTaskRegistrar。
其实 ScheduledAnnotationBeanPostProcessor 里面的 registrar 和 Spring 初始化的 ScheduledTaskRegistrar 不是同一个。当 ScheduledAnnotationBeanPostProcessor 被创建出来时，构造器中就会通过 new 的方式创建出 ScheduledTaskRegistrar。

public ScheduledAnnotationBeanPostProcessor() {this.registrar = new ScheduledTaskRegistrar();}

好了，不卖关子，当 ScheduledAnnotationBeanPostProcessor 创建完成之后，Spring 上下文启动之后会回调 onApplicationEvent 方法。

@Override
public void onApplicationEvent(ContextRefreshedEvent event) {if (event.getApplicationContext() == this.applicationContext) {finishRegistration();}
}

在这个 onApplicationEvent 中会调用 finishRegistration，下面就看下这个方法的逻辑。

private void finishRegistration() {// 如果设置了调度器if (this.scheduler != null) {// 把调度器设置到 registrar 里面this.registrar.setScheduler(this.scheduler);}// BeanFactory 一般都会是这个类型if (this.beanFactory instanceof ListableBeanFactory) {// 获取所有 SchedulingConfigurer 类型的 beanMap<String, SchedulingConfigurer> beans =((ListableBeanFactory) this.beanFactory).getBeansOfType(SchedulingConfigurer.class);List<SchedulingConfigurer> configurers = new ArrayList<>(beans.values());// 进行排序AnnotationAwareOrderComparator.sort(configurers);// 然后遍历这些实现类for (SchedulingConfigurer configurer : configurers) {// 调用 configureTasks 方法configurer.configureTasks(this.registrar);}}// 如果实现类里面添加了任务并且没有设置 taskSchedulerif (this.registrar.hasTasks() && this.registrar.getScheduler() == null) {Assert.state(this.beanFactory != null, "BeanFactory must be set to find scheduler by type");try {// 从 Spring 里面找到 TaskScheduler 类型的 bean，然后设置到当前调度器上面this.registrar.setTaskScheduler(resolveSchedulerBean(this.beanFactory, TaskScheduler.class, false));}catch (NoUniqueBeanDefinitionException ex) {if (logger.isTraceEnabled()) {logger.trace("Could not find unique TaskScheduler bean - attempting to resolve by name: " +ex.getMessage());}try {this.registrar.setTaskScheduler(resolveSchedulerBean(this.beanFactory, TaskScheduler.class, true));}catch (NoSuchBeanDefinitionException ex2) {if (logger.isInfoEnabled()) {logger.info("More than one TaskScheduler bean exists within the context, and " +"none is named 'taskScheduler'. Mark one of them as primary or name it 'taskScheduler' " +"(possibly as an alias); or implement the SchedulingConfigurer interface and call " +"ScheduledTaskRegistrar#setScheduler explicitly within the configureTasks() callback: " +ex.getBeanNamesFound());}}}catch (NoSuchBeanDefinitionException ex) {if (logger.isTraceEnabled()) {logger.trace("Could not find default TaskScheduler bean - attempting to find ScheduledExecutorService: " +ex.getMessage());}// Search for ScheduledExecutorService bean next...try {this.registrar.setScheduler(resolveSchedulerBean(this.beanFactory, ScheduledExecutorService.class, false));}catch (NoUniqueBeanDefinitionException ex2) {if (logger.isTraceEnabled()) {logger.trace("Could not find unique ScheduledExecutorService bean - attempting to resolve by name: " +ex2.getMessage());}try {this.registrar.setScheduler(resolveSchedulerBean(this.beanFactory, ScheduledExecutorService.class, true));}catch (NoSuchBeanDefinitionException ex3) {if (logger.isInfoEnabled()) {logger.info("More than one ScheduledExecutorService bean exists within the context, and " +"none is named 'taskScheduler'. Mark one of them as primary or name it 'taskScheduler' " +"(possibly as an alias); or implement the SchedulingConfigurer interface and call " +"ScheduledTaskRegistrar#setScheduler explicitly within the configureTasks() callback: " +ex2.getBeanNamesFound());}}}catch (NoSuchBeanDefinitionException ex2) {if (logger.isTraceEnabled()) {logger.trace("Could not find default ScheduledExecutorService bean - falling back to default: " +ex2.getMessage());}// Giving up -> falling back to default scheduler within the registrar...logger.info("No TaskScheduler/ScheduledExecutorService bean found for scheduled processing");}}}// 再调用 afterPropertiesSet 方法this.registrar.afterPropertiesSet();
}

在这个方法中会从 Spring 容器里面获取所有实现了 SchedulingConfigurer 的类，然后调用这些类的 configureTasks 方法，我们设置的 Executor 和 triggerTask 就是在这里添加到调度器中的。最后调用调度器的 afterPropertiesSet 方法，在这个方法里面通过调度器调度任务。
好了，到这里我们就看到了，调度器是如何初始化的，同时也看到了添加的任务是如何被调度的。那么我们知道 ScheduledTaskRegistrar 里面的 taskScheduler 有很多种类型，这些类型有什么不同呢？

7. taskScheduler 类型

我们先来看下这个方法，在我们自己实现的 SchedulingConfigurer 接口类中，就通过 taskRegistrar.setScheduler(taskExecutor()) 添加了一个调度器。上面 6.3 的逻辑中如果没有设置这玩意，就会到 Spring 里面根据类型或者名字去找。如果设置了就不会，就用我们自己设置的。
当我们自己设置了 taskScheduler，ScheduledTaskRegistrar 就不会再创建一个默认的单线程的 ConcurrentTaskScheduler 作为 taskScheduler 去执行任务。

/**
* 设置任务调度器，当我们使用接口动态配置的时候* @param scheduler*/
public void setScheduler(@Nullable Object scheduler) {if (scheduler == null) {this.taskScheduler = null;}else if (scheduler instanceof TaskScheduler) {this.taskScheduler = (TaskScheduler) scheduler;}else if (scheduler instanceof ScheduledExecutorService) {// JDK 的 ScheduledExecutorService 类型，需要封装this.taskScheduler = new ConcurrentTaskScheduler(((ScheduledExecutorService) scheduler));}else {throw new IllegalArgumentException("Unsupported scheduler type: " + scheduler.getClass());}
}

其实说到底 taskScheduler 也就是上面这三种类型。

7.1 ConcurrentTaskScheduler

我们直接看实现类，其实这几个都是差不多的，我们直接看 ConcurrentTaskScheduler 的。因为上面代码中如果 scheduler instanceof ScheduledExecutorService，就会封转成 ConcurrentTaskScheduler 类型，我们先看下构造器。

public ConcurrentTaskScheduler(ScheduledExecutorService scheduledExecutor) {// 父类是 ConcurrentTaskExecutorsuper(scheduledExecutor);// 初始化调度器initScheduledExecutor(scheduledExecutor);
}private void initScheduledExecutor(@Nullable ScheduledExecutorService scheduledExecutor) {if (scheduledExecutor != null) {// 设置调度器this.scheduledExecutor = scheduledExecutor;// ManagedScheduledExecutorService 是 Java 并发实用工具（Concurrency Utilities）为 Java EE 环境提供的一种可管理的调度执行服务// 扩展了 Java SE 的 ScheduledExecutorService，为在 Java EE 环境中提交延迟或周期性任务的执行提供了额外的方法和管理功能this.enterpriseConcurrentScheduler = (managedScheduledExecutorServiceClass != null &&managedScheduledExecutorServiceClass.isInstance(scheduledExecutor));}else {// 创建一个单线程的 ScheduledExecutorServicethis.scheduledExecutor = Executors.newSingleThreadScheduledExecutor();this.enterpriseConcurrentScheduler = false;}
}

这个方法接收了一个 ScheduledExecutorService ，设置到父类里面，然后初始化这个调度器。如果我们没有设置 scheduledExecutor，就会创建一个单线程的 ScheduledExecutorService。
然后再看下核心的调度方法，也就是 schedule 方法：

/**
* 调度任务的方法* @param task the Runnable to execute whenever the trigger fires* @param trigger an implementation of the {@link Trigger} interface,* e.g. a {@link org.springframework.scheduling.support.CronTrigger} object* wrapping a cron expression* @return*/
@Override
@Nullable
public ScheduledFuture<?> schedule(Runnable task, Trigger trigger) {try {// 如果设置了 ManagedScheduledExecutorService，就用 EnterpriseConcurrentTriggerScheduler 去调度，一般也用不上if (this.enterpriseConcurrentScheduler) {return new EnterpriseConcurrentTriggerScheduler().schedule(decorateTask(task, true), trigger);}else {// 使用 ReschedulingRunnable 去调度，这个 errorHandler 是处理异常的ErrorHandler errorHandler =(this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));return new ReschedulingRunnable(task, trigger, this.clock, this.scheduledExecutor, errorHandler).schedule();}}catch (RejectedExecutionException ex) {throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);}
}/*** 调度任务* @param task the Runnable to execute whenever the trigger fires* @param startTime the desired execution time for the task* (if this is in the past, the task will be executed immediately, i.e. as soon as possible)* @return*/
@Override
public ScheduledFuture<?> schedule(Runnable task, Date startTime) {// 获取当前任务还有多久执行long delay = startTime.getTime() - this.clock.millis();try {// 使用 ScheduledExecutorService 去执行延时任务return this.scheduledExecutor.schedule(decorateTask(task, false), delay, TimeUnit.MILLISECONDS);}catch (RejectedExecutionException ex) {throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);}
}@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, Date startTime, long period) {long initialDelay = startTime.getTime() - this.clock.millis();try {return this.scheduledExecutor.scheduleAtFixedRate(decorateTask(task, true), initialDelay, period, TimeUnit.MILLISECONDS);}catch (RejectedExecutionException ex) {throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);}
}@Override
public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, long period) {try {return this.scheduledExecutor.scheduleAtFixedRate(decorateTask(task, true), 0, period, TimeUnit.MILLISECONDS);}catch (RejectedExecutionException ex) {throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex);}
}

这里面 scheduleAtFixedRate 其实就是调用的 ScheduledExecutorService 的方法来调度任务的。而对于 schedule 方法，会通过创建 ReschedulingRunnable 去调度任务。那我们就看下 ReschedulingRunnable 的 schedule 方法。

@Nullable
public ScheduledFuture<?> schedule() {synchronized (this.triggerContextMonitor) {// 计算出下一次的调用时间this.scheduledExecutionTime = this.trigger.nextExecutionTime(this.triggerContext);if (this.scheduledExecutionTime == null) {return null;}// 获取下一次执行时间距离当前的延时long delay = this.scheduledExecutionTime.getTime() - this.triggerContext.getClock().millis();// 调用 ScheduledExecutorService 方法去调度，延时时间是 delaythis.currentFuture = this.executor.schedule(this, delay, TimeUnit.MILLISECONDS);return this;}
}

在这个方法中通过触发器获取下一次任务的执行事件，然后计算出当前时间距离下一次任务执行时间的延时，最后通过 ScheduledExecutorService 去调度任务。然后再来看下这个 ReschedulingRunnable 的 run 方法，因为调度任务其实就是执行 run 方法。

/*** 任务调度的核心逻辑*/
@Override
public void run() {// 获取当前的时间Date actualExecutionTime = new Date(this.triggerContext.getClock().millis());// 执行父类的方法，父类其实就是调用 task.run，也就是调用我们设置的 @Scheduled 方法super.run();// 任务的完成时间Date completionTime = new Date(this.triggerContext.getClock().millis());synchronized (this.triggerContextMonitor) {Assert.state(this.scheduledExecutionTime != null, "No scheduled execution");// 更新调度器上下文，也就是调度时间、实际的调度时间、完成时间this.triggerContext.update(this.scheduledExecutionTime, actualExecutionTime, completionTime);// 如果任务没有取消if (!obtainCurrentFuture().isCancelled()) {// 在里面通过 ScheduledExecutorService 进行方法的调度// 也就是说 spring 启动的时候会先调用一次 run 方法，后续再通过 ScheduledExecutorService 调度任务schedule();}}
}

方法很简单，就是执行具体的任务逻辑，然后更新调度器上下文，把调度时间、实际的调度时间、完成时间都存起来，接着再调用上面的 schedule() 方法，再把任务重新放到 ScheduledExecutorService 里面去执行。

7.2 ThreadPoolTaskScheduler

好了，上面的 ConcurrentTaskScheduler 核心逻辑已经说完了，下面再来看下 ThreadPoolTaskScheduler 的逻辑，因为我们在自定义 SchedulingConfigurer 实现类中设置了ThreadPoolTaskScheduler 作为调度类，所以这里简单看下这个调度类是怎么调度任务的。

public class ThreadPoolTaskScheduler extends ExecutorConfigurationSupportimplements AsyncListenableTaskExecutor, SchedulingTaskExecutor, TaskScheduler {...
}

ThreadPoolTaskScheduler 里面没有提供构造器，就是使用的无参构造器，我们看下自定义 SchedulingConfigurer 实现类中是怎么设置这玩意的。

@Configuration
public class BeanConfig {@Beanpublic ThreadPoolTaskScheduler taskScheduler() {ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();scheduler.setPoolSize(10); // 设置线程池大小scheduler.setThreadNamePrefix("MyTaskScheduler-"); // 设置线程名称前缀scheduler.initialize(); // 初始化调度器return scheduler;}
}private Executor taskExecutor() {return BeanUtils.getBean(ThreadPoolTaskScheduler.class);
}

我们在 BeanConfig 里面注册了类型为 ThreadPoolTaskScheduler 的 bean，然后在 ScheduledConfig 里面通过 BeanUtils.getBean 获取我们注册的 bean。
所以我们就看下 ThreadPoolTaskScheduler 的 父类 ExecutorConfigurationSupport 里面的 afterPropertiesSet()。因为 Spring 创建 ThreadPoolTaskScheduler 的时候会去创建父类 ExecutorConfigurationSupport，而父类 ExecutorConfigurationSupport 实现了 InitializingBean 接口，实现 afterPropertiesSet() 方法。

@Override
public void afterPropertiesSet() {initialize();
}public void initialize() {if (logger.isDebugEnabled()) {logger.debug("Initializing ExecutorService" + (this.beanName != null ? " '" + this.beanName + "'" : ""));}if (!this.threadNamePrefixSet && this.beanName != null) {setThreadNamePrefix(this.beanName + "-");}this.executor = initializeExecutor(this.threadFactory, this.rejectedExecutionHandler);
}

其实 ThreadPoolTaskScheduler 里面调度任务用的就是 ExecutorConfigurationSupport 里面的 executor。我们看下这个 executor 的初始化逻辑。

@Override
protected ExecutorService initializeExecutor(ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {this.scheduledExecutor = createExecutor(this.poolSize, threadFactory, rejectedExecutionHandler);if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor) {ScheduledThreadPoolExecutor scheduledPoolExecutor = (ScheduledThreadPoolExecutor) this.scheduledExecutor;if (this.removeOnCancelPolicy) {scheduledPoolExecutor.setRemoveOnCancelPolicy(true);}if (this.continueExistingPeriodicTasksAfterShutdownPolicy) {scheduledPoolExecutor.setContinueExistingPeriodicTasksAfterShutdownPolicy(true);}if (!this.executeExistingDelayedTasksAfterShutdownPolicy) {scheduledPoolExecutor.setExecuteExistingDelayedTasksAfterShutdownPolicy(false);}}return this.scheduledExecutor;
}protected ScheduledExecutorService createExecutor(int poolSize, ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {return new ScheduledThreadPoolExecutor(poolSize, threadFactory, rejectedExecutionHandler);
}

在初始化 ThreadPoolTaskScheduler 方法中会创建一个 ScheduledExecutorService 来调度任务。然后配置一些参数。

8. 小结

到这里我们终于把 @Scheduled 的原理学习完了，其实这个注解最底层还是用的 ScheduledExecutorService 来调度。

如有错误，欢迎指出！！！