quartz数据库集群锁机制

quartz数据库集群锁机制

一、quartz数据库锁

其中，QRTZ_LOCKS就是Quartz集群实现同步机制的行锁表，其表结构如下： 1. 2. 3. 4. 5. 6. 7. 8. 9. 10. 11. 12. 13. 14. 15. 16. 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 / 6

--QRTZ_LOCKS表结构

1、executeInNonManagedTXLock方法的含义是自己管理事务，不让容器管理事务的加锁方法。 1. 2. 3.

executeInNonManagedTXLock( String lockName,

TransactionCallback txCallback , final TransactionValidator 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() { @Override public Boolean execute(Connection conn) throws JobPersistenceException { return txValidator.validate(conn, result); } })) { throw e; } } 2 / 6

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

二、源码分析锁

目前代码中行锁只用到了STATE_ACCESS 和TRIGGER_ACCESS 这两种。

1、TRIGGER_ACCESS

触发器的操作主要用主线程StdScheduleThread来完成，不管是获取需要触发的30S内的触发器，还是触发过程。select

3 / 6

和update触发器表时都会先加锁，后解锁。如果数据库资源竞争比较大的话，锁会影响整个性能。可以考虑将任务信息放在分布式内存，如redis上进行处理。数据库只是定时从redis上load数据下来做统计。

实现都在JobStoreSupport类

加锁类型 加锁方法 底层数据库操作 selectTriggerToAcquire acquireNextTrigger selectTrigger selectJobDetail insertFiredTrigger selectJobDetail for执行 triggerFired executeInNonManagedTXLock selectCalendar updateFiredTrigger triggerExists updateTrigger updateTriggerStatesFromOtherStates hasMisfiredTriggersInState doUrecoverJobs pdateOfMisfiredTrigger selectTriggersForRecoveringJobs selectTriggersInState deleteFiredTriggers updateTriggerStateFromOtherStreleaseAcquiredTrigger ate deleteFiredTrigger retryExecuteInNonManagedTXLock triggeredJobComplete selectTriggerStatus removeTrigger updateTriggerS触发JOB任务完成后的处tate deleteFiredTrigger 重新执行misfired的recoverMisfiredJobs obtainLock hasMisfiredTriggersInState doUtrigger pdateOfMisfiredTrigger 可以在启动时执行，也可以由misfired线程定期执行。 clusterRecover selectInstancesFiredTriggerReco集群有结点faied，让JOB 4 / 6

备注 查询需要点火的trigger 选择需要执行的trigger加入到fired_trigger表 点火trigger 修改trigger状态为可执行状态。 非集群环境下重新执行 failedtrigger 与misfired的异常情况下重新释放trigger到初使状态。 理。

rds updateTriggerStatesForJobFromOtherState storeTrigger deleteFiredTriggers selectFiredTriggerRecords removeTrigger deleteSchedulerState updateJobDetail insertJobDetail storeJobAndTrigger triggerExists selectJobDetail updateTrigger insertTrigger storeJob removeJob removeJobs removeTriggers executeInLock 数据库集群里等同于 executeInNonManagedremoveTrigger TXLock replaceTrigger storeCalendar removeCalendar clearAllSchedulingData pauseTrigger pauseJob pauseJobs resumeTrigger 恢复触发器 批量停止任务 停止任务 停止触发器 清除所有定时数据 删除定时日期 保存定时日期 storeJobsAndTriggers 保存JOB和多个trigger配置 删除trigger 替换trigger 批量删除triggers 批量删除JOB 删除JOB 保存JOB 保存JOB和TRIGGER配置 能重新执行。 5 / 6

2、STATE_TRIGGER

实现都在JobStoreSupport类 加锁类型 加锁方法 底层数据库操作 备注 判断集群状态 obtainLock doCheckin clusterCheckIn 先用LOCK_STATE_ACCESS锁集群状态 再用LOCK_TRIGGER_ACCESS恢复集群运行

本文作者：tim_java

resumeTriggers pauseAll resumeAll 恢复所有 停止所有 resumeJobs pauseTriggers 批量恢复触发器 批量停止触发器 resumeJob 批量恢复任务 恢复任务 6 / 6