Spring定时任务Scheduled分布式/集群应用插件,支持redis缓存、zookeeper、数据库等
Spring 自带定时器Scheduled是单应用服务上的,不支持分布式环境。目前Spring集群环境下的定时任务调度执行的现成解决方案有 quartz 和 spring cloud task两种,不知道还有没有其它的解决方案,如果有麻烦请告知一下。quartz 和 spring cloud task都是只能用数据库解决,并且如果现有的项目一直用的是@Scheduled方式做定时任务,因为要做..
Spring 自带定时器Scheduled是单应用服务上的,不支持分布式环境。目前Spring集群环境下的定时任务调度执行的现成解决方案有 quartz
和 spring cloud task
两种,不知道还有没有其它的解决方案,如果有麻烦请告知一下。
quartz
和 spring cloud task
都是只能用数据库解决,并且如果现有的项目一直用的是@Scheduled
方式做定时任务,因为要做集群而改用 quartz
或 spring cloud task
,那么全部定时任务都要重写改成对应的实现方式。所以,这就有了插件spring-scheduling-cluster
。
spring-scheduling-cluster
是对Spring原生的@Scheduled
添加集群控制,原理是:使用如数据库、缓存等中间做分布式锁,对每个任务添加锁操作,控制每个任务每次只有一个服务器可以执行。
spring-scheduling-cluster文档:https://gitee.com/lnkToKing/spring-scheduling-cluster
将插件应用到项目中
添加插件
<dependency>
<groupId>pres.lnk.springframework</groupId>
<artifactId>spring-scheduling-cluster</artifactId>
<version>1.0-BATE</version>
</dependency>
#### 启用插件
使用用注解@EnableClusterScheduling
代替注解@EnableScheduling
继承AbstractScheduler类实现中间件,并注册成Spring Bean
这里贴一下redis缓存做中间件代码,将这段代码添加到项目的java文件里就可以用了。文档里还有mysql和zookeeper的实现和AbstractScheduler
类说明
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Component;
import java.lang.reflect.Method;
import java.util.concurrent.TimeUnit;
/**
* redis调度器中间件
*
* @Author lnk
* @Date 2018/2/28
*/
@Component
public class RedisSchedulerImpl extends AbstractScheduler {
private static final String CACHE_PREFIX = "scheduler_";
private static final String MAX_LEVEL = "maxLevel";
@Autowired
private RedisTemplate redisTemplate;
@Override
public boolean check(String id) {
Long time = getCache(id);
return time == null || currentTimeMillis() > time;
}
@Override
public boolean lock(String id, long timeoutMillis) {
String key = prefixKey(id);
long nextTimeMillis = currentTimeMillis() + timeoutMillis;
boolean flag = redisTemplate.opsForValue().setIfAbsent(key, nextTimeMillis);
if(flag){
redisTemplate.expire(key, timeoutMillis < 0 ? 1 : timeoutMillis, TimeUnit.MILLISECONDS);
}
return flag;
}
@Override
public void relock(String id, long timeoutMillis) {
String key = prefixKey(id);
redisTemplate.expire(key, timeoutMillis < 0 ? 1 : timeoutMillis, TimeUnit.MILLISECONDS);
}
@Override
public long currentTimeMillis() {
RedisConnection connection = redisTemplate.getConnectionFactory().getConnection();
long time = connection.time();
connection.close();
return time;
}
/*
* ------------- 下面3个重写方法可选 -------------
* 如果启用了主从服务器则 keepAlive() 和 getMaxAliveLevel() 必须重写
*/
@Override
public void keepAlive() {
int level = getLevel();
String key = prefixKey(MAX_LEVEL);
Integer maxLevel = getCache(MAX_LEVEL);
//如果maxLevel为null,就将当前level写进去
if (maxLevel == null) {
//控制只能写一个,防止并发低级别把高级别覆盖了
boolean result = redisTemplate.opsForValue().setIfAbsent(key, level);
if (result) {
//写入成功,添加过期时间,过期时间追加5秒,避免出现误差,在下次刷新时间前就失效了,被低级别的服务器抢先执行任务
redisTemplate.expire(key, getHeartTime() + 5, TimeUnit.SECONDS);
return;
} else {
//写入失败,则获取最高级别跟当前级别做比较,避免低级别抢先写入成功
maxLevel = getCache(key);
}
}
//如果当前级别比缓存里级别还要高,则覆盖它
if (maxLevel > level) {
redisTemplate.delete(key);
keepAlive();
return;
}
//如果当前级别跟最高级别同级,则刷新过期时间
if (maxLevel == level) {
Long expire = redisTemplate.getExpire(prefixKey(MAX_LEVEL));
//只有过期时间有还有10秒的时候才刷新时间,避免多个服务器同时刷新
if (expire < 10) {
//加5秒理由同上
redisTemplate.expire(key, getHeartTime() + 5, TimeUnit.SECONDS);
}
}
}
@Override
public int getMaxAliveLevel() {
Integer level = getCache(MAX_LEVEL);
if (level != null) {
return level;
}
//如果没有最高级别则返回当前服务器级别
return getLevel();
}
/**
* 任务执行结束后调用的方法,可以写日志,不需要做后续处理可不重写
*/
@Override
public void executed(Method method, Object targer, long startTimeMillis, long endTimeMillis, String description) throws Exception {
//可写任务执行日志
switch (getStatus()){
case AbstractScheduler.SUCCESS :
// 执行成功
case AbstractScheduler.FAIL_CHECK :
// 未执行任务,已有服务器执行过
case AbstractScheduler.FAIL_LOCK :
// 未执行任务,获取锁失败
case AbstractScheduler.FAIL_LEVEL :
// 未执行任务,级别低
case AbstractScheduler.ERROR :
// 执行任务出现异常,如果不想处理可以抛回由Spring处理
throw getException();
}
}
public <T> T getCache(String key) {
key = prefixKey(key);
return (T)redisTemplate.opsForValue().get(key);
}
private static String prefixKey(String key) {
return CACHE_PREFIX.concat(key.replaceAll("\\W+", "_"));
}
}
更多推荐
所有评论(0)