Quartz集群实战及原理解析

标签: quartz 集群 原理 | 发表时间:2015-05-20 17:38 | 作者:wenniuwuren
出处:http://blog.csdn.net

  选Quartz的团队基本上是冲着Quartz本身实现的集群去的, 不然JDK自带Timer就可以实现相同的功能, 而Timer存在的单点故障是生产环境上所不能容忍的。 在自己造个有负载均衡和支持集群(高可用、伸缩性)的调度框架又影响项目的进度, 所以大多数团队都直接使用了Quartz来作为调度框架。


一、 Quartz集群的架构图:



二、 Quartz集群配置:

	<!-- 调度任务 -->
	<bean id="jobDetail"
			class="org.springframework.scheduling.quartz.JobDetailFactoryBean">
			<property name="jobClass" value="全类名" />
			<property name="durability" value="true"/>
			<property name="targetMethod" value="execute" />
			<property name="concurrent" value="true" /> -->
			<!-- <property name="shouldRecover" value="true" /> -->
	</bean>
	
	<!-- 调度工厂 -->
	<bean id="scheduler" lazy-init="false" autowire="no"
		class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
		
		<!-- 注册JobDetails -->
		<property name="jobDetails">
			<list>
				<ref bean="jobDetail"/>
			</list>
		</property>
		
		<!--可选,QuartzScheduler 启动时更新己存在的Job,这样就不用每次修改targetObject后删除qrtz_job_details表对应记录了 -->
		<property name="overwriteExistingJobs" value="true"/>
		
		<!-- 属性 -->
		<property name="quartzProperties">
			<props>
				<!-- 集群要求必须使用持久化存储 -->
				<prop key="org.quartz.jobStore.class">org.quartz.impl.jdbcjobstore.JobStoreCMT</prop>
				
				<prop key="org.quartz.scheduler.instanceName">EventScheduler</prop>
				<!-- 每个集群节点要有独立的instanceId -->
				<prop key="org.quartz.scheduler.instanceId">AUTO</prop>
				
				<!-- Configure ThreadPool -->
				<prop key="org.quartz.threadPool.class">org.quartz.simpl.SimpleThreadPool</prop>
				<prop key="org.quartz.threadPool.threadCount">50</prop>
				<prop key="org.quartz.threadPool.threadPriority">5</prop>
				<prop key="org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread">true</prop>
				<!-- Configure JobStore -->
				<prop key="org.quartz.jobStore.misfireThreshold">60000</prop>
				<prop key="org.quartz.jobStore.driverDelegateClass">org.quartz.impl.jdbcjobstore.StdJDBCDelegate</prop>
				<prop key="org.quartz.jobStore.tablePrefix">SCHEDULER_</prop>
				<prop key="org.quartz.jobStore.maxMisfiresToHandleAtATime">10</prop>
				<!-- 开启集群 -->
				<prop key="org.quartz.jobStore.isClustered">true</prop>
				<prop key="org.quartz.jobStore.clusterCheckinInterval">20000</prop>
				<prop key="org.quartz.jobStore.dontSetAutoCommitFalse">true</prop>
				<prop key="org.quartz.jobStore.txIsolationLevelSerializable">false</prop>
				<prop key="org.quartz.jobStore.dataSource">myDS</prop>
				<prop key="org.quartz.jobStore.nonManagedTXDataSource">myDS</prop>
				<prop key="org.quartz.jobStore.useProperties">false</prop>
				<!-- Configure Datasources  -->
				<prop key="org.quartz.dataSource.myDS.driver">com.mysql.jdbc.Driver</prop>
				<prop key="org.quartz.dataSource.myDS.URL">${db.url}</prop>
				<prop key="org.quartz.dataSource.myDS.user">${db.username}</prop>
				<prop key="org.quartz.dataSource.myDS.password">${db.password}</prop>
				<prop key="org.quartz.dataSource.myDS.maxConnections">10</prop>
				<prop key="org.quartz.dataSource.myDS.validationQuery">select 0 from dual</prop>
			</props>
		</property>
		<property name="applicationContextSchedulerContextKey" value="applicationContext" />
	</bean>


三、 集群源码分析

Quartz如何保证多个节点的应用只进行一次调度(即某一时刻的调度任务只由其中一台服务器执行)?


正如上面架构图所示, Quartz的集群是在同一个数据库下, 由数据库的数据来确定调度任务是否正在执行, 正在执行则其他服务器就不能去执行该行调度数据。 这个跟很多项目是用Zookeeper做集群不一样, 这些项目是靠Zookeeper选举出来的的服务器去执行, 可以理解为Quartz靠数据库选举一个服务器来执行。


如果之前看过这篇 Quartz按时启动原理就应该了解到Quartz最主要的一个类QuartzSchedulerThread职责是触发任务, 是一个不断运行的Quartz主线程, 还是从这里入手了解集群原理。


集群配置里面有一个配置项:

<prop key="org.quartz.jobStore.class">org.quartz.impl.jdbcjobstore.JobStoreCMT</prop>
源码可以看到JobStoreCMT extends JobStoreSupport, 在QuartzSchedulerThread的run方法里面调用的acquireNextTriggers、 triggersFired、 releaseAcquiredTrigger方法都进行了加锁处理。

以acquireNextTriggers为例:



而LOCK_TRIGGER_ACCESS其实就是一个JAVA常量

protected static final String LOCK_TRIGGER_ACCESS = "TRIGGER_ACCESS";


这个常量传入加锁的核心方法executeInNonManagedTXLock: 处理逻辑前获取锁, 处理完成后在finally里面释放锁(一种典型的同步处理方法)

 protected <T> T executeInNonManagedTXLock(
            String lockName, 
            TransactionCallback<T> txCallback, final TransactionValidator<T> txValidator) throws JobPersistenceException {
        boolean transOwner = false;
        Connection conn = null;
        try {
            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);
            }
        }
    }


getLockHandler那么可以思考下这个LockHandler怎么来的?

最后发现在JobStoreSupport的initail方法赋值了:

public void initialize(ClassLoadHelper loadHelper,
            SchedulerSignaler signaler) throws SchedulerConfigException {

        ...
        

        // If the user hasn't specified an explicit lock handler, then 
        // choose one based on CMT/Clustered/UseDBLocks.
        if (getLockHandler() == null) {
            
            // If the user hasn't specified an explicit lock handler, 
            // then we *must* use DB locks with clustering
            if (isClustered()) {
                setUseDBLocks(true);
            }
            
            if (getUseDBLocks()) {
                ...
				// 在初始化方法里面赋值了
                setLockHandler(new StdRowLockSemaphore(getTablePrefix(), getInstanceName(), getSelectWithLockSQL()));
            } else {
                getLog().info(
                    "Using thread monitor-based data access locking (synchronization).");
                setLockHandler(new SimpleSemaphore());
            }
        }

    }

可以在StdRowLockSemaphore里面看到:

public static final String SELECT_FOR_LOCK = "SELECT * FROM "
            + TABLE_PREFIX_SUBST + TABLE_LOCKS + " WHERE " + COL_SCHEDULER_NAME + " = " + SCHED_NAME_SUBST
            + " AND " + 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 + ", ?)"; 
可以看出采用了悲观锁的方式对triggers表进行行加锁, 以保证任务同步的正确性。

当线程使用上述的SQL对表中的数据执行操作时,数据库对该行进行行加锁; 于此同时, 另一个线程对该行数据执行操作前需要获取锁, 而此时已被占用, 那么这个线程就只能等待, 直到该行锁被释放。


Quartz的锁存放在:

CREATE TABLE `scheduler_locks` (
  `SCHED_NAME` varchar(120) NOT NULL COMMENT '调度名',
  `LOCK_NAME` varchar(40) NOT NULL COMMENT '锁名',
  PRIMARY KEY (`SCHED_NAME`,`LOCK_NAME`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8


锁名和上述常量一一对应:



有可能你的任务不能支持并发执行(因为有可能任务还没执行完, 下一轮就trigger了, 如果没做同步处理可能造成严重的数据问题), 那么在任务类加上注解:

@DisallowConcurrentExecution

设置@DisallowConcurrentExecution以后程序会等任务执行完毕以后再去执行



四、 参考资料


Quartz官网: http://quartz-scheduler.org/documentation/quartz-2.x/tutorials/tutorial-lesson-11


作者:wenniuwuren 发表于2015/5/20 9:38:51 原文链接
阅读:168 评论:0 查看评论

相关 [quartz 集群 原理] 推荐:

Quartz集群实战及原理解析

- - CSDN博客推荐文章
  选Quartz的团队基本上是冲着Quartz本身实现的集群去的, 不然JDK自带Timer就可以实现相同的功能, 而Timer存在的单点故障是生产环境上所不能容忍的. 在自己造个有负载均衡和支持集群(高可用、伸缩性)的调度框架又影响项目的进度, 所以大多数团队都直接使用了Quartz来作为调度框架.

Quartz应用与集群原理分析

- - 美团技术团队
美团CRM系统中每天有大量的后台任务需要调度执行,如构建索引、统计报表、周期同步数据等等,要求任务调度系统具备高可用性、负载均衡特性,可以管理并监控任务的执行流程,以保证任务的正确执行. 美团CRM系统的任务调度模块经历了以下历史方案. 每天晚上运行定时任务,通过SQL脚本+crontab方式执行,例如,.

spring quartz 集群配置

- - CSDN博客推荐文章
Quartz 是一个开源的作业调度框架,它完全由 Java 写成,并设计用于 J2SE 和 J2EE 应用中. 它提供了巨大的灵活性而不牺牲简单性. 你能够用它来为执行一个作业而创建简单的或复杂的调度. 在项目中有大量的后台任务需要调度执行,如构建索引、统计报表、周期同步数据等等,要求任务调度系统具备高可用性、负载均衡特性,使用Quartz 会很方便.

quartz集群调度机制调研及源码分析

- - CSDN博客推荐文章
quartz2.2.1集群调度机制调研及源码分析. quratz是目前最为成熟,使用最广泛的java任务调度框架,功能强大配置灵活.在企业应用中占重要地位.quratz在集群环境中的使用方式是每个企业级系统都要考虑的问题.早在2006年,在ITeye上就有一篇关于quratz集群方案的讨论: http://www.iteye.com/topic/40970 ITeye创始人@Robbin在8楼给出了自己对quartz集群应用方案的意见..

quartz集群分布式(并发)部署解决方案-Spring

- - 企业架构 - ITeye博客
项目中使用分布式并发部署定时任务,多台跨JVM,按照常理逻辑每个JVM的定时任务会各自运行,这样就会存在问题,多台分布式JVM机器的应用服务同时干活,一个是加重服务负担,另外一个是存在严重的逻辑问题,. 比如需要回滚的数据,就回滚了多次,刚好quartz提供很好的解决方案. 集群分布式并发环境中使用QUARTZ定时任务调度,会在各个节点会上报任务,存到数据库中,执行时会从数据库中取出触发器来执行,如果触发器的名称和执行时间相同,则只有一个节点去执行此任务.

quartz 任务的增删改

- - 编程语言 - ITeye博客
从网上找了好多例子,要么太老,要么用不了. * 添加一个定时任务,使用默认的任务组名,触发器名,触发器组名. * @param jobName 任务名. * @param cls 任务. * @param time 时间设置,参考quartz说明文档. Scheduler scheduler = schedulerFactory.getScheduler();//创建一个触发器表.

Quartz任务监控管理 (1)

- - Java - 编程语言 - ITeye博客
Quartz任务监控管理,类似Windows任务管理器,可以获得运行时的实时监控,查看任务运行状态,动态增加任务,暂停、恢复、移除任务等. 对于动态增加任务,可以参加我的前一篇文章《Quartz如何在Spring动态配置时间》,本文在前文的基础上扩展,增加暂停、恢复、移除任务等功能,实现Quartz任务监控管理.

Quartz实现动态定时任务

- - 编程语言 - ITeye博客
转http://my.oschina.net/u/1177710/blog/284608.      由于最近工作要实现定时任务的执行,而且要求定时周期是不固定的,所以就用到了quartz来实现这个功能;.      spring3.1以下的版本必须使用quartz1.x系列,3.1以上的版本才支持quartz 2.x,不然会出错.

quartz spring 实现动态定时任务

- - 企业架构 - ITeye博客
在实际项目应用中经常会用到定时任务,可以通过quartz和spring的简单配置即可完成,但如果要改变任务的执行时间、频率,废弃任务等就需要改变配置甚至代码需要重启服务器,这里介绍一下如何通过quartz与spring的组合实现动态的改变定时任务的状态的一个实现. 参考文章: http://www.meiriyouke.net/?p=82.

使用Quartz和Obsidian来调度任务

- - Java译站
在介绍使用到的Quartz和Obsidian的API之前,首先我得声明一下,一般来说使用API并不是调度任务的最佳方式. Quartz提供了一个通过XML来配置作业的机制,而Obsidian则为你提供了一套完整的管理和监控的WEB应用. 然而,有一些使用场景还是强烈推荐使用API的,我们来看一下吧.