精进Quartz源码—Quartz中JobStore保存JonDetail和Trigger源码分析（一）

=================================================

对人工智能感兴趣的伙伴，分享一个我朋友的人工智能教程。零基础！通俗易懂！风趣幽默！大家可以看看是否对自己有帮助，点击这里查看教程。

=================================================

欢迎关注我的公众号： Java编程技术乐园。分享技术，一起精进Quartz!

做一个积极的人

编码、改bug、提升自己

我有一个乐园，面向编程，春暖花开！

我都是分析的jobStore 方式为jdbc的SimpleTrigger！RAM的方式类似分析方式！

说明：

• {0} ：表的前缀 ，如表qrtz_trigger ,{0}== qrtz_




• {1}：quartz.properties 中配置的 org.quartz.scheduler.instanceName: myInstanceName ，{1} ==myInstanceName

如果要使用定时器，那么正常企业级开发中一定是要将任务持久到数据库，那么我们一定明白JobDetail和Trigger是如何操作进入数据库，如何注册到Scheduler中！

一、从一个小Demo开始

那么下面就开始分析源码，分析源码前首先加一个小的demo，方面解释！

//1.创建Scheduler的工厂
SchedulerFactory sf = new StdSchedulerFactory();
//2.从工厂中获取调度器实例
Scheduler scheduler = sf.getScheduler();
//3.创建JobDetail
JobDetail jb = JobBuilder.newJob(RAMJob.class)
.withDescription("this is a ram job") //job的描述
.withIdentity("ramJob", "ramGroup") //job 的name和group
.build();
//任务运行的时间，SimpleSchedle类型触发器有效
long time= System.currentTimeMillis() + 3*1000L; //3秒后启动任务
Date statTime = new Date(time);
//4.创建Trigger
//使用SimpleScheduleBuilder或者CronScheduleBuilder
Trigger t = TriggerBuilder.newTrigger()
.withDescription("")
.withIdentity("ramTrigger", "ramTriggerGroup")
//.withSchedule(SimpleScheduleBuilder.simpleSchedule())
.startAt(statTime) //默认当前时间启动
.withSchedule(CronScheduleBuilder.cronSchedule("0/2 * * * * ?")) //两秒执行一次
.build();
//5.注册任务和定时器
scheduler.scheduleJob(jb, t);//源码分析
//6.启动 调度器
scheduler.start();


二、开始源码分析

主要分析入口 scheduleJob ，从这里进去，看看源码里面到底是怎么进行处理的！

//5.注册任务和定时器
scheduler.scheduleJob(jb, t);//源码分析


2.1、scheduler.scheduleJob(jb, t);

开始一个任务调度 。scheduler.scheduleJob(jb, t);//源码分析

/** *

* Calls the equivalent method on the 'proxied' QuartzScheduler. * 调用“代理”上的等效方法。 * * 实现 在 StdScheduler.scheduleJob(JobDetail jobDetail, Trigger trigger) *

2.2、ched.scheduleJob(jobDetail, trigger);

保存JobDetail和Trigger。ched.scheduleJob(jobDetail, trigger);//保存JobDetail和Trigger

/** *

* Add the {@link org.quartz.Job} identified by the given * {@link org.quartz.JobDetail} to the Scheduler, and * associate the given {@link org.quartz.Trigger} with it. *

2.3、 resources.getJobStore().storeJobAndTrigger(jobDetail, trig);

存储给定的org.quartz.JobDetail和org.quartz.Trigger。 resources.getJobStore().storeJobAndTrigger(jobDetail, trig);

public void storeJobAndTrigger(final JobDetail newJob,
final OperableTrigger newTrigger)
throws JobPersistenceException {

executeInLock(
(isLockOnInsert()) ? LOCK_TRIGGER_ACCESS : null,
new VoidTransactionCallback() {

public void executeVoid(Connection conn) throws JobPersistenceException {

//（1）保存JobDetail
storeJob(conn, newJob, false);
//（2）保存Trigger
storeTrigger(conn, newTrigger, newJob, false,
Constants.STATE_WAITING, false, false);
}
});
}


2.3.1 storeJob(conn, newJob, false);

保存JobDetail 。storeJob(conn, newJob, false);

首先判断JobDetail是否已经存在，存在则更新，否则插入！

protected void storeJob(Connection conn,
JobDetail newJob, boolean replaceExisting)
throws JobPersistenceException {

//判断JobDetail是否已经存在，根据name和group
boolean existingJob = jobExists(conn, newJob.getKey());
try {

if (existingJob) {

if (!replaceExisting) {

throw new ObjectAlreadyExistsException(newJob);
}
//更新JobDetail
getDelegate().updateJobDetail(conn, newJob);
"UPDATE {0}JOB_DETAILS SET DESCRIPTION = ?, JOB_CLASS_NAME = ?, IS_DURABLE = ?, IS_NOnCONCURRENT= ?, IS_UPDATE_DATA = ?, REQUESTS_RECOVERY = ?, JOB_DATA = ? WHERE SCHED_NAME = {1} AND JOB_NAME = ? AND JOB_GROUP = ?"
} else {

//插入JobDetail
getDelegate().insertJobDetail(conn, newJob);
"INSERT INTO {0}JOB_DETAILS (SCHED_NAME, JOB_NAME, JOB_GROUP, DESCRIPTION, JOB_CLASS_NAME, IS_DURABLE, IS_NONCONCURRENT, IS_UPDATE_DATA, REQUESTS_RECOVERY, JOB_DATA) VALUES({1}, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
}
} catch (IOException e) {

throw new JobPersistenceException("Couldn't store job: "
+ e.getMessage(), e);
} catch (SQLException e) {

throw new JobPersistenceException("Couldn't store job: "
+ e.getMessage(), e);
}
}


2.３.２　storeTrigger(conn, newTrigger, newJob, false,Constants.STATE_WAITING, false, false);

保存Trigger。 storeTrigger(conn, newTrigger, newJob, false,Constants.STATE_WAITING, false, false);

protected void storeTrigger(Connection conn,
OperableTrigger newTrigger, JobDetail job, boolean replaceExisting, String state,
boolean forceState, boolean recovering)
throws JobPersistenceException {

//判断Trigger是否已经存在，根据name和group
boolean existingTrigger = triggerExists(conn, newTrigger.getKey());
if ((existingTrigger) && (!replaceExisting)) {

throw new ObjectAlreadyExistsException(newTrigger);
}
try {

boolean shouldBepaused;
//进行一些状态的判断
if (!forceState) {

shouldBepaused = getDelegate().isTriggerGroupPaused(
conn, newTrigger.getKey().getGroup());
if(!shouldBepaused) {

shouldBepaused = getDelegate().isTriggerGroupPaused(conn,
ALL_GROUPS_PAUSED);
if (shouldBepaused) {

getDelegate().insertPausedTriggerGroup(conn, newTrigger.getKey().getGroup());
}
}
if (shouldBepaused && (state.equals(STATE_WAITING) || state.equals(STATE_ACQUIRED))) {

state = STATE_PAUSED;
}
}
//若job为null，重新获取！
if(job == null) {

job = getDelegate().selectJobDetail(conn, newTrigger.getJobKey(), getClassLoadHelper());
"SELECT * FROM {0}JOB_DETAILS WHERE SCHED_NAME = {1} AND JOB_NAME = ? AND JOB_GROUP = ?"
}
if (job == null) {

throw new JobPersistenceException("The job ("
+ newTrigger.getJobKey()
+ ") referenced by the trigger does not exist.");
}
//判断是否有DisallowConcurrentExecution注解，recovering恢复
if (job.isConcurrentExectionDisallowed() && !recovering) {

state = checkBlockedState(conn, job.getKey(), state);
"if (jobName != null) {

ps = conn.prepareStatement(rtp(SELECT_FIRED_TRIGGERS_OF_JOB));
"SELECT * FROM {0}FIRED_TRIGGERS WHERE SCHED_NAME = {1} AND JOB_NAME = ? AND JOB_GROUP = ?"
ps.setString(1, jobName);
ps.setString(2, groupName);
} else {

ps = conn
.prepareStatement(rtp(SELECT_FIRED_TRIGGERS_OF_JOB_GROUP));
"SELECT * FROM {0}FIRED_TRIGGERS WHERE SCHED_NAME = {1} AND JOB_GROUP = ?"
ps.setString(1, groupName);
}"
}
if (existingTrigger) {

//更新trigger
getDelegate().updateTrigger(conn, newTrigger, state, job);
" 大概贴出一些，具体想看完整的进入源码看
if(updateJobData) {

ps = conn.prepareStatement(rtp(UPDATE_TRIGGER));
"UPDATE {0}TRIGGERS SET JOB_NAME = ?, JOB_GROUP = ?, DESCRIPTION = ?, NEXT_FIRE_TIME = ?, PREV_FIRE_TIME = ?, TRIGGER_STATE = ?, TRIGGER_TYPE = ?, START_TIME = ?, END_TIME = ?, CALENDAR_NAME = ?, MISFIRE_INSTR = ?, PRIORITY = ?, JOB_DATA = ? WHERE SCHED_NAME = {1} AND TRIGGER_NAME = ? AND TRIGGER_GROUP = ?"
} else {

ps = conn.prepareStatement(rtp(UPDATE_TRIGGER_SKIP_DATA));
"UPDATE {0}TRIGGERS SET JOB_NAME = ?, JOB_GROUP = ?, DESCRIPTION = ?, NEXT_FIRE_TIME = ?, PREV_FIRE_TIME = ?, TRIGGER_STATE = ?, TRIGGER_TYPE = ?, START_TIME = ?, END_TIME = ?, CALENDAR_NAME = ?, MISFIRE_INSTR = ?, PRIORITY = ? WHERE SCHED_NAME = {1} AND TRIGGER_NAME = ? AND TRIGGER_GROUP = ?"
}
"
} else {

//插入trigger
getDelegate().insertTrigger(conn, newTrigger, state, job);
"INSERT INTO {0}TRIGGERS (SCHED_NAME, TRIGGER_NAME, TRIGGER_GROUP, JOB_NAME, JOB_GROUP, DESCRIPTION, NEXT_FIRE_TIME, PREV_FIRE_TIME, TRIGGER_STATE, TRIGGER_TYPE, START_TIME, END_TIME, CALENDAR_NAME, MISFIRE_INSTR, JOB_DATA, PRIORITY) VALUES({1}, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)"
}
} catch (Exception e) {

throw new JobPersistenceException("Couldn't store trigger '" + newTrigger.getKey() + "' for '"
+ newTrigger.getJobKey() + "' job:" + e.getMessage(), e);
}
}


到这里，保存就结束了！将任务保存成功后，就可以启动了，期待下一篇源码的分析：Scheduler的start()和QuartzSchedulerThread的run()！

三、参考资料

quartz2.2源码分析4-JobStore

Quartz专栏系列

1.精进Quartz——Quartz大致介绍（一）
2.精进Quartz——Quartz简单入门Demo（二）
3.精进Quartz——Spring和Quartz集成详解
4.精进Quartz——SSMM(Spring+SpringMVC+Mybatis+Mysql)和Quartz集成详解（四）
5.精进Quartz源码——JobStore保存JonDetail和Trigger源码分析（一）
6.精进Quartz源码——scheduler.start()启动源码分析（二）
7.精进Quartz源码——QuartzSchedulerThread.run() 源码分析（三）
8.精进Quartz源码——Quartz调度器的Misfire处理规则（四）

谢谢你的阅读，如果您觉得这篇博文对你有帮助，请点赞或者喜欢，让更多的人看到！祝你每天开心愉快！

不管做什么，只要坚持下去就会看到不一样！在路上，不卑不亢!

博客首页 : http://blog.csdn.net/u010648555

愿你我在人生的路上能都变成最好的自己，能够成为一个独挡一面的人

© 每天都在变得更好的阿飞云