springboot-Quartz 集成源码分析和demo

╰半夏微凉° 2022-04-08 14:28 336阅读 0赞

### springboot-Quartz 集成源码跟踪 ###

在pom.xml里加入

<dependency>
     <groupId>org.springframework.boot</groupId>
     <artifactId>spring-boot-starter-quartz</artifactId>
    </dependency>

利用starter的原理在  
spring.factories里有

org.springframework.boot.autoconfigure.quartz.QuartzAutoConfiguration

则启动的时候就会加载QuartzAutoConfiguration，下面看下QuartzAutoConfiguration

@Configuration
    @ConditionalOnClass({  Scheduler.class, SchedulerFactoryBean.class,
    		PlatformTransactionManager.class })
    		//加入QuartzProperties.class
    @EnableConfigurationProperties(QuartzProperties.class)
    @AutoConfigureAfter({  DataSourceAutoConfiguration.class,
    		HibernateJpaAutoConfiguration.class })
    public class QuartzAutoConfiguration { 
    
    private final QuartzProperties properties;
    
    //构造器注入QuartzProperties
    public QuartzAutoConfiguration(QuartzProperties properties,
    			ObjectProvider<SchedulerFactoryBeanCustomizer> customizers,
    			ObjectProvider<JobDetail[]> jobDetails,
    			ObjectProvider<Map<String, Calendar>> calendars,
    			ObjectProvider<Trigger[]> triggers, ApplicationContext applicationContext) { 
    		this.properties = properties;
    		this.customizers = customizers;
    		this.jobDetails = jobDetails.getIfAvailable();
    		this.calendars = calendars.getIfAvailable();
    		this.triggers = triggers.getIfAvailable();
    		this.applicationContext = applicationContext;
    	}
    	
    //定义一个SchedulerFactoryBean，前提是没有SchedulerFactoryBean 
    @Bean
    	@ConditionalOnMissingBean
    	public SchedulerFactoryBean quartzScheduler() { 
    	//new 了一个SchedulerFactoryBean
    		SchedulerFactoryBean schedulerFactoryBean = new SchedulerFactoryBean();
    		SpringBeanJobFactory jobFactory = new SpringBeanJobFactory();
    		jobFactory.setApplicationContext(this.applicationContext);
    		schedulerFactoryBean.setJobFactory(jobFactory);
    		
    		//这里可以定义很多quartz.properteis里的属性
    		if (this.properties.getSchedulerName() != null) { 
    			schedulerFactoryBean.setSchedulerName(this.properties.getSchedulerName());
    		}
    		schedulerFactoryBean.setAutoStartup(this.properties.isAutoStartup());
    		schedulerFactoryBean
    				.setStartupDelay((int) this.properties.getStartupDelay().getSeconds());
    		schedulerFactoryBean.setWaitForJobsToCompleteOnShutdown(
    				this.properties.isWaitForJobsToCompleteOnShutdown());
    		schedulerFactoryBean
    				.setOverwriteExistingJobs(this.properties.isOverwriteExistingJobs());
    		if (!this.properties.getProperties().isEmpty()) { 
    			schedulerFactoryBean
    					.setQuartzProperties(asProperties(this.properties.getProperties()));
    		}
    		if (this.jobDetails != null && this.jobDetails.length > 0) { 
    			schedulerFactoryBean.setJobDetails(this.jobDetails);
    		}
    		if (this.calendars != null && !this.calendars.isEmpty()) { 
    			schedulerFactoryBean.setCalendars(this.calendars);
    		}
    		if (this.triggers != null && this.triggers.length > 0) { 
    			schedulerFactoryBean.setTriggers(this.triggers);
    		}
    		customize(schedulerFactoryBean);
    		return schedulerFactoryBean;
    	}	
    	
    	
    }

以下是SchedulerFactoryBean，可以看到是  
implements FactoryBean，也就是这是Scheduler 这个定义的生成者

FactoryBean和BeanFactory的区别，其实2者除了名字颠倒外，没有必然的关系。BeanFactory是个工厂类，顾名思义就是生产Bean的工厂。而FactoryBean是个生成BeanDefinition的类。

当代码里有这种@Autowired注入Scheduler的时候，如下

@Autowired
     Scheduler scheduler;

BeanFactory会根据name为`Scheduler`来获取Scheduler,最终会通过SchedulerFactoryBean.getObject()来得到Scheduler。

通过分析得出FactoryBean.getObject()里得到的是this.scheduler.

但这个Scheduler 在哪里被构造处理呢？

SchedulerFactoryBean implements `InitializingBean`,在`InitializingBean`里有`afterPropertiesSet()`

public class SchedulerFactoryBean extends SchedulerAccessor implements FactoryBean<Scheduler>,
    		BeanNameAware, ApplicationContextAware, InitializingBean, DisposableBean, SmartLifecycle { 
    		
    		public void afterPropertiesSet() throws Exception { 
            ...
    		// Initialize the Scheduler instance...
    		//这里初始化了Scheduler
    		this.scheduler = prepareScheduler(prepareSchedulerFactory());
    		try { 
    			registerListeners();
    			registerJobsAndTriggers();
    		}
    		catch (Exception ex) { 
    			...
    		}
    	}
    	
    }

bean 生命周期简单理解如下，注意InitializingBean\#afterPropertiesSet()所处的位置

->construcor
    ->initialization(各种autowired)
    ->BeanPostProcessor#postProcessBeforeInitialization
    ->@postConsruct 或 InitializingBean#afterPropertiesSet() 或 @Bean(initMethod="xxx")
    ->BeanPostProcessor#postProcessAfterInitialization
    ->@PreDestroy

看prepareSchedulerFactory()方法，最后返回了`SchedulerFactory`，赋值给了`SchedulerFactoryBean`的scheduler

public class SchedulerFactoryBean{ 
        
    	private Class<? extends SchedulerFactory> schedulerFactoryClass = 
    	StdSchedulerFactory.class;
    	
        private SchedulerFactory prepareSchedulerFactory() throws SchedulerException, IOException { 
            	private SchedulerFactory schedulerFactory;
            //这里这个SchedulerFactory肯定为空，当然有办法可以让它不为空，通过定义SchedulerFactoryBeanCustomizer来实现 
    		SchedulerFactory schedulerFactory = this.schedulerFactory;
    		if (schedulerFactory == null) { 
    			// Create local SchedulerFactory instance (typically a StdSchedulerFactory)
    			//这里也写了这里是实例化出StdSchedulerFactory
    			schedulerFactory = BeanUtils.instantiateClass(this.schedulerFactoryClass);
    			if (schedulerFactory instanceof StdSchedulerFactory) { 
    			//调用initSchedulerFactory来填充StdSchedulerFactory)
    			//看过Quartz的官方demo，就知道StdSchedulerFactory用来生产出sheduler
    			initSchedulerFactory((StdSchedulerFactory) schedulerFactory);
    			}
    			else if (this.configLocation != null || this.quartzProperties != null ||
    					this.taskExecutor != null || this.dataSource != null) { 
    				throw new IllegalArgumentException(
    						"StdSchedulerFactory required for applying Quartz properties: " + schedulerFactory);
    			}
    			// Otherwise, no local settings to be applied via StdSchedulerFactory.initialize(Properties)
    		}
    		// Otherwise, assume that externally provided factory has been initialized with appropriate settings
    		return schedulerFactory;
    	}
    }

initSchedulerFactory主要是将配置信息配置到schedulerFactory里

private void initSchedulerFactory(StdSchedulerFactory schedulerFactory) throws SchedulerException, IOException { 
    		Properties mergedProps = new Properties();
            ...    
    		CollectionUtils.mergePropertiesIntoMap(this.quartzProperties, mergedProps);
            ...
            //这里很重要，
            //可以直接通过 application.properties里的配置来配置quartz.properties里的配置spring.quartz.properties.xxx
    		schedulerFactory.initialize(mergedProps);
    	}

只所以看这里，是由于springboot的quartz整合官方文档实在是太过简单，只有寥寥数语

Quartz Scheduler configuration can be customized using spring.quartz properties and SchedulerFactoryBeanCustomizer beans, 
    which allow programmatic SchedulerFactoryBean customization.
    Advanced Quartz configuration properties can be customized using spring.quartz.properties.*.

翻译过来就是

> Quartz Scheduler可以通过使用spring.quartz 的properties 和写SchedulerFactoryBeanCustomizer这个类来达到客户化SchedulerFactoryBean 的配置。更进一步的配置可以通过spring.quartz.properties.\*

> 这样写连例子都没有，未免太过简略了。没有看源码根本不知道说的什么

prepareScheduler(prepareSchedulerFactory()) 在一些列的调用后会到StdSchedulerFactory的  
private Scheduler instantiate()。这是个很长的方法  
，但逻辑还算简单，各种初始化,下面只列出jobStore的配置

private Scheduler instantiate() throws SchedulerException { 
      //cfg 就是上个代码片段里的mergedProps
            if (cfg == null) { 
                initialize();
            }
            //jobstore,如果不配，默认是RAMJobStore
            //在application.properties里可以配置为
            //spring.quartz.properties.org.quartz.jobStore.class=org.quartz.impl.jdbcjobstore.JobStoreTX
               String jsClass = cfg.getStringProperty(PROP_JOB_STORE_CLASS,
                    RAMJobStore.class.getName());
        
        try { 
                js = (JobStore) loadHelper.loadClass(jsClass).newInstance();
            } catch (Exception e) { 
                initException = new SchedulerException("JobStore class '" + jsClass
                        + "' could not be instantiated.", e);
                throw initException;
            }
    
            SchedulerDetailsSetter.setDetails(js, schedName, schedInstId);
    
        //这里可以获取到spring.quartz.properties.org.quartz.jobStore.xxx
        //xxx是jobStore这个类的里属性，比如isClustered，clusterCheckinInterval等
        //在applicaiton.properties配置成spring.quartz.properties.org.quartz.jobStore.isClustered = true
            tProps = cfg.getPropertyGroup(PROP_JOB_STORE_PREFIX, true, new String[] { PROP_JOB_STORE_LOCK_HANDLER_PREFIX});
            try { 
                setBeanProps(js, tProps);
            } catch (Exception e) { 
                initException = new SchedulerException("JobStore class '" + jsClass
                        + "' props could not be configured.", e);
                throw initException;
            }
    
            
            
    }

因此我们要配置集群就可以在application.properties配置成如下：

spring.quartz.properties.org.quartz.jobStore.class=org.springframework.scheduling.quartz.LocalDataSourceJobStore
    spring.quartz.properties.org.quartz.jobStore.isClustered = true
    spring.quartz.properties.org.quartz.jobStore.clusterCheckinInterval = 10000
    spring.quartz.properties.org.quartz.scheduler.instanceId = AUTO

### Demo 1 ###

这个例子主要是Quartz的简单应用。Quartz api 写的很人性化，只要看下方法名就知道什么意思，几乎不需要查文档。  
需求 如下图需要能定制不同的执行计划  
![定期执行的选项][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b194bA_size_16_color_FFFFFF_t_70]

###### code ######

public static void main(String[] args) throws SchedulerException { 
            // Grab the Scheduler instance from the Factory
            Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
            // and start it off
            scheduler.start();
    
            // define the job and tie it to our MyJob class
            JobDetail job = newJob(MyJob.class)
                    .withIdentity("job1", "group1")
                    .build();
    
            //DAILY on12:00 START 【now】- END 【2019-3-1】
            GregorianCalendar calendar=new GregorianCalendar();
            calendar.set(2019,3,1);
            Date endTime=calendar.getTime();
            Trigger triggerDaily = newTrigger()
                    .withIdentity("triggerDaily", "group1")
                    .startNow()
                    .endAt(endTime)
                    .withSchedule(dailyAtHourAndMinute(12,0)).build();
             //DAILY exclude Sunday on 12:00 START 【now】- END 【2019-3-1】 
             Trigger triggerDaily2 = newTrigger()
                    .withIdentity("triggerDaily2", "group1")
                    .startNow()
                    .endAt(endTime)
                    .withSchedule(atHourAndMinuteOnGivenDaysOfWeek(12,0,2,3,4,5,6,7)).build();
            //WEELKY monday 12:00 START 【now】- END 【2019-3-1】
            Trigger triggerWeekly = newTrigger()
                    .withIdentity("triggerWeekly", "group1")
                    .startNow()
                    .endAt(endTime)
                    .withSchedule(weeklyOnDayAndHourAndMinute(DateBuilder.MONDAY,12,0)).build();
            //MONTHLY 31th 12:00 START 【now】- END 【2019-3-1】
            Trigger triggerMonthly = newTrigger()
                    .withIdentity("triggerMonthly", "group1")
                    .startNow()
                    .endAt(endTime)
                    .withSchedule(monthlyOnDayAndHourAndMinute(31,12,0)).build();
    
            // Tell quartz to schedule the job using our trigger
            //one job can only have one trigger
            //scheduler.scheduleJob(job, triggerDaily);
            //scheduler.scheduleJob(job, triggerDaily1);
            //scheduler.scheduleJob(job, triggerWeekly);
            scheduler.scheduleJob(job, triggerMonthly);
        }

### Demo 2 ###

##### 需求： #####

*  动态的增加，删除schedule任务。
 *  支持集群，只能集群里的一台机器执行。
 *  当其中一台机器宕机，其它机器要能继续执行。

通过springmvc来做这个例子。

###### 执行sql ######

在maven lib里找到org.quartz-scheduler:quartz:2.3.0里的tables\_mysql\_innodb.sql

###### Code ######

application.properies

server.port=9999
    spring.datasource.url=jdbc:mysql://127.0.0.1:3306/ssmdb?serverTimezone=UTC
    spring.datasource.username=root
    spring.datasource.password=devuser
    spring.datasource.driverClassName=com.mysql.cj.jdbc.Driver
    
    spring.quartz.job-store-type=jdbc
    #开启集群
    spring.quartz.properties.org.quartz.jobStore.isClustered = true
    spring.quartz.properties.org.quartz.jobStore.clusterCheckinInterval = 10000
    spring.quartz.properties.org.quartz.scheduler.instanceId = AUTO

@SpringBootApplication
    public class SchedualBootstrap { 
        private Logger logger = LoggerFactory.getLogger(this.getClass());
        public static void main(String[] args) { 
            SpringApplicationBuilder builder =new SpringApplicationBuilder(SchedualBootstrap.class);
            builder.run(args);
        }
    
       @Autowired
       Scheduler scheduler;
       
       //加入监听
       @PostConstruct
       public void init(){ 
                scheduler.getListenerManager().addSchedulerListener(new SchedulerListenerSupport(){ 
                    @Override
                    public void jobDeleted(JobKey jobKey) { 
                        logger.info(jobKey.getName()+" is deleted");
                    }
                    public void jobAdded(JobDetail jobDetail) { 
                        logger.info(jobDetail.getKey().getName()+" is add");
                    }
                    public void jobScheduled(Trigger trigger) { 
                        logger.info(trigger.getJobKey()+" is trigger");
                    }
                });
        }

public class JobBean { 
        private String jobName;
        private int intervalSecond;
        //...setter and getter
    }

rest 方式增加，减少schedual task

@RestController
    public class schedualController { 
        @Autowired
        Scheduler scheduler;
    
        @PostMapping("/schedual/add")
        public void addSchedual(@RequestBody JobBean jobBean){ 
            try { 
                JobDetail jobDetail = newJob(MyJob.class)
                        .withIdentity(jobBean.getJobName(), "group1")
                        .usingJobData("jobName", jobBean.getJobName())
                        .storeDurably()
                        .build();
                scheduler.addJob(jobDetail,true);
                // Trigger the job to run now, and then repeat every 2 seconds
                Trigger trigger = newTrigger()
                        .withIdentity("trigger-"+jobBean.getJobName(), "group1")
                        .forJob(jobDetail)
                        .startNow()
                        .withSchedule(simpleSchedule()
                                .withIntervalInSeconds(jobBean.getIntervalSecond())
                                .repeatForever())
                        .build();
    
                scheduler.scheduleJob(trigger);
            } catch (SchedulerException e) { 
                e.printStackTrace();
            }
        }
    
        @PostMapping("/schedual/{key}/delete")
        public void deleteSchedual(@PathVariable String key){ 
            JobKey jobKey = new JobKey(key,"group1");
            try { 
                if(scheduler.checkExists(jobKey)){ 
                    scheduler.deleteJob(jobKey);
                }
            } catch (SchedulerException e) { 
                e.printStackTrace();
            }
    
        }
    }

job

@DisallowConcurrentExecution
    public class MyJob implements org.quartz.Job{ 
    
        Logger logger = LoggerFactory.getLogger(this.getClass());
    
        private String JobName;
    
        public String getJobName() { 
            return JobName;
        }
    
        public void setJobName(String jobName) { 
            JobName = jobName;
        }
    
        public MyJob() { 
        }
    
        public void execute(JobExecutionContext context) throws JobExecutionException { 
            JobKey key = context.getJobDetail().getKey();
            logger.info(key.getName());
            logger.info(getJobName()+" : [Hello World! MyJob is executing.]");
        }
    }

###### How to run ######

1.  修改application.properties server.port=9999 启动
2.  修改application.properties server.port=9998 启动
3.  打开postman,http://127.0.0.1:9998/schedual/add  
    body

{
    	"jobName":"job1",
    	"intervalSecond":1
    }

> 增加1个job1 ,每1秒运行一次

1.  打开postman,http://127.0.0.1:9999/schedual/add  
    body

{
    	"jobName":"job2",
    	"intervalSecond":3
    }

> 增加1个job2 ,没3秒运行一次

看控制台输出,9998 运行job1,9999运行job2

5.关闭9999，会发现9998,运行job1和job2

[github code][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2d1b194bA_size_16_color_FFFFFF_t_70]: /images/20220404/2330181a442046898a9b036f097b7992.png
[github code]: https://github.com/rechardguo/spring-learn/tree/master/springboot/springboot-schedual