1.场景描述
老项目需要多机部署，项目中有几十个定时任务，一旦多机部署，定时任务就会重复执行，固定ip与错开时间方案都存在较大弊端，最终采用的方案是:AOP+排他锁的方式，软件老王已验证通过，介绍下，有需要的朋友可以参考下。
2.解决方案
软件老王基本方案是采用：AOP+排他锁的方式。
（1）目前老项目有几十个定时任务，采用AOP的方式，可以保证代码的无侵入（即使简单的微侵入，例如增加几行代码，测试验证的工作量也会比较大的）。
（2）采用排他锁的方式，保证批处理的高可用，不重复执行。
2.1 AOP编程
Aop的概念就不说了，就是面向切面编程，通俗点就是统一处理一类问题，比如日志、请求鉴权等，刚开始不确定是否可行，系统中的批处理是使用spring注解的方式@Scheduled进行批处理，采用aop对注解@Scheduled进行编程，统一拦截批处理，代码如下：
/**
* 软件老王-AOP处理类
*/
@Aspect
@Component
public class ScheduledAspect {
@Autowired
ScheduleService scheduleService ;
@Pointcut( "@annotation(org.springframework.scheduling.annotation.Scheduled)")
public void scheduled() {
}
@Around("scheduled()")
public Object scheduled(ProceedingJoinPoint pjd) {
Object result = null;
String taskName = pjd.getSignature().getName();
try {
if (scheduleService.isInvoke(taskName)){
return result;
}
result = pjd.proceed();
scheduleService.end(taskName);
} catch (Throwable e) {
throw new RuntimeException(e);
}
return result;
}
}
说明：
（1）面向标签编程
@Pointcut( "@annotation(org.springframework.scheduling.annotation.Scheduled)")
这样注解会拦截标签@Scheduled。
（2）使用aop的环绕标签 @Around("scheduled()")
@before标签拿不到执行完成状态，需要使用环绕标签@@Around，在标签中可以拿到执行完成后状态，以便放开锁。
result = pjd.proceed();
（3）结合排他锁使用
@Autowired
ScheduleService scheduleService ;
2.2 排他锁
排他锁，简单来说就是通过数据库总的标志位+版版号进行的控制.
软件老王的代码如下，：
/**
* 软件老王-排他锁服务类
*/
@Service
public class ScheduleService {
@Autowired
ScheduleClusterMapper scheduleClusterMapper;
public boolean isInvoke(String taskName) {
boolean isValid = false;
try {
ScheduleCluster carIndexEntity = scheduleClusterMapper.selectByTaskName(taskName);
int execute = carIndexEntity.getExecute();
String ip = InetAddress.getLocalHost().getHostAddress();
long currentTimeMillis = System.currentTimeMillis();
long time = carIndexEntity.getUpdatedate().getTime();
if (execute == 0) {
isValid = start(taskName, carIndexEntity.getVersion(), ip);
}
} catch (UnknownHostException e) {
e.printStackTrace();
}
return isValid;
}
//执行锁机制，软件老王
public boolean start(String taskName, int version, String ip) {
ScheduleCluster scheduleCluster = new ScheduleCluster();
scheduleCluster.setVersion(version);
scheduleCluster.setExecuteIp(ip);
scheduleCluster.setUpdatedate(DateUtil.getCurrentTime());
scheduleCluster.setTaskName(taskName);
scheduleCluster.setExecute(1);
int count = scheduleClusterMapper.updateByTaskName(scheduleCluster);
if (count > 0) {
return true;
}
return false;
}
//执行解锁机制，软件老王
public void end(String taskName) {
ScheduleCluster scheduleCluster = new ScheduleCluster();
scheduleCluster.setUpdatedate(DateUtil.getCurrentTime());
scheduleCluster.setTaskName(taskName);
scheduleCluster.setExecute(0);
scheduleClusterMapper.updateNormalByTaskName(scheduleCluster);
}
}
说明：
大的原理是在where条件后带上版本号，在update中更新version+1，这样通过影响数据库的影响条数，来判断是否拿到锁。
（1）主类中调用start方法，该方法是更新批处理状态，软件老王这里设置了一个小点，在updateByTaskName的mybatis方法中，有个version+1的更新；
（2）end方法放在更新完成后，释放锁。
（3）其实还有一个点，可以考虑下，需要有个机制，比如出现异常情况，刚好批处理执行中，重启服务了等，下次批处理执行前，假如锁还未释放，代码中增加释放锁的机制。
2.3 数据库相关
（1）数据库表设计
（2）mybatis相关方法
（1）第一个是start对应方法，执行锁和version增加。
<update id="updateByTaskName" parameterType="com.yutong.dmp.entity.ScheduleCluster">
update t_schedule_cluster
<set>
<if test="executeIp != null">
execute_ip = #{executeIp,jdbcType=VARCHAR},
</if>
<if test="version != null">
version = #{version,jdbcType=INTEGER} + 1,
</if>
<if test="execute != null">
execute = #{execute,jdbcType=INTEGER},
</if>
<if test="status != null">
status = #{status,jdbcType=VARCHAR},
</if>
<if test="createby != null">
createby = #{createby,jdbcType=VARCHAR},
</if>
<if test="createdate != null">
createdate = #{createdate,jdbcType=TIMESTAMP},
</if>
<if test="updateby != null">
updateby = #{updateby,jdbcType=VARCHAR},
</if>
<if test="updatedate != null">
updatedate = #{updatedate,jdbcType=TIMESTAMP},
</if>
</set>
where task_name = #{taskName,jdbcType=VARCHAR}
and version = #{version,jdbcType=INTEGER}
and status ='1'
</update>
（2）第二个是释放锁，更改excute为0。
<update id="updateNormalByTaskName" parameterType="com.yutong.dmp.entity.ScheduleCluster">
update t_schedule_cluster
<set>
<if test="executeIp != null">
execute_ip = #{executeIp,jdbcType=VARCHAR},
</if>
<if test="version != null">
version = #{version,jdbcType=INTEGER},
</if>
<if test="execute != null">
execute = #{execute,jdbcType=INTEGER},
</if>
<if test="status != null">
status = #{status,jdbcType=VARCHAR},
</if>
<if test="createby != null">
createby = #{createby,jdbcType=VARCHAR},
</if>
<if test="createdate != null">
createdate = #{createdate,jdbcType=TIMESTAMP},
</if>
<if test="updateby != null">
updateby = #{updateby,jdbcType=VARCHAR},
</if>
<if test="updatedate != null">
updatedate = #{updatedate,jdbcType=TIMESTAMP},
</if>
</set>
where task_name = #{taskName,jdbcType=VARCHAR}
and status ='1'
</update>
I’m 「软件老王」，如果觉得还可以的话，关注下呗，后续更新秒知！欢迎讨论区、同名公众号留言交流！
1、Oracle
思路：Insert表内容时，主键重复，无法插入
范围：适用于执行频率低的定时任务
方法：新建一张任务执行表，每次执行定时任务之前先insert该表。
比如：每天执行一次批处理操作。主键可以设置为批处理name+日期（年月日）。集群服务器可能会同时去insert该表，而只有一台服务器能插入成功，则只让这一台服务器执行该批处理任务。
2、Redis
范围：适用于执行频率高的定时任务
有如下两种思路方法：
思路1：采用的是redis中list的push和pop操作。因为Redis是单线程，所有命令依次执行，所以不会出现多台服务器同时访问的情况。
方法：系统初始化时，向Redis中lpush一个list(key,1)，作为标识Flag。每台服务器执行批处理之前,先去rpop该list，只有获取到该标识的服务器才能执行批处理任务，执行完毕后再lpush回一个list(key,1)。
比如：执行任务前先执行rpopRedisFlag，任务完成后执行lpushRedisFlag.
public class AutoJobActDis extends QuartzJobBean {
@Override
protected void executeInternal(JobExecutionContext arg0) throws JobExecutionException {
if (rpopRedisFlag("AUTO_JOB_ACTDIS")) {
this.executeJob(this);
lpushRedisFlag("AUTO_JOB_ACTDIS");
}
}
/**
* 从缓存中取得Flag，获取定时任务执行权限
*
* @param redisKey
* @return
*/
public boolean rpopRedisFlag(String redisKey) {
String redisValue = redisService.rpopRedisList(redisKey);
if (StringUtil.isEmpty(redisValue)) {
return false;
} else {
return true;
}
}
/**
* 缓存中增加Flag
*
* @param redisKey
*/
public void lpushRedisFlag(String redisKey) {
redisService.lpushRedisList(redisKey, "1");
}
｝
public class RedisService {
/**
* 存储Redis队列，顺序存储
* @param key redis键名
* @param value 键值
*/
public void lpushRedisList(String key, String value) {
Jedis jedis = null;
try {
jedis = jedisClient.getClient();
jedis.lpush(key, value);
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
if (jedis != null) {
jedis.close();
}
}
}
/**
* 移除并获取列表最后一个元素
* @param key
* @return
*/
public String rpopRedisList(String key) {
Jedis jedis = null;
try {
jedis = jedisClient.getClient();
return jedis.rpop(key);
} catch (Exception e) {
logger.error(e.getMessage(), e);
} finally {
if (jedis != null) {
jedis.close();
}
}
return null;
}
｝
思路2：采用的是redis中String的incr和expire操作。因为Redis是单线程，所有命令依次执行，所以不会出现多台服务器同时访问的情况。
语法：INCR key 将 key 中储存的数字值增一。
Redis Incr 命令将 key 中储存的数字值增一。
如果 key 不存在，那么 key 的值会先被初始化为 0 ，然后再执行 INCR 操作。
如果值包含错误的类型，或字符串类型的值不能表示为数字，那么返回一个错误。
本操作的值限制在 64 位(bit)有符号数字表示之内。
EXPIRE key seconds 为给定 key 设置过期时间。
Redis Expire 命令用于设置 key 的过期时间。key 过期后将不再可用。
方法：每台服务器执行批处理之前,先去incr一个固定key，只有返回值等于1的服务器才能执行批处理任务，然后将该key值设置过期时间。
比如：
jedis = jedisTool.getJedis();
if(jedis.incr(PRE_AGAIN_REDIS+batchId)!=1) {
logger.info("其他机器已经重跑");
return ;
} else {
jedis.expire(PRE_AGAIN_REDIS+batchId, 60);//设置60s过期
}
————————————————
版权声明：本文为CSDN博主「Java小农」的原创文章，遵循CC 4.0 BY-SA版权协议，转载请附上原文出处链接及本声明。
原文链接：https://blog.csdn.net/wyongan1128/article/details/81208725