v一、前言
定时任务一般是项目中都需要用到的,可以用于定时处理一些特殊的任务。这篇文章主要给大家介绍了关于spring boot定时任务的原理及动态创建的相关内容，下面来一起看看详细的介绍吧
上周工作遇到了一个需求，同步多个省份销号数据，解绑微信粉丝。分省定时将销号数据放到sftp服务器上，我需要开发定时任务去解析文件。因为是多省份，服务器、文件名规则、数据规则都不一定，所以要做成可配置是有一定难度的。数据规则这块必须强烈要求统一，服务器、文件名规则都可以从配置中心去读。每新增一个省份的配置，后台感知到后，动态生成定时任务。
v二、springboot引入定时任务核心配置
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
|
@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(); }} |
接下来主要看一下这个核心后置处理器：scheduledannotationbeanpostprocessor 。
|
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
|
@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;} |
1、处理scheduled注解，通过scheduledtaskregistrar注册定时任务。
|
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
|
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();} |
1、通过一系列的schedulingconfigurer动态配置scheduledtaskregistrar。
2、向scheduledtaskregistrar注册一个taskscheduler（用于对runnable的任务进行调度，它包含有多种触发规则）。
3、registrar.afterpropertiesset()，在这开始安排所有的定时任务开始执行了。
|
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
|
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)); } }} |
1、triggertask：动态定时任务。通过trigger#nextexecutiontime 给定的触发上下文确定下一个执行时间。
2、crontask：动态定时任务，triggertask子类。通过cron表达式确定的时间触发下一个任务执行。
3、intervaltask：一定时间延迟之后，周期性执行的任务。
4、taskscheduler 如果为空，默认是concurrenttaskscheduler，并使用默认单线程的scheduledexecutor。
v三、主要看一下crontask工作原理
|
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
|
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(); } }} |
1、最终将task和trigger都封装到了reschedulingrunnable中。
2、reschedulingrunnable实现了任务重复调度（schedule方法中调用调度器executor并传入自身对象，executor会调用run方法，run方法又调用了schedule方法）。
3、reschedulingrunnable schedule方法加了同步锁，只能有一个线程拿到下次执行时间并加入执行器的调度。
4、不同的reschedulingrunnable对象之间在线程池够用的情况下是不会相互影响的，也就是说满足线程池的条件下，taskscheduler的schedule方法的多次调用是可以交叉执行的。
|
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
|
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(); }} |
scheduledfuturetask 工作原理如下图所示【太懒了，不想画图了，盗图一张】。
1、scheduledfuturetask会放入优先阻塞队列：scheduledthreadpoolexecutor.delayedworkqueue（二叉最小堆实现）
2、上图中的thread对象即threadpoolexecutor.worker，实现了runnable接口
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
/** * 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);} |
1、worker中维护了thread对象，thread对象的runnable实例即worker自身
2、threadpoolexecutor#addworker方法中会创建worker对象，然后拿到worker中的thread实例并start，这样就创建了线程池中的一个线程实例
3、worker的run方法会调用threadpoolexecutor#runworker方法，这才是任务最终被执行的地方，该方法示意如下
（1）首先取传入的task执行，如果task是null，只要该线程池处于运行状态，就会通过gettask方法从workqueue中取任务。threadpoolexecutor的execute方法会在无法产生core线程的时候向 workqueue队列中offer任务。
gettask方法从队列中取task的时候会根据相关配置决定是否阻塞和阻塞多久。如果gettask方法结束，返回的是null，runworker循环结束，执行processworkerexit方法。
至此，该线程结束自己的使命，从线程池中“消失”。
（2）在开始执行任务之前，会调用worker的lock方法，目的是阻止task正在被执行的时候被interrupt，通过调用clearinterruptsfortaskrun方法来保证的(后面可以看一下这个方法)，该线程没有自己的interrupt set了。
（3）beforeexecute和afterexecute方法用于在执行任务前后执行一些自定义的操作，这两个方法是空的，留给继承类去填充功能。
我们可以在beforeexecute方法中抛出异常，这样task不会被执行，而且在跳出该循环的时候completedabruptly的值是true，表示the worker died due to user exception，会用decrementworkercount调整wc。
（4）因为runnable的run方法不能抛出throwables异常，所以这里重新包装异常然后抛出，抛出的异常会使当当前线程死掉，可以在afterexecute中对异常做一些处理。
（5）afterexecute方法也可能抛出异常，也可能使当前线程死掉。
v四、动态创建定时任务
v taskconfiguration 配置类
|
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
|
@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; } }} |
1、保证concurrenttaskscheduler不使用默认单线程的scheduledexecutor，而是corepoolsize=5的线程池
2、自定义线程池工厂类
v dynamictask 动态定时任务
|
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
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
|
@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; } }} |
1、taskconstants 动态任务列表
2、scheduledtaskregistrar#addtriggertask 添加动态周期定时任务，检测动态任务列表的变化
|
1
2
3
4
|
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); |
3、动态创建cron定时任务，拿到scheduledfuture实例并缓存起来
4、在刷新任务列表时，通过缓存的scheduledfuture实例和crontask实例，来决定是否取消、移除失效的动态定时任务。
v dynamictasktest 动态定时任务测试类
|
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
|
@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); }} |
总结
以上就是这篇文章的全部内容了，希望本文的内容对大家的学习或者工作具有一定的参考学习价值，如果有疑问大家可以留言交流，谢谢大家对服务器之家的支持。
原文链接：https://www.cnblogs.com/hujunzheng/p/10353390.html
