集群环境下的单例任务调度基类

最近为公司框架提供了基于ID的单例的任务调度基类，用于集群的环境下，一个任务同一时间只能被调度一次。

 

接口声明
继承该基类的任务需要覆盖两个抽象函数：
        protected abstract String getServiceID();//服务ID,与数据库中的和timeservice.xml中的要配置相同       
        protected abstract void doTask();//执行任务，需覆盖的业务方法

 

实现机理

每个任务要有唯一的ServiceId，该任务在t_s_sys_timeservice 中有一条记录，同时有一个state记录着状态信息，state为0表示未占用，state为1表示已占用。

设计上还要考虑处理异常的情况，通过引入一个超时时间，如果已经超时，则认为改状态为未被占用

 

关键代码

 

package com.itown.framework.schedule;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;

import org.apache.log4j.Logger;

import com.itown.framework.exception.ServiceInvokeException;
import com.itown.framework.impl.ThreadContext;
import com.itown.framework.persistence.DataSourceFactory;
import com.itown.framework.persistence.ParamsMaker;
import com.itown.framework.persistence.SingleDBHelper;
import com.itown.framework.transaction.FrameworkTransaction;
import com.itown.framework.transaction.JDBCTransaction;

/**
*	timeout_interval用来设定超时时间，单位是秒，极端情况state可能为1，当达到超时时间时，再次执行，要求超时时间一定要大于task的执行时间
*/
public abstract class SingletonTimeService extends TimeService {
	
	private static final Logger	logger	= Logger.getLogger(SingletonTimeService.class);
	
	private String selectSql = "select * from t_s_sys_timeservice where (id = ? and state = '0') or (id = ? and sysdate > update_date + timeout_interval/86400 )"; 
	private String updateLockSql = "update t_s_sys_timeservice set update_date = sysdate,state = 1 where (id = ? and state = '0') or (id = ? and sysdate > update_date + timeout_interval/86400 )"; 
	private String updateUnlockSql = "update t_s_sys_timeservice set update_date = sysdate,state = '1' where id = ?"; 
	
	public void doTimeService() {
        String funcName = "SingletonTimeService.doTimeService";
        logger.info(funcName + " begin time:" + System.currentTimeMillis());
		ThreadContext ctx = ThreadContext.getThreadContext();
		FrameworkTransaction _tr = null;
		Connection _conn = null;
		try {
			_conn = DataSourceFactory.getDefaultConnection();			
			_tr = new JDBCTransaction(_conn);
			_tr.begin();			

			//争任务，此处无法采用框架提供的封装
			PreparedStatement ps = null;
			ResultSet rs = null;
			try {
				logger.info(funcName + " selectSql:" + selectSql);
				logger.info(funcName + " id:" + getServiceID());
				ps = _conn.prepareStatement(selectSql);
				ps.setString(1, getServiceID());
				ps.setString(2, getServiceID());
				rs = ps.executeQuery();
			} catch (SQLException ex) {
				logger.error(ex);
				throw ex;
			}
			int result =0;//result > 0 表示抢到任务
			if(rs.next()){
				SingleDBHelper helper = new SingleDBHelper(_conn);				
				//锁定任务
				try{
					logger.info(funcName + " updateLockSql:" + updateLockSql);
					logger.info(funcName + " id:" + getServiceID());
					result=helper.executeUpdate(updateLockSql,ParamsMaker.makeParams(getServiceID()));
					_tr.commit();
					
				}catch(Exception e){
					logger.error(e);
				}
				
				//执行任务
				try{
					_tr.begin();
					logger.info(funcName + " 开始执行timeservice(id="+getServiceID()+")任务...");
					if(result>0){
						doTask();
					}
					logger.info(funcName + " 结束执行timeservice(id="+getServiceID()+")任务。");
				}catch(Throwable e){
					logger.error(e);
					throw e;
					
				}				

				//解锁
				//注释此timeservice解锁业务，超时会自动满足timeservice执行条件；

				try{
					logger.info(funcName + " updateUnlockSql:" + updateUnlockSql);
					logger.info(funcName + " id:" + getServiceID());
					if(result>0){
						helper.executeUpdate(updateUnlockSql,ParamsMaker.makeParams(getServiceID()));
					}
				}catch(Exception e){
					logger.error(e);
				}
				
			}
			
			//事务相关操作
			_tr.commit();							
		} catch (Throwable ex) {
			if (_tr != null) {
				try {
					_tr.rollback();
				} catch (Exception ex1) {
					logger.error(ex1);
				}
			}
			throw new ServiceInvokeException("调度异常", ex);
		} finally {
			ctx.runShieldStatements();
			DataSourceFactory.release(_conn);
		}
	}

	protected abstract String getServiceID();
	
	protected abstract void doTask();
}