Spring的定時(shí)任務(wù)@Scheduled源碼詳解

更新時(shí)間：2023年09月25日 09:20:46 作者：HengTian_real

這篇文章主要介紹了Spring的定時(shí)任務(wù)@Scheduled源碼詳解,@Scheduled注解是包org.springframework.scheduling.annotation中的一個(gè)注解,主要是用來(lái)開啟定時(shí)任務(wù),本文提供了部分實(shí)現(xiàn)代碼與思路,需要的朋友可以參考下

Spring的定時(shí)任務(wù)@Scheduled源碼詳解

EnableScheduling

@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Import({SchedulingConfiguration.class})	//@Import注解將SchedulingConfiguration導(dǎo)入到IOC中
@Documented
public @interface EnableScheduling {
}

SchedulingConfiguration

將ScheduledAnnotationBeanPostProcessor加入到Spring容器中 @Configuration //當(dāng)結(jié)合@Bean注解時(shí)，@Bean注解的類可以類比在spring.xml中定義

@Role(2)
public class SchedulingConfiguration {
    public SchedulingConfiguration() {
    }
    @Bean(
        name = {"org.springframework.context.annotation.internalScheduledAnnotationProcessor"}
    )
    @Role(2)
    public ScheduledAnnotationBeanPostProcessor scheduledAnnotationProcessor() {
        return new ScheduledAnnotationBeanPostProcessor();
    }
}

ScheduledAnnotationBeanPostProcessor

1、先看下ScheduledAnnotationBeanPostProcessor有哪些屬性

//string屬性解析器,用來(lái)解析${}對(duì)應(yīng)的配置文件的屬性,aware接口注入
@Nullable
private StringValueResolver embeddedValueResolver;
@Nullable
private String beanName;
@Nullable
private BeanFactory beanFactory;	//aware接口注入
@Nullable
private ApplicationContext applicationContext;	//aware接口注入
@Nullable
//定時(shí)任務(wù)線程池,如果不為空使用這個(gè)scheduler當(dāng)作ScheduledTaskRegistrar的線程池
private Object scheduler;
//定時(shí)任務(wù)的注冊(cè)器,通過(guò)這個(gè)類將定時(shí)任務(wù)委托給定時(shí)任務(wù)線程池
private final ScheduledTaskRegistrar registrar = new ScheduledTaskRegistrar();
//已檢測(cè)的沒(méi)有scheduled注解的類的集合
private final Set<Class<?>> nonAnnotatedClasses = Collections.newSetFromMap(new ConcurrentHashMap(64));
//保存class與scheduled方法的映射
private final Map<Object, Set<ScheduledTask>> scheduledTasks = new IdentityHashMap(16);

2、根據(jù)上面的定時(shí)任務(wù)流程，在每個(gè)Bean的屬性填充完之后，調(diào)用postProcessAfterInitialization方法，將帶有@Scheduled 注解的方法，在拿到@Scheduled注解的方法后，調(diào)用processScheduled

public Object postProcessAfterInitialization(Object bean, String beanName) {
    Class<?> targetClass = AopProxyUtils.ultimateTargetClass(bean);
    if (!this.nonAnnotatedClasses.contains(targetClass)) {
        Map<Method, Set<Scheduled>> annotatedMethods = MethodIntrospector.selectMethods(targetClass, (method) -> {
            Set<Scheduled> scheduledMethods = AnnotatedElementUtils.getMergedRepeatableAnnotations(method, Scheduled.class, Schedules.class);
            return !scheduledMethods.isEmpty() ? scheduledMethods : null;
        });
        if (annotatedMethods.isEmpty()) {
            this.nonAnnotatedClasses.add(targetClass);
            if (this.logger.isTraceEnabled()) {
                this.logger.trace("No @Scheduled annotations found on bean class: " + bean.getClass());
            }
        } else {
            annotatedMethods.forEach((method, scheduledMethods) -> {
                scheduledMethods.forEach((scheduled) -> {
                    //這里調(diào)用
                    this.processScheduled(scheduled, method, bean);
                });
            });
            if (this.logger.isDebugEnabled()) {
                this.logger.debug(annotatedMethods.size() + " @Scheduled methods processed on bean '" + beanName + "': " + annotatedMethods);
            }
        }
    }
    return bean;
}

3、要了解processScheduled方法做了什么，可以先看下@Scheduled 注解的定義

@Target({ElementType.METHOD, ElementType.ANNOTATION_TYPE})
@Retention(RetentionPolicy.RUNTIME)
@Documented
@Repeatable(Schedules.class)
public @interface Scheduled {
    String cron() default "";
    String zone() default "";
    long fixedDelay() default -1L;
    String fixedDelayString() default "";
    long fixedRate() default -1L;
    String fixedRateString() default "";
    long initialDelay() default -1L;
    String initialDelayString() default "";
}

可以看到，processScheduled方法主要是使用embeddedValueResolver對(duì)帶String后綴的屬性進(jìn)行從配置文件讀取的操作，根據(jù)每個(gè)方法上使用的注解判斷定時(shí)任務(wù)的類型是CronTask還是FixedRateTask，將這些任務(wù)添加到ScheduledTaskRegistrar中的unresolvedTasks

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);
        long initialDelay = scheduled.initialDelay();
        String initialDelayString = scheduled.initialDelayString();
        if (StringUtils.hasText(initialDelayString)) {
            Assert.isTrue(initialDelay < 0L, "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 var25) {
                    throw new IllegalArgumentException("Invalid initialDelayString value \"" + initialDelayString + "\" - cannot parse into long");
                }
            }
        }
        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 == -1L, "'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))));
            }
        }
        //省略了一部分解析過(guò)程,和解析cron是一樣的
        ...
            ...
            Assert.isTrue(processedSchedule, errorMessage);
        synchronized(this.scheduledTasks) {
            Set<ScheduledTask> registeredTasks = (Set)this.scheduledTasks.get(bean);
            if (registeredTasks == null) {
                registeredTasks = new LinkedHashSet(4);
                this.scheduledTasks.put(bean, registeredTasks);
            }
            ((Set)registeredTasks).addAll(tasks);
        }
    } catch (IllegalArgumentException var26) {
        throw new IllegalStateException("Encountered invalid @Scheduled method '" + method.getName() + "': " + var26.getMessage());
    }
}

對(duì)于3中，比較關(guān)鍵的代碼

tasks.add(this.registrar.scheduleCronTask(new CronTask(runnable, new CronTrigger(cron, timeZone))));

跟蹤到this.registrar.scheduleCronTask(),這里跳轉(zhuǎn)到ScheduledTaskRegistrar類的scheduleCronTask()

public ScheduledTask scheduleCronTask(CronTask task) {
    ScheduledTask scheduledTask = (ScheduledTask)this.unresolvedTasks.remove(task);
    boolean newTask = false;
    if (scheduledTask == null) {
        //創(chuàng)建ScheduledTask
        scheduledTask = new ScheduledTask(task);
        newTask = true;
    }
    //可以看到ScheduledTaskRegistrar的初始化方法中沒(méi)有對(duì)taskScheduler賦值
    //所以此時(shí)this.taskScheduler = null
    if (this.taskScheduler != null) {
        scheduledTask.future = this.taskScheduler.schedule(task.getRunnable(), task.getTrigger());
    } else {
        //進(jìn)入這里
        this.addCronTask(task);
        this.unresolvedTasks.put(task, scheduledTask);
    }
    return newTask ? scheduledTask : null;
}

4、在所有單例的Bean實(shí)例化完成后，調(diào)用afterSingletonsInstantiated() ，在Spring容器初始化完成后，觸發(fā)ContextRefreshedEvent 事件，調(diào)用onApplicationEvent方法，執(zhí)行finishRegistration()

private void finishRegistration() {
    	//對(duì)應(yīng)a
       if (this.scheduler != null) {
           this.registrar.setScheduler(this.scheduler);
       }
   	//對(duì)應(yīng)b
       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);
           Iterator var3 = configurers.iterator();
           while(var3.hasNext()) {
               SchedulingConfigurer configurer = (SchedulingConfigurer)var3.next();
               configurer.configureTasks(this.registrar);
           }
       }
   	//對(duì)應(yīng)c
       if (this.registrar.hasTasks() && this.registrar.getScheduler() == null) {
           Assert.state(this.beanFactory != null, "BeanFactory must be set to find scheduler by type");
           try {
               this.registrar.setTaskScheduler((TaskScheduler)this.resolveSchedulerBean(this.beanFactory, TaskScheduler.class, false));
           } catch (NoUniqueBeanDefinitionException var9) {
               this.logger.debug("Could not find unique TaskScheduler bean", var9);
               try {
                   this.registrar.setTaskScheduler((TaskScheduler)this.resolveSchedulerBean(this.beanFactory, TaskScheduler.class, true));
               } catch (NoSuchBeanDefinitionException var8) {
                   if (this.logger.isInfoEnabled()) {
                       this.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: " + var9.getBeanNamesFound());
                   }
               }
           } catch (NoSuchBeanDefinitionException var10) {
               this.logger.debug("Could not find default TaskScheduler bean", var10);
               try {
                   this.registrar.setScheduler(this.resolveSchedulerBean(this.beanFactory, ScheduledExecutorService.class, false));
               } catch (NoUniqueBeanDefinitionException var6) {
                   this.logger.debug("Could not find unique ScheduledExecutorService bean", var6);
                   try {
                       this.registrar.setScheduler(this.resolveSchedulerBean(this.beanFactory, ScheduledExecutorService.class, true));
                   } catch (NoSuchBeanDefinitionException var5) {
                       if (this.logger.isInfoEnabled()) {
                           this.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: " + var6.getBeanNamesFound());
                       }
                   }
               } catch (NoSuchBeanDefinitionException var7) {
                   this.logger.debug("Could not find default ScheduledExecutorService bean", var7);
                   this.logger.info("No TaskScheduler/ScheduledExecutorService bean found for scheduled processing");
               }
           }
       }
       this.registrar.afterPropertiesSet();
   }

這個(gè)方法主要實(shí)現(xiàn)的內(nèi)容是：

用容器中的SchedulingConfigurer配置ScheduledTaskRegistrar,這里是根據(jù)ScheduledTaskRegistrar的引用，調(diào)用其set方法設(shè)置一些屬性

public interface SchedulingConfigurer {
void configureTasks(ScheduledTaskRegistrar var1);
}

如果此時(shí)ScheduledTaskRegistrar的scheduler還是空，就從容器中取TaskScheduler（byName和byType），如果沒(méi)有取到就根據(jù)容器中的ScheduledExecutorService實(shí)例化TaskScheduler

this.registrar.afterPropertiesSet();

所以在容器中注入TaskScheduler或ScheduledExecutorService的類或者實(shí)現(xiàn)SchedulingConfigurer接口都可以配置定時(shí)任務(wù)的線程池

5、afterPropertiesSet

public void afterPropertiesSet() {
    this.scheduleTasks();
}
protected void scheduleTasks() {
    if (this.taskScheduler == null) {
        this.localExecutor = Executors.newSingleThreadScheduledExecutor();
        this.taskScheduler = new ConcurrentTaskScheduler(this.localExecutor);
    }
    //省略了內(nèi)容
    ...
        ...
        if (this.cronTasks != null) {
            var1 = this.cronTasks.iterator();
            while(var1.hasNext()) {
                CronTask task = (CronTask)var1.next();
                this.addScheduledTask(this.scheduleCronTask(task));
            }
        }
    //省略了內(nèi)容
    ...
    ...
}

又進(jìn)入了熟悉的方法scheduleCronTask，在這里將任務(wù)提交給taskScheduler

public ScheduledTask scheduleCronTask(CronTask task) {
    ScheduledTask 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 {
        this.addCronTask(task);
        this.unresolvedTasks.put(task, scheduledTask);
    }
    return newTask ? scheduledTask : null;
}

6、如果想看taskScheduler是怎么執(zhí)行定時(shí)任務(wù)的，可以看taskScheduler的一個(gè)默認(rèn)實(shí)現(xiàn)ConcurrentTaskScheduler，大體是有一個(gè)任務(wù)隊(duì)列WorkerQueue，這個(gè)隊(duì)列是按小頂堆排序的，排序規(guī)則是任務(wù)執(zhí)行的時(shí)間，每次取出任務(wù)時(shí)，將任務(wù)提交給線程池執(zhí)行，在執(zhí)行任務(wù)的時(shí)候，計(jì)算下一次執(zhí)行的時(shí)間，提交隊(duì)列…

到此這篇關(guān)于Spring的定時(shí)任務(wù)@Scheduled源碼詳解的文章就介紹到這了,更多相關(guān)Spring定時(shí)任務(wù)@Scheduled內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

您可能感興趣的文章: