Java应用集群下的定时任务处理方案(mysql)

今天来讲一个Java多机部署下定时任务的处理方案。

需求: 有两台服务器同时部署了同一套代码， 代码中写有spring自带的定时任务，可是每次执行定时任务时只须要一台机器去执行。

当拿到这个需求时我脑子中立马出现了两个简单的解决方案：

• 利用ip进行判断， 两台机器ip确定不同， 指定某一台机器的ip运行。
• 只在一台机器上部署定时任务的代码。

最后两个方案又都被本身否决了。 第一条，若是指定ip的机器出现了问题怎么办？ 例如说宕机了， 那么该制定ip的机器上的定时任务是否是就没法运行了？若是之后该服务器迁移致使ip变化怎么办？
第二条， 同上， 还有就是要维护两套代码很不方便。

由于上面两个假设都不成立， 只能另找他法。 因而便想到利用mysql去解决， 以前了解过一点mysql的锁机制， 知道若是有同时的两个任务去写数据库中同一条记录， 只有一条会成功， 这是利用了mysql的排他锁。

下面就开始代码演示， 这里主要想给你们的是一个思路的提示， 代码仍是很简单的。

• 首先须要单首创建一张表

CREATE TABLE `t_schedule_cluster` (
  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '@cname:主键',
  `execute` int(1) NOT NULL COMMENT '@cname:执行状态',
  `version` int(11) NOT NULL COMMENT '@cname:版本号\r\n            ',
  `task_name` varchar(128) NOT NULL COMMENT '@cname:任务名称\r\n            ',
  `execute_ip` varchar(32) DEFAULT NULL COMMENT '@cname:执行ip\r\n            ',
  `update_time` datetime DEFAULT NULL COMMENT '@cname:修改时间\r\n            ',
  PRIMARY KEY (`id`),
  KEY `Index_series_id` (`execute`)
) ENGINE=InnoDB AUTO_INCREMENT=8 DEFAULT CHARSET=utf8 COMMENT='@cname:多机定时任务调度';

看一下建成后的表结构， 注释写的已经很清楚了， 初始化时须要添加一些定时任务的名称（task_name）， 这个要和你代码中保持一致， 后面会提到：

• 代码
  首先看下我代码中用到的spring定时任务：

代码
首先看下我代码中用到的spring定时任务：

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns=
xmlns:xsi=
       xmlns:task=
       xsi:schemaLocation=
       default-lazy-init="true">
    <description>使用Spring的 Scheduled的定时任务配置</description>
    <!--支持annotation的方式-->
    <task:annotation-driven  />
    <task:scheduler id="springScheduler"  pool-size="10"/>
    <task:scheduled-tasks scheduler="springScheduler">
        <!-- 测试使用， 项目启动后每隔一分钟执行一次 -->
        <task:scheduled ref="listCarAction" method="listCar" cron="0 0/1 0 * * ?"/>
        <task:scheduled ref="listCarAction" method="listCar" cron="0 0/1 0 * * ?"/>
    </task:scheduled-tasks>
</beans>

相信你们都是用过这种定时任务的设置方法， 由于它是spring自带的， 因此使用起来很方便， 这里我指定了两个定时任务来模拟两台机器的状况， 两个定时任务都是项目启动后每隔一分钟执行一次。

而后看看这个listCar中的代码：

//定时任务的名称， 这个和数据库中的task_name是保持一致的， 保证要执行该定时任务。
public static final String LIST_CAR_TASK = "listCarTask";
private ScheduleClusterTask scheduleClusterTask;
//这个时间是根据spring-scheduler.xml中配置的定时刷新时间, 好比说咱们之后要设置这个定时任务时4小时刷新一次
public static final long maxExpireTime = 4 * 3600;

public void listCar() {
        if (scheduleClusterTask.isValidMachine(maxExpireTime, CommonConstants.ScheduleTaskName.LIST_CAR_TASK)) {
            //执行具体的task方法， 
            doTask();
            //将execute状态更新为0
            scheduleClusterTask.end(LIST_CAR_TASK);
        }
    }

最后看下最核心的代码：ScheduleClusterTask.java

/**
 * 多机定时任务工具类
 * Created by WangMeng on 2017/4/12.
 */
@Service
public class ScheduleClusterTask {

    private ScheduleClusterEntityService scheduleClusterEntityService;

    /**
     * 这里由于两台机器都有一样的定时任务， 会同时执行这个方法，只有一台机器能够执行成功，返回true。
     * @param maxExpireTime 最大的检查时间。
     * @param taskName 任务名称。
     * @return
     */
    public boolean isValidMachine(long maxExpireTime, String taskName) {
        boolean isValid = false;
        try {
            //经过taskName去数据库中查找到该条记录， 若是你们使用的是mybatis这里须要改一下， 就是一个简单的查询操做
            ScheduleClusterEntity carIndexEntity = scheduleClusterEntityService.findOne(ScheduleClusterEntity.Fields.taskName.eq(taskName));
            int execute = carIndexEntity.getExecute();
            String ip = InetAddress.getLocalHost().getHostAddress();
            long currentTimeMillis = System.currentTimeMillis();
            long time = carIndexEntity.getUpdateTime().getTime();
            if (execute == 0 && time + maxExpireTime - 1000 < currentTimeMillis) {
                isValid = checkMachine(taskName, carIndexEntity, ip);
            } else if (time + maxExpireTime - 1000 < currentTimeMillis){
                //这里要判断下， 若是上一次执行出现异常致使execute没有更新为0， 那么这里要判断上一次更新时间的间隔。
                isValid = checkMachine(taskName, carIndexEntity, ip);
            }
        } catch (UnknownHostException e) {
            e.printStackTrace();
        }
        return isValid;
    }

    /**
     * end方法主要是将excute(是否正在执行的标志位，0:没有执行， 1:正在执行)更新为0
     * @param taskName
     * @return
     */
    public boolean end (String taskName) {
        ScheduleClusterEntity carIndexEntity = scheduleClusterEntityService.findOne(ScheduleClusterEntity.Fields.taskName.eq(taskName));
        //将execute状态更新为0
        return scheduleClusterEntityService.end(carIndexEntity);
    }

    private boolean checkMachine(String taskName, ScheduleClusterEntity carIndexRefresh, String ip) {
        return scheduleClusterEntityService.start(taskName, carIndexRefresh.getVersion(), ip);
    }

    @Autowired
    public void setScheduleClusterEntityService(ScheduleClusterEntityService scheduleClusterEntityService) {
        this.scheduleClusterEntityService = scheduleClusterEntityService;
    }
}

这里还有start方法， 看看怎样的操做：

@Repository
public class DefaultScheduleClusterEntityDao extends AbstractDao<ScheduleClusterEntity> implements ScheduleClusterEntityDao {
    @Override
    public boolean start(String taskName, int version, String ip) {
        String sql = "update t_schedule_cluster set execute = 1, " +
                "version = ?, execute_ip = ?, update_time = ?" +
                " where task_name = ? and version = ?";
        Sql s = new Sql(sql);
        s.addParam(version + 1);
        s.addParam(ip);
        s.addParam(SqlTimeUtils.nowTimestamp());
        s.addParam(taskName);
        s.addParam(version);
        return 1 == executeUpdate(s);
    }
}

核心的代码到了这里就没有了， 代码确实是很是很是的简单， 有兴趣的话你们能够在本地测试一下就能够。
固然还有更多很好地解决方案， 我这里秉承的是最简单的处理方式， 若是你们对我这个方案有疑问或者作的很差的地方都但愿你们可以提出来， 谢谢了，