在业务系统中有很多这样的场景:
账单日或者还款日上午 10 点,给每个信用卡客户发送账单通知,还款通知。如何判断客户的账单日、还款日,完成通知的发送?
银行业务系统,夜间要完成跑批的一系列流程,清理数据,下载文件,解析文件,对账清算、切换结算日期等等。如何触发一系列流程的执行?
金融机构跟人民银行二代支付系统对接,人民银行要求低于 5W 的金额(小额支付)半个小时打一次包发送,以缓解并发压力。所以,银行的跨行转账分成了多个流程:录入、复核、发送。如何把半个小时以内的所有数据一次性发送?
类似于这种基于准确的时刻或者固定的时间间隔触发的任务,或者有批量数据需要处理,或者要实现两个动作解耦的场景,我们都可以用任务调度来实现。
任务调度需求分析:
任务调度的实现方式有很多,如果要实现我们的调度需求,我们对这个工具有什么样的基本要求呢?
任务调度工具对比:
操作系统 Linux crontab:Windows 计划任务,只能执行简单脚本或者命令
数据库 MySQL、Oracle 可以操作数据。不能执行 Java 代码
工具 Kettle 可以操作数据,执行脚本。没有集中配置
开发语言 JDK Timer、ScheduledThreadPool Timer:单线程JDK1.5 之后:ScheduledThreadPool(Cache、Fiexed、Single):没有集中配置,日程管理不够灵活
容器 Spring Task、@Scheduled 不支持集群
分布式框架 XXL-JOB,Elastic-Job
其中JDK Timer的案例如下:
public class TestTimerTask extends TimerTask {
/**
* 此计时器任务要执行的操作。
*/
public void run() {
Date executeTime = new Date(this.scheduledExecutionTime());
String dateStr = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date());
System.out.println("任务执行了:" + dateStr);
}
}
public class TestTimer {
public static void main(String[] args) {
Timer timer = new Timer();
TimerTask task = new TestTimerTask();
timer.schedule(task, 5000L, 1000L);
}
}
@Scheduled 也是用 JUC 的 ScheduledExecutorService 实现的Scheduled(cron = “0 15 10 15 * ?”)
Quatz 是一个特性丰富的,开源的任务调度库,它几乎可以嵌入所有的 Java 程序,从很小的独立应用程序到大型商业系统。Quartz 可以用来创建成百上千的简单的或者复杂的任务,这些任务可以用来执行任何程序可以做的事情。Quartz 拥有很多企业级的特性,包括支持 JTA 事务和集群。Quartz 的目的就是让任务调度更加简单,开发人员只需要关注业务即可。他是用 Java 语言编写的(也有.NET 的版本)。Java 代码能做的任何事情,Quartz 都可以调度。特点:
Quzartz Java 编程:
1.引入依赖:
2.默认配置文件
org.quartz包下,有一个默认的配置文件,quartz.properties。当我们没有定义一个同名的配置文件的时候,就会使用默认配置文件里面的配置。
org.quartz.scheduler.instanceName: DefaultQuartzScheduler
org.quartz.scheduler.rmi.export: false
org.quartz.scheduler.rmi.proxy: false
org.quartz.scheduler.wrapJobExecutionInUserTransaction: false
//线程池
org.quartz.threadPool.class: org.quartz.simpl.SimpleThreadPool
org.quartz.threadPool.threadCount:
org.quartz.threadPool.threadPriority:
org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread: true
org.quartz.jobStore.misfireThreshold:
//内存持久化
org.quartz.jobStore.class: org.quartz.simpl.RAMJobStore
3.创建 Job
实现唯一的方法 execute(),方法中的代码就是任务执行的内容。此处仅输出字符串。
public class MyJob1 implements Job {
public void execute(JobExecutionContext context) throws JobExecutionException {
Date date = new Date();
SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
JobDataMap dataMap = context.getJobDetail().getJobDataMap();
System.out.println( " " + sf.format(date) + " 任务1执行了,"+dataMap.get("name"));
}
}
Job 进一步包装成 JobDetail。必须要指定 JobName 和 groupName,两个合起来是唯一标识符。可以携带 KV 的数据(JobDataMap),用于扩展属性,在运行的时候可以从 context获取到。
// JobDetail
JobDetail jobDetail = JobBuilder.newJob(MyJob1.class)
.withIdentity("myJob1", "group1")
.usingJobData("name", "wuzz")
.usingJobData("moon", 5.21F)
.build();
4 创建 Trigger
基于 SimpleTrigger 定义了一个每 2 秒钟运行一次、不断重复的 Trigger:
// Trigger
Trigger trigger = TriggerBuilder.newTrigger()
.withIdentity("trigger1", "group1")
.startNow()
.withSchedule(SimpleScheduleBuilder.simpleSchedule()
.withIntervalInSeconds()
.repeatForever())
.build();
5.创建 Scheduler
通过 Factory 获取调度器的实例,把 JobDetail 和 Trigger绑定,注册到容器中。Scheduler 先启动后启动无所谓,只要有 Trigger 到达触发条件,就会执行任务。
// SchedulerFactory
SchedulerFactory factory = new StdSchedulerFactory();
// Scheduler
Scheduler scheduler = factory.getScheduler();
// 绑定关系是1:N
scheduler.scheduleJob(jobDetail, trigger);
scheduler.start();
注意这里,调度器一定是单例的。
JobDetail : 我们创建一个实现 Job 接口的类,使用 JobBuilder 包装成 JobDetail,它可以携带KV 的数据。
Trigger : 定义任务的触发规律,Trigger,使用 TriggerBuilder 来构建。JobDetail 跟 Trigger 是 1:N 的关系。Trigger 接口在 Quartz 有 4 个继承的子接口:
Trigger dailyTimeIntervalTrigger = TriggerBuilder.newTrigger()
.withIdentity("trigger1", "group1")
.startNow()
.withSchedule(DailyTimeIntervalScheduleBuilder.dailyTimeIntervalSchedule()
.startingDailyAt(TimeOfDay.hourAndMinuteOfDay(, )) //第天9:00开始
.endingDailyAt(TimeOfDay.hourAndMinuteOfDay(, )) //16:00 结束
.onDaysOfTheWeek(,,,,) //周一至周五执行
.withIntervalInHours() //每间隔1小时执行一次
.withRepeatCount())//最多重复100次(实际执行100+1次)
.build();
上面我们定义的都是在什么时间执行,但是我们有一些在什么时间不执行的需求,比如:理财周末和法定假日购买不计息;证券公司周末和法定假日休市。是不是要把日期写在数据库中,然后读取基于当前时间判断呢?
基于 Calendar 的 排除规则
如果要在触发器的基础上,排除一些时间区间不执行任务,就要用到 Quartz 的Calendar 类(注意不是 JDK 的 Calendar)。可以按年、月、周、日、特定日期、Cron表达式排除。调用 Trigger 的 modifiedByCalendar()添加到触发器中,并且调用调度器的addCalendar()方法注册排除规则。
// 定义日历
AnnualCalendar holidays = new AnnualCalendar();// 排除中秋节
Calendar midAutumn = new GregorianCalendar(, , );
holidays.setDayExcluded(midAutumn, true);
// 排除圣诞节
Calendar christmas = new GregorianCalendar(, , );
holidays.setDayExcluded(christmas, true);
// 调度器添加日历
scheduler.addCalendar("holidays", holidays, false, false);
JobDetail jobDetail = JobBuilder.newJob(MyJob1.class)
.withIdentity("job1", "group1")
.usingJobData("name","wuzz")
.build();
Trigger trigger = TriggerBuilder.newTrigger()
.withIdentity("trigger1", "group1")
.startNow()
.modifiedByCalendar("holidays")
.withSchedule(SimpleScheduleBuilder.simpleSchedule()
.withIntervalInSeconds()
.repeatForever())
.build();
调度器,是 Quartz 的指挥官,由 StdSchedulerFactory 产生。它是单例的。并且是 Quartz 中最重要的 API,默认是实现类是 StdScheduler,里面包含了一个QuartzScheduler。QuartzScheduler 里面又包含了一个 QuartzSchedulerThread。
Scheduler 中的方法主要分为三大类:
操作调度器本身,例如调度器的启动 start()、调度器的关闭 shutdown()。
操作 Trigger,例如 pauseTriggers()、resumeTrigger()。
操作 Job,例如 scheduleJob()、unscheduleJob()、rescheduleJob()
这些方法非常重要,可以实现任务的动态调度。
我们有这么一种需求,在每个任务运行结束之后发送通知给运维管理员。那是不是要在每个任务的最后添加一行代码呢?这种方式对原来的代码造成了入侵,不利于维护。如果代码不是写在任务代码的最后一行,怎么知道任务执行完了呢?或者说,怎么监测到任务的生命周期呢?观察者模式:定义对象间一种一对多的依赖关系,使得每当一个对象改变状态,则所有依赖它的对象都会得到通知并自动更新。Quartz 中提供了三种 Listener,监听 Scheduler 的,监听 Trigger 的,监听 Job 的。只需要创建类实现相应的接口,并在 Scheduler 上注册 Listener,便可实现对核心对象的监听。
JobListener
TriggerListener
SchedulerListener
以 JobListener 为例:
public class MyJobListener implements JobListener {
public String getName() {
String name = getClass().getSimpleName();
System.out.println( "Method 111111 :"+ "获取到监听器名称:"+name);
return name;
}
public void jobToBeExecuted(JobExecutionContext context) {
String jobName = context.getJobDetail().getKey().getName();
System.out.println("Method 222222 :"+ jobName + " ——任务即将执行 ");
}
public void jobExecutionVetoed(JobExecutionContext context) {
String jobName = context.getJobDetail().getKey().getName();
System.out.println("Method 333333 :"+ jobName + " ——任务被否决 ");
}
public void jobWasExecuted(JobExecutionContext context, JobExecutionException jobException) {
String jobName = context.getJobDetail().getKey().getName();
System.out.println("Method 444444 :"+ jobName + " ——执行完毕 ");
System.out.println("------------------");
}
}
public static void main(String[] args) throws SchedulerException {
// JobDetail
JobDetail jobDetail = JobBuilder.newJob(MyJob1.class).withIdentity("job1", "group1").build();
// Trigger
Trigger trigger = TriggerBuilder.newTrigger().withIdentity("trigger1", "group1").startNow()
.withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds().repeatForever()).build();
// SchedulerFactory
SchedulerFactory factory = new StdSchedulerFactory();
// Scheduler
Scheduler scheduler = factory.getScheduler();
scheduler.scheduleJob(jobDetail, trigger);
// 创建并注册一个全局的Job Listener
scheduler.getListenerManager().addJobListener(new MyJobListener(), EverythingMatcher.allJobs());
scheduler.start();
}
工具类:ListenerManager,用于添加、获取、移除监听器。Matcher,主要是基于 groupName 和 keyName 进行匹配。
Jobstore 用来存储任务和触发器相关的信息,例如所有任务的名称、数量、状态等等。Quartz 中有两种存储任务的方式,一种在在内存,一种是在数据库。
RAMJobStore:
Quartz 默认的 JobStore 是 RAMJobstore,也就是把任务和触发器信息运行的信息存储在内存中,用到了 HashMap、TreeSet、HashSet 等等数据结构。如果程序崩溃或重启,所有存储在内存中的数据都会丢失。所以我们需要把这些数据持久化到磁盘。
JDBCJobStore:
JDBCJobStore 可以通过 JDBC 接口,将任务运行数据保存在数据库中。JDBC 的实现方式有两种,JobStoreSupport 类的两个子类:
JobStoreTX:在独立的程序中使用,自己管理事务,不参与外部事务。
JobStoreCMT:(Container Managed Transactions (CMT),如果需要容器管理事务时,使用它。
使用 JDBCJobSotre 时,需要配置数据库信息:
org.quartz.jobStore.class:org.quartz.impl.jdbcjobstore.JobStoreTX
org.quartz.jobStore.driverDelegateClass:org.quartz.impl.jdbcjobstore.StdJDBCDelegate
org.quartz.jobStore.useProperties:true
#数据库中 quartz 表的表名前缀
org.quartz.jobStore.tablePrefix:QRTZ_
org.quartz.jobStore.dataSource:myDS
#配置数据源
org.quartz.dataSource.myDS.driver:com.mysql.jdbc.Driver
org.quartz.dataSource.myDS.URL:jdbc:mysql://localhost:3306/quartz?useUnicode=true&characterEncoding=utf8
org.quartz.dataSource.myDS.user:root
org.quartz.dataSource.myDS.password:
org.quartz.dataSource.myDS.validationQuery=select from dual
问题来了?需要建什么表?表里面有什么字段?字段类型和长度是什么?在官网的 Downloads 链接中,提供了 11 张表的建表语句:quartz-2.2.3-distribution\quartz-2.2.3\docs\dbTables2.3 的版本在这个路径下:src\org\quartz\impl\jdbcjobstore
表名与作用:
传统的 Spring 方式集成,由于任务信息全部配置在 xml 文件中,如果需要操作任务或者修改任务运行频率,只能重新编译、打包、部署、重启,如果有紧急问题需要处理,会浪费很多的时间。有没有可以动态调度任务的方法?比如停止一个 Job?启动一个 Job?修改 Job 的触发频率?读取配置文件、写入配置文件、重启 Scheduler 或重启应用明显是不可取的。对于这种频繁变更并且需要实时生效的配置信息,我们可以放到哪里?可以存放于ZK、Redis、DB tables。并且,我们可以提供一个界面,实现对数据表的轻松操作。
这里我们用最简单的数据库的实现。问题 1:建一张什么样的表?参考 JobDetail 的属性。
CREATE TABLE `sys_job` (
`id` int() NOT NULL AUTO_INCREMENT COMMENT 'ID',
`job_name` varchar() NOT NULL COMMENT '任务名称',
`job_group` varchar() NOT NULL COMMENT '任务组名',
`job_cron` varchar() NOT NULL COMMENT '时间表达式',
`job_class_path` varchar() NOT NULL COMMENT '类路径,全类型',
`job_data_map` varchar() DEFAULT NULL COMMENT '传递 map 参数',
`job_status` int() NOT NULL COMMENT '状态:1 启用 0 停用',
`job_describe` varchar() DEFAULT NULL COMMENT '任务功能描述',
PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT= DEFAULT CHARSET=utf8;
操作数据表非常简单,SSM 增删改查。但是在修改了表的数据之后,怎么让调度器知道呢?调度器的接口:Scheduler。在我们的需求中,我们需要做的事情:
新增一个任务
删除一个任务
启动、停止一个任务
修改任务的信息(包括调度规律)
因 此 可 以 把 相 关 的 操 作 封 装 到 一 个 工 具 类 中 。我这里贴出来基本的增删改的几个方法:
public static void addJob(String jobClassName, String jobName, String jobGroupName, String cronExpression, String jobDataMap) throws Exception {
// 通过SchedulerFactory获取一个调度器实例
SchedulerFactory sf = new StdSchedulerFactory();
Scheduler scheduler = sf.getScheduler();
// 启动调度器
scheduler.start();
// 构建job信息
JobDetail jobDetail = JobBuilder.newJob(getClass(jobClassName).getClass())
.withIdentity(jobName, jobGroupName).build();
// JobDataMap用于传递任务运行时的参数,比如定时发送邮件,可以用json形式存储收件人等等信息
if (StringUtils.isNotEmpty(jobDataMap)) {
JSONObject jb = JSONObject.parseObject(jobDataMap);
Map
for (Map.Entry
jobDetail.getJobDataMap().put(m.getKey(), m.getValue());
}
}
// 表达式调度构建器(即任务执行的时间)
CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cronExpression);
// 按新的cronExpression表达式构建一个新的trigger
CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(jobName, jobGroupName)
.withSchedule(scheduleBuilder).startNow().build();
try {
scheduler.scheduleJob(jobDetail, trigger);
} catch (SchedulerException e) {
logger.info("创建定时任务失败" + e);
throw new Exception("创建定时任务失败");
}
}
/\*\*
\* 停用一个定时任务
\*
\* @param jobName 任务名称
\* @param jobGroupName 组别
\* @throws Exception
\*/
public static void jobPause(String jobName, String jobGroupName) throws Exception {
// 通过SchedulerFactory获取一个调度器实例
SchedulerFactory sf = new StdSchedulerFactory();
Scheduler scheduler = sf.getScheduler();
scheduler.pauseJob(JobKey.jobKey(jobName, jobGroupName));
}
/\*\*
\* 启用一个定时任务
\*
\* @param jobName 任务名称
\* @param jobGroupName 组别
\* @throws Exception
\*/
public static void jobresume(String jobName, String jobGroupName) throws Exception {
// 通过SchedulerFactory获取一个调度器实例
SchedulerFactory sf = new StdSchedulerFactory();
Scheduler scheduler = sf.getScheduler();
scheduler.resumeJob(JobKey.jobKey(jobName, jobGroupName));
}
/\*\*
\* 删除一个定时任务
\*
\* @param jobName 任务名称
\* @param jobGroupName 组别
\* @throws Exception
\*/
public static void jobdelete(String jobName, String jobGroupName) throws Exception {
// 通过SchedulerFactory获取一个调度器实例
SchedulerFactory sf = new StdSchedulerFactory();
Scheduler scheduler = sf.getScheduler();
scheduler.pauseTrigger(TriggerKey.triggerKey(jobName, jobGroupName));
scheduler.unscheduleJob(TriggerKey.triggerKey(jobName, jobGroupName));
scheduler.deleteJob(JobKey.jobKey(jobName, jobGroupName));
}
容器启动:因为任务没有定义在 ApplicationContext.xml 中,而是放到了数据库中,SpringBoot 启动时,怎么读取任务信息?或者,怎么在 Spring 启动完成的时候做一些事情?创建一个类,实现 CommandLineRunner 接口,实现 run 方法。从表中查出状态是 1 的任务,然后构建。
Service 类注入到 Job 中 : Spring Bean 如何注入到实现了 Job 接口的类中?例如在 TaskJob 中,需要注入 XXXXService(自定义的service),查询数据库进行操作。如果没有任何配置,注入会报空指针异常。
解决这个问题的步骤:
1.定义一个 AdaptableJobFactory,实现 JobFactory 接口,实现接口定义的newJob 方法,在这里面返回 Job 实例,具体实现可以参考 源码 org.springframework.scheduling.quartz.AdaptableJobFactory 类,直接拷贝就可以了。
2.定义一个 MyJobFactory,继承 AdaptableJobFactory。使用 Spring 的 AutowireCapableBeanFactory,把 Job 实例注入到容器中。
@Component
public class MyJobFactory extends AdaptableJobFactory {
@Autowired
private AutowireCapableBeanFactory capableBeanFactory;
protected Object createJobInstance(TriggerFiredBundle bundle) throws Exception {
//调用父类的方法
Object jobInstance = super.createJobInstance(bundle);
capableBeanFactory.autowireBean(jobInstance);
return jobInstance;
}
}
3.指定 Scheduler 的 JobFactory 为自定义的 JobFactory。
scheduler.setJobFactory(myJobFactory);
为什么 需要集群?
集群需要 解决的问题?
任务重跑,因为节点部署的内容是一样的,到 10 点的时候,每个节点都会执行相同的操作,引起数据混乱。比如跑批,绝对不能执行多次。
任务漏跑,假如任务是平均分配的,本来应该在某个节点上执行的任务,因为节点故障,一直没有得到执行。
水平集群需要注意时间同步问题
Quartz 使用的是随机的负载均衡算法,不能指定节点执行所以必须要有一种共享数据或者通信的机制。在分布式系统的不同节点中,我们可以采用什么样的方式,实现数据共享?两两通信,或者基于分布式的服务,实现数据共享。例如:ZK、Redis、DB。
在 Quartz 中,提供了一种简单的方式,基于数据库共享任务执行信息。也就是说,一个节点执行任务的时候,会操作数据库,其他的节点查询数据库,便可以感知到了。同样的问题:建什么表?哪些字段?依旧使用系统自带的 11 张表。
配置集群需要先配置如下信息,另外需要配置数据库连接持久化的信息 ,具体如上文所示。
#如果使用集群,instanceId必须唯一,设置成AUTO
org.quartz.scheduler.instanceId = AUTO
#是否使用集群
org.quartz.jobStore.isClustered = true
如果设置的数量>1,并且使用 JDBC JobStore(RAMJobStore 不支持分布式,只有 一 个 调 度 器 实 例 , 所 以 不 加 锁 ) , 则 属 性org.quartz.jobStore.acquireTriggersWithinLock 应设置为 true。否则不加锁会导致任务重复执行。
//代表 Scheduler 一次拉取trigger 的最大数量,默认是 1
org.quartz.scheduler.batchTriggerAcquisitionMaxCount=
org.quartz.jobStore.acquireTriggersWithinLock=true
Job 没有继承 Thread 和实现 Runnable,是怎么被调用的?通过反射还是什么?
任务是什么时候被调度的?是谁在监视任务还是监视 Trigger?
任务是怎么被调用的?谁执行了任务?
任务本身有状态吗?还是触发器有状态?
看源码的入口
Scheduler scheduler = factory.getScheduler();
scheduler.scheduleJob(jobDetail, trigger);
scheduler.start();
factory.getScheduler() 调用到 StdSchedulerFactory 的方法,如下:
public Scheduler getScheduler() throws SchedulerException {
if (cfg == null) {
// 读取 quartz.properties 配置文件
initialize();
}
// 这个类是一个 HashMap,用来基于调度器的名称保证调度器的唯一性
SchedulerRepository schedRep = SchedulerRepository.getInstance();
Scheduler sched = schedRep.lookup(getSchedulerName());
// 如果调度器已经存在了
if (sched != null) {
// 调度器关闭了,移除
if (sched.isShutdown()) {
schedRep.remove(getSchedulerName());
} else {
// 返回调度器
return sched;
}
}
// 调度器不存在,初始化
sched = instantiate();
return sched;
}
instantiate()方法中做了初始化的所有工作:
// 存储任务信息的 JobStore
JobStore js = null;
// 创建线程池,默认是 SimpleThreadPool
ThreadPool tp = null;
// 创建调度器
QuartzScheduler qs = null;
// 连接数据库的连接管理器
DBConnectionManager dbMgr = null;
// 自动生成 ID
String instanceIdGeneratorClass = null;
// 创建线程执行器,默认为 DefaultThreadExecutor
ThreadExecutor threadExecutor;
创建线程池 (包工头)即上面代码的 ThreadPool tp = null; 在该方法830行和839行,创建了一个线程池,默认是配置文件中指定的SimpleThreadPool。
String tpClass = cfg.getStringProperty(PROP_THREAD_POOL_CLASS, SimpleThreadPool.class.getName());
tp = (ThreadPool) loadHelper.loadClass(tpClass).newInstance();
SimpleThreadPool 里面维护了三个 list,分别存放所有的工作线程、空闲的工作线程和忙碌的工作线程。我们可以把 SimpleThreadPool 理解为包工头。
private List
private LinkedList
private LinkedList
tp 的 runInThread()方法是线程池运行线程的接口方法。参数 Runnable 是执行的任务内容。取出 WorkerThread 去执行参数里面的 runnable(JobRunShell)。
WorkerThread wt = (WorkerThread)availWorkers.removeFirst();
busyWorkers.add(wt);
wt.run(runnable);
WorkerThread (工人)是 SimpleThreadPool 的 内 部 类 , 用 来 执 行 任 务 。 我 们 把WorkerThread理解为工人。在WorkerThread的run方法中,执行传入的参数runnable任务:
runnable.run();
创建调度线程 (项目经理 )QuartzScheduler:
qs = new QuartzScheduler(rsrcs, idleWaitTime, dbFailureRetry);
在 QuartzScheduler 的构造函数中,创建了 QuartzSchedulerThread,我们把它理解为项目经理,它会调用包工头的工人资源,给他们安排任务。并 且 创 建 了 线 程 执 行 器 schedThreadExecutor , 执 行 了 这 个QuartzSchedulerThread,也就是调用了它的 run 方法。
// 创建一个线程,resouces 里面有线程名称
this.schedThread = new QuartzSchedulerThread(this, resources);
// 线程执行器
ThreadExecutor schedThreadExecutor = resources.getThreadExecutor();
//执行这个线程,也就是调用了线程的 run 方法
schedThreadExecutor.execute(this.schedThread);
点开 QuartzSchedulerThread 类,找到 run 方法,这个是 Quartz 任务调度的核心方法:
public void run() {
int acquiresFailed = ;
// 检查 scheuler 是否为停止状态
while (!halted.get()) {
try {
// check if we're supposed to pause…
synchronized (sigLock) {
// 检查是否为暂停状态
while (paused && !halted.get()) {
try {
// wait until togglePause(false) is called…
// 暂停的话会尝试去获得信号锁,并 wait 一会
sigLock.wait(1000L);
} catch (InterruptedException ignore) {
}
// reset failure counter when paused, so that we don't
// wait again after unpausing
acquiresFailed = ;
}
if (halted.get()) {
break;
}
}
// wait a bit, if reading from job store is consistently
// failing (e.g. DB is down or restarting)..
// 从 JobStore 获取 Job 持续失败,sleep 一下
if (acquiresFailed > ) {
try {
long delay = computeDelayForRepeatedErrors(qsRsrcs.getJobStore(), acquiresFailed);
Thread.sleep(delay);
} catch (Exception ignore) {
}
}
// 从线程池获取可用的线程
int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
if(availThreadCount > ) { // will always be true, due to semantics of blockForAvailableThreads...
List<OperableTrigger> triggers;
long now = System.currentTimeMillis();
clearSignaledSchedulingChange();
try {
// 获取需要下次执行的 triggers
// idleWaitTime: 默认 30s
// availThreadCount:获取可用(空闲)的工作线程数量,总会大于 1,因为该方法会一直阻塞,直到有工作线程空闲下来。
// maxBatchSize:一次拉取 trigger 的最大数量,默认是 1
// batchTimeWindow:时间窗口调节参数,默认是 0
// misfireThreshold: 超过这个时间还未触发的 trigger,被认为发生了 misfire,默认 60s
// 调度线程一次会拉取 NEXT\_FIRETIME 小于(now + idleWaitTime +batchTimeWindow),大于(now - misfireThreshold)的,min(availThreadCount,maxBatchSize)个 triggers,默认情况下,会拉取未来 30s、
过去 60s 之间还未 fire 的 个 trigger
triggers = qsRsrcs.getJobStore().acquireNextTriggers(
now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());
acquiresFailed = ;
if (log.isDebugEnabled())
log.debug("batch acquisition of " + (triggers == null ? : triggers.size()) + " triggers");
} catch (JobPersistenceException jpe) {
if (acquiresFailed == ) {
qs.notifySchedulerListenersError(
"An error occurred while scanning for the next triggers to fire.",
jpe);
}
if (acquiresFailed < Integer.MAX_VALUE)
acquiresFailed++;
continue;
} catch (RuntimeException e) {
if (acquiresFailed == ) {
getLog().error("quartzSchedulerThreadLoop: RuntimeException "
+e.getMessage(), e);
}
if (acquiresFailed < Integer.MAX_VALUE)
acquiresFailed++;
continue;
}
if (triggers != null && !triggers.isEmpty()) {
now = System.currentTimeMillis();
long triggerTime = triggers.get().getNextFireTime().getTime();
long timeUntilTrigger = triggerTime - now;
while(timeUntilTrigger > ) {
synchronized (sigLock) {
if (halted.get()) {
break;
}
if (!isCandidateNewTimeEarlierWithinReason(triggerTime, false)) {
try {
// we could have blocked a long while
// on 'synchronize', so we must recompute
now = System.currentTimeMillis();
timeUntilTrigger = triggerTime - now;
if(timeUntilTrigger >= )
sigLock.wait(timeUntilTrigger);
} catch (InterruptedException ignore) {
}
}
}
if(releaseIfScheduleChangedSignificantly(triggers, triggerTime)) {
break;
}
now = System.currentTimeMillis();
timeUntilTrigger = triggerTime - now;
}
// this happens if releaseIfScheduleChangedSignificantly decided to release triggers
if(triggers.isEmpty())
continue;
// set triggers to 'executing'
List<TriggerFiredResult> bndles = new ArrayList<TriggerFiredResult>();
boolean goAhead = true;
synchronized(sigLock) {
goAhead = !halted.get();
}
if(goAhead) {
try {
// 触发 Trigger,把 ACQUIRED 状态改成 EXECUTING
// 如果这个 trigger 的 NEXTFIRETIME 为空,也就是未来不再触发,就将其状态改为COMPLETE
// 如果trigger不允许并发执行(即Job的实现类标注了@DisallowConcurrentExecution),则将状态变为 BLOCKED,否则就将状态改为 WAITING
List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);
if(res != null)
bndles = res;
} catch (SchedulerException se) {
qs.notifySchedulerListenersError(
"An error occurred while firing triggers '"
+ triggers + "'", se);
//QTZ-179 : a problem occurred interacting with the triggers from the db
//we release them and loop again
// 循环处理 Trigger
for (int i = ; i < triggers.size(); i++) {
qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
}
continue;
}
}
for (int i = ; i < bndles.size(); i++) {
TriggerFiredResult result = bndles.get(i);
TriggerFiredBundle bndle = result.getTriggerFiredBundle();
Exception exception = result.getException();
if (exception instanceof RuntimeException) {
getLog().error("RuntimeException while firing trigger " + triggers.get(i), exception);
qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
continue;
}
// it's possible to get 'null' if the triggers was paused,
// blocked, or other similar occurrences that prevent it being
// fired at this time... or if the scheduler was shutdown (halted)
if (bndle == null) {
qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
continue;
}
JobRunShell shell = null;
try {
// 根据 trigger 信息实例化 JobRunShell(implements Runnable),同时依据JOB\_CLASS\_NAME 实例化 Job,随后我们将 JobRunShell 实例丢入工作线
shell = qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle);
shell.initialize(qs);
} catch (SchedulerException se) {
qsRsrcs.getJobStore().triggeredJobComplete(triggers.get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET\_ALL\_JOB\_TRIGGERS\_ERROR);
continue;
}
// 执行 JobRunShell 的 run 方法
if (qsRsrcs.getThreadPool().runInThread(shell) == false) {
// this case should never happen, as it is indicative of the
// scheduler being shutdown or a bug in the thread pool or
// a thread pool being used concurrently - which the docs
// say not to do...
getLog().error("ThreadPool.runInThread() return false!");
qsRsrcs.getJobStore().triggeredJobComplete(triggers.get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET\_ALL\_JOB\_TRIGGERS\_ERROR);
}
}
continue; // while (!halted)
}
} else { // if(availThreadCount > 0)
// should never happen, if threadPool.blockForAvailableThreads() follows contract
continue; // while (!halted)
}
long now = System.currentTimeMillis();
long waitTime = now + getRandomizedIdleWaitTime();
long timeUntilContinue = waitTime - now;
synchronized(sigLock) {
try {
if(!halted.get()) {
// QTZ-336 A job might have been completed in the mean time and we might have
// missed the scheduled changed signal by not waiting for the notify() yet
// Check that before waiting for too long in case this very job needs to be
// scheduled very soon
if (!isScheduleChanged()) {
sigLock.wait(timeUntilContinue);
}
}
} catch (InterruptedException ignore) {
}
}
} catch(RuntimeException re) {
getLog().error("Runtime error occurred in main trigger firing loop.", re);
}
} // while (!halted)
// drop references to scheduler stuff to aid garbage collection...
qs = null;
qsRsrcs = null;
}
JobRunShell 的作用:JobRunShell 用来为 Job 提供安全的运行环境的,执行 Job 中所有的作业,捕获运行中的异常,在任务执行完毕的时候更新 Trigger 状态,等等。JobRunShell 实例是用 JobRunShellFactory 为 QuartzSchedulerThread 创建的,在调度器决定一个 Job 被触发的时候,它从线程池中取出一个线程来执行任务。
线程模型总结:
SimpleThreadPool:包工头,管理所有 WorkerThread
WorkerThread:工人,把 Job 包装成 JobRunShell,执行
QuartSchedulerThread:项目经理,获取即将触发的 Trigger,从包工头出拿到worker,执行 Trigger 绑定的任务
绑定 l JobDetail 和 和 Trigger(scheduler.scheduleJob(jobDetail, trigger);)
// 存储 JobDetail 和 Trigger
resources.getJobStore().storeJobAndTrigger(jobDetail, trig);
// 通知相关的 Listener
notifySchedulerListenersJobAdded(jobDetail);
notifySchedulerThread(trigger.getNextFireTime().getTime());
notifySchedulerListenersSchduled(trigger);
启动调度器scheduler.start();
public void start() throws SchedulerException {
if (shuttingDown|| closed) {
throw new SchedulerException(
"The Scheduler cannot be restarted after shutdown() has been called.");
}
// QTZ-212 : calling new schedulerStarting() method on the listeners
// right after entering start()
// 通知监听器
notifySchedulerListenersStarting();
if (initialStart == null) {
initialStart = new Date();
this.resources.getJobStore().schedulerStarted();
startPlugins();
} else {
resources.getJobStore().schedulerResumed();
}
// 通知 QuartzSchedulerThread 不再等待,开始干活
schedThread.togglePause(false);
getLog().info(
"Scheduler " + resources.getUniqueIdentifier() + " started.");
notifySchedulerListenersStarted();
}
getScheduler 方法创建线程池 ThreadPool,创建调度器 QuartzScheduler,创建调度线程 QuartzSchedulerThread,调度线程初始处于暂停状态。scheduleJob 将任务添加到 JobStore 中。scheduler.start()方法激活调度器,QuartzSchedulerThread 从 timeTrriger 取出待触发的任务,并包装成 TriggerFiredBundle,然后由 JobRunShellFactory 创建TriggerFiredBundle 的 执 行 线 程 JobRunShell , 调 度 执 行 通 过 线 程 池SimpleThreadPool去执行JobRunShell,而JobRunShell执行的就是任务类的execute方法:job.execute(JobExecutionContext context)。
1、一个任务10秒钟触发一次,但是每次执行需要60秒,在第20秒的时候,会同时运行两个任务吗?怎么禁止一个任务并发运行?需要注解 @DisallowConcurrentExecution
@DisallowConcurrentExecution
public class MyJob4 implements Job {
public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
Date date = new Date();
SimpleDateFormat sf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
try {
System.out.println(" " + sf.format(date) + " 任务4开始执行了,请等待3秒");
TimeUnit.SECONDS.sleep();
} catch (InterruptedException e) {
e.printStackTrace();
}
}
}
2、现在有三个任务,任务A、任务B、任务C,怎么让多个任务串行执行,例如A执行完了之后再执行B,B执行完了再执行C?
如果仅仅只是串行的话,可以把三个任务放在同一个任务当中。
3、除了执行本地的代码之外,怎么调用其他系统的任务?
鉴权的HTTP,RPC
4、任务在什么时候会错过触发?错过触发怎么办?
没有可用线程,Trigger 被暂停,系统重启,禁止并发执行的任务在到达触发时间时,上次执行还没有结束。
忽略,立即跑一次,下次跑。
每一种 Trigger 都定义了自己的 Misfire 策略,不同的策略通过不同的方法来设置。
// Trigger1
Trigger trigger = TriggerBuilder.newTrigger().withIdentity("trigger1", "group1").startNow()
.withSchedule(SimpleScheduleBuilder.simpleSchedule().
withMisfireHandlingInstructionNowWithExistingCount().
withIntervalInSeconds().
repeatForever()).build();
手机扫一扫
移动阅读更方便
你可能感兴趣的文章