spring+Quartz集群

54 阅读 0 评论 36 点赞

我是靠谱客的博主迷人树叶，最近开发中收集的这篇文章主要介绍spring+Quartz集群，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

最近项目中使用了 spring +Quartz定时任务、但是项目最近要集群部署、多个APP下如何利用Quartz 协调处理任务。
   大家可以思考一下、现在有 A、B、C三个应用同时作为集群服务器对外统一提供服务、每个应用下各有一个Quartz、它们会按照既定的时间自动执行各自的任务。我们先不说实现什么功能，就说这样的架构其实有点像多线程。那多线程里就会存在“资源竞争”的问题，即可能产生脏读，脏写，由于三台应用里都有 Quartz，因此会存在重复处理任务的现象。
   解决方案一：只在一台应用上装 Quartz，其它两台不装，这样集群就没有意义了。
   解决方案二：使用其实Quartz自身可以实例化数据库的特性就可以解决问题
本方案优点：
1.    每台作为集群点的应用上都可以布署 Quartz ；
2.    Quartz 的 TASK （ 12 张表）实例化如数据库，基于数据库引擎及 High-Available 的策略（集群的一种策略）自动协调每个节点的 QUARTZ ，当任一一节点的 QUARTZ 非正常关闭或出错时，另几个节点的 QUARTZ 会自动启动；
3. 无需开发人员更改原已经实现的 QUARTZ ，使用 SPRING+ 类反射的机制对原有程序作切面重构；
解决方案：
1:去官网下载最新的 quartz 解压在目录 docsdbTables 下就会找到 tables_mysql.sql  文件、建立数据库Quartz 并导入数据库。

  2：生成 quartz.properties 文件，把它放在工程的 src 目录下修改配置文件如下：

#==============================================================
#Configure Main Scheduler Properties
#==============================================================
org.quartz.scheduler.instanceName = quartzScheduler
org.quartz.scheduler.instanceId = AUTO
#==============================================================
#Configure ThreadPool
#==============================================================
org.quartz.threadPool.class = org.quartz.simpl.SimpleThreadPool
org.quartz.threadPool.threadCount = 10
org.quartz.threadPool.threadPriority = 5
org.quartz.threadPool.threadsInheritContextClassLoaderOfInitializingThread = true
#==============================================================
#Configure JobStore
#==============================================================
org.quartz.jobStore.class = org.quartz.impl.jdbcjobstore.JobStoreTX
org.quartz.jobStore.driverDelegateClass = org.quartz.impl.jdbcjobstore.StdJDBCDelegate
org.quartz.jobStore.tablePrefix = QRTZ_
org.quartz.jobStore.isClustered = true
org.quartz.jobStore.clusterCheckinInterval = 20000
org.quartz.jobStore.dataSource = myDS
#==============================================================
#Configure DataSource （此处填你自己的数据库连接信息）
#==============================================================
org.quartz.dataSource.myDS.driver = com.mysql.jdbc.Driver
org.quartz.dataSource.myDS.URL = jdbc:mysql://localhost:3306/quartz?useUnicode=true&characterEncoding=UTF-8
org.quartz.dataSource.myDS.user = root
org.quartz.dataSource.myDS.password = 123
org.quartz.dataSource.myDS.maxConnections =30

复制代码

3：重写 quartz 的 QuartzJobBean 类
原因是在使用 quartz + spring 把 quartz 的 task 实例化进入数据库时，会产生： serializable 的错误，原因在于：
这个 MethodInvokingJobDetailFactoryBean 类中的 methodInvoking 方法，是不支持序列化的，因此在把 QUARTZ 的 TASK 序列化进入数据库时就会抛错。网上有说把 SPRING 源码拿来，修改一下这个方案，然后再打包成 SPRING.jar 发布，这些都是不好的方法，是不安全的。
必须根据 QuartzJobBean 来重写一个自己的类。
BootstrapJob. java : 引导Job，通过Spring容器获取任务的Job，根据注入的 target Job,该Job必须实现Job2接口

/**
* 引导Job，通过Spring容器获取任务的Job，根据注入的targetJob,该Job必须实现Job2接口
* @author zzp
* @date 2014-7-7
*/
public class BootstrapJob implements Serializable{
private String targetJob ;
public void executeInternal(ApplicationContext cxt) {
Job2 job = (Job2)cxt.getBean(this.targetJob);
job.executeInternal() ;
}
public String getTargetJob() {
return targetJob;
}
public void setTargetJob(String targetJob) {
this.targetJob = targetJob;
}
}

复制代码

Job2.java:

/**
* Quartz 与 Spring 集成时，自定义的Job可以拥有Spring的上下文，
* 因此定义了该接口，自定义的Job需要实现该接口，并实现executeInternal的task，
* 这样解决了Quartz 与Spring 在集群环境下，可以不需要序列化，
* 只需要在executeInternal获取Spring 上下文中的target job bean.
* 调用其相关的处理函数，来处理任务
* @author zzp
* @date 2014-7-7
*/
public interface Job2 extends Serializable{
/**
* 处理任务的核心函数
*
* @param cxt Spring 上下文
*/
void executeInternal();
}

复制代码

重写 MethodInvokingJobDetailFactoryBean类方法如下：

<p>public void execute(JobExecutionContext context) throws JobExecutionException
{
try
{
logger.debug("start");
String targetClass = context.getMergedJobDataMap().getString("targetClass");
//logger.debug("targetClass is "+targetClass);
Class targetClassClass = null;
if(targetClass!=null)
{
targetClassClass = Class.forName(targetClass); // Could throw ClassNotFoundException
}
Object targetObject = context.getMergedJobDataMap().get("targetObject");
if(targetObject instanceof BootstrapJob){
//Job2 job = (Job2)targetObject;
//job.executeInternal(context.getScheduler().getContext().)
ApplicationContext ac = (ApplicationContext)context.getScheduler().getContext().get("applicationContext");
BootstrapJob target = (BootstrapJob)targetObject ;
target.executeInternal(ac);
}else{
//logger.debug("targetObject is "+targetObject);
String targetMethod = context.getMergedJobDataMap().getString("targetMethod");
//logger.debug("targetMethod is "+targetMethod);
String staticMethod = context.getMergedJobDataMap().getString("staticMethod");
//logger.debug("staticMethod is "+staticMethod);
Object[] arguments = (Object[])context.getMergedJobDataMap().get("arguments");
//logger.debug("arguments are "+arguments);
//logger.debug("creating MethodInvoker");
MethodInvoker methodInvoker = new MethodInvoker();
methodInvoker.setTargetClass(targetClassClass);
methodInvoker.setTargetObject(targetObject);
methodInvoker.setTargetMethod(targetMethod);
methodInvoker.setStaticMethod(staticMethod);
methodInvoker.setArguments(arguments);
methodInvoker.prepare();
//logger.info("Invoking: "+methodInvoker.getPreparedMethod().toGenericString());
methodInvoker.invoke();
}
}
catch(Exception e)
{
throw new JobExecutionException(e);
}
finally
{
logger.debug("end");
}
}
}</p><p>
</p>

复制代码

QuartzDeleteQueAction 任务类、一定要实现接口job2、只做参考。

public class QuartzDeleteQueAction implements Job2 {
private static final long serialVersionUID = 1L;
private IQuesGroupService quesGroupService;
public void executeInternal(){
LogUtil.jobInfo("Quartz的任务调度执行删除教研组试题html文件开始");
try {
ServletContext context = ContextLoader.getCurrentWebApplicationContext().getServletContext();
String pathHtml = context.getRealPath(Constants.PATH_HTML);
//获取被删除试题No
List<Object> list = quesGroupService.queryDeleteQues();
for(Object obj:list){
String quesName = pathHtml+"ques_"+obj.toString()+".html";
FileUtil.delFile(quesName);//删除无用html文件
}
} catch (Exception e) {
e.printStackTrace();
}
LogUtil.jobInfo("Quartz的任务调度执行删除教研组试题html文件结束");
}
public IQuesGroupService getQuesGroupService() {
return quesGroupService;
}
public void setQuesGroupService(IQuesGroupService quesGroupService) {
this.quesGroupService = quesGroupService;
}
}

复制代码

4：配置 applicationContext-job. xml ：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xmlns:context="http://www.springframework.org/schema/context"
xmlns:aop="http://www.springframework.org/schema/aop" xmlns:tx="http://www.springframework.org/schema/tx"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.0.xsd
http://www.springframework.org/schema/aop http://www.springframework.org/schema/aop/spring-aop-3.0.xsd
http://www.springframework.org/schema/tx http://www.springframework.org/schema/tx/spring-tx-3.0.xsd
http://www.springframework.org/schema/context http://www.springframework.org/schema/context/spring-context-3.0.xsd" default-autowire="byName" default-lazy-init="true">
<bean id="quartzJob" class="com.web.action.QuartzDeleteQueAction"></bean>
<bean id="bootstrapJob" class="com.acts.web.quartz.BootstrapJob">
<property name="targetJob" value="quartzJob" />
</bean>
<bean id="jobTask" class="com.acts.web.quartz.MethodInvokingJobDetailFactoryBean">
<property name="concurrent" value="true" />
<property name="targetObject" ref="bootstrapJob" />
</bean>
<bean id="doTime" class="org.springframework.scheduling.quartz.CronTriggerBean">
<property name="jobDetail">
<ref bean="jobTask"/>
</property>
<property name="cronExpression">
<value>0 0/10 5-20 * * ?</value>
</property>
</bean>
<bean id="startQuertz" lazy-init="false" autowire="no" class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
<property name="configLocation" value="classpath:quartz.properties" />
<property name="dataSource" ref="dataSourceQuartz" />
<property name="triggers">
<list>
<ref bean="doTime"/>
</list>
</property>
<property name="applicationContextSchedulerContextKey" value="applicationContext" />
</bean>
</beans>

复制代码

集群环境下测试：
   三个个节点都带有 Quartz 任务，监控控制台、此时只有一台 quartz 在运行，另几个节点上的 quartz 没有运行。
   此时手动停掉那台运行 QUARTZ 过了 10分钟左右，另一个节点的 quartz 自动监测到了集群中运行着的 quartz 的 instance 已经停掉，因此 quartz 集群会自动把任一台可用的 APP上启动起一个 quartz job 的任务。
   至此 Quartz使用集群策略已经ok，不用改原有代码，配置一下我们就可做到 Quartz的集群与自动错误冗余。