引言
在一些業(yè)務(wù)場景中需要執(zhí)行定時操作來完成一些周期性的任務(wù)，比如每隔一周刪除一周前的某些歷史數(shù)據(jù)以及定時進(jìn)行某項(xiàng)檢測任務(wù)等等。
在日常開發(fā)中比較簡單的實(shí)現(xiàn)方式就是使用Spring的@Scheduled(具體使用方法不再贅述)注解。
但是在修改服務(wù)器時間時會導(dǎo)致定時任務(wù)不執(zhí)行情況的發(fā)生，解決的辦法是當(dāng)修改服務(wù)器時間后，將服務(wù)進(jìn)行重啟就可以避免此現(xiàn)象的發(fā)生。
本文將主要探討服務(wù)器時間修改導(dǎo)致@Scheduled注解失效的原因，同時找到在修改服務(wù)器時間后不重啟服務(wù)的情況下，定時任務(wù)仍然正常執(zhí)行的方法。
- @Scheduled失效原因分析
- 解析流程圖
- 使用新的方法
1.@Scheduled失效原因
（1）首先我們一起看一下@Scheduled注解的源碼，主要說明了注解可使用的參數(shù)形式，在注解中使用了Schedules這個類。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
|
@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 "";} |
（2）接下來我們來看下，Spring容器是如何解析@Scheduled注解的。
|
1
2
3
4
5
6
|
public class ScheduledAnnotationBeanPostProcessor implements MergedBeanDefinitionPostProcessor, DestructionAwareBeanPostProcessor, Ordered, EmbeddedValueResolverAware, BeanNameAware, BeanFactoryAware, ApplicationContextAware, SmartInitializingSingleton, ApplicationListener<ContextRefreshedEvent>, DisposableBean { ... } |
Spring容器加載完bean之后，postProcessAfterInitialization將攔截所有以@Scheduled注解標(biāo)注的方法。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
|
@Overridepublic 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 { // 循環(huán)處理包含@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;} |
再往下繼續(xù)看，Spring是如何處理帶有@Schedule注解的方法的。processScheduled獲取scheduled類參數(shù)，之后根據(jù)參數(shù)類型、相應(yīng)的延時時間、對應(yīng)的時區(qū)將定時任務(wù)放入不同的任務(wù)列表中。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
|
protected void processScheduled(Scheduled scheduled, Method method, Object bean) { try { Assert.isTrue(method.getParameterCount() == 0, "Only no-arg methods may be annotated with @Scheduled"); //獲取調(diào)用的方法 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參數(shù) 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(); } //加入到定時任務(wù)列表中 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))); } } // 執(zhí)行頻率的類型為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()); } } |
滿足條件時將定時任務(wù)添加到定時任務(wù)列表中，在加入任務(wù)列表的同時對定時任務(wù)進(jìn)行注冊。ScheduledTaskRegistrar這個類為Spring容器的定時任務(wù)注冊中心。以下為ScheduledTaskRegistrar部分源碼，主要說明該類中包含的屬性。Spring容器通過線程處理注冊的定時任務(wù)。
|
1
2
3
4
5
6
7
8
9
10
11
12
|
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); ......} |
ScheduledTaskRegistrar類中在處理定時任務(wù)時會調(diào)用scheduleCronTask方法初始化定時任務(wù)。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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); } |
在ThreadPoolTaskShcedule這個類中，進(jìn)行線程池的初始化。在創(chuàng)建線程池時會創(chuàng)建 DelayedWorkQueue()阻塞隊(duì)列，定時任務(wù)會被提交到線程池，由線程池進(jìn)行相關(guān)的操作，線程池初始化大小為1。當(dāng)有多個線程需要執(zhí)行時，是需要進(jìn)行任務(wù)等待的，前面的任務(wù)執(zhí)行完了才可以進(jìn)行后面任務(wù)的執(zhí)行。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
@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; } |
根本原因，jvm啟動之后會記錄系統(tǒng)時間，然后jvm根據(jù)CPU ticks自己來算時間，此時獲取的是定時任務(wù)的基準(zhǔn)時間。如果此時將系統(tǒng)時間進(jìn)行了修改，當(dāng)Spring將之前獲取的基準(zhǔn)時間與當(dāng)下獲取的系統(tǒng)時間進(jìn)行比對時，就會造成Spring內(nèi)部定時任務(wù)失效。因?yàn)榇藭r系統(tǒng)時間發(fā)生變化了，不會觸發(fā)定時任務(wù)。
|
1
2
3
4
5
6
7
8
9
10
11
12
|
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; } } |
2.解析流程圖
3.使用新的方法
為了避免使用@Scheduled注解，在修改服務(wù)器時間導(dǎo)致定時任務(wù)不執(zhí)行情況的發(fā)生。在項(xiàng)目中需要使用定時任務(wù)場景的情況下，使ScheduledThreadPoolExecutor進(jìn)行替代，它任務(wù)的調(diào)度是基于相對時間的，原因是它在任務(wù)的內(nèi)部 存儲了該任務(wù)距離下次調(diào)度還需要的時間（使用的是基于 System.nanoTime實(shí)現(xiàn)的相對時間 ，不會因?yàn)橄到y(tǒng)時間改變而改變，如距離下次執(zhí)行還有10秒，不會因?yàn)閷⑾到y(tǒng)時間調(diào)前6秒而變成4秒后執(zhí)行）。
schedule定時任務(wù)修改表達(dá)式無效
真是鬼了。 就那么個cron表達(dá)式，難道還能錯了。
對了無數(shù)遍，cron表達(dá)式?jīng)]問題。 但就是無效。
擴(kuò)展下思路，有沒有用到zookeeper，zookeeper是會緩存配置信息的。
看了下，果然是緩存了。 清空后，重啟項(xiàng)目有效了。
以上為個人經(jīng)驗(yàn)，希望能給大家一個參考，也希望大家多多支持服務(wù)器之家。
原文鏈接：https://blog.csdn.net/Diamond_Tao/article/details/80628440
