博客

spring定时器@Scheduled的原理和实现分析

  • Post author:
  • Post category:其他



目录


一 简单介绍


二 为什么要加@EnableScheduling， @Scheduled才生效


三 什么时候什么地方解析@Scheduled


四，思考

一 简单介绍

我们可以通过如下方式使用定时器 

@Component
@EnableScheduling
public class TestSchedule {
    DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
    @Scheduled(cron = "*/1 * * * * *")
    private void testTask() throws InterruptedException {
        Thread.sleep(2000);
        System.out.println("定时任务执行时间：" + Thread.currentThread().getName());
        System.out.println("定时任务执行时间：" + LocalDateTime.now().format(dateTimeFormatter));
    }
}

更多使用详细，请点击

spring 定时器Scheduled和ScheduledExecutorService使用_yegeg的专栏-CSDN博客

思考下


1. 为什么要加@EnableScheduling， @Scheduled才生效


2.什么时候什么地方解析@Scheduled，这些方法被解析放到什么地方


3.解析完@Scheduled注解的方法后，什么时候调用，后面是怎么按照约定的定时规则调用的

二 为什么要加@EnableScheduling， @Scheduled才生效

对于第一个问题，为什么要加@EnableScheduling， @Scheduled才生效

我们查看@EnableScheduling注解到底是什么 

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

}

我们发现了①处有个 @Import(SchedulingConfiguration.class)的注解，该注解属于spring容器一个扩展点，使用这个注解可以把我们自己的某个类交给spring容器来管理和创建，接下来我们继续查看SchedulingConfiguration源代码 

@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();
	}

}

发现这个配置文件中有个用@Bean注解的方法，该注解也会被spring容器去管理自己new出来的对象，到这里 ，相当于把ScheduledAnnotationBeanPostProcessor类通过自己new 出的的对象交给了spring容器管理，我们继续查看ScheduledAnnotationBeanPostProcessor源码

① 从类的实现接口，我们可以看出，ScheduledAnnotationBeanPostProcessor实现了spring ioc容器各种扩展点，在spring容器初始化时候，会调用相应的扩展点，从而实现了代码执行的入口。

② registrar保存了通过@Scheduled注解过的任务，后面提到的

registrar

就是指此处

这就是为什么我们要使用@EnableScheduling后@Scheduled才生效 ，因为使用了@EnableScheduling ，ScheduledAnnotationBeanPostProcessor就会被spring容器管理，这个类也实现了@Scheduled的解析、保存、执行启动等。

三 什么时候什么地方解析@Scheduled

我继续在ScheduledAnnotationBeanPostProcessor找解析@Scheduled的地方

① 我们发现了postProcessAfterInitialization方法 ，它在BeanPostProcessor接口中声明，

这个是springioc容器扩展的回调，会在初始化bean后回调，也就给执行这段代码找到入口。

② 此处就是去找@Scheduled的方法

③ 把找到@Scheduled注解的方法交给processScheduled方法处理，接下来我们看看processScheduled方法的实现 

protected void processScheduled(Scheduled scheduled, Method method, Object bean) {
		try {
            // 把我们的对象和方法封装层Runnable ，定时器就可以直接调用Runnable来执行任务，调用Runnable实际就通过反射来执行我们的对象和方法
			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);

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

			// 如果我们的@Scheduled里面是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;
					if (!Scheduled.CRON_DISABLED.equals(cron)) {
						TimeZone timeZone;
						if (StringUtils.hasText(zone)) {
							timeZone = StringUtils.parseTimeZoneString(zone);
						}
						else {
							timeZone = TimeZone.getDefault();
						}
                        // 把我们的cronTask放到 registrar的 cronTasks列表中
						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;
			}

			// 处理 fixed delay 也放到 registrar的 cronTasks列表中
			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)));
				}
			}

			// 处理fixed rate  也放到 registrar的 cronTasks列表中
			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> regTasks = this.scheduledTasks.computeIfAbsent(bean, key -> new LinkedHashSet<>(4));
				regTasks.addAll(tasks);
			}
		}
		catch (IllegalArgumentException ex) {
			throw new IllegalStateException(
					"Encountered invalid @Scheduled method '" + method.getName() + "': " + ex.getMessage());
		}
	}

上面 processScheduled方法主要是把我们找到的定时任务按类型重新封装保存到

registrar


（前面所提）

的一个列表中，那么这些任务是什么时候执行呢，继续从ScheduledAnnotationBeanPostProcessor类中找 

onApplicationEvent 也属于springioc容器扩展之一，他会在实例化完成剩下的非惰性单例后的刷新事件中调用，到这里，我们的ioc容器已经准备好了该有的bean，然后就回调该方法，在finishRegistration方法的最后，有一个registrar的afterPropertiesSet的方法调用

该方法在ScheduledTaskRegistrar类中，

afterPropertiesSet继续调用scheduleTasks，我们来看下该方法的源代码 

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));
			}
		}
         // cronTasks处理
		if (this.cronTasks != null) {
			for (CronTask task : this.cronTasks) {
                // addScheduledTask 主要把我们创建的ScheduledTask任务放到scheduledTasks集合中
               // scheduleCronTask 把第一次解析的任务拿出来交给taskScheduler执行
				addScheduledTask(scheduleCronTask(task));  // ①
			}
		}
        //fixedRateTasks 处理
		if (this.fixedRateTasks != null) {
			for (IntervalTask task : this.fixedRateTasks) {
				addScheduledTask(scheduleFixedRateTask(task)); // ②
			}
		}
       // fixedDelayTasks 处理
		if (this.fixedDelayTasks != null) {
			for (IntervalTask task : this.fixedDelayTasks) {
				addScheduledTask(scheduleFixedDelayTask(task)); //③
			}
		}
	}

①②③处功能都差不多

1.按照不同类型，通过addScheduledTask方法把ScheduleTask务放到scheduledTasks集合中

2.处理不同类型的任务 用②中的scheduleFixedRateTask来分析， 

	@Nullable
	public ScheduledTask scheduleFixedRateTask(IntervalTask task) {
		FixedRateTask taskToUse = (task instanceof FixedRateTask ? (FixedRateTask) task :
				new FixedRateTask(task.getRunnable(), task.getInterval(), task.getInitialDelay()));
		return scheduleFixedRateTask(taskToUse);// ①
	}

继续看看①scheduleFixedRateTask方法的源码 

	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) {
            // 本方法 第二次执行，去调用this.taskScheduler.scheduleAtFixedRate开始执行任务
			if (task.getInitialDelay() > 0) {
				Date 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);
            // 本方法 第一次执行 把任务放到unresolvedTasks集合中
			this.unresolvedTasks.put(task, scheduledTask);
		}
		return (newTask ? scheduledTask : null);
	}

这个方法有两个地方调用，第一次是在解析@Sheduled时候，把任务放到一个叫unresolvedTasks集合中，第二次就是上面所说的地方进行调用，此时的调用是把unresolvedTasks集合的任务移除，并且通过this.taskScheduler对象去第一次执行

代码中的①②fixedRate真正执行开始，①有第一次执行有延时

接下来，我们查看下scheduleAtFixedRate的源代码

this.taskScheduler.scheduleAtFixedRate()方法就是我们任务执行的开始，这里taskScheduler是ConcurrentTaskScheduler的实例，我们可以看下ConcurrentTaskScheduler的scheduleAtFixedRate 

	@Override
	public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, long period) {
		try {
            // 调用scheduledExecutor的scheduleAtFixedRate方法去执行任务
			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);
		}
	}

那么我们①处的scheduledExecutor是什么呢 他和ScheduledThreadPoolExecutor有什么关系，

这里用到了一个静态代理模式  scheduledExecutor.scheduleAtFixedRate执行的就是ScheduledThreadPoolExecutor对象的scheduleAtFixedRate 

 static class DelegatedScheduledExecutorService
            extends DelegatedExecutorService
            implements ScheduledExecutorService {
        private final ScheduledExecutorService e;
        DelegatedScheduledExecutorService(ScheduledExecutorService executor) {
            super(executor);
            e = executor;
        }
        public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
            return e.schedule(command, delay, unit);
        }
        public <V> ScheduledFuture<V> schedule(Callable<V> callable, long delay, TimeUnit unit) {
            return e.schedule(callable, delay, unit);
        }
        public ScheduledFuture<?> scheduleAtFixedRate(Runnable command, long initialDelay, long period, TimeUnit unit) {
            return e.scheduleAtFixedRate(command, initialDelay, period, unit);
        }
        public ScheduledFuture<?> scheduleWithFixedDelay(Runnable command, long initialDelay, long delay, TimeUnit unit) {
            return e.scheduleWithFixedDelay(command, initialDelay, delay, unit);
        }
    }

到这里，我们清楚了@Scheduled解析地方，也知道了他的运行原理，就是把原对象和方法封装成runnable 执行的时候就是反射调用我们要执行的对象的方法，这里定时用了ScheduledThreadPoolExecutor。他里面就提供了FixedRate和FixedDelay

四，思考

1 我们怎么动态修改@Scheduled(cron = “*/1 * * * * *”)里面的cron等参数呢，比如通过调用接口或者修改数据库的方式动态改变cron

2 ScheduledThreadPoolExecutor 定时器原理是什么


https://blog.csdn.net/yegeg/article/details/121691961


版权声明：本文为yegeg原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。