Spring 定时任务

EnableScheduling

  • 首先关注的类为启动定时任务的注解@EnableScheduling
  1. @Target(ElementType.TYPE)
  2. @Retention(RetentionPolicy.RUNTIME)
  3. @Import(SchedulingConfiguration.class)
  4. @Documented
  5. public @interface EnableScheduling {
  7. }

SchedulingConfiguration

  • 注册定时任务相关信息
  1. @Configuration
  2. @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  3. public class SchedulingConfiguration {
  5.     /**
  6.      * 开启定时任务
  7.      * @return
  8.      */
  9.     @Bean(name = TaskManagementConfigUtils.SCHEDULED_ANNOTATION_PROCESSOR_BEAN_NAME)
  10.     @Role(BeanDefinition.ROLE_INFRASTRUCTURE)
  11.     public ScheduledAnnotationBeanPostProcessor scheduledAnnotationProcessor() {
  12.         // 注册 ScheduledAnnotationBeanPostProcessor
  13.         return new ScheduledAnnotationBeanPostProcessor();
  14.     }
  16. }

ScheduledAnnotationBeanPostProcessor

  • 关注 application 事件,以及 spring 生命周期相关的接口实现
  1.   /**
  2.      * application 事件
  3.      * @param event the event to respond to
  4.      */
  5.     @Override
  6.     public void onApplicationEvent(ContextRefreshedEvent event) {
  7.         if (event.getApplicationContext() == this.applicationContext) {
  8.             // Running in an ApplicationContext -> register tasks this late...
  9.             // giving other ContextRefreshedEvent listeners a chance to perform
  10.             // their work at the same time (e.g. Spring Batch's job registration).
  11.             // 注册定时任务
  12.             finishRegistration();
  13.         }
  14.     }
  1. @Override
  2.     public Object postProcessAfterInitialization(Object bean, String beanName) {
  3.         if (bean instanceof AopInfrastructureBean || bean instanceof TaskScheduler ||
  4.                 bean instanceof ScheduledExecutorService) {
  5.             // Ignore AOP infrastructure such as scoped proxies.
  6.             return bean;
  7.         }
  9.         // 当前类
  10.         Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);
  11.         if (!this.nonAnnotatedClasses.contains(targetClass)) {
  12.             // 方法扫描,存在 Scheduled、Schedules  注解的全部扫描
  13.             Map<Method, Set<Scheduled>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
  14.                     (MethodIntrospector.MetadataLookup<Set<Scheduled>>) method -> {
  16.                         Set<Scheduled> scheduledMethods = AnnotatedElementUtils.getMergedRepeatableAnnotations(
  17.                                 method, Scheduled.class, Schedules.class);
  18.                         return (!scheduledMethods.isEmpty() ? scheduledMethods : null);
  19.                     });
  20.             if (annotatedMethods.isEmpty()) {
  21.                 this.nonAnnotatedClasses.add(targetClass);
  22.                 if (logger.isTraceEnabled()) {
  23.                     logger.trace("No @Scheduled annotations found on bean class: " + targetClass);
  24.                 }
  25.             }
  26.             else {
  27.                 // Non-empty set of methods
  28.                 annotatedMethods.forEach((method, scheduledMethods) ->
  29.                         // 处理 scheduled 相关信息
  30.                         scheduledMethods.forEach(scheduled -> processScheduled(scheduled, method, bean)));
  31.                 if (logger.isTraceEnabled()) {
  32.                     logger.trace(annotatedMethods.size() + " @Scheduled methods processed on bean '" + beanName +
  33.                             "': " + annotatedMethods);
  34.                 }
  35.             }
  36.         }
  37.         return bean;
  38.     }
  • 处理定时任务注解
  1. protected void processScheduled(Scheduled scheduled, Method method, Object bean) {
  2.         try {
  3.             Runnable runnable = createRunnable(bean, method);
  4.             boolean processedSchedule = false;
  5.             String errorMessage =
  6.                     "Exactly one of the 'cron', 'fixedDelay(String)', or 'fixedRate(String)' attributes is required";
  8.             Set<ScheduledTask> tasks = new LinkedHashSet<>(4);
  10.             // Determine initial delay
  11.             // 是否延迟执行
  12.             long initialDelay = scheduled.initialDelay();
  13.             // 延迟执行时间
  14.             String initialDelayString = scheduled.initialDelayString();
  15.             // 是否有延迟执行的时间
  16.             if (StringUtils.hasText(initialDelayString)) {
  17.                 Assert.isTrue(initialDelay < 0, "Specify 'initialDelay' or 'initialDelayString', not both");
  18.                 if (this.embeddedValueResolver != null) {
  19.                     initialDelayString = this.embeddedValueResolver.resolveStringValue(initialDelayString);
  20.                 }
  21.                 if (StringUtils.hasLength(initialDelayString)) {
  22.                     try {
  23.                         initialDelay = parseDelayAsLong(initialDelayString);
  24.                     }
  25.                     catch (RuntimeException ex) {
  26.                         throw new IllegalArgumentException(
  27.                                 "Invalid initialDelayString value \"" + initialDelayString + "\" - cannot parse into long");
  28.                     }
  29.                 }
  30.             }
  32.             // Check cron expression
  33.             // 获取cron表达式
  34.             String cron = scheduled.cron();
  35.             // cron表达式是否存在
  36.             if (StringUtils.hasText(cron)) {
  37.                 // 获取时区
  38.                 String zone = scheduled.zone();
  39.                 if (this.embeddedValueResolver != null) {
  40.                     // 字符串转换
  41.                     cron = this.embeddedValueResolver.resolveStringValue(cron);
  42.                     zone = this.embeddedValueResolver.resolveStringValue(zone);
  43.                 }
  44.                 if (StringUtils.hasLength(cron)) {
  45.                     // cron 是否延迟
  46.                     Assert.isTrue(initialDelay == -1, "'initialDelay' not supported for cron triggers");
  47.                     processedSchedule = true;
  48.                     if (!Scheduled.CRON_DISABLED.equals(cron)) {
  49.                         TimeZone timeZone;
  50.                         if (StringUtils.hasText(zone)) {
  51.                             // 时区解析
  52.                             timeZone = StringUtils.parseTimeZoneString(zone);
  53.                         }
  54.                         else {
  55.                             // 默认时区获取
  56.                             timeZone = TimeZone.getDefault();
  57.                         }
  58.                         // 创建任务
  59.                         tasks.add(this.registrar.scheduleCronTask(new CronTask(runnable, new CronTrigger(cron, timeZone))));
  60.                     }
  61.                 }
  62.             }
  64.             // At this point we don't need to differentiate between initial delay set or not anymore
  65.             if (initialDelay < 0) {
  66.                 initialDelay = 0;
  67.             }
  69.             // Check fixed delay
  70.             // 获取间隔调用时间
  71.             long fixedDelay = scheduled.fixedDelay();
  72.             // 间隔时间>0
  73.             if (fixedDelay >= 0) {
  74.                 Assert.isTrue(!processedSchedule, errorMessage);
  75.                 processedSchedule = true;
  76.                 // 创建任务,间隔时间定时任务
  77.                 tasks.add(this.registrar.scheduleFixedDelayTask(new FixedDelayTask(runnable, fixedDelay, initialDelay)));
  78.             }
  79.             // 延迟时间
  80.             String fixedDelayString = scheduled.fixedDelayString();
  81.             if (StringUtils.hasText(fixedDelayString)) {
  82.                 if (this.embeddedValueResolver != null) {
  83.                     fixedDelayString = this.embeddedValueResolver.resolveStringValue(fixedDelayString);
  84.                 }
  85.                 if (StringUtils.hasLength(fixedDelayString)) {
  86.                     Assert.isTrue(!processedSchedule, errorMessage);
  87.                     processedSchedule = true;
  88.                     try {
  89.                         fixedDelay = parseDelayAsLong(fixedDelayString);
  90.                     }
  91.                     catch (RuntimeException ex) {
  92.                         throw new IllegalArgumentException(
  93.                                 "Invalid fixedDelayString value \"" + fixedDelayString + "\" - cannot parse into long");
  94.                     }
  95.                     // 创建延迟时间任务
  96.                     tasks.add(this.registrar.scheduleFixedDelayTask(new FixedDelayTask(runnable, fixedDelay, initialDelay)));
  97.                 }
  98.             }
  100.             // Check fixed rate
  101.             // 获取调用频率
  102.             long fixedRate = scheduled.fixedRate();
  103.             if (fixedRate >= 0) {
  104.                 Assert.isTrue(!processedSchedule, errorMessage);
  105.                 processedSchedule = true;
  106.                 // 创建调用频率的定时任务
  107.                 tasks.add(this.registrar.scheduleFixedRateTask(new FixedRateTask(runnable, fixedRate, initialDelay)));
  108.             }
  109.             String fixedRateString = scheduled.fixedRateString();
  110.             if (StringUtils.hasText(fixedRateString)) {
  111.                 if (this.embeddedValueResolver != null) {
  112.                     fixedRateString = this.embeddedValueResolver.resolveStringValue(fixedRateString);
  113.                 }
  114.                 if (StringUtils.hasLength(fixedRateString)) {
  115.                     Assert.isTrue(!processedSchedule, errorMessage);
  116.                     processedSchedule = true;
  117.                     try {
  118.                         fixedRate = parseDelayAsLong(fixedRateString);
  119.                     }
  120.                     catch (RuntimeException ex) {
  121.                         throw new IllegalArgumentException(
  122.                                 "Invalid fixedRateString value \"" + fixedRateString + "\" - cannot parse into long");
  123.                     }
  124.                     tasks.add(this.registrar.scheduleFixedRateTask(new FixedRateTask(runnable, fixedRate, initialDelay)));
  125.                 }
  126.             }
  128.             // Check whether we had any attribute set
  129.             Assert.isTrue(processedSchedule, errorMessage);
  131.             // Finally register the scheduled tasks
  132.             synchronized (this.scheduledTasks) {
  133.                 // 定时任务注册
  134.                 Set<ScheduledTask> regTasks = this.scheduledTasks.computeIfAbsent(bean, key -> new LinkedHashSet<>(4));
  135.                 regTasks.addAll(tasks);
  136.             }
  137.         }
  138.         catch (IllegalArgumentException ex) {
  139.             throw new IllegalStateException(
  140.                     "Encountered invalid @Scheduled method '" + method.getName() + "': " + ex.getMessage());
  141.         }
  142.     }

定时任务

  • CronTask
    • cron 定时任务
  • FixedDelayTask
    • 间隔时间的定时任务
  • FixedRateTask
    • 调用频率的定时任务
  • ScheduledTask
    • 定时任务对象

cron 表达式解析

  • org.springframework.scheduling.support.CronSequenceGenerator.doParse
  1.     private void doParse(String[] fields) {
  2.         setNumberHits(this.seconds, fields[0], 0, 60);
  3.         setNumberHits(this.minutes, fields[1], 0, 60);
  4.         setNumberHits(this.hours, fields[2], 0, 24);
  5.         setDaysOfMonth(this.daysOfMonth, fields[3]);
  6.         setMonths(this.months, fields[4]);
  7.         setDays(this.daysOfWeek, replaceOrdinals(fields[5], "SUN,MON,TUE,WED,THU,FRI,SAT"), 8);
  9.         if (this.daysOfWeek.get(7)) {
  10.             // Sunday can be represented as 0 or 7
  11.             this.daysOfWeek.set(0);
  12.             this.daysOfWeek.clear(7);
  13.         }
  14.     }

执行定时任务

  • 这里以 CronTask 任务进行分析,其他定时任务同理
    • org.springframework.scheduling.config.ScheduledTaskRegistrar.scheduleCronTask
  1.     @Nullable
  2.     public ScheduledTask scheduleCronTask(CronTask task) {
  3.         // 从未执行的任务列表中删除,并且获取这个任务
  4.         ScheduledTask scheduledTask = this.unresolvedTasks.remove(task);
  5.         boolean newTask = false;
  6.         // 没有这个任务
  7.         if (scheduledTask == null) {
  8.             scheduledTask = new ScheduledTask(task);
  9.             newTask = true;
  10.         }
  11.         // 任务调度器是否为空
  12.         if (this.taskScheduler != null) {
  14.             scheduledTask.future = this.taskScheduler.schedule(task.getRunnable(), task.getTrigger());
  15.         }
  16.         else {
  17.             // 添加到cron任务列表
  18.             addCronTask(task);
  19.             // 保存到没有执行的任务中
  20.             this.unresolvedTasks.put(task, scheduledTask);
  21.         }
  22.         return (newTask ? scheduledTask : null);
  23.     }