北京建设工程信息网官网入口,东莞网络优化专业乐云seo,wordpress登陆图标修改,网站信息建设总结一、问题背景 美团CRM系统中每天有大量的后台任务需要调度执行#xff0c;如构建索引、统计报表、周期同步数据等等#xff0c;要求任务调度系统具备高可用性、负载均衡特性#xff0c;可以管理并监控任务的执行流程#xff0c;以保证任务的正确执行。 二、历史方案 美团CR… 一、问题背景 美团CRM系统中每天有大量的后台任务需要调度执行如构建索引、统计报表、周期同步数据等等要求任务调度系统具备高可用性、负载均衡特性可以管理并监控任务的执行流程以保证任务的正确执行。 二、历史方案 美团CRM系统的任务调度模块经历了以下历史方案。 1. CrontabSQL 每天晚上运行定时任务通过SQL脚本crontab方式执行例如 #crm
0 2 * * * /xxx/mtcrm/shell/mtcrm_daily_stat.sql //每天凌晨2:00执行统计
30 7 * * * /xxx/mtcrm/shell/mtcrm_data_fix.sql //每天早上7:30执行数据修复该方案存在以下问题 直接访问数据库各系统业务接口没有重用。完成复杂业务需求时会引入过多中间表。业务逻辑计算完全依赖SQL增大数据库压力。任务失败无法自动恢复。2. PythonSQL 采用python脚本(多数据源)SQL方式执行例如 def connectCRM():return MySQLdb.Connection(host1, uname, xxx, crm, 3306, charsetutf8)def connectTemp():return MySQLdb.Connection(host1, uname, xxx, temp, 3306, charsetutf8)该方案存在问题 直接访问数据需要理解各系统的数据结构无法满足动态任务问题各系统业务接口没有重用。无负载均衡。任务失败无法恢复。在JAVA语言开发中出现异构且很难统一到自动部署系统中。3. SpringJDK Timer 该方案使用springJDK Timer方式调用接口完成定时任务在分布式部署环境下防止多个节点同时运行任务需要写死host控制在一台服务器上执行task。 bean idaccountStatusTaskScanner classxxx.crm.service.impl.AccountStatusTaskScanner /task:scheduler idtaskScheduler pool-size5 /task:scheduled-tasks schedulertaskSchedulertask:scheduled refaccountStatusTaskScanner methodexecute cron0 0 1 * * ? /
/task:scheduled-tasks该方案较方案12有很大改进但仍存在以下问题 步骤复杂、分散任务量增大的情况下很难扩展使用写死服务器Host的方式执行task存在单点风险负载均衡手动完成。应用重启任务无法自动恢复。CRM系统定时任务走过了很多弯路定时任务多种实现方式使配置和代码分散在多处难以维护和监控任务执行过程没有保证没有错误恢复任务执行异常没有反馈(邮件)没有集群支持、负载均衡。CRM系统需要分布式的任务调度框架统一解决问题Java可以使用的任务调度框架有QuartzJcrontabcron4j我们选择了Quartz。 三、为什么选择Quartz Quartz是Java领域最著名的开源任务调度工具。Quartz提供了极为广泛的特性如持久化任务集群和分布式任务等其特点如下 完全由Java写成方便集成(Spring)伸缩性负载均衡高可用性四、Quartz集群部署实践 CRM中Quartz与Spring结合使用Spring通过提供org.springframework.scheduling.quartz下的封装类对Quartz支持。 Quartz集群部署 Quartz集群中的每个节点是一个独立的Quartz应用它又管理着其他的节点。该集群需要分别对每个节点分别启动或停止不像应用服务器的集群独立的Quartz节点并不与另一个节点或是管理节点通信。Quartz应用是通过数据库表来感知到另一应用。只有使用持久的JobStore才能完成Quqrtz集群。 基于Spring的集群配置 !-- 调度工厂 --
bean idquartzSchedulerclassorg.springframework.scheduling.quartz.SchedulerFactoryBeanproperty namedataSource refdataSource /property namequartzPropertiespropsprop keyorg.quartz.scheduler.instanceNameCRMscheduler/propprop keyorg.quartz.scheduler.instanceIdAUTO/prop!-- 线程池配置 --prop keyorg.quartz.threadPool.classorg.quartz.simpl.SimpleThreadPool/propprop keyorg.quartz.threadPool.threadCount20/propprop keyorg.quartz.threadPool.threadPriority5/prop!-- JobStore 配置 --prop keyorg.quartz.jobStore.classorg.quartz.impl.jdbcjobstore.JobStoreTX/prop!-- 集群配置 --prop keyorg.quartz.jobStore.isClusteredtrue/propprop keyorg.quartz.jobStore.clusterCheckinInterval15000/propprop keyorg.quartz.jobStore.maxMisfiresToHandleAtATime1/propprop keyorg.quartz.jobStore.misfireThreshold120000/propprop keyorg.quartz.jobStore.tablePrefixQRTZ_/prop/props/propertyproperty nameschedulerName valueCRMscheduler /!--必须的QuartzScheduler 延时启动应用启动完后 QuartzScheduler 再启动 --property namestartupDelay value30 /property nameapplicationContextSchedulerContextKey valueapplicationContextKey /!--可选QuartzScheduler 启动时更新己存在的Job这样就不用每次修改targetObject后删除qrtz_job_details表对应记录了 --property nameoverwriteExistingJobs valuetrue /!-- 设置自动启动 --property nameautoStartup valuetrue /!-- 注册触发器 --property nametriggerslistref beanuserSyncScannerTrigger /....../list/property!-- 注册jobDetail --property namejobDetailslist/list/propertyproperty nameschedulerListenerslistref beanquartzExceptionSchedulerListener //list/property
/beanorg.quartz.jobStore.class属性为JobStoreTX将任务持久化到数据中。因为集群中节点依赖于数据库来传播Scheduler实例的状态你只能在使用JDBC JobStore时应用Quartz集群。org.quartz.jobStore.isClustered属性为true通知Scheduler实例要它参与到一个集群当中。org.quartz.jobStore.clusterCheckinInterval属性定义了Scheduler实例检入到数据库中的频率(单位毫秒)。Scheduler检查是否其他的实例到了它们应当检入的时候未检入这能指出一个失败的Scheduler实例且当前 Scheduler会以此来接管任何执行失败并可恢复的Job。通过检入操作Scheduler 也会更新自身的状态记录。clusterChedkinInterval越小Scheduler节点检查失败的Scheduler实例就越频繁。默认值是 15000 (即15 秒)。其余参数在后文将会详细介绍。Quartz监控 CRM后台目前可以做到对Quartz实例的监控、操作以及动态部署Trigger. Triggers监控 JobDetails监控 五、Quartz集群原理分析 1. Quartz集群数据库表 Quartz的集群部署方案在架构上是分布式的没有负责集中管理的节点而是利用数据库锁的方式来实现集群环境下进行并发控制。BTW分布式部署时需要保证各个节点的系统时间一致。 Quartz数据库核心表如下 Table NameDescriptionQRTZ_CALENDARS存储Quartz的Calendar信息QRTZ_CRON_TRIGGERS存储CronTrigger包括Cron表达式和时区信息QRTZ_FIRED_TRIGGERS存储与已触发的Trigger相关的状态信息以及相联Job的执行信息QRTZ_PAUSED_TRIGGER_GRPS存储已暂停的Trigger组的信息QRTZ_SCHEDULER_STATE存储少量的有关Scheduler的状态信息和别的Scheduler实例QRTZ_LOCKS存储程序的悲观锁的信息QRTZ_JOB_DETAILS存储每一个已配置的Job的详细信息QRTZ_JOB_LISTENERS存储有关已配置的JobListener的信息QRTZ_SIMPLE_TRIGGERS存储简单的Trigger包括重复次数、间隔、以及已触的次数QRTZ_BLOG_TRIGGERSTrigger作为Blob类型存储QRTZ_TRIGGER_LISTENERS存储已配置的TriggerListener的信息QRTZ_TRIGGERS存储已配置的Trigger的信息其中QRTZ_LOCKS就是Quartz集群实现同步机制的行锁表其表结构如下 --QRTZ_LOCKS表结构
CREATE TABLE QRTZ_LOCKS (LOCK_NAME varchar(40) NOT NULL,PRIMARY KEY (LOCK_NAME)
) ENGINEInnoDB DEFAULT CHARSETutf8;--QRTZ_LOCKS记录
-----------------
| LOCK_NAME |
-----------------
| CALENDAR_ACCESS |
| JOB_ACCESS |
| MISFIRE_ACCESS |
| STATE_ACCESS |
| TRIGGER_ACCESS |
-----------------可以看出QRTZ_LOCKS中有5条记录代表5把锁分别用于实现多个Quartz Node对Job、Trigger、Calendar访问的同步控制。 2. Quartz线程模型 在Quartz中有两类线程Scheduler调度线程和任务执行线程。*任务执行线程*Quartz不会在主线程(QuartzSchedulerThread)中处理用户的Job。Quartz把线程管理的职责委托给ThreadPool一般的设置使用SimpleThreadPool。SimpleThreadPool创建了一定数量的WorkerThread实例来使得Job能够在线程中进行处理。WorkerThread是定义在SimpleThreadPool类中的内部类它实质上就是一个线程。例如CRM中配置如下 !-- 线程池配置 --prop keyorg.quartz.threadPool.classorg.quartz.simpl.SimpleThreadPool/propprop keyorg.quartz.threadPool.threadCount20/propprop keyorg.quartz.threadPool.threadPriority5/prop*QuartzSchedulerThread调度主线程*QuartzScheduler被创建时创建一个QuartzSchedulerThread实例。 3. 集群源码分析 Quartz究竟是如何保证集群情况下trgger处理的信息同步 下面跟着源码一步一步分析QuartzSchedulerThread包含有决定何时下一个Job将被触发的处理循环主要逻辑在其run()方法中 public void run() {boolean lastAcquireFailed false;while (!halted.get()) {......int availThreadCount qsRsrcs.getThreadPool().blockForAvailableThreads();if(availThreadCount 0) { ......//调度器在trigger队列中寻找30秒内一定数目的trigger(需要保证集群节点的系统时间一致)triggers qsRsrcs.getJobStore().acquireNextTriggers(now idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());......//触发triggerListTriggerFiredResult res qsRsrcs.getJobStore().triggersFired(triggers);......//释放triggerfor (int i 0; i triggers.size(); i) {qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));}}
}由此可知QuartzScheduler调度线程不断获取trigger触发trigger释放trigger。下面分析trigger的获取过程qsRsrcs.getJobStore()返回对象是JobStore集群环境配置如下 !-- JobStore 配置 --
prop keyorg.quartz.jobStore.classorg.quartz.impl.jdbcjobstore.JobStoreTX/propJobStoreTX继承自JobStoreSupport而JobStoreSupport的acquireNextTriggers、triggersFired、releaseAcquiredTrigger方法负责具体trigger相关操作都必须获得TRIGGER_ACCESS锁。核心逻辑在executeInNonManagedTXLock方法中 protected T T executeInNonManagedTXLock(String lockName, TransactionCallbackT txCallback, final TransactionValidatorT txValidator) throws JobPersistenceException {boolean transOwner false;Connection conn null;try {if (lockName ! null) {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 TransactionCallbackBoolean() {Overridepublic 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);}}
}由上代码可知Quartz集群基于数据库锁的同步操作流程如下图所示 一个调度器实例在执行涉及到分布式问题的数据库操作前首先要获取QUARTZ_LOCKS表中对应的行级锁获取锁后即可执行其他表中的数据库操作随着操作事务的提交行级锁被释放供其他调度实例获取。集群中的每一个调度器实例都遵循这样一种严格的操作规程。 getLockHandler()方法返回的对象类型是Semaphore获取锁和释放锁的具体逻辑由该对象维护 public interface Semaphore {boolean obtainLock(Connection conn, String lockName) throws LockException;void releaseLock(String lockName) throws LockException;boolean requiresConnection();
}该接口的实现类完成具体操作锁的逻辑在JobStoreSupport的初始化方法中注入的Semaphore具体类型是StdRowLockSemaphore setLockHandler(new StdRowLockSemaphore(getTablePrefix(), getInstanceName(), getSelectWithLockSQL()));StdRowLockSemaphore的源码如下所示 public class StdRowLockSemaphore extends DBSemaphore {
//锁定SQL语句
public static final String SELECT_FOR_LOCK SELECT * FROM TABLE_PREFIX_SUBST TABLE_LOCKS WHERE COL_LOCK_NAME ? FOR UPDATE;public static final String INSERT_LOCK INSERT INTO TABLE_PREFIX_SUBST TABLE_LOCKS ( COL_SCHEDULER_NAME , COL_LOCK_NAME ) VALUES ( SCHED_NAME_SUBST , ?); //指定锁定SQL
protected void executeSQL(Connection conn, String lockName, String expandedSQL) throws LockException {PreparedStatement ps null;ResultSet rs null;try {ps conn.prepareStatement(expandedSQL);ps.setString(1, lockName);......rs ps.executeQuery();if (!rs.next()) {throw new SQLException(Util.rtp(No row exists in table TABLE_PREFIX_SUBST TABLE_LOCKS for lock named: lockName, getTablePrefix()));}} catch (SQLException sqle) {} finally {...... //release resources}}
}//获取QRTZ_LOCKS行级锁
public boolean obtainLock(Connection conn, String lockName) throws LockException {lockName lockName.intern();if (!isLockOwner(conn, lockName)) {executeSQL(conn, lockName, expandedSQL);getThreadLocks().add(lockName);}return true;
}//释放QRTZ_LOCKS行级锁
public void releaseLock(Connection conn, String lockName) {lockName lockName.intern();if (isLockOwner(conn, lockName)) {getThreadLocks().remove(lockName);}......
}至此总结一下Quartz集群同步机制每当要进行与某种业务相关的数据库操作时先去QRTZ_LOCKS表中查询操作相关的业务对象所需要的锁在select语句之后加for update来实现。例如TRIGGER_ACCESS表示对任务触发器相关的信息进行修改、删除操作时所需要获得的锁。这时执行查询这个表数据的SQL形如 select * from QRTZ_LOCKS t where t.lock_nameTRIGGER_ACCESS for update当一个线程使用上述的SQL对表中的数据执行查询操作时若查询结果中包含相关的行数据库就对该行进行ROW LOCK若此时另外一个线程使用相同的SQL对表的数据进行查询由于查询出的数据行已经被数据库锁住了此时这个线程就只能等待直到拥有该行锁的线程完成了相关的业务操作执行了commit动作后数据库才会释放了相关行的锁这个线程才能继续执行。 通过这样的机制在集群环境下结合悲观锁的机制就可以防止一个线程对数据库数据的操作的结果被另外一个线程所覆盖从而可以避免一些难以觉察的错误发生。当然达到这种效果的前提是需要把Connection设置为手动提交即autoCommit为false。 六、参考资料 http://quartz-scheduler.org/documentation Quartz Documentationhttp://www.ibm.com/developerworks/cn/opensource/os-cn-quartz 基于Quartz开发企业级任务调度应用
