Java应用集群下的定时任务处理

概述

需求：有两台服务器同时部署了同一套代码。代码中写有Spring自带的定时任务，但是每次执行定时任务都只需要一台及其去执行。

解决：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:多机定时任务调度';

代码中的Spring定时任务

<?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:task="http://www.springframework.org/schema/task"
xsi:schemaLocation="http://www.springframework.org/schema/beans http://www.springframework.org/schema/beans/spring-beans-3.2.xsd http://www.springframework.org/schema/task http://www.springframework.org/schema/task/spring-task-3.2.xsd" 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>
</bean

这里指定了两个定时任务来模拟两台及其的情况，两个定时任务都是项目驱动后每隔一分钟执行一次。

再看看这个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);
}
}

核心代码

/**
* 多机定时任务工具类
* 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);
}
}

通过redis实现任务调度思路 ： http://www.jianshu.com/p/48c5b11b80cd

出自： http://www.cnblogs.com/wang-meng/p/6718199.html