作者 | 付政委
博客 | https://bugstack.cn
微信公众号:bugstack虫洞栈 | 博客:https://bugstack.cn 沉淀、分享、成长,专注于原创专题案例,以最易学习编程的方式分享知识,让自己和他人都能有所收获。目前已完成的专题有;Netty4.x实战专题案例、用Java实现JVM、基于JavaAgent的全链路监控、手写RPC框架、架构设计专题案例[Ing]等。 你用剑?、我用刀?,好的代码都很烧?,望你不吝出招?!
在日常开发中经常会用到定时任务,用来;库表扫描发送MQ、T+n账单结算、缓存数据更新、秒杀活动状态变更,等等。因为有了Spring的Schedule极大的方便了我们对这类场景的使用。那么,除了应用你还了解它多少呢;
蒙圈了吧,是不感觉平时只是使用了,根本没关注过这些。有种冲动赶紧搜索答案吧!但只是知道答案是没有多少意义的,扛不住问不说,也不了解原理。所以,如果你想真的提升自己技能,还是要从根本搞定。
为了更好的做源码分析,我们将平时用的定时任务服务单独抽离出来。工程下载,关注公众号:bugstack虫洞栈,回复:源码分析
1itstack-demo-code-schedule
2└── src
3 ├── main
4 │ ├── java
5 │ │ └── org.itstack.demo
6 │ │ ├── DemoTask.java
7 │ │ └── JobImpl.java
8 │ └── resources
9 │ ├── props
10 │ │ └── config.properties
11 │ ├── spring
12 │ │ └── spring-config-schedule-task.xml
13 │ ├── logback.xml
14 │ └── spring-config.xml
15 └── test
16 └── java
17 └── org.itstack.demo.test
18 ├── ApiTest.java
19 ├── MyQuartz.java
20 └── MyTask.java
1<dependency>
2 <groupId>org.quartz-scheduler</groupId>
3 <artifactId>quartz</artifactId>
4 <version>2.3.2</version>
5</dependency>
依赖于Spring版本升级quartz选择2.3.2,同时如果你如本文案例中所示使用xml配置任务。那么会有如下更改;
Spring 3.x/org.springframework.scheduling.quart.CronTriggerBean
1 <bean id="taskTrigger" class="org.springframework.scheduling.quartz.CronTriggerBean">
2 <property name="jobDetail" ref="taskHandler"/>
3 <property name="cronExpression" value="0/5 * * * * ?"/>
4 </bean>
Spring 4.x/org.springframework.scheduling.quartz.CronTriggerFactoryBean
1 <bean id="taskTrigger" class="org.springframework.scheduling.quartz.CronTriggerFactoryBean">
2 <property name="jobDetail" ref="taskHandler"/>
3 <property name="cronExpression" value="0/5 * * * * ?"/>
4 </bean>
在正式分析前,可以看下quartz的默认配置,很多初始化动作都要从这里取得参数,同样你可以配置自己的配置文件。例如,当你的任务很多时,默认初始化的10个线程组不满足你的业务需求,就可以按需调整。
quart.properties
1# Default Properties file for use by StdSchedulerFactory
2# to create a Quartz Scheduler Instance, if a different
3# properties file is not explicitly specified.
4#
5
6org.quartz.scheduler.instanceName: DefaultQuartzScheduler
7org.quartz.scheduler.rmi.export: false
8org.quartz.scheduler.rmi.proxy: false
9org.quartz.scheduler.wrapJobExecutionInUserTransaction: false
10
11org.quartz.threadPool.class: org.quartz.simpl.SimpleThreadPool
12org.quartz.threadPool.threadCount: 10
13org.quartz.threadPool.threadPriority: 5
14org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread: true
15
16org.quartz.jobStore.misfireThreshold: 60000
17
18org.quartz.jobStore.class: org.quartz.simpl.RAMJobStore
平时我们使用Schedule基本都是注解或者xml配置文件,但是为了可以更简单的分析代码,我们从一个简单的Demo入手,放到main函数中。
DemoTask.java & 定义一个等待被执行的任务
1public class DemoTask {
2
3 private Logger logger = LoggerFactory.getLogger(DemoTask.class);
4
5 public void execute() throws Exception{
6 logger.info("定时处理用户信息任务:0/5 * * * * ?");
7 }
8
9}
MyTask.java & 测试类,将配置在xml中的代码抽离出来
1public class MyTask {
2
3 public static void main(String[] args) throws Exception {
4
5 DemoTask demoTask = new DemoTask();
6
7 // 定义了;执行的内容
8 MethodInvokingJobDetailFactoryBean methodInvokingJobDetailFactoryBean = new MethodInvokingJobDetailFactoryBean();
9 methodInvokingJobDetailFactoryBean.setTargetObject(demoTask);
10 methodInvokingJobDetailFactoryBean.setTargetMethod("execute");
11 methodInvokingJobDetailFactoryBean.setConcurrent(true);
12 methodInvokingJobDetailFactoryBean.setName("demoTask");
13 methodInvokingJobDetailFactoryBean.afterPropertiesSet();
14
15 // 定义了;执行的计划
16 CronTriggerFactoryBean cronTriggerFactoryBean = new CronTriggerFactoryBean();
17 cronTriggerFactoryBean.setJobDetail(methodInvokingJobDetailFactoryBean.getObject());
18 cronTriggerFactoryBean.setCronExpression("0/5 * * * * ?");
19 cronTriggerFactoryBean.setName("demoTask");
20 cronTriggerFactoryBean.afterPropertiesSet();
21
22 // 实现了;执行的功能
23 SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
24 schedulerFactoryBean.setTriggers(cronTriggerFactoryBean.getObject());
25 schedulerFactoryBean.setAutoStartup(true);
26 schedulerFactoryBean.afterPropertiesSet();
27
28 schedulerFactoryBean.start();
29
30 // 暂停住
31 System.in.read();
32
33 }
34
35}
如果一切顺利,那么会有如下结果:
12020-01-04 10:47:16.369 [main] INFO org.quartz.impl.StdSchedulerFactory[1220] - Using default implementation for ThreadExecutor
22020-01-04 10:47:16.421 [main] INFO org.quartz.core.SchedulerSignalerImpl[61] - Initialized Scheduler Signaller of type: class org.quartz.core.SchedulerSignalerImpl
32020-01-04 10:47:16.422 [main] INFO org.quartz.core.QuartzScheduler[229] - Quartz Scheduler v.2.3.2 created.
42020-01-04 10:47:16.423 [main] INFO org.quartz.simpl.RAMJobStore[155] - RAMJobStore initialized.
52020-01-04 10:47:16.424 [main] INFO org.quartz.core.QuartzScheduler[294] - Scheduler meta-data: Quartz Scheduler (v2.3.2) 'QuartzScheduler' with instanceId 'NON_CLUSTERED'
6 Scheduler class: 'org.quartz.core.QuartzScheduler' - running locally.
7 NOT STARTED.
8 Currently in standby mode.
9 Number of jobs executed: 0
10 Using thread pool 'org.quartz.simpl.SimpleThreadPool' - with 10 threads.
11 Using job-store 'org.quartz.simpl.RAMJobStore' - which does not support persistence. and is not clustered.
12
132020-01-04 10:47:16.424 [main] INFO org.quartz.impl.StdSchedulerFactory[1374] - Quartz scheduler 'QuartzScheduler' initialized from an externally provided properties instance.
142020-01-04 10:47:16.424 [main] INFO org.quartz.impl.StdSchedulerFactory[1378] - Quartz scheduler version: 2.3.2
152020-01-04 10:47:16.426 [main] INFO org.quartz.core.QuartzScheduler[2293] - JobFactory set to: org.springframework.scheduling.quartz.AdaptableJobFactory@3e9b1010
162020-01-04 10:47:16.651 [main] INFO org.quartz.core.QuartzScheduler[547] - Scheduler QuartzScheduler_$_NON_CLUSTERED started.
17一月 04, 2020 10:47:16 上午 org.springframework.scheduling.quartz.SchedulerFactoryBean startScheduler
18信息: Starting Quartz Scheduler now
192020-01-04 10:47:20.321 [QuartzScheduler_Worker-1] INFO org.itstack.demo.DemoTask[11] - 定时处理用户信息任务:0/5 * * * * ?
202020-01-04 10:47:25.001 [QuartzScheduler_Worker-2] INFO org.itstack.demo.DemoTask[11] - 定时处理用户信息任务:0/5 * * * * ?
212020-01-04 10:47:30.000 [QuartzScheduler_Worker-3] INFO org.itstack.demo.DemoTask[11] - 定时处理用户信息任务:0/5 * * * * ?
222020-01-04 10:47:35.001 [QuartzScheduler_Worker-4] INFO org.itstack.demo.DemoTask[11] - 定时处理用户信息任务:0/5 * * * * ?
232020-01-04 10:47:40.000 [QuartzScheduler_Worker-5] INFO org.itstack.demo.DemoTask[11] - 定时处理用户信息任务:0/5 * * * * ?
24
25Process finished with exit code -1
26
1// 定义了;执行的内容
2MethodInvokingJobDetailFactoryBean methodInvokingJobDetailFactoryBean = new MethodInvokingJobDetailFactoryBean();
3methodInvokingJobDetailFactoryBean.setTargetObject(demoTask);
4methodInvokingJobDetailFactoryBean.setTargetMethod("execute");
5methodInvokingJobDetailFactoryBean.setConcurrent(true);
6methodInvokingJobDetailFactoryBean.setName("demoTask");
7methodInvokingJobDetailFactoryBean.afterPropertiesSet();
这块内容主要将我们的任务体(即待执行任务DemoTask)交给MethodInvokingJobDetailFactoryBean管理,首先设置必要信息;
最后我们通过手动调用 afterPropertiesSet() 来模拟初始化。如果我们的类是交给 Spring 管理的,那么在实现了 InitializingBean 接口的类,在类配置信息加载后会自动执行 afterPropertiesSet() 。一般实现了 InitializingBean 接口的类,同时也会去实现 FactoryBean 接口,因为这个接口实现后就可以通过 T getObject() 获取自己自定义初始化的类。这也常常用在一些框架开发中。
MethodInvokingJobDetailFactoryBean.afterPropertiesSet()
1public void afterPropertiesSet() throws ClassNotFoundException, NoSuchMethodException {
2 prepare();
3 // Use specific name if given, else fall back to bean name.
4 String name = (this.name != null ? this.name : this.beanName);
5 // Consider the concurrent flag to choose between stateful and stateless job.
6 Class<?> jobClass = (this.concurrent ? MethodInvokingJob.class : StatefulMethodInvokingJob.class);
7 // Build JobDetail instance.
8 JobDetailImpl jdi = new JobDetailImpl();
9 jdi.setName(name);
10 jdi.setGroup(this.group);
11 jdi.setJobClass((Class) jobClass);
12 jdi.setDurability(true);
13 jdi.getJobDataMap().put("methodInvoker", this);
14 this.jobDetail = jdi;
15
16 postProcessJobDetail(this.jobDetail);
17}
微信公众号:bugstack虫洞栈 & Schedule.xml
1// 定义了;执行的计划
2CronTriggerFactoryBean cronTriggerFactoryBean = new CronTriggerFactoryBean();
3cronTriggerFactoryBean.setJobDetail(methodInvokingJobDetailFactoryBean.getObject());
4cronTriggerFactoryBean.setCronExpression("0/5 * * * * ?");
5cronTriggerFactoryBean.setName("demoTask");
6cronTriggerFactoryBean.afterPropertiesSet();
这一块主要定义任务的执行计划,并将任务执行内容交给 CronTriggerFactoryBean 管理,同时设置必要信息;
CronTriggerFactoryBean.afterPropertiesSet()
1@Override
2public void afterPropertiesSet() throws ParseException {
3
4 // ... 校验属性信息
5
6 CronTriggerImpl cti = new CronTriggerImpl();
7 cti.setName(this.name);
8 cti.setGroup(this.group);
9 if (this.jobDetail != null) {
10 cti.setJobKey(this.jobDetail.getKey());
11 }
12 cti.setJobDataMap(this.jobDataMap);
13 cti.setStartTime(this.startTime);
14 cti.setCronExpression(this.cronExpression);
15 cti.setTimeZone(this.timeZone);
16 cti.setCalendarName(this.calendarName);
17 cti.setPriority(this.priority);
18 cti.setMisfireInstruction(this.misfireInstruction);
19 cti.setDescription(this.description);
20 this.cronTrigger = cti;
21}
CronTriggerFactoryBean.getObject()
1@Override
2public CronTrigger getObject() {
3 return this.cronTrigger;
4}
1// 调度了;执行的计划(scheduler)
2SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
3schedulerFactoryBean.setTriggers(cronTriggerFactoryBean.getObject());
4schedulerFactoryBean.setAutoStartup(true);
5schedulerFactoryBean.afterPropertiesSet();
6
7schedulerFactoryBean.start();
这一部分如名字一样调度工厂,相当于一个指挥官,可以从全局做调度,比如监听哪些trigger已经ready、分配线程等等,同样也需要设置必要的属性信息;
这个过程较长包括:调度工厂、线程池、注册任务等等,整体核心加载流程如下;
微信公众号:bugstack虫洞栈 & 调度工程初始化流程
SchedulerFactoryBean.afterPropertiesSet()
1public void afterPropertiesSet() throws Exception {
2 if (this.dataSource == null && this.nonTransactionalDataSource != null) {
3 this.dataSource = this.nonTransactionalDataSource;
4 }
5 if (this.applicationContext != null && this.resourceLoader == null) {
6 this.resourceLoader = this.applicationContext;
7 }
8 // Initialize the Scheduler instance...
9 this.scheduler = prepareScheduler(prepareSchedulerFactory());
10 try {
11 registerListeners();
12 registerJobsAndTriggers();
13 }
14 catch (Exception ex) {
15 try {
16 this.scheduler.shutdown(true);
17 }
18 catch (Exception ex2) {
19 logger.debug("Scheduler shutdown exception after registration failure", ex2);
20 }
21 throw ex;
22 }
23}
SchedulerAccessor.addTriggerToScheduler() & SchedulerAccessor 是SchedulerFactoryBean的父类
1private boolean addTriggerToScheduler(Trigger trigger) throws SchedulerException {
2 boolean triggerExists = (getScheduler().getTrigger(trigger.getKey()) != null);
3 if (triggerExists && !this.overwriteExistingJobs) {
4 return false;
5 }
6 // Check if the Trigger is aware of an associated JobDetail.
7 JobDetail jobDetail = (JobDetail) trigger.getJobDataMap().remove("jobDetail");
8 if (triggerExists) {
9 if (jobDetail != null && !this.jobDetails.contains(jobDetail) && addJobToScheduler(jobDetail)) {
10 this.jobDetails.add(jobDetail);
11 }
12 try {
13 getScheduler().rescheduleJob(trigger.getKey(), trigger);
14 }
15 catch (ObjectAlreadyExistsException ex) {
16 if (logger.isDebugEnabled()) {
17 logger.debug("Unexpectedly encountered existing trigger on rescheduling, assumably due to " +
18 "cluster race condition: " + ex.getMessage() + " - can safely be ignored");
19 }
20 }
21 }
22 else {
23 try {
24 if (jobDetail != null && !this.jobDetails.contains(jobDetail) &&
25 (this.overwriteExistingJobs || getScheduler().getJobDetail(jobDetail.getKey()) == null)) {
26 getScheduler().scheduleJob(jobDetail, trigger);
27 this.jobDetails.add(jobDetail);
28 }
29 else {
30 getScheduler().scheduleJob(trigger);
31 }
32 }
33 catch (ObjectAlreadyExistsException ex) {
34 if (logger.isDebugEnabled()) {
35 logger.debug("Unexpectedly encountered existing trigger on job scheduling, assumably due to " +
36 "cluster race condition: " + ex.getMessage() + " - can safely be ignored");
37 }
38 if (this.overwriteExistingJobs) {
39 getScheduler().rescheduleJob(trigger.getKey(), trigger);
40 }
41 }
42 }
43 return true;
44}
案例中使用硬编码方式调用 schedulerFactoryBean.start() 启动线程服务。线程的协作通过Object sigLock来实现,关于sigLock.wait()方法都在QuartzSchedulerThread的run方法里面,所以sigLock唤醒的是只有线程QuartzSchedulerThread。核心流程如下;
微信公众号:bugstack虫洞栈 & 调度启动流程
这个启动过程中,核心的代码类,如下;
QuartzScheduler.start() & 启动
1public void start() throws SchedulerException {
2
3 if (shuttingDown|| closed) {
4 throw new SchedulerException(
5 "The Scheduler cannot be restarted after shutdown() has been called.");
6 }
7
8 // QTZ-212 : calling new schedulerStarting() method on the listeners
9 // right after entering start()
10 notifySchedulerListenersStarting();
11
12 if (initialStart == null) {
13 initialStart = new Date();
14 this.resources.getJobStore().schedulerStarted();
15 startPlugins();
16 } else {
17 resources.getJobStore().schedulerResumed();
18 }
19
20 // 唤醒线程
21 schedThread.togglePause(false);
22
23 getLog().info(
24 "Scheduler " + resources.getUniqueIdentifier() + " started.");
25
26 notifySchedulerListenersStarted();
27}
QuartzSchedulerThread.run() & 执行过程
1@Override
2public void run() {
3 int acquiresFailed = 0;
4
5 // 只有调用了halt()方法,才会退出这个死循环
6 while (!halted.get()) {
7 try {
8
9 // 一、如果是暂停状态,则循环超时等待1000毫秒
10
11 // wait a bit, if reading from job store is consistently failing (e.g. DB is down or restarting)..
12
13 // 阻塞直到有空闲的线程可用并返回可用的数量
14 int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
15 if(availThreadCount > 0) {
16
17 List<OperableTrigger> triggers;
18 long now = System.currentTimeMillis();
19 clearSignaledSchedulingChange();
20
21 try {
22 // 二、获取acquire状态的Trigger列表,也就是即将执行的任务
23 triggers = qsRsrcs.getJobStore().acquireNextTriggers(
24 now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBat
25 acquiresFailed = 0;
26 if (log.isDebugEnabled())
27 log.debug("batch acquisition of " + (triggers == null ? 0 : triggers
28 } catch(){//...}
29
30 if (triggers != null && !triggers.isEmpty()) {
31
32 // 三:获取List第一个Trigger的下次触发时刻
33 long triggerTime = triggers.get(0).getNextFireTime().getTime();
34
35 // 四:获取任务触发集合
36 List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);
37
38 // 五:设置Triggers为'executing'状态
39 qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
40
41 // 六:创建JobRunShell
42 qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle);
43
44 // 七:执行Job
45 qsRsrcs.getThreadPool().runInThread(shell)
46
47 continue; // while (!halted)
48 }
49 } else { // if(availThreadCount > 0)
50 // should never happen, if threadPool.blockForAvailableThreads() follows con
51 continue; // while (!halted)
52 }
53
54
55 } catch(RuntimeException re) {
56 getLog().error("Runtime error occurred in main trigger firing loop.", re);
57 }
58 }
59
60 qs = null;
61 qsRsrcs = null;
62}
JobRunShell.run() & 从上面WorkerThread.run(),调用到这里执行
1public void run() {
2 qs.addInternalSchedulerListener(this);
3
4 try {
5 OperableTrigger trigger = (OperableTrigger) jec.getTrigger();
6 JobDetail jobDetail = jec.getJobDetail();
7
8 do {
9 // ...
10
11 long startTime = System.currentTimeMillis();
12 long endTime = startTime;
13
14 // execute the job
15 try {
16 log.debug("Calling execute on job " + jobDetail.getKey());
17
18 // 执行业务代码,也就是我们的task
19 job.execute(jec);
20
21 endTime = System.currentTimeMillis();
22 } catch (JobExecutionException jee) {
23 endTime = System.currentTimeMillis();
24 jobExEx = jee;
25 getLog().info("Job " + jobDetail.getKey() +
26 " threw a JobExecutionException: ", jobExEx);
27 } catch (Throwable e) {
28 endTime = System.currentTimeMillis();
29 getLog().error("Job " + jobDetail.getKey() +
30 " threw an unhandled Exception: ", e);
31 SchedulerException se = new SchedulerException(
32 "Job threw an unhandled exception.", e);
33 qs.notifySchedulerListenersError("Job ("
34 + jec.getJobDetail().getKey()
35 + " threw an exception.", se);
36 jobExEx = new JobExecutionException(se, false);
37 }
38
39 jec.setJobRunTime(endTime - startTime);
40
41 // 其他代码
42 } while (true);
43
44 } finally {
45 qs.removeInternalSchedulerListener(this);
46 }
47}
QuartzJobBean.execte() & 继续往下走
1public final void execute(JobExecutionContext context) throws JobExecutionException {
2 try {
3 BeanWrapper bw = PropertyAccessorFactory.forBeanPropertyAccess(this);
4 MutablePropertyValues pvs = new MutablePropertyValues();
5 pvs.addPropertyValues(context.getScheduler().getContext());
6 pvs.addPropertyValues(context.getMergedJobDataMap());
7 bw.setPropertyValues(pvs, true);
8 }
9 catch (SchedulerException ex) {
10 throw new JobExecutionException(ex);
11 }
12 executeInternal(context);
13}
MethodInvokingJobDetailFactoryBean->MethodInvokingJob.executeInternal(JobExecutionContext context)
1protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
2 try {
3 // 反射执行业务代码
4 context.setResult(this.methodInvoker.invoke());
5 }
6 catch (InvocationTargetException ex) {
7 if (ex.getTargetException() instanceof JobExecutionException) {
8 // -> JobExecutionException, to be logged at info level by Quartz
9 throw (JobExecutionException) ex.getTargetException();
10 }
11 else {
12 // -> "unhandled exception", to be logged at error level by Quartz
13 throw new JobMethodInvocationFailedException(this.methodInvoker, ex.getTargetException());
14 }
15 }
16 catch (Exception ex) {
17 // -> "unhandled exception", to be logged at error level by Quartz
18 throw new JobMethodInvocationFailedException(this.methodInvoker, ex);
19 }
20}