开发踩坑记录之二：谨慎使用Spring中的@Scheduled注解

@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Repeatable(Schedules.class)
public @interface Scheduled {
  /**
   * A cron-like expression, extending the usual UN*X definition to include
   * triggers on the second as well as minute, hour, day of month, month
   * and day of week.  e.g. {@code "0 * * * * MON-FRI"} means once per minute on
   * weekdays (at the top of the minute - the 0th second).
   * @return an expression that can be parsed to a cron schedule
   * @see org.springframework.scheduling.support.CronSequenceGenerator
   */
  String cron() default "";
  /**
   * A time zone for which the cron expression will be resolved. By default, this
   * attribute is the empty String (i.e. the server's local time zone will be used).
   * @return a zone id accepted by {@link java.util.TimeZone#getTimeZone(String)},
   * or an empty String to indicate the server's default time zone
   * @since 4.0
   * @see org.springframework.scheduling.support.CronTrigger#CronTrigger(String, java.util.TimeZone)
   * @see java.util.TimeZone
   */
  String zone() default "";
  /**
   * Execute the annotated method with a fixed period in milliseconds between the
   * end of the last invocation and the start of the next.
   * @return the delay in milliseconds
   */
  long fixedDelay() default -1;
  /**
   * Execute the annotated method with a fixed period in milliseconds between the
   * end of the last invocation and the start of the next.
   * @return the delay in milliseconds as a String value, e.g. a placeholder
   * @since 3.2.2
   */
  String fixedDelayString() default "";
  /**
   * Execute the annotated method with a fixed period in milliseconds between
   * invocations.
   * @return the period in milliseconds
   */
  long fixedRate() default -1;
  /**
   * Execute the annotated method with a fixed period in milliseconds between
   * invocations.
   * @return the period in milliseconds as a String value, e.g. a placeholder
   * @since 3.2.2
   */
  String fixedRateString() default "";
  /**
   * Number of milliseconds to delay before the first execution of a
   * {@link #fixedRate()} or {@link #fixedDelay()} task.
   * @return the initial delay in milliseconds
   * @since 3.2
   */
  long initialDelay() default -1;
  /**
   * Number of milliseconds to delay before the first execution of a
   * {@link #fixedRate()} or {@link #fixedDelay()} task.
   * @return the initial delay in milliseconds as a String value, e.g. a placeholder
   * @since 3.2.2
   */
  String initialDelayString() default "";
}

public class ScheduledAnnotationBeanPostProcessor
    implements MergedBeanDefinitionPostProcessor, DestructionAwareBeanPostProcessor,
    Ordered, EmbeddedValueResolverAware, BeanNameAware, BeanFactoryAware, ApplicationContextAware,
    SmartInitializingSingleton, ApplicationListener<ContextRefreshedEvent>, DisposableBean {
    ...
    }

  @Override
  public Object postProcessAfterInitialization(final Object bean, String beanName) {
    Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);
    if (!this.nonAnnotatedClasses.contains(targetClass)) {
      //获取含有@Scheduled注解的方法
      Map<Method, Set<Scheduled>> annotatedMethods = MethodIntrospector.selectMethods(targetClass,
          (MethodIntrospector.MetadataLookup<Set<Scheduled>>) method -> {
            Set<Scheduled> scheduledMethods = AnnotatedElementUtils.getMergedRepeatableAnnotations(
                method, Scheduled.class, Schedules.class);
            return (!scheduledMethods.isEmpty() ? scheduledMethods : null);
          });
      if (annotatedMethods.isEmpty()) {
        this.nonAnnotatedClasses.add(targetClass);
        if (logger.isTraceEnabled()) {
          logger.trace("No @Scheduled annotations found on bean class: " + bean.getClass());
        }
      }
      else {
        // 循环处理包含@Scheduled注解的方法
        annotatedMethods.forEach((method, scheduledMethods) ->
            scheduledMethods.forEach(scheduled -> processScheduled(scheduled, method, bean)));
        if (logger.isDebugEnabled()) {
          logger.debug(annotatedMethods.size() + " @Scheduled methods processed on bean '" + beanName +
              "': " + annotatedMethods);
        }
      }
    }
    return bean;
  }

protected void processScheduled(Scheduled scheduled, Method method, Object bean) {
    try {
      Assert.isTrue(method.getParameterCount() == 0,
          "Only no-arg methods may be annotated with @Scheduled");
      //获取调用的方法
      Method invocableMethod = AopUtils.selectInvocableMethod(method, bean.getClass());
      //处理线程
      Runnable runnable = new ScheduledMethodRunnable(bean, invocableMethod);
      boolean processedSchedule = false;
      String errorMessage =
          "Exactly one of the 'cron', 'fixedDelay(String)', or 'fixedRate(String)' attributes is required";
      Set<ScheduledTask> tasks = new LinkedHashSet<>(4);
      // Determine initial delay
      long initialDelay = scheduled.initialDelay();
      String initialDelayString = scheduled.initialDelayString();
      if (StringUtils.hasText(initialDelayString)) {
        Assert.isTrue(initialDelay < 0, "Specify 'initialDelay' or 'initialDelayString', not both");
        if (this.embeddedValueResolver != null) {
          initialDelayString = this.embeddedValueResolver.resolveStringValue(initialDelayString);
        }
        if (StringUtils.hasLength(initialDelayString)) {
          try {
            initialDelay = parseDelayAsLong(initialDelayString);
          }
          catch (RuntimeException ex) {
            throw new IllegalArgumentException(
                "Invalid initialDelayString value \"" + initialDelayString + "\" - cannot parse into long");
          }
        }
      }
      // 获取cron参数
      String cron = scheduled.cron();
      if (StringUtils.hasText(cron)) {
        String zone = scheduled.zone();
        if (this.embeddedValueResolver != null) {
          cron = this.embeddedValueResolver.resolveStringValue(cron);
          zone = this.embeddedValueResolver.resolveStringValue(zone);
        }
        if (StringUtils.hasLength(cron)) {
          Assert.isTrue(initialDelay == -1, "'initialDelay' not supported for cron triggers");
          processedSchedule = true;
          TimeZone timeZone;
          if (StringUtils.hasText(zone)) {
            timeZone = StringUtils.parseTimeZoneString(zone);
          }
          else {
            timeZone = TimeZone.getDefault();
          }
          //加入到定时任务列表中
          tasks.add(this.registrar.scheduleCronTask(new CronTask(runnable, new CronTrigger(cron, timeZone))));
        }
      }
      // At this point we don't need to differentiate between initial delay set or not anymore
      if (initialDelay < 0) {
        initialDelay = 0;
      }
      // Check fixed delay
      long fixedDelay = scheduled.fixedDelay();
      if (fixedDelay >= 0) {
        Assert.isTrue(!processedSchedule, errorMessage);
        processedSchedule = true;
        tasks.add(this.registrar.scheduleFixedDelayTask(new FixedDelayTask(runnable, fixedDelay, initialDelay)));
      }
      String fixedDelayString = scheduled.fixedDelayString();
      if (StringUtils.hasText(fixedDelayString)) {
        if (this.embeddedValueResolver != null) {
          fixedDelayString = this.embeddedValueResolver.resolveStringValue(fixedDelayString);
        }
        if (StringUtils.hasLength(fixedDelayString)) {
          Assert.isTrue(!processedSchedule, errorMessage);
          processedSchedule = true;
          try {
            fixedDelay = parseDelayAsLong(fixedDelayString);
          }
          catch (RuntimeException ex) {
            throw new IllegalArgumentException(
                "Invalid fixedDelayString value \"" + fixedDelayString + "\" - cannot parse into long");
          }
          tasks.add(this.registrar.scheduleFixedDelayTask(new FixedDelayTask(runnable, fixedDelay, initialDelay)));
        }
      }
      // 执行频率的类型为long
      long fixedRate = scheduled.fixedRate();
      if (fixedRate >= 0) {
        Assert.isTrue(!processedSchedule, errorMessage);
        processedSchedule = true;
        tasks.add(this.registrar.scheduleFixedRateTask(new FixedRateTask(runnable, fixedRate, initialDelay)));
      }
      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 {
            fixedRate = parseDelayAsLong(fixedRateString);
          }
          catch (RuntimeException ex) {
            throw new IllegalArgumentException(
                "Invalid fixedRateString value \"" + fixedRateString + "\" - cannot parse into long");
          }
          tasks.add(this.registrar.scheduleFixedRateTask(new FixedRateTask(runnable, fixedRate, initialDelay)));
        }
      }
      // Check whether we had any attribute set
      Assert.isTrue(processedSchedule, errorMessage);
      // Finally register the scheduled tasks
      synchronized (this.scheduledTasks) {
        Set<ScheduledTask> registeredTasks = this.scheduledTasks.get(bean);
        if (registeredTasks == null) {
          registeredTasks = new LinkedHashSet<>(4);
          this.scheduledTasks.put(bean, registeredTasks);
        }
        registeredTasks.addAll(tasks);
      }
    }
    catch (IllegalArgumentException ex) {
      throw new IllegalStateException(
          "Encountered invalid @Scheduled method '" + method.getName() + "': " + ex.getMessage());
    }
  }

public class ScheduledTaskRegistrar implements InitializingBean, DisposableBean {
  private TaskScheduler taskScheduler;
  private ScheduledExecutorService localExecutor;
  private List<TriggerTask> triggerTasks;
  private List<CronTask> cronTasks;
  private List<IntervalTask> fixedRateTasks;
  private List<IntervalTask> fixedDelayTasks;
  private final Map<Task, ScheduledTask> unresolvedTasks = new HashMap<Task, ScheduledTask>(16);
  private final Set<ScheduledTask> scheduledTasks = new LinkedHashSet<ScheduledTask>(16);
  ......
}

public ScheduledTask scheduleCronTask(CronTask task) {
    ScheduledTask scheduledTask = this.unresolvedTasks.remove(task);
    boolean newTask = false;
    if (scheduledTask == null) {
      scheduledTask = new ScheduledTask();
      newTask = true;
    }
    if (this.taskScheduler != null) {
      scheduledTask.future = this.taskScheduler.schedule(task.getRunnable(), task.getTrigger());
    }
    else {
      addCronTask(task);
      this.unresolvedTasks.put(task, scheduledTask);
    }
    return (newTask ? scheduledTask : null);
  }

@Override
  protected ExecutorService initializeExecutor(
      ThreadFactory threadFactory, RejectedExecutionHandler rejectedExecutionHandler) {
    this.scheduledExecutor = createExecutor(this.poolSize, threadFactory, rejectedExecutionHandler);
    if (this.removeOnCancelPolicy) {
      if (this.scheduledExecutor instanceof ScheduledThreadPoolExecutor) {
        ((ScheduledThreadPoolExecutor) this.scheduledExecutor).setRemoveOnCancelPolicy(true);
      }
      else {
        logger.info("Could not apply remove-on-cancel policy - not a Java 7+ ScheduledThreadPoolExecutor");
      }
    }
    return this.scheduledExecutor;
  }

public ScheduledFuture<?> schedule() {
    synchronized (this.triggerContextMonitor) {
      this.scheduledExecutionTime = this.trigger.nextExecutionTime(this.triggerContext);
      if (this.scheduledExecutionTime == null) {
        return null;
      }
      //获取时间差
      long initialDelay = this.scheduledExecutionTime.getTime() - System.currentTimeMillis();
      this.currentFuture = this.executor.schedule(this, initialDelay, TimeUnit.MILLISECONDS);
      return this;
    }
  }