http://blog.itpub.NET/11627468/viewspace-1764753/
其中,QRTZ_LOCKS就是Quartz集群实现同步机制的行锁表,其表结构如下:
点击(此处)折叠或打开
- --QRTZ_LOCKS表结构
- CREATE TABLE `QRTZ_LOCKS` (
- `LOCK_NAME` varchar(40) NOT NULL,
- PRIMARY KEY (`LOCK_NAME`)
- ) ENGINE=InnoDB DEFAULT CHARSET=utf8;
- --QRTZ_LOCKS记录
- +-----------------+
- | LOCK_NAME |
- +-----------------+
- | CALENDAR_ACCESS |
- | JOB_ACCESS |
- | MISFIRE_ACCESS |
- | STATE_ACCESS |
- | TRIGGER_ACCESS |
- +-----------------+
注:此表结构在2.2版本有新增字段,这里暂时不考虑。
可以看出QRTZ_LOCKS中有5条记录,代表5把锁,分别用于实现多个Quartz Node对Job、Trigger、Calendar访问的同步控制。
关于行锁的机制:
1、mysql > set autocommit=0; //先把mysql设置为不自动提交。
2、 select * from es_locks where lock_name = 'TRIGGER_ACCESS' for update ; //线程一通过for update 可以把这行锁住
3、 select * from es_locks where lock_name = 'TRIGGER_ACCESS' for update ; //线程二通过for update 无法获得锁,线程等待。
4、commit; //线程一通过commit 释放锁
5、 //线程二可以访问到数据,线程不再等待。
所以,通过这个机制,一次只能有一个线程来操作 加锁 - 操作 - 释放锁。 如果 操作 的时间过长的话,会带来集群间的主线程等待。
数据库行锁是一种悲观锁,锁表时其它线程无法查询。
源码中关于数据库集群加锁的方法有如下几种:
1、executeInNonManagedTXLock方法的含义是自己管理事务,不让容器管理事务的加锁方法。
点击(此处)折叠或打开
- executeInNonManagedTXLock(
- String lockName,
- TransactionCallback<T> txCallback , final TransactionValidator<T> txValidator )
三个参数lockName的值是上面所说的TRIGGER_ACCESS,表示要加锁的类型。
txCallback是加锁后再回调的方法。
txValidator是验证方法,一般为null
函数先执行加锁,再回调要操作的方法,然后再解锁。
看一下源码:
点击(此处)折叠或打开
- if (lockName != null) {
- // If we aren't using db locks, then delay getting DB connection
- // until after acquiring the lock since it isn't needed.
- if (getLockHandler().requiresConnection()) {
- conn = getNonManagedTXConnection();
- }
- transOwner = getLockHandler().obtainLock(conn, lockName);
- }
- if (conn == null) {
- conn = getNonManagedTXConnection();
- }
- final T result = txCallback.execute(conn);
- try {
- commitConnection(conn);
- } catch (JobPersistenceException e) {
- rollbackConnection(conn);
- if (txValidator == null || !retryExecuteInNonManagedTXLock(lockName, new TransactionCallback<Boolean>() {
- @Override
- public Boolean execute(Connection conn) throws JobPersistenceException {
- return txValidator.validate(conn, result);
- }
- })) {
- throw e;
- }
- }
- Long sigTime = clearAndGetSignalSchedulingChangeOnTxCompletion();
- if(sigTime != null && sigTime >= 0) {
- signalSchedulingChangeImmediately(sigTime);
- }
- return result;
- } catch (JobPersistenceException e) {
- rollbackConnection(conn);
- throw e;
- } catch (RuntimeException e) {
rollbackConnection(conn);
throw new JobPersistenceException("Unexpected runtime exception: "
+ e.getMessage(), e);
} finally {
try {
releaseLock(lockName, transOwner);
} finally {
cleanupConnection(conn);
}
}
2、如果不是通过这种回调方法的加锁,一般是:
getLockHandler().obtainLock
执行
commitConnection(conn)
releaseLock
cleanupConnection
二、源码分析锁
1、TRIGGER_ACCESS
先了解一篇文章,通过源码来分析quartz是如何通过加锁来实现集群环境,触发器状态的一致性。
http://www.360doc.com/content/14/0926/08/15077656_412418636.shtml可以看到触发器的操作主要用主线程StdScheduleThread来完成,不管是获取需要触发的30S内的触发器,还是触发过程。select和update触发器表时
都会先加锁,后解锁。如果数据库资源竞争比较大的话,锁会影响整个性能。可以考虑将任务信息放在分布式内存,如redis上进行处理。数据库只是定时从redis上load数据下来做统计。
参考:quartz详解2:quartz由浅入深 查看第四章第1,2节
实现都在JobStoreSupport类
加锁类型 | 加锁方法 | 底层数据库操作 | 备注 |
executeInNonManagedTXLock | acquireNextTrigger | selectTriggerToAcquire selectTrigger selectJobDetail insertFiredTrigger |
查询需要点火的trigger 选择需要执行的trigger加入到fired_trigger表 |
for执行 triggerFired | selectJobDetail selectCalendar updateFiredTrigger triggerExists updateTrigger |
点火trigger 修改trigger状态为可执行状态。 |
|
recoverJobs | updateTriggerStatesFromOtherStates hasMisfiredTriggersInState doUpdateOfMisfiredTrigger selectTriggersForRecoveringJobs selectTriggersInState deleteFiredTriggers |
非集群环境下重新执行 failed与misfired的trigger |
|
retryExecuteInNonManagedTXLock | releaseAcquiredTrigger | updateTriggerStateFromOtherState deleteFiredTrigger |
异常情况下重新释放trigger到初使状态。 |
triggeredJobComplete | selectTriggerStatus removeTrigger updateTriggerState deleteFiredTrigger |
触发JOB任务完成后的处理。 | |
obtainLock | recoverMisfiredJobs | hasMisfiredTriggersInState doUpdateOfMisfiredTrigger | 重新执行misfired的trigger 可以在启动时执行,也可以由misfired线程定期执行。 |
clusterRecover | selectInstancesFiredTriggerRecords updateTriggerStatesForJobFromOtherState storeTrigger deleteFiredTriggers selectFiredTriggerRecords removeTrigger deleteSchedulerState |
集群有结点faied,让JOB能重新执行。 | |
executeInLock 数据库集群里等同于 executeInNonManagedTXLock |
storeJobAndTrigger | updateJobDetail insertJobDetail triggerExists selectJobDetail updateTrigger insertTrigger |
保存JOB和TRIGGER配置 |
storeJob | 保存JOB | ||
removeJob | 删除JOB | ||
removeJobs | 批量删除JOB | ||
removeTriggers | 批量删除triggers | ||
storeJobsAndTriggers | 保存JOB和多个trigger配置 | ||
removeTrigger | 删除trigger | ||
replaceTrigger | 替换trigger | ||
storeCalendar | 保存定时日期 | ||
removeCalendar | 删除定时日期 | ||
clearAllSchedulingData | 清除所有定时数据 | ||
pauseTrigger | 停止触发器 | ||
pauseJob | 停止任务 | ||
pauseJobs | 批量停止任务 | ||
resumeTrigger | 恢复触发器 | ||
resumeJob | 恢复任务 | ||
resumeJobs | 批量恢复任务 | ||
pauseTriggers | 批量停止触发器 | ||
resumeTriggers | 批量恢复触发器 | ||
pauseAll | 停止所有 | ||
resumeAll | 恢复所有 |
---
2、STATE_TRIGGER
实现都在JobStoreSupport类
加锁类型 | 加锁方法 | 底层数据库操作 | 备注 |
obtainLock | doCheckin | clusterCheckIn | 判断集群状态 先用LOCK_STATE_ACCESS锁集群状态 再用LOCK_TRIGGER_ACCESS恢复集群运行 |
---