当前位置：首页 > news >正文

@EnableScheduling 和 @Scheduled 实现定时任务的任务延期问题

news 2024/9/27 10:22:06

前言

在复盘 ieg 一面看到定时任务阻塞的问题时，研究了下 @EnableScheduling 的源码，觉得可以单开一篇文章讲一讲

本文主要讲述了使用 @EnableScheduling 可能出现的线程阻塞导致定时任务延期的问题，也顺便解释了动态定时任务源码上的实现

引用文章：

@Schedule定时任务+分布式环境：@Schedule定时任务+分布式环境，这些坑你一定得注意！！！ (qq.com)

java 中的线程池参数：java中四种线程池及poolSize、corePoolSize、maximumPoolSize_maximum-pool-size-CSDN博客

线程池的拒绝策略：线程池的RejectedExecutionHandler（拒绝策略）-CSDN博客

Java 中实现定时任务：Java中实现定时任务，有多少种解决方案？好久没更新博客了，最近上班做了点小东西，总结复盘一下。主要介绍了定时任务的三种 - 掘金 (juejin.cn)

线程阻塞问题

问题根源

Java中使用 Springboot 自带的定时任务 @EnableScheduling 和 @Scheduled 注解，会装配一个 SchedulingConfiguration 的类

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

@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 的 Bean

在这个类的无参构造中又初始化了一个 ScheduledTaskRegistrar 的对象

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

在创建单例或刷新上下文之后，会执行 finishRegistration 方法，最后执行 registrar 的 afterPropertiesSet 方法：

@Override  
public void afterSingletonsInstantiated() {  // Remove resolved singleton classes from cache  this.nonAnnotatedClasses.clear();  if (this.applicationContext == null) {  // Not running in an ApplicationContext -> register tasks early...  finishRegistration();  }  
}  @Override  
public void onApplicationEvent(ContextRefreshedEvent event) {  if (event.getApplicationContext() == this.applicationContext) {  // Running in an ApplicationContext -> register tasks this late...  // giving other ContextRefreshedEvent listeners a chance to perform  // their work at the same time (e.g. Spring Batch's job registration).  finishRegistration();  }  
}  private void finishRegistration() {  if (this.scheduler != null) {  this.registrar.setScheduler(this.scheduler);  }  // ...this.registrar.afterPropertiesSet();  
}

ScheduledTaskRegistrar 的成员变量包括任务的执行器以及几种类型的定时任务列表

@Nullable  
private TaskScheduler taskScheduler;  @Nullable  
private ScheduledExecutorService localExecutor;  @Nullable  
private List<TriggerTask> triggerTasks;  @Nullable  
private List<CronTask> cronTasks;

afterPropertiesSet 方法会获取一个执行器

@Override  
public void afterPropertiesSet() {  scheduleTasks();  
}  /**  
* Schedule all registered tasks against the underlying  
* {@linkplain #setTaskScheduler(TaskScheduler) task scheduler}.  
*/  
@SuppressWarnings("deprecation")  
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));  }  }  
}

进入 newSingleThreadScheduledExecutor 可以看到，默认使用了一个 corePoolSize 为 1, maximumPoolSize 为 Integer.MAX_VALUE 的线程池

public static ScheduledExecutorService newSingleThreadScheduledExecutor() {  return new DelegatedScheduledExecutorService  (new ScheduledThreadPoolExecutor(1));  
}

public ScheduledThreadPoolExecutor(int corePoolSize) {  super(corePoolSize, Integer.MAX_VALUE,  DEFAULT_KEEPALIVE_MILLIS, MILLISECONDS,  new DelayedWorkQueue());  
}

public ThreadPoolExecutor(int corePoolSize,  int maximumPoolSize,  long keepAliveTime,  TimeUnit unit,  BlockingQueue<Runnable> workQueue) {  this(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue,  Executors.defaultThreadFactory(), defaultHandler);  
}

而线程池主要有几个重要的参数分别是：

corePoolSize：线程池的基本大小。
maximumPoolSize：线程池中允许的最大线程数。
poolSize：线程池中当前线程的数量。

当提交一个新任务时，若

poolSize < corePoolSize : 创建新线程处理该任务
poolSize = corePoolSize : 将任务置于阻塞队列中
阻塞队列的容量达到上限，且这时 poolSize < maximumPoolSize :
阻塞队列满了，且 poolSize = maximumPoolSize : 那么线程池已经达到极限，会根据饱和策略 RejectedExecutionHandler 拒绝新的任务，默认是 AbortPolicy 会丢掉任务并抛出异常

解决方案

注入自己编写的线程池，自行设置参数：

@Configuration  public class MyTheadPoolConfig {  @Bean  public TaskExecutor taskExecutor() {  ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();  //设置核心线程数  executor.setCorePoolSize(10);  //设置最大线程数  executor.setMaxPoolSize(20);  //缓冲队列200：用来缓冲执行任务的队列  executor.setQueueCapacity(200);  //线程活路时间 60 秒  executor.setKeepAliveSeconds(60);  //线程池名的前缀：设置好了之后可以方便我们定位处理任务所在的线程池  // 这里我继续沿用 scheduling 默认的线程名前缀  executor.setThreadNamePrefix("nzc-create-scheduling-");  //设置拒绝策略  executor.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());  executor.setWaitForTasksToCompleteOnShutdown(true);  return executor;  }  }

在定时任务的类上再加一个 @EnableAsync 注解，给方法添加一个 @Async 即可

@Slf4j  
@Component  
@EnableAsync  
@EnableScheduling  
public class ScheduleService {  @Autowired  TaskExecutor taskExecutor;  @Async(value = "taskExecutor")  @Scheduled(cron = "0/5 * * * * ? ")  public void testSchedule() {  try {  Thread.sleep(10000);  log.info("当前执行任务的线程号ID===>{}", Thread.currentThread().getId());  } catch (Exception e) {  e.printStackTrace();  }   }  
}

动态定时任务

上面提到了

@EnableScheduling 导入了 SchedulingConfiguration，SchedulingConfiguration 又创建了 ScheduledAnnotationBeanPostProcessor 的Bean，ScheduledAnnotationBeanPostProcessor 又实例化了 ScheduledTaskRegistrar 对象，即

@EnableScheduling -> SchedulingConfiguration -> ScheduledAnnotationBeanPostProcessor -> ScheduledTaskRegistrar

实际上，在 ScheduledAnnotationBeanPostProcessor 的 finishRegistration 方法中，会先获取所有实现了 SchedulingConfigurer 接口的 Bean，并执行他们的 configureTasks 方法


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);  }  }// ...
}

我们可以通过配置一个实现了 SchedulingConfigurer 接口的 Bean，实现动态加载定时任务的执行时间

@Data  
@Slf4j  
@Component  
@RequiredArgsConstructor  
@PropertySource("classpath:task-config.ini")  
public class ScheduleTask implements SchedulingConfigurer {  // private Long timer = 100 * 1000L;  @Value("${printTime.cron}")  private String cron;  @Override  public void configureTasks(ScheduledTaskRegistrar taskRegistrar) {  // 间隔触发的任务  taskRegistrar.addTriggerTask(new Runnable() {  @Override  public void run(){  // ...}}, new Trigger() {  @Override  public Date nextExecutionTime(TriggerContext triggerContext) {  // 使用CronTrigger触发器，可动态修改cron表达式来操作循环规则  CronTrigger cronTrigger = new CronTrigger(cron);  Date nextExecutionTime = cronTrigger.nextExecutionTime(triggerContext);  return nextExecutionTime;  // 使用PerodicTrigger触发器，修改timer变量指定操作间隔，单位为毫秒// PeriodicTrigger periodicTrigger = new PeriodicTrigger(timer);  // Date nextExecutionTime = periodicTrigger.nextExecutionTime(triggerContext);  // return nextExecutionTime;  }  });  }  
}