quartz 运行机制分析
2017-07-13 11:13
260 查看
quartz主要由:scheduler,job,jobDetail,trigger,jobBuilder(定义和创建jobDetail实例的接口),triggerBuilder(定义和创建trigger) 组成。当job的一个trigger被触发后,job的 execute()方法会被scheduler的一个工作线程调用;传递给execute()方法的JobExecutionContext对象中保存着该job运行时的一些信息 -
执行job的scheduler的引用,触发job的trigger的引用,JobDetail对象引用,以及一些其它信息。JobDetail对象是在将job加入scheduler时,由客户端程序(你的程序)创建的。它包含job的各种属性设置,以及用于存储job实例状态信息的JobDataMap。Trigger用于触发Job的执行。一个Job可以对应多个Trigger, 但一个Trigger只能对应一个Job。当你准备调度一个job时,你创建一个Trigger的实例,然后设置调度相关的属性。Trigger也有一个相关联的JobDataMap,用于给Job传递一些触发相关的参数。Quartz自带了各种不同类型的Trigger,最常用的主要是SimpleTrigger和CronTrigger。
quartz运行原理,通过schedule线程管理它下面注册的trigger,不断循环执行筛选出需要执行的trigger,交付给线程池进行执行
quartz 运行机制分析:
1、通过schedulerFactory的getScheduler()方法创建scheduler,
2、scheduler.scheduleJob(jobDetail,trigger);将设置好的jobDetail 和 trigger 添加到schedule中,一个job可以对应多个trigger,但是一个trigger只能对应一个job
3、scheduler.start(); 主要是将scheduler 中变量paused 设为 FALSE,并唤起阻塞的scheduler线程
4、重点分析一下schedule.run()方法,方法比较长选取重要部分
quartz 的集群实现原理:
通过使用事务,对Lock_table 表中的数据使用 select... for update 的方式进行加锁。获得锁后,保存到threadLocal 变量中,执行job中的execute 方法。执行完成提交事务,释放锁 (remove threadLocal 中的锁变量)
quartz misfire(错过job的处理策略):
MISFIRE_INSTRUCTION_SMART_POLICY
智能根据trigger属性选择策略:
repeatCount为0,则策略同MISFIRE_INSTRUCTION_FIRE_NOW
repeatCount为REPEAT_INDEFINITELY,则策略同MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_REMAINING_COUNT
否则策略同MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_EXISTING_REPEAT_COUNT
MISFIRE_INSTRUCTION_FIRE_NOW
以当前时间为触发频率立即触发执行
MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_EXISTING_REPEAT_COUNT
以当前时间为触发频率立即触发执行
以总次数-已执行次数作为剩余周期次数,重新计算FinalTime
调整后的FinalTime会略大于根据starttime计算的到的FinalTime值
MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_REMAINING_REPEAT_COUNT
以当前时间为触发频率立即触发执行
保持FinalTime不变,重新计算剩余周期次数(相当于错过的当做已执行)
MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_EXISTING_COUNT
不触发立即执行
等待下次触发频率周期时刻执行
以总次数-已执行次数作为剩余周期次数,重新计算FinalTime
调整后的FinalTime会略大于根据starttime计算的到的FinalTime值
MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_REMAINING_COUNT
不触发立即执行
等待下次触发频率周期时刻,执行至FinalTime的剩余周期次数
保持FinalTime不变,重新计算剩余周期次数(相当于错过的当做已执行)
总结一下:
1、带有_NOW字样的策略,就是立即执行;反之,带有_NEXT的策略,则会等到下一个触发周期才会执行。
2、带有WITH_EXISTING_REPEAT_COUNT字样的,则是确保周期总数不变,用周期总数-已执行数作为剩余周期数,因此FinalTime会适当延后;
例如,repeatCount为3次(总计4次),已执行1次,错过2次,则后续仍会执行4-1=3次。
3、带有WITH_REMAINING_COUNT则是按原定计划执行,FinalTime不变,已错过的忽略。
例如,repeatCount为3次(总计4次),已执行1次,错过2次,则后续会执行4-1-2=1次。
执行job的scheduler的引用,触发job的trigger的引用,JobDetail对象引用,以及一些其它信息。JobDetail对象是在将job加入scheduler时,由客户端程序(你的程序)创建的。它包含job的各种属性设置,以及用于存储job实例状态信息的JobDataMap。Trigger用于触发Job的执行。一个Job可以对应多个Trigger, 但一个Trigger只能对应一个Job。当你准备调度一个job时,你创建一个Trigger的实例,然后设置调度相关的属性。Trigger也有一个相关联的JobDataMap,用于给Job传递一些触发相关的参数。Quartz自带了各种不同类型的Trigger,最常用的主要是SimpleTrigger和CronTrigger。
quartz运行原理,通过schedule线程管理它下面注册的trigger,不断循环执行筛选出需要执行的trigger,交付给线程池进行执行
quartz 运行机制分析:
1、通过schedulerFactory的getScheduler()方法创建scheduler,
public Scheduler getScheduler() throws SchedulerException { if (cfg == null) { initialize(); } //SchedulerRepository为一个hashmap用于保存所有的Scheduler SchedulerRepository schedRep = SchedulerRepository.getInstance(); //查看SchedulerRepository是否有该scheduler Scheduler sched = schedRep.lookup(getSchedulerName()); if (sched != null) { if (sched.isShutdown()) { schedRep.remove(getSchedulerName()); } else { return sched; } } //实例化一个Scheduler,过程主要包括 //1、读取配置属性 //2、创建线程池 //3、如果集群,配置DataSource相关信息 //4、启动 SchedulerPlugins //5、启动JobListeners,TriggerListeners //6、将以上所有的启动创建的实例都保存到 QuartzSchedulerResources 中,一个schedule对应一个QuartzSchedulerResources //7、创建 QuartzScheduler 实例 //8、通过 QuartzSchedulerResources 和 QuartzScheduler 创建 Scheduler,Scheduler作为一个thread会开始被线程池启动执行 sched = instantiate(); return sched; }
2、scheduler.scheduleJob(jobDetail,trigger);将设置好的jobDetail 和 trigger 添加到schedule中,一个job可以对应多个trigger,但是一个trigger只能对应一个job
public Date scheduleJob(JobDetail jobDetail, Trigger trigger) throws SchedulerException { validateState(); ... if (trigger.getJobKey() == null) { trig.setJobKey(jobDetail.getKey()); } else if (!trigger.getJobKey().equals(jobDetail.getKey())) { throw new SchedulerException( "Trigger does not reference given job!"); } trig.validate(); Calendar cal = null; if (trigger.getCalendarName() != null) { cal = resources.getJobStore().retrieveCalendar(trigger.getCalendarName()); } Date ft = trig.computeFirstFireTime(cal); if (ft == null) { throw new SchedulerException( "Based on configured schedule, the given trigger '" + trigger.getKey() + "' will never fire."); } //最重要的步骤,将jobDetail, trig 添加导QuartzSchedulerResources的jobStore中 //jobstore 中有个 timeTriggers 用于存储即将要执行的trigger resources.getJobStore().storeJobAndTrigger(jobDetail, trig); notifySchedulerListenersJobAdded(jobDetail); notifySchedulerThread(trigger.getNextFireTime().getTime()); notifySchedulerListenersSchduled(trigger); return ft; }
3、scheduler.start(); 主要是将scheduler 中变量paused 设为 FALSE,并唤起阻塞的scheduler线程
4、重点分析一下schedule.run()方法,方法比较长选取重要部分
public void run() { boolean lastAcquireFailed = false; while (!halted.get()) { try { // check if we're supposed to pause... synchronized (sigLock) { //支持对schedule进行暂停,退出操作 while (paused && !halted.get()) { try { // wait until togglePause(false) is called... sigLock.wait(1000L); } catch (InterruptedException ignore) { } } if (halted.get()) { break; } } int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads(); if(availThreadCount > 0) { // will always be true, due to semantics of blockForAvailableThreads... List<OperableTrigger> triggers = null; long now = System.currentTimeMillis(); clearSignaledSchedulingChange(); try { //从timeTriggers根据NextFireTime获取当前时刻需要执行的trigger,其中包括有misfire机制添加进来的trigger。 triggers = qsRsrcs.getJobStore().acquireNextTriggers( now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow()); lastAcquireFailed = false; ... if (triggers != null && !triggers.isEmpty()) { now = System.currentTimeMillis(); long triggerTime = triggers.get(0).getNextFireTime().getTime(); long timeUntilTrigger = triggerTime - now; //获取的触发器的触发时间较大,判断该trigger是否有必要执行 while(timeUntilTrigger > 2) { 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 >= 1) 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中的NextFireTime值,如果NextFireTime不为空则添加进timeTriggers中,作为下次筛选执行的依据 List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers); if(res != null) bndles = res; } ... //依次执行要trigger中的的job for (int i = 0; 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 { //构造要执行的jobRunShell线程 shell = qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle); shell.initialize(qs); } ... //将要执行的job放入线程池中去执行 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) } .... }
quartz 的集群实现原理:
通过使用事务,对Lock_table 表中的数据使用 select... for update 的方式进行加锁。获得锁后,保存到threadLocal 变量中,执行job中的execute 方法。执行完成提交事务,释放锁 (remove threadLocal 中的锁变量)
quartz misfire(错过job的处理策略):
MISFIRE_INSTRUCTION_SMART_POLICY
智能根据trigger属性选择策略:
repeatCount为0,则策略同MISFIRE_INSTRUCTION_FIRE_NOW
repeatCount为REPEAT_INDEFINITELY,则策略同MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_REMAINING_COUNT
否则策略同MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_EXISTING_REPEAT_COUNT
MISFIRE_INSTRUCTION_FIRE_NOW
以当前时间为触发频率立即触发执行
MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_EXISTING_REPEAT_COUNT
以当前时间为触发频率立即触发执行
以总次数-已执行次数作为剩余周期次数,重新计算FinalTime
调整后的FinalTime会略大于根据starttime计算的到的FinalTime值
MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_REMAINING_REPEAT_COUNT
以当前时间为触发频率立即触发执行
保持FinalTime不变,重新计算剩余周期次数(相当于错过的当做已执行)
MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_EXISTING_COUNT
不触发立即执行
等待下次触发频率周期时刻执行
以总次数-已执行次数作为剩余周期次数,重新计算FinalTime
调整后的FinalTime会略大于根据starttime计算的到的FinalTime值
MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_REMAINING_COUNT
不触发立即执行
等待下次触发频率周期时刻,执行至FinalTime的剩余周期次数
保持FinalTime不变,重新计算剩余周期次数(相当于错过的当做已执行)
总结一下:
1、带有_NOW字样的策略,就是立即执行;反之,带有_NEXT的策略,则会等到下一个触发周期才会执行。
2、带有WITH_EXISTING_REPEAT_COUNT字样的,则是确保周期总数不变,用周期总数-已执行数作为剩余周期数,因此FinalTime会适当延后;
例如,repeatCount为3次(总计4次),已执行1次,错过2次,则后续仍会执行4-1=3次。
3、带有WITH_REMAINING_COUNT则是按原定计划执行,FinalTime不变,已错过的忽略。
例如,repeatCount为3次(总计4次),已执行1次,错过2次,则后续会执行4-1-2=1次。
相关文章推荐
- LUA通过C语言运行时的错误处理机制分析
- LUA通过C语言运行时的错误处理机制分析
- IOS4的多任务机制概述,后台运行详细分析
- uIP TCP Server 运行机制分析
- Struts2运行机制(MVC)的分析:
- ASP.NET分析(一)--运行机制
- Windows程序内部运行机制简易分析
- java运行机制的分析
- vlc内部运行机制以及架构分析
- Intent与Intent的核心运行机制--通过SD卡插拔事件分析
- [转]分析Cache的运行机制和设计理念
- jQuery的运行机制和设计理念分析
- jQuery的运行机制和设计理念分析
- WCF学习笔记3(客户端内部运行机制分析)
- 分析CSLA.Net 4.* 开源框架的源码,深入理解框架内部运行机制
- 分析Cache的运行机制和设计理念
- java程序的运行机制详细分析
- vlc内部运行机制以及架构分析
- Erlang运行时源码分析之——线程进度机制
- uIP TCP Server 运行机制分析