前言
早在实习的时候,笔者就接触了spring-task和quartz框架,相信任何java程序员都会有定时任务的需求,在单机上使用定时任务是非常简单的,但是在集群环境中就显得比较棘手:
- 如何限定只有一台机器在执行定时任务
- 某台服务宕机以后如何进行故障转移
- 如何确定正在执行的是哪一台服务
此问题官方肯定有解决方案,资料也不少:quartz集群分布式(并发)部署解决方案-Spring
不过笔者是个比较懒的人,上述的解决方案需要配置不少的quartz表,并且业务里不需要动态配置定时任务,对quartz框架并无依赖,所以就考虑使用其他方式来实现定时任务的分布式调度。
初尝试Uncode-Schedule
Uncode-Schedule 是笔者在开源中国中看到的一个开源的分布式调度任务组件,与spring集成比较容易,使用方便,借助了zookeeper。在此我不多做评价,毕竟是个人的一个开源框架,使用过程中还是踩了一些坑,并且文档尚未齐全,用了一段时间后决定弃用了。有兴趣的同学可以了解下:分布式任务调度组件 Uncode-Schedule
所幸在填坑过程中对这个任务调度的过程有了一定了解,于是决定自己实现一个简单的任务调度器。
通过redis实现任务调度思路
实现功能之前,回顾下之前遇到的三个问题:单点执行,故障转移,服务状态。结合着redis的一些接口特性,解决思路如下:
- 使用redis作为任务调度中心,采用了redis的自动过期与分布式锁特性
- 每个服务的ip加项目名作为每台服务的唯一别名
- 通过redis中对应key值中的value来判定执行的是哪台服务: 如redis中key为 schedular_root:projectA, value为192.168.1.187. 意为项目projectA当前执行任务的节点为192.168.1.187这台机器上的服务
- 每次执行任务之前判定下redis中schedular_root:projectA是否为空,如果为空,则设置当前ip进去,设置一定时间的有效期,并执行定时任务;如果不为空,判断是否与本机ip相同,相同则执行定时任务,否则跳过
- 设置有效期是为了某台机器发生故障时能进行故障转移
核心流程代码
此解决方案非常简单,核心代码也十分容易集成,为了减少耦合度,我们采用了spring的aop进行实现。
核心AOP代码
@Aspect
@Component
@Log4j
public class QuartzAop {
public boolean checkStatus(){
String key = "schedular_root:projectA";
try {
// 这个接口必然是并发的,所以加分布式锁
while (true) {
// 一秒的超时时间
boolean lock = RedisUtil.checkLock(key,1);
if (lock) {
// 获取到锁,才能跳出
break;
}
}
String ip = InetAddress.getLocalHost().getHostAddress();
// 获取服务器上的工作ip
String currentIp = RedisUtil.get(key);
// 如果为空的时候,设置进去
if(currentIp == null){
RedisUtil.setex(key, ip, 10);
return true;
}
// 就是当前机器,则返回true
if(currentIp.equals(ip)){
return true;
}else{
return false;
}
} catch (Exception e) {
log.error(e);
return false;
} finally {
RedisUtil.unLock(key);
}
}
@Around("@annotation(org.springframework.scheduling.annotation.Scheduled)")
public void around(ProceedingJoinPoint jp) throws Throwable{
if(checkStatus()){
String ip = InetAddress.getLocalHost().getHostAddress();
log.info("现在正在执行"+jp.getSignature()+":"+ip);
jp.proceed();
}
}
}
RedisUtil中的锁代码
public static boolean checkLock(String key,int second) {
String lockKey = "lock:" + key;
try {
// 1表示之前不存在,设置成功
if (setnx(lockKey, "lock") == 1) {
// 设置有限期
setExpiredTime(lockKey, second);
return true;
} else {
// 50毫秒的延迟,避免过多请求
try {
Thread.sleep(50L);
} catch (InterruptedException e) {
log.error(e);
}
return false;
}
} catch (RedisException e) {
log.error(e);
return true;
}
}
核心代码已经全部放送,关于redis实现分布式锁,会在另外找个时间和大家分享,欢迎关注~
网友评论