quartz+spring-集群版任务动态载入及批改
quartz+spring-集群版任务动态载入及修改
关于单机版的配置描述前文已经提过了，有兴趣的可以看下
http://lee1177.iteye.com/blog/2171379
这里介绍对于集群做的一个简单实现。
首先其实quartz针对就群有自己的实现了，本人这个实现不过是简化并在前文基础上做的一个集群实现。
本实现目的是在集群中任一一台作为运行节点，如果该节点不可用，自动切换为其他节点继续执行。同时自动检测定时任务设定被修改的重新加载。未考虑负载均衡及异常任务恢复情况。
下面说如何做的：
首先是SchedulerJob的定义比之前多了一个更新时间的属性
public class SchedulerJob {
public static final String JOB_GOURP = "JOB_GOURP"; //任务组
public static final int STATUS_USED = 1; //可用状态
/** 任务id spring的bean名称*/
private String jobId;
/** 任务名称 */
private String jobName;
/** 任务状态 0停用 1启用 */
private Integer jobStatus;
/** 任务运行时间表达式 */
private String cronExpression;
/** 更新时间，用于集群自动动态更新任务修改*/
private Long updateTime;
对应任务SchedulerJobBean同前文一致；
然后自定义一个集群quartz运行工厂类，来执行定时任务实现
/**
* 集群任务执行工厂
* @ClassName: ClusterQuartzJobFactory
*/
@DisallowConcurrentExecution
public class ClusterQuartzJobFactory implements Job {
private static final Logger log = LoggerFactory.getLogger(ClusterQuartzJobFactory.class);
public void execute(JobExecutionContext context) throws JobExecutionException {
SchedulerJob scheduleJob = (SchedulerJob)context.getMergedJobDataMap().get("scheduleJob");
if(scheduleJob.getJobId().equals(ClusterQuartzManager.SCHEDULER_CHECK_JOB)
||ClusterQuartzManager.isRun()){ //如果是检查任务或该服务正在运行
try{
SchedulerJobBean jobBean = (SchedulerJobBean) ContextHolder.getBean(scheduleJob.getJobId());
jobBean.execute(context);
} catch (Exception ex){
log.error("====================Scheduler-error-begin====================");
log.error(ex.toString());
StackTraceElement[] element = ex.getStackTrace();
for(int i=0;i<element.length;i++){
log.error("位置:"+element[i]);
}
log.error("====================Scheduler-error-end====================");
}
}
}
}
然后集群任务管理器
/**
* Quartz集群管理器
* @Create In 2014年12月31日 By lee
*/
public class ClusterQuartzManager extends AbstractQuartzManager{
private static final Logger log = LoggerFactory.getLogger(ClusterQuartzManager.class);
public static final int STATE_RUN = 1; //运行状态
public static final int STATE_WAIT = 0; //等待状态
public static final String SCHEDULER_CHECK_JOB = "schedulerCheckJob"; //定时检查任务Id
private static String jobFactoryKey = null; //每个服务实例一个单独KEY
private static int jobState = -1; //该服务实例状态
/**
* 启动所有定时任务
* @throws SchedulerException
* @Methods Name startAll
* @Create In 2014年9月23日 By lee
*/
public static void startAll() throws SchedulerException{
if(jobFactoryKey==null){
jobFactoryKey = UUID.randomUUID().toString();
}
if(jobState==-1){
jobState = STATE_WAIT;
}
SchedulerServerService schedulerServerService = ContextHolder.getBean(SchedulerServerService.class);
schedulerServerService.save(new SchedulerServer(jobFactoryKey,jobState,DateUtils.getCurrentDateTime().getTime()));
create(ClusterQuartzManager.instanceCheckJob());
log.debug("=====定时任务启动完成=====");
}
/**
* 构造默认的检查任务
* @Methods Name instanceCheckJob
* @Create In 2014年12月31日 By lee
* @return
*/
private static SchedulerJob instanceCheckJob(){
SchedulerJob checkJob = new SchedulerJob();
checkJob.setCronExpression("0/5 * * * * ?"); //每5秒运行检查一次
checkJob.setJobId(SCHEDULER_CHECK_JOB);
checkJob.setJobStatus(SchedulerJob.STATUS_USED);
return checkJob;
}
/**
* 当前是否为运行状态
* @Methods Name isRun
* @Create In 2014年12月30日 By lee
* @return
*/
public static boolean isRun(){
return jobState==STATE_RUN;
}
public static String getFactoryKey(){
return jobFactoryKey;
}
/**
* 启动
* @Methods Name startUp
* @Create In 2015年1月4日 By lee
* @return 首次启动返回true,如果已启动返回false
* @throws SchedulerException
*/
public static boolean startUp() {
if(jobState != STATE_RUN){
SchedulerJobService schedulerJobService = ContextHolder.getBean(SchedulerJobService.class);
List<SchedulerJob> jobs = schedulerJobService.findAll();
for(SchedulerJob job : jobs){
if(SchedulerJob.STATUS_USED==job.getJobStatus()){
try {
create(job);
} catch (SchedulerException e) {
e.printStackTrace();
}
}
}
log.debug("=====定时任务启动完成=====");
jobState = STATE_RUN;
return true;
}
return false;
}
/**
* 设置为等待
* @Methods Name toWait
* @Create In 2015年1月4日 By lee
*/
public static void toWait() {
if(jobState == STATE_RUN){
SchedulerJobService schedulerJobService = ContextHolder.getBean(SchedulerJobService.class);
List<SchedulerJob> jobs = schedulerJobService.findAll();
for(SchedulerJob job : jobs){
try {
delete(job);
} catch (SchedulerException e) {
e.printStackTrace();
}
}
log.debug("=====定时任务启动完成=====");
jobState = STATE_WAIT;
}
}
}
可以看到在管理器启动时默认只启动了一个检查任务，这个实现就是利用这检查任务来实现集群中节点的状态检查及激活更新等动作，检查任务如下
/**
* 定时任务检查任务
* @ClassName: SchedulerCheckJob
* @Create In 2014年12月30日 By lee
*/
@Component
@Transactional
public class SchedulerCheckJob implements SchedulerJobBean{
//private static Log logger = LogFactory.getLog(SchedulerCheckJob.class);
@Autowired
private SchedulerServerService schedulerServerService;
@Autowired
private SchedulerJobService schedulerJobService;
@Override
public void execute(JobExecutionContext context) {
Long curTime = context.getScheduledFireTime().getTime();
if(checkRun(curTime)){ //如果当前服务启动
if(!ClusterQuartzManager.startUp()){ //如果不是首次启动,检查是否有更新
Long previousTime = null;
if(context.getPreviousFireTime()!=null){
previousTime = context.getPreviousFireTime().getTime();
}else{
previousTime = curTime+5000;
}
checkJobUpdate(previousTime,curTime);
}
//logger.info("=====["+QuartzManager.getFactoryKey()+"]=====RUN=====");
}else{
ClusterQuartzManager.toWait();
//logger.info("=====["+QuartzManager.getFactoryKey()+"]=====WAIT=====");
}
}
/**
* 检查任务是否有更新
* @Methods Name checkJobUpdate
* @Create In 2014年12月30日 By lee
* @param previousTime
* @param curTime
*/
private void checkJobUpdate(Long previousTime,Long curTime) {
List<SchedulerJob> updateJobs = schedulerJobService.findByUpdateTimeBetween(previousTime,curTime);
for(SchedulerJob job : updateJobs){
if(SchedulerJob.STATUS_USED==job.getJobStatus()){
try {
ClusterQuartzManager.update(job);
} catch (SchedulerException e) {
e.printStackTrace();
}
}
}
}
/**
* 检查当前服务是否用运行服务
* @Methods Name checkRun
* @Create In 2014年12月30日 By lee
* @param curTime
* @return
*/
private boolean checkRun(Long curTime){
String curServerKey = ClusterQuartzManager.getFactoryKey();
List<SchedulerServer> servers = schedulerServerService.findAll();
SchedulerServer readServer = null; //备用服务
SchedulerServer runServer = null; //运行服务
SchedulerServer curServer = null; //当前服务
boolean runServerError = true; //运行服务异常
for(SchedulerServer server : servers){
if(server.getJobServer().equals(curServerKey)){
curServer = server;
}
Long updateTime = server.getUpdateTime();
long afterTime = curTime-updateTime;
//logger.info("=====["+DateUtils.getCurrentDateTimeStr()+"]=====["+curServerKey+"]=====["+afterTime+"]=====");
if(updateTime!=null&&(afterTime<=5000)){//如果更新时间小于等于5S，说明该服务可用
if(server.getState()==SchedulerServer.STATE_RUN){
runServer = server;
runServerError = false;
}
if(readServer==null&&server.getState()==SchedulerServer.STATE_WAIT){
readServer = server;
}
}else{
if(!server.getJobServer().equals(curServerKey)){
schedulerServerService.delete(server);
}
}
}
//如果当前服务未被实例到数据库，放弃本次运行检查
if(curServer==null){
curServer = new SchedulerServer(curServerKey,SchedulerServer.STATE_WAIT,curTime);
schedulerServerService.save(curServer);
return false;
}else{
if(runServerError){ //如果运行服务不可用
/**
* 如果当前服务为备用服务
*/
curServer.setUpdateTime(curTime);
if(readServer!=null&&readServer.getJobServer().equals(curServerKey)){
if(runServer!=null){
schedulerServerService.delete(runServer);
}
curServer.setState(SchedulerServer.STATE_RUN);
schedulerServerService.save(curServer);
return true;
}else{
schedulerServerService.save(curServer);
return false;
}
}else{ //如果可用，只更新当前服务状态
curServer.setUpdateTime(curTime);
schedulerServerService.save(readServer);
if(runServer.getJobServer().equals(curServerKey)){
return true;
}else{
return false;
}
}
}
}
}
另涉及到一个集群节点状态实体如下
public class SchedulerServer {
private static final long serialVersionUID = 1204654463096914772L;
public static final int STATE_RUN = 1;
public static final int STATE_WAIT = 0;
private String jobServer; //定时任务服务（集群内每个服务启动一个）
private Integer state; //状态
private Long updateTime; //更新时间
springxml配置与前文一致
主要完成以下几个功能：
- 减少quartz配置；
- 通过配置文件动态修改quartz是否运行（我的配置文件也是动态的）；
- 动态修改定时任务的状态预计运行规则；
- 实现集群中只有一个节点运行任务；
- 实现集群中运行节点异常其他节点自动唤醒运行方式；