例子

import org.quartz.*;
import org.quartz.impl.StdSchedulerFactory;import java.util.Date;/*** quartz定时器测试** @author leizhimin 2009-7-23 8:49:01*/
class MyJob implements Job {public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {System.out.println(new Date() + ": doing something...");}
}class Test {public static void main(String[] args) {//1、创建JobDetial对象JobDetail jobDetail = new JobDetail();//设置工作项jobDetail.setJobClass(MyJob.class);jobDetail.setName("MyJob_1");jobDetail.setGroup("JobGroup_1");//2、创建Trigger对象SimpleTrigger strigger = new SimpleTrigger();strigger.setName("Trigger_1");strigger.setGroup("Trigger_Group_1");strigger.setStartTime(new Date());//设置重复停止时间，并销毁该Trigger对象java.util.Calendar c = java.util.Calendar.getInstance();c.setTimeInMillis(System.currentTimeMillis() + 1000 * 1L);strigger.setEndTime(c.getTime());strigger.setFireInstanceId("Trigger_1_id_001");//设置重复间隔时间strigger.setRepeatInterval(1000 * 1L);//设置重复执行次数strigger.setRepeatCount(3);//3、创建Scheduler对象，并配置JobDetail和Trigger对象SchedulerFactory sf = new StdSchedulerFactory();Scheduler scheduler = null;try {scheduler = sf.getScheduler();scheduler.scheduleJob(jobDetail, strigger);//4、并执行启动、关闭等操作scheduler.start();} catch (SchedulerException e) {e.printStackTrace();}
//                try {
//                        //关闭调度器
//                        scheduler.shutdown(true);
//                } catch (SchedulerException e) {
//                        e.printStackTrace();
//                } }
}

先创建好JobDetail类的实例，这个实例主要是用于描述Job的详细信息，包括类，name和group是什么等等的用于描述这个job的相关信息。

接着创建了一个Trigger用于描述触发器的，执行时间啊，频次啊之类的。

Scheduler

这个类中主要涉及到了任务的调度了，那么总的来看的话，在创建一个Scheduler的时候，会创建QuartzSchedulerThread的线程池，然后这个线程池不断的取出trigger来判断这个是不是已经到了执行的时间了，然后会调用已经设置进来的job执行。

try {scheduler = sf.getScheduler();scheduler.scheduleJob(jobDetail, strigger);//4、并执行启动、关闭等操作scheduler.start();

调用这个scheduler的start方法、

public void start() throws SchedulerException {if (!this.shuttingDown && !this.closed) {if (this.initialStart == null) {this.initialStart = new Date();this.resources.getJobStore().schedulerStarted();this.startPlugins();}this.schedThread.togglePause(false);this.getLog().info("Scheduler " + this.resources.getUniqueIdentifier() + " started.");} else {throw new SchedulerException("The Scheduler cannot be restarted after shutdown() has been called.");}}

运行togglePause方法

void togglePause(boolean pause) {Object var2 = this.sigLock;synchronized(this.sigLock) {this.paused = pause;if (this.paused) {this.signalSchedulingChange(0L);} else {this.sigLock.notifyAll();}}}

这里调用了sigLock的notifyAll方法来运行。

再次之前已经创建了一个线程池，然后用于任务的调度。

public void run() {boolean lastAcquireFailed = false;while(!this.halted) {try {Object var2 = this.sigLock;synchronized(this.sigLock) {while(this.paused && !this.halted) {try {this.sigLock.wait(1000L);} catch (InterruptedException var28) {;}}if (this.halted) {break;}}

当刚才的方法notifyAll的时候，会激活线程去运行，同时传入的参数为false，然后，既然传入了false，在唤醒的时候，这个循环就会退出了。

int availTreadCount = this.qsRsrcs.getThreadPool().blockForAvailableThreads();if (availTreadCount > 0) {Trigger trigger = null;long now = System.currentTimeMillis();this.clearSignaledSchedulingChange();try {trigger = this.qsRsrcs.getJobStore().acquireNextTrigger(this.ctxt, now + this.idleWaitTime);lastAcquireFailed = false;} catch (JobPersistenceException var30) {。。。} catch (RuntimeException var31) {。。。}if (trigger == null) {long now = System.currentTimeMillis();long waitTime = now + this.getRandomizedIdleWaitTime();long timeUntilContinue = waitTime - now; // 继续等待Object var9 = this.sigLock;synchronized(this.sigLock) {try {this.sigLock.wait(timeUntilContinue);} catch (InterruptedException var16) {;}}} else {now = System.currentTimeMillis();long triggerTime = trigger.getNextFireTime().getTime();// 这里是一个for循环，然后遍历triggerTime时间不断的计算与当前时间的差值，如果当前差值>1可以休眠处理否则就退出循环for(long timeUntilTrigger = triggerTime - now; timeUntilTrigger > 0L; timeUntilTrigger = triggerTime - now) {Object var10 = this.sigLock;synchronized(this.sigLock) {try {now = System.currentTimeMillis();timeUntilTrigger = triggerTime - now;if (timeUntilTrigger >= 1L) {this.sigLock.wait(timeUntilTrigger);}} catch (InterruptedException var26) {;}}if (this.isScheduleChanged() && this.isCandidateNewTimeEarlierWithinReason(triggerTime)) {try {this.qsRsrcs.getJobStore().releaseAcquiredTrigger(this.ctxt, trigger);} catch (JobPersistenceException var24) {。。。。}trigger = null;break;}now = System.currentTimeMillis();}if (trigger != null) {TriggerFiredBundle bndle = null;boolean goAhead = true;Object var12 = this.sigLock;synchronized(this.sigLock) {goAhead = !this.halted;}if (goAhead) {try {bndle = this.qsRsrcs.getJobStore().triggerFired(this.ctxt, trigger);} catch (SchedulerException var21) {。。。。}}if (bndle == null) {try {this.qsRsrcs.getJobStore().releaseAcquiredTrigger(this.ctxt, trigger);} catch (SchedulerException var20) {。。。。}} else {var12 = null;JobRunShell shell;try {shell = this.qsRsrcs.getJobRunShellFactory().borrowJobRunShell();shell.initialize(this.qs, bndle);} catch (SchedulerException var29) {try {this.qsRsrcs.getJobStore().triggeredJobComplete(this.ctxt, trigger, bndle.getJobDetail(), 6);} catch (SchedulerException var19) {。。。。}continue;}if (!this.qsRsrcs.getThreadPool().runInThread(shell)) {// 真正执行在这里try {this.getLog().error("ThreadPool.runInThread() return false!");this.qsRsrcs.getJobStore().triggeredJobComplete(this.ctxt, trigger, bndle.getJobDetail(), 6);} catch (SchedulerException var18) {。。。。。}}}}}}} catch (RuntimeException var33) {this.getLog().error("Runtime error occured in main trigger firing loop.", var33);}}this.qs = null;this.qsRsrcs = null;}acquireNextTrigger(this.ctxt, now + this.idleWaitTime);lastAcquireFailed = false;} catch (JobPersistenceException var30) {。。。} catch (RuntimeException var31) {。。。}if (trigger == null) {long now = System.currentTimeMillis();long waitTime = now + this.getRandomizedIdleWaitTime();long timeUntilContinue = waitTime - now; // 继续等待Object var9 = this.sigLock;synchronized(this.sigLock) {try {this.sigLock.wait(timeUntilContinue);} catch (InterruptedException var16) {;}}} else {now = System.currentTimeMillis();long triggerTime = trigger.getNextFireTime().getTime();// 这里是一个for循环，然后遍历triggerTime时间不断的计算与当前时间的差值，如果当前差值>1可以休眠处理否则就退出循环for(long timeUntilTrigger = triggerTime - now; timeUntilTrigger > 0L; timeUntilTrigger = triggerTime - now) {Object var10 = this.sigLock;synchronized(this.sigLock) {try {now = System.currentTimeMillis();timeUntilTrigger = triggerTime - now;if (timeUntilTrigger >= 1L) {this.sigLock.wait(timeUntilTrigger);}} catch (InterruptedException var26) {;}}if (this.isScheduleChanged() && this.isCandidateNewTimeEarlierWithinReason(triggerTime)) {try {this.qsRsrcs.getJobStore().releaseAcquiredTrigger(this.ctxt, trigger);} catch (JobPersistenceException var24) {。。。。}trigger = null;break;}now = System.currentTimeMillis();}if (trigger != null) {TriggerFiredBundle bndle = null;boolean goAhead = true;Object var12 = this.sigLock;synchronized(this.sigLock) {goAhead = !this.halted;}if (goAhead) {try {bndle = this.qsRsrcs.getJobStore().triggerFired(this.ctxt, trigger);} catch (SchedulerException var21) {。。。。}}if (bndle == null) {try {this.qsRsrcs.getJobStore().releaseAcquiredTrigger(this.ctxt, trigger);} catch (SchedulerException var20) {。。。。}} else {var12 = null;JobRunShell shell;try {shell = this.qsRsrcs.getJobRunShellFactory().borrowJobRunShell();shell.initialize(this.qs, bndle);} catch (SchedulerException var29) {try {this.qsRsrcs.getJobStore().triggeredJobComplete(this.ctxt, trigger, bndle.getJobDetail(), 6);} catch (SchedulerException var19) {。。。。}continue;}if (!this.qsRsrcs.getThreadPool().runInThread(shell)) {// 真正执行在这里try {this.getLog().error("ThreadPool.runInThread() return false!");this.qsRsrcs.getJobStore().triggeredJobComplete(this.ctxt, trigger, bndle.getJobDetail(), 6);} catch (SchedulerException var18) {。。。。。}}}}}}} catch (RuntimeException var33) {this.getLog().error("Runtime error occured in main trigger firing loop.", var33);}}this.qs = null;this.qsRsrcs = null;}

以上代码基本上做的就是以下轮询(服务器启动后不断地执行run方法)：
qsRsrcs.getJobStore().acquireNextTriggers【查找即将触发的Trigger】 ---->
sigLock.wait(timeUntilTrigger)【等待执行】 ---->
qsRsrcs.getJobStore().triggersFired(triggers)【执行】---->

qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i)) 【释放Trigger】

这个地方其实是Scheduler线程来调度的，在while的死循环里面不断的进行检查的。

然后调用的默认JobStore用的是RAMJobStore，在这里面实际上使用的是TreeSet来存储的每一个触发器包装器。

 public TriggerFiredBundle triggerFired(SchedulingContext ctxt, Trigger trigger) {Object var3 = this.lock;synchronized(this.lock) {TriggerWrapper tw = (TriggerWrapper)this.triggersByFQN.get(TriggerWrapper.getTriggerNameKey(trigger));if (tw != null && tw.trigger != null) {if (tw.state != 1) {return null;} else {Calendar cal = null;if (tw.trigger.getCalendarName() != null) {cal = this.retrieveCalendar(ctxt, tw.trigger.getCalendarName());if (cal == null) {return null;}}Date prevFireTime = trigger.getPreviousFireTime();this.timeTriggers.remove(tw);tw.trigger.triggered(cal);trigger.triggered(cal);tw.state = 0;TriggerFiredBundle bndle = new TriggerFiredBundle(this.retrieveJob(ctxt, trigger.getJobName(), trigger.getJobGroup()), trigger, cal, false, new Date(), trigger.getPreviousFireTime(), prevFireTime, trigger.getNextFireTime());JobDetail job = bndle.getJobDetail();if (job.isStateful()) {ArrayList trigs = this.getTriggerWrappersForJob(job.getName(), job.getGroup());TriggerWrapper ttw;for(Iterator itr = trigs.iterator(); itr.hasNext(); this.timeTriggers.remove(ttw)) {ttw = (TriggerWrapper)itr.next();if (ttw.state == 0) {ttw.state = 5;}if (ttw.state == 4) {ttw.state = 6;}}this.blockedJobs.add(JobWrapper.getJobNameKey(job));} else if (tw.trigger.getNextFireTime() != null) {Object var16 = this.lock;synchronized(this.lock) {this.timeTriggers.add(tw);}}return bndle;}} else {return null;}}}

实际执行不在这里，这里只是做了一个封装。

在SimpleThreadPool里面执行

public boolean runInThread(Runnable runnable) {if (runnable == null) {return false;} else {Object var2 = this.nextRunnableLock;synchronized(this.nextRunnableLock) {this.handoffPending = true;while(this.availWorkers.size() < 1 && !this.isShutdown) {try {this.nextRunnableLock.wait(500L);} catch (InterruptedException var5) {;}}SimpleThreadPool.WorkerThread wt;if (!this.isShutdown) {wt = (SimpleThreadPool.WorkerThread)this.availWorkers.removeFirst();this.busyWorkers.add(wt);wt.run(runnable);} else {wt = new SimpleThreadPool.WorkerThread(this, this.threadGroup, "WorkerThread-LastJob", this.prio, this.isMakeThreadsDaemons(), runnable);this.busyWorkers.add(wt);this.workers.add(wt);wt.start();}this.nextRunnableLock.notifyAll();this.handoffPending = false;return true;}}}

获取一个线程来运行，然后这个任务会进行启动然后去执行。

Quartz源码解读-任务是如何定时执行的相关推荐

jpcsp源码解读10：指令的执行
这次要说的是处理器类: public class Processor 主要的成员变量: public CpuState cpu = new CpuState(); public static fina ...
Zeus源码解读之定时任务执行与手动执行任务的过程分析
Zeus源码解读之定时任务执行与手动执行任务的过程分析 zeus集群依赖任务执行模式宙斯中任务出去任务独立调度之外,支持任务直接的复杂依赖调度,如下图一所示: 图1 A为根任务,B,C依赖A任务 ...
KClient——kafka消息中间件源码解读
目录 kclient消息中间件 kclient-processor top.ninwoo.kclient.app.KClientApplication top.ninwoo.kclient.app.K ...
Redis 源码解读之 Rehash 的调用时机
Redis 源码解读之 Rehash 的调用时机背景和问题本文想要解决的问题什么时机触发 Rehash 操作? 什么时机实际执行 Rehash 函数? 结论什么时机触发 Rehash 操作? ...
k8s与日志--journalbeat源码解读 1
前言对于日志系统的重要性不言而喻,参照沪江的一篇关于日志系统的介绍,基本上日志数据在以下几方面具有非常重要的作用: 数据查找:通过检索日志信息,定位相应的 bug ,找出解决方案服务诊断:通过对 ...
并发编程之 Executor 线程池原理与源码解读
并发编程之 Executor 线程池原理与源码解读线程是调度 CPU 资源的最小单位,线程模型分为 KLT 模型与 ULT 模型,JVM使用的是 KLT 模型.java线程与 OS 线程保持 1:1 ...
Bert系列（二）——源码解读之模型主体
本篇文章主要是解读模型主体代码modeling.py.在阅读这篇文章之前希望读者们对bert的相关理论有一定的了解,尤其是transformer的结构原理,网上的资料很多,本文内容对原理部分就不做过多 ...
Bert系列（三）——源码解读之Pre-train
https://www.jianshu.com/p/22e462f01d8c pre-train是迁移学习的基础,虽然Google已经发布了各种预训练好的模型,而且因为资源消耗巨大,自己再预训练也不现 ...
linux下free源码,linux命令free源码解读：Procps free.c
linux命令free源码解读 linux命令free源码解读:Procps free.c 作者:isayme 发布时间:September 26, 2011 分类:Linux 我们讨论的是linux ...

Quartz源码解读-任务是如何定时执行的

例子

Scheduler

Quartz源码解读-任务是如何定时执行的相关推荐

最新文章

热门文章