Spring boot定时任务的原理及动态创建详解

@Target(ElementType.TYPE)@Retention(RetentionPolicy.RUNTIME)@Import(SchedulingConfiguration.class)@Documentedpublic @interface EnableScheduling {}@Configuration@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(); }}

@Overridepublic Object postProcessAfterInitialization(Object bean, String beanName) { if (bean instanceof AopInfrastructureBean || bean instanceof TaskScheduler ||  bean instanceof ScheduledExecutorService) { // Ignore AOP infrastructure such as scoped proxies. return bean; } Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean); if (!this.nonAnnotatedClasses.contains(targetClass)) { 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: " + targetClass);  } } else {  // Non-empty set of methods  annotatedMethods.forEach((method, scheduledMethods) ->   scheduledMethods.forEach(scheduled -> processScheduled(scheduled, method, bean)));  if (logger.isTraceEnabled()) {  logger.trace(annotatedMethods.size() + " @Scheduled methods processed on bean '" + beanName +   "': " + annotatedMethods);  } } } return bean;}

private void finishRegistration() { if (this.scheduler != null) { this.registrar.setScheduler(this.scheduler); } if (this.beanFactory instanceof ListableBeanFactory) { Map<String, SchedulingConfigurer> beans =  ((ListableBeanFactory) this.beanFactory).getBeansOfType(SchedulingConfigurer.class); List<SchedulingConfigurer> configurers = new ArrayList<>(beans.values()); AnnotationAwareOrderComparator.sort(configurers); for (SchedulingConfigurer configurer : configurers) {  configurer.configureTasks(this.registrar); } } if (this.registrar.hasTasks() && this.registrar.getScheduler() == null) { Assert.state(this.beanFactory != null, "BeanFactory must be set to find scheduler by type"); try {  // Search for TaskScheduler bean...  this.registrar.setTaskScheduler(resolveSchedulerBean(this.beanFactory, TaskScheduler.class, false)); } catch (NoUniqueBeanDefinitionException ex) {  logger.trace("Could not find unique TaskScheduler bean", ex);  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) {  logger.trace("Could not find default TaskScheduler bean", ex);  // Search for ScheduledExecutorService bean next...  try {  this.registrar.setScheduler(resolveSchedulerBean(this.beanFactory, ScheduledExecutorService.class, false));  }  catch (NoUniqueBeanDefinitionException ex2) {  logger.trace("Could not find unique ScheduledExecutorService bean", ex2);  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) {  logger.trace("Could not find default ScheduledExecutorService bean", ex2);  // Giving up -> falling back to default scheduler within the registrar...  logger.info("No TaskScheduler/ScheduledExecutorService bean found for scheduled processing");  } } } this.registrar.afterPropertiesSet();}

protected void scheduleTasks() { if (this.taskScheduler == null) { this.localExecutor = Executors.newSingleThreadScheduledExecutor(); this.taskScheduler = new ConcurrentTaskScheduler(this.localExecutor); } if (this.triggerTasks != null) { for (TriggerTask task : this.triggerTasks) {  addScheduledTask(scheduleTriggerTask(task)); } } if (this.cronTasks != null) { for (CronTask task : this.cronTasks) {  addScheduledTask(scheduleCronTask(task)); } } if (this.fixedRateTasks != null) { for (IntervalTask task : this.fixedRateTasks) {  addScheduledTask(scheduleFixedRateTask(task)); } } if (this.fixedDelayTasks != null) { for (IntervalTask task : this.fixedDelayTasks) {  addScheduledTask(scheduleFixedDelayTask(task)); } }}

ScheduledTaskRegistrar.java@Nullablepublic ScheduledTask scheduleCronTask(CronTask task) { ScheduledTask 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 { addCronTask(task); this.unresolvedTasks.put(task, scheduledTask); } return (newTask ? scheduledTask : null);}ConcurrentTaskScheduler.java@Override@Nullablepublic ScheduledFuture<?> schedule(Runnable task, Trigger trigger) { try { if (this.enterpriseConcurrentScheduler) {  return new EnterpriseConcurrentTriggerScheduler().schedule(decorateTask(task, true), trigger); } else {  ErrorHandler errorHandler =   (this.errorHandler != null ? this.errorHandler : TaskUtils.getDefaultErrorHandler(true));  return new ReschedulingRunnable(task, trigger, this.scheduledExecutor, errorHandler).schedule(); } } catch (RejectedExecutionException ex) { throw new TaskRejectedException("Executor [" + this.scheduledExecutor + "] did not accept task: " + task, ex); }}ReschedulingRunnable.java@Nullablepublic 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; }}private ScheduledFuture<?> obtainCurrentFuture() { Assert.state(this.currentFuture != null, "No scheduled future"); return this.currentFuture;}@Overridepublic void run() { Date actualExecutionTime = new Date(); super.run(); Date completionTime = new Date(); synchronized (this.triggerContextMonitor) { Assert.state(this.scheduledExecutionTime != null, "No scheduled execution"); this.triggerContext.update(this.scheduledExecutionTime, actualExecutionTime, completionTime); if (!obtainCurrentFuture().isCancelled()) {  schedule(); } }}

ScheduledThreadPoolExecutor.javapublic ScheduledFuture<?> schedule(Runnable command,     long delay,     TimeUnit unit) { if (command == null || unit == null) throw new NullPointerException(); RunnableScheduledFuture<?> t = decorateTask(command, new ScheduledFutureTask<Void>(command, null,     triggerTime(delay, unit))); delayedExecute(t); return t;}private void delayedExecute(RunnableScheduledFuture<?> task) { if (isShutdown()) reject(task); else { super.getQueue().add(task); if (isShutdown() &&  !canRunInCurrentRunState(task.isPeriodic()) &&  remove(task))  task.cancel(false); else  ensurePrestart(); }}

/** * Creates with given first task and thread from ThreadFactory. * @param firstTask the first task (null if none) */Worker(Runnable firstTask) { setState(-1); // inhibit interrupts until runWorker this.firstTask = firstTask; this.thread = getThreadFactory().newThread(this);}/** Delegates main run loop to outer runWorker */public void run() { runWorker(this);}

@Configuration@EnableScheduling@Role(BeanDefinition.ROLE_INFRASTRUCTURE)public class TaskConfiguration { @Bean(name = ScheduledAnnotationBeanPostProcessor.DEFAULT_TASK_SCHEDULER_BEAN_NAME) @Role(BeanDefinition.ROLE_INFRASTRUCTURE) public ScheduledExecutorService scheduledAnnotationProcessor() { return Executors.newScheduledThreadPool(5, new DefaultThreadFactory()); } private static class DefaultThreadFactory implements ThreadFactory { private static final AtomicInteger poolNumber = new AtomicInteger(1); private final ThreadGroup group; private final AtomicInteger threadNumber = new AtomicInteger(1); private final String namePrefix; DefaultThreadFactory() {  SecurityManager s = System.getSecurityManager();  group = (s != null) ? s.getThreadGroup() :   Thread.currentThread().getThreadGroup();  namePrefix = "pool-" +   poolNumber.getAndIncrement() +   "-schedule-"; } @Override public Thread newThread(Runnable r) {  Thread t = new Thread(group, r,   namePrefix + threadNumber.getAndIncrement(),   0);  if (t.isDaemon()) {  t.setDaemon(false);  }  if (t.getPriority() != Thread.NORM_PRIORITY) {  t.setPriority(Thread.NORM_PRIORITY);  }  return t; } }}

@Configurationpublic class DynamicTask implements SchedulingConfigurer { private static Logger LOGGER = LoggerFactory.getLogger(DynamicTask.class); private static final ExecutorService es = new ThreadPoolExecutor(10, 20,   0L, TimeUnit.MILLISECONDS,   new LinkedBlockingQueue<>(10),   new DynamicTaskConsumeThreadFactory()); private volatile ScheduledTaskRegistrar registrar; private final ConcurrentHashMap<String, ScheduledFuture<?>> scheduledFutures = new ConcurrentHashMap<>(); private final ConcurrentHashMap<String, CronTask> cronTasks = new ConcurrentHashMap<>(); private volatile List<TaskConstant> taskConstants = Lists.newArrayList(); @Override public void configureTasks(ScheduledTaskRegistrar registrar) {  this.registrar = registrar;  this.registrar.addTriggerTask(() -> {     if (!CollectionUtils.isEmpty(taskConstants)) {      LOGGER.info("检测动态定时任务列表...");      List<TimingTask> tts = new ArrayList<>();      taskConstants        .forEach(taskConstant -> {         TimingTask tt = new TimingTask();         tt.setExpression(taskConstant.getCron());         tt.setTaskId("dynamic-task-" + taskConstant.getTaskId());         tts.add(tt);        });      this.refreshTasks(tts);     }    }    , triggerContext -> new PeriodicTrigger(5L, TimeUnit.SECONDS).nextExecutionTime(triggerContext)); } public List<TaskConstant> getTaskConstants() {  return taskConstants; } private void refreshTasks(List<TimingTask> tasks) {  //取消已经删除的策略任务  Set<String> taskIds = scheduledFutures.keySet();  for (String taskId : taskIds) {   if (!exists(tasks, taskId)) {    scheduledFutures.get(taskId).cancel(false);   }  }  for (TimingTask tt : tasks) {   String expression = tt.getExpression();   if (StringUtils.isBlank(expression) || !CronSequenceGenerator.isValidExpression(expression)) {    LOGGER.error("定时任务DynamicTask cron表达式不合法: " + expression);    continue;   }   //如果配置一致，则不需要重新创建定时任务   if (scheduledFutures.containsKey(tt.getTaskId())     && cronTasks.get(tt.getTaskId()).getExpression().equals(expression)) {    continue;   }   //如果策略执行时间发生了变化，则取消当前策略的任务   if (scheduledFutures.containsKey(tt.getTaskId())) {    scheduledFutures.remove(tt.getTaskId()).cancel(false);    cronTasks.remove(tt.getTaskId());   }   CronTask task = new CronTask(tt, expression);   ScheduledFuture<?> future = registrar.getScheduler().schedule(task.getRunnable(), task.getTrigger());   cronTasks.put(tt.getTaskId(), task);   scheduledFutures.put(tt.getTaskId(), future);  } } private boolean exists(List<TimingTask> tasks, String taskId) {  for (TimingTask task : tasks) {   if (task.getTaskId().equals(taskId)) {    return true;   }  }  return false; } @PreDestroy public void destroy() {  this.registrar.destroy(); } public static class TaskConstant {  private String cron;  private String taskId;  public String getCron() {   return cron;  }  public void setCron(String cron) {   this.cron = cron;  }  public String getTaskId() {   return taskId;  }  public void setTaskId(String taskId) {   this.taskId = taskId;  } } private class TimingTask implements Runnable {  private String expression;  private String taskId;  public String getTaskId() {   return taskId;  }  public void setTaskId(String taskId) {   this.taskId = taskId;  }  @Override  public void run() {   //设置队列大小10   LOGGER.error("当前CronTask: " + this);   DynamicBlockingQueue queue = new DynamicBlockingQueue(3);   es.submit(() -> {    while (!queue.isDone() || !queue.isEmpty()) {     try {      String content = queue.poll(500, TimeUnit.MILLISECONDS);      if (StringUtils.isBlank(content)) {       return;      }      LOGGER.info("DynamicBlockingQueue 消费：" + content);      TimeUnit.MILLISECONDS.sleep(500);     } catch (InterruptedException e) {      e.printStackTrace();     }    }   });   //队列放入数据   for (int i = 0; i < 5; ++i) {    try {     queue.put(String.valueOf(i));     LOGGER.info("DynamicBlockingQueue 生产：" + i);    } catch (InterruptedException e) {     e.printStackTrace();    }   }   queue.setDone(true);  }  public String getExpression() {   return expression;  }  public void setExpression(String expression) {   this.expression = expression;  }  @Override  public String toString() {   return ReflectionToStringBuilder.toString(this     , ToStringStyle.JSON_STYLE     , false     , false     , TimingTask.class);  } } /**  * 队列消费线程工厂类  */ private static class DynamicTaskConsumeThreadFactory implements ThreadFactory {  private static final AtomicInteger poolNumber = new AtomicInteger(1);  private final ThreadGroup group;  private final AtomicInteger threadNumber = new AtomicInteger(1);  private final String namePrefix;  DynamicTaskConsumeThreadFactory() {   SecurityManager s = System.getSecurityManager();   group = (s != null) ? s.getThreadGroup() :     Thread.currentThread().getThreadGroup();   namePrefix = "pool-" +     poolNumber.getAndIncrement() +     "-dynamic-task-";  }  @Override  public Thread newThread(Runnable r) {   Thread t = new Thread(group, r,     namePrefix + threadNumber.getAndIncrement(),     0);   if (t.isDaemon()) {    t.setDaemon(false);   }   if (t.getPriority() != Thread.NORM_PRIORITY) {    t.setPriority(Thread.NORM_PRIORITY);   }   return t;  } } private static class DynamicBlockingQueue extends LinkedBlockingQueue<String> {  DynamicBlockingQueue(int capacity) {   super(capacity);  }  private volatile boolean done = false;  public boolean isDone() {   return done;  }  public void setDone(boolean done) {   this.done = done;  } }}

CronTask task = new CronTask(tt, expression);ScheduledFuture<?> future = registrar.getScheduler().schedule(task.getRunnable(), task.getTrigger());cronTasks.put(tt.getTaskId(), task);scheduledFutures.put(tt.getTaskId(), future);

@RunWith(SpringRunner.class)@SpringBootTestpublic class DynamicTaskTest { @Autowired private DynamicTask dynamicTask; @Test public void test() throws InterruptedException {  List<DynamicTask.TaskConstant> taskConstans = dynamicTask.getTaskConstants();  DynamicTask.TaskConstant taskConstant = new DynamicTask.TaskConstant();  taskConstant.setCron("0/5 * * * * ?");  taskConstant.setTaskId("test1");  taskConstans.add(taskConstant);  DynamicTask.TaskConstant taskConstant1 = new DynamicTask.TaskConstant();  taskConstant1.setCron("0/5 * * * * ?");  taskConstant1.setTaskId("test2");  taskConstans.add(taskConstant1);  DynamicTask.TaskConstant taskConstant2 = new DynamicTask.TaskConstant();  taskConstant2.setCron("0/5 * * * * ?");  taskConstant2.setTaskId("test3");  taskConstans.add(taskConstant2);  TimeUnit.SECONDS.sleep(40);  //移除并添加新的配置  taskConstans.remove(taskConstans.size() - 1);  DynamicTask.TaskConstant taskConstant3 = new DynamicTask.TaskConstant();  taskConstant3.setCron("0/5 * * * * ?");  taskConstant3.setTaskId("test4");  taskConstans.add(taskConstant3);//  TimeUnit.MINUTES.sleep(50); }}