集群环境下定时调度的解决方案之Quartz集群

时间:2022-12-28 07:14:52

集群环境可能出现的问题

在上一篇博客我们介绍了如何在自己的项目中从无到有的添加了Quartz定时调度引擎,其实就是一个Quartz 和Spring的整合过程,很容易实现,但是我们现在企业中项目通常都是部署在集群环境中的,这样我们之前的定时调度就会出现问题了,因为我们的定时任务都加载在内存中的,每个集群节点中的调度器都会去执行,这就会存在重复执行和资源竞争的问题,那么如何来解决这样的问题呢,往下面看吧...

解决方案

在一般的企业中解决类似的问题一般都是在一个note上部署Quartz其他note不部署(或者是在其他几个机器加IP地址过滤),但是这样集群对于定时任务来说就没有什么意义了,而且存在着单点故障的隐患,也就是这台部署着Quartz的机器一旦挂了,我们的定时任务就停止服务了,这绝对不是我们想要的。

Quartz本身是支持集群的,我们通过Quartz的集群方式来解决这样的问题。

Quartz集群

集群环境下定时调度的解决方案之Quartz集群

虽然单个 Quartz 实例能给予你很好的 Job调度能力,但它不能令典型的企业需求,如可伸缩性、高可靠性满足。假如你需要故障转移的能力并能运行日益增多的 Job,Quartz 集群势必成为你方言的一部分了,并且即使是其中一台机器在最糟的时间崩溃了也能确保所有的 Job 得到执行。 QuartzJob Scheduling Framework

了解了Quartz集群的好处,接下来就对我们之前的工程进行改造,增加Quartz集群特性。

Quartz集群中节点依赖于数据库来传播 Scheduler 实例的状态,你只能在使用 JDBC JobStore 时应用 Quartz 集群

所以我们集群的第一步就是建立Quartz所需要的12张表:

1、建表

在quartz核心包里面通过quartz提供的建表语句建立相关表结构

集群环境下定时调度的解决方案之Quartz集群

生成的表结构如下

集群环境下定时调度的解决方案之Quartz集群

这几张表是用于存储任务信息,触发器,调度器,集群节点等信息

详细解释: 
QRTZ_CALENDARS 以Blob 类型存储Quartz 的Calendar 信息 
QRTZ_CRON_TRIGGERS 存储Cron Trigger,包括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_TRIGGERS Trigger 作为Blob 类型存储(用于Quartz 用户用JDBC 创建他们自己定制的Trigger 类型,JobStore并不知道如何存储实例的时候) 
QRTZ_TRIGGER_LISTENERS 存储已配置的TriggerListener 的信息 
QRTZ_TRIGGERS 存储已配置的Trigger 的信息 
所有的表默认以前缀QRTZ_开始。可以通过在quartz.properties配置修改(org.quartz.jobStore.tablePrefix= QRTZ_)。

2、编写quartz.properties文件

建立 quartz.properties文件把它放在工程的 src 目录下,内容如下:

 #============================================================================

 # Configure Main Scheduler Properties 

 #============================================================================

 org.quartz.scheduler.instanceName = Mscheduler

 org.quartz.scheduler.instanceId = AUTO
12
13 org.quartz.jobStore.clusterCheckinInterval=20000 #============================================================================ # Configure ThreadPool #============================================================================ org.quartz.threadPool.class = org.quartz.simpl.SimpleThreadPool org.quartz.threadPool.threadCount = 3 org.quartz.threadPool.threadPriority = 5 #============================================================================ # Configure JobStore #============================================================================ #org.quartz.jobStore.class = org.quartz.simpl.RAMJobStore org.quartz.jobStore.class = org.quartz.impl.jdbcjobstore.JobStoreTX #org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.StdJDBCDelegate org.quartz.jobStore.useProperties = true #org.quartz.jobStore.dataSource = myDS org.quartz.jobStore.tablePrefix = QRTZ_ org.quartz.jobStore.isClustered = true org.quartz.jobStore.maxMisfiresToHandleAtATime=1 #============================================================================ # Configure Datasources #============================================================================ #mysql #org.quartz.dataSource.myDS.driver = com.ibm.db2.jcc.DB2Driver #org.quartz.dataSource.myDS.URL = jdbc:db2://localhost:50000/db #org.quartz.dataSource.myDS.user = db2 #org.quartz.dataSource.myDS.password = db2 #org.quartz.dataSource.myDS.maxConnections = 5 #oracle #org.quartz.dataSource.myDS.driver = oracle.jdbc.driver.OracleDriver #org.quartz.dataSource.myDS.URL = jdbc:oracle:thin:@localhost:1521:orcl #org.quartz.dataSource.myDS.user = scott #org.quartz.dataSource.myDS.password = shao #org.quartz.dataSource.myDS.maxConnections = 5 #For Tomcat org.quartz.jobStore.driverDelegateClass =org.quartz.impl.jdbcjobstore.oracle.OracleDelegate #For Weblogic & Websphere #org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.WebLogicDelegate org.quartz.jobStore.useProperties = false org.quartz.jobStore.dataSource = myDS #JNDI MODE #For Tomcat org.quartz.dataSource.myDS.jndiURL=java:comp/env/jdbc/oracle #For Weblogic & Websphere #org.quartz.dataSource.myDS.jndiURL=jdbc/oracle #============================================================================ # Configure Plugins #============================================================================ #org.quartz.plugin.triggHistory.class = org.quartz.plugins.history.LoggingJobHistoryPlugin #org.quartz.plugin.jobInitializer.class = org.quartz.plugins.xml.JobInitializationPlugin #org.quartz.plugin.jobInitializer.fileNames = jobs.xml #org.quartz.plugin.jobInitializer.overWriteExistingJobs = true #org.quartz.plugin.jobInitializer.failOnFileNotFound = true #org.quartz.plugin.jobInitializer.scanInterval = 10 #org.quartz.plugin.jobInitializer.wrapInUserTransaction = false 

红色加粗部分是集群需要的配置

核心配置解释如下:

org.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 实例就越频繁。默认值是 20000 (即20 秒)

3、修改spring-time.xml文件

 <?xmlversion="1.0"encoding="UTF-8"?>
<!DOCTYPEbeansPUBLIC"-//SPRING//DTD BEAN//EN"
"http://www.springframework.org/dtd/spring-beans.dtd">
<beans> <!-- 调度器lazy-init='false'那么容器启动就会执行调度程序 -->
<beanid="startQuertz"lazy-init="false"autowire="no"class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
<property name="dataSource" ref="dataSource"/>
9 <property name="configLocation" value="classpath:quartz.properties" />

<propertyname="triggers">
<list>
<refbean="doTime"/>
</list>
</property>
<!-- 允许在Quartz上下文中使用Spring实例工厂 -->
<propertyname="applicationContextSchedulerContextKey"value="applicationContext"/>
</bean> <!-- 触发器 -->
<beanid="doTime"class="org.springframework.scheduling.quartz.CronTriggerBean">
<propertyname="jobDetail"ref="jobtask"></property>
<!-- cron表达式 -->
<propertyname="cronExpression"value="10,15,20,25,30,35,40,45,50,55 * * * * ?"></property>
</bean> <!-- 任务 -->
<beanid="jobtask"class="org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean">
<propertyname="targetObject"ref="synUsersJob"></property>
<propertyname="targetMethod"value="execute"></property>
</bean> <!-- 要调用的工作类 -->
<beanid="synUsersJob"class="org.leopard.core.quartz.job.SynUsersJob"></bean> </beans>

增加红色加粗部分代码,注入数据源和加载quartz.properties文件

OK Quartz集群的配置只有这几步,我们来启动项目。。。

我们启着启着….报错了!

17:00:59,718 ERROR ContextLoader:215 - Context initialization failed

org.springframework.beans.factory.BeanCreationException: Error creating bean with name 'startQuertz' defined in class path resource [config/spring/spring-time.xml]: Invocation of init method failed; nested exception is org.quartz.JobPersistenceException:Couldn't store job: Unable to serialize JobDataMap for insertion into database because the value of property 'methodInvoker' is not serializable: org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean [See nested exception: java.io.NotSerializableException: Unable to serialize JobDataMap for insertion into database because the value of property 'methodInvoker' is not serializable: org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean]

我们主要来看红色部分,主要原因就是这个MethodInvokingJobDetailFactoryBean 类中的 methodInvoking 方法,是不支持序列化的,因此在把 quartz 的 task 序列化进入数据库时就会抛这个serializable的错误

4、解决serializable错误解决方案

网上查了一下,解决这个问题,目前主要有两种方案:

4.1.修改Spring的源码

博客地址:http://jira.springframework.org/browse/SPR-3797

作者重写了MethodInvokingJobDetailFactoryBean

4.2.通过AOP反射对Spring源码进行切面重构

博客地址:http://blog.csdn.net/lifetragedy/article/details/6212831

根据 QuartzJobBean 来重写一个自己的类,然后使用 SPRING 把这个重写的类(我们就名命它为: MyDetailQuartzJobBean )注入 appContext 中后,再使用 AOP 技术反射出原有的 quartzJobx( 就是开发人员原来已经做好的用于执行 QUARTZ 的 JOB 的执行类 ) 。

两种方式我都进行了测试,都可以解决问题,我们这里先通过第二种方式解决这个bug,没有修改任何Spring的源码

4.2.1、增加MyDetailQuartzJobBean.Java

 package org.leopard.core.quartz;

 import java.lang.reflect.Method;

 import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.quartz.QuartzJobBean; /**
* 解决Spring和Quartz整合bug
*
*/
public class MyDetailQuartzJobBean extends QuartzJobBean {
protected final Log logger = LogFactory.getLog(getClass()); private String targetObject;
private String targetMethod;
private ApplicationContext ctx; protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
try { logger.info("execute [" + targetObject + "] at once>>>>>>");
Object otargetObject = ctx.getBean(targetObject);
Method m = null;
try {
m = otargetObject.getClass().getMethod(targetMethod, new Class[] {}); m.invoke(otargetObject, new Object[] {});
} catch (SecurityException e) {
logger.error(e);
} catch (NoSuchMethodException e) {
logger.error(e);
} } catch (Exception e) {
throw new JobExecutionException(e);
} } public void setApplicationContext(ApplicationContext applicationContext) {
this.ctx = applicationContext;
} public void setTargetObject(String targetObject) {
this.targetObject = targetObject;
} public void setTargetMethod(String targetMethod) {
this.targetMethod = targetMethod;
} }

5、再次修改spring-time.xml文件解决serializable问题

修改后的spring-time.xml文件内容如下:

 <?xmlversion="1.0"encoding="UTF-8"?>
<!DOCTYPEbeansPUBLIC"-//SPRING//DTD BEAN//EN"
"http://www.springframework.org/dtd/spring-beans.dtd">
<beans> <!-- 调度器lazy-init='false'那么容器启动就会执行调度程序 -->
<beanid="startQuertz"lazy-init="false"autowire="no"class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
<propertyname="dataSource"ref="dataSource"/>
<propertyname="configLocation"value="classpath:quartz.properties"/>
<propertyname="triggers">
<list>
<refbean="doTime"/>
</list>
</property>
<!--这个是必须的,QuartzScheduler延时启动,应用启动完后 QuartzScheduler再启动-->
<propertyname="startupDelay"value="30"/>
<!--这个是可选,QuartzScheduler启动时更新己存在的Job,这样就不用每次修改targetObject后删除qrtz_job_details表对应记录了-->
<propertyname="overwriteExistingJobs"value="true"/>
<!-- 允许在Quartz上下文中使用Spring实例工厂 -->
<propertyname="applicationContextSchedulerContextKey"value="applicationContext"/>
</bean> <!-- 触发器 -->
<beanid="doTime"class="org.springframework.scheduling.quartz.CronTriggerBean">
<propertyname="jobDetail"ref="jobtask"></property>
<!-- cron表达式 -->
<propertyname="cronExpression"value="10,15,20,25,30,35,40,45,50,55 * * * * ?"></property>
</bean> <!-- 任务 -->
<beanid="jobtask"class="org.springframework.scheduling.quartz.JobDetailBean">
<propertyname="jobClass">
33 <value>org.leopard.core.quartz.MyDetailQuartzJobBean</value>
34 </property>

<propertyname="jobDataAsMap">
<map>
<entrykey="targetObject"value="synUsersJob"/>
<entrykey="targetMethod"value="execute"/>
</map>
</property>
</bean> <!-- 要调用的工作类 -->
<beanid="synUsersJob"class="org.leopard.core.quartz.job.SynUsersJob"></bean> </beans>

主要看红色加粗部分...

测试

Ok  配置完成,我们把oa_ssh部署到两台tomcat上面,分别启动。

集群环境下定时调度的解决方案之Quartz集群

可以看到我们先启动的tomcat控制台打印出日志

集群环境下定时调度的解决方案之Quartz集群

另外一台没有打印日志

集群环境下定时调度的解决方案之Quartz集群

这时我们把执行定时任务的那台tomcat停止,可以看到等了一会之后,我们的另外一台tomcat会把之前tomcat执行的定时任务接管过来继续执行,我们的集群是成功的。

集群环境下定时调度的解决方案之Quartz集群