源码回顾
调度中心触发任务之后，他的调用链如下
RemoteHttpJobBean> executeInternal > XxlJobTrigger > trigger ,
通过之前的分析xxl-job 源码解读 （二） ， 我们可以了解到，xxl-job他的路由策略主要发生在trigger这个方法中
public static void trigger(int jobId) {
// 通过JobId从数据库中查询该任务的具体信息
XxlJobInfo jobInfo = XxlJobDynamicScheduler.xxlJobInfoDao.loadById(jobId); // job info
if (jobInfo == null) {
logger.warn(">>>>>>>>>>>> trigger fail, jobId invalid，jobId={}", jobId);
return;
}
// 获取该类型的执行器信息
XxlJobGroup group = XxlJobDynamicScheduler.xxlJobGroupDao.load(jobInfo.getJobGroup()); // group info
// 匹配运行模式
ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION); // block strategy
// 匹配失败后的处理模式
ExecutorFailStrategyEnum failStrategy = ExecutorFailStrategyEnum.match(jobInfo.getExecutorFailStrategy(), ExecutorFailStrategyEnum.FAIL_ALARM); // fail strategy
// 获取路由策略
ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null); // route strategy
// 获取该执行器的集群机器列表
ArrayList<String> addressList = (ArrayList<String>) group.getRegistryList();
// 判断路由策略 是否为 分片广播模式
if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum && CollectionUtils.isNotEmpty(addressList)) {
for (int i = 0; i < addressList.size(); i++) {
String address = addressList.get(i);
//定义日志信息
XxlJobLog jobLog = new XxlJobLog();
// .....省略
ReturnT<String> triggerResult = new ReturnT<String>(null);
if (triggerResult.getCode() == ReturnT.SUCCESS_CODE) {
// 4.1、trigger-param
TriggerParam triggerParam = new TriggerParam();
triggerParam.setJobId(jobInfo.getId());
triggerParam.setBroadcastIndex(i); // 设置分片标记
triggerParam.setBroadcastIndex(addressList.size());// 设置分片总数
// ......省略组装参数的过程
// 根据参数以及 机器地址，向执行器发送执行信息 ， 此处将会详细讲解runExecutor 这个方法
triggerResult = runExecutor(triggerParam, address);
}
// 将日志ID，放入队列，便于日志监控线程来监控任务的执行状态
JobFailMonitorHelper.monitor(jobLog.getId());
logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId());
}
} else {
// 出分片模式外，其他的路由策略均走这里
//定义日志信息
XxlJobLog jobLog = new XxlJobLog();
jobLog.setJobGroup(jobInfo.getJobGroup());
// .....省略
ReturnT<String> triggerResult = new ReturnT<String>(null);
if (triggerResult.getCode() == ReturnT.SUCCESS_CODE) {
// 4.1、trigger-param
TriggerParam triggerParam = new TriggerParam();
triggerParam.setJobId(jobInfo.getId());
triggerParam.setExecutorHandler(jobInfo.getExecutorHandler());
triggerParam.setBroadcastIndex(0); // 默认分片标记为0
triggerParam.setBroadcastTotal(1); // 默认分片总数为1
// .... 省略组装参数的过程
// 此处使用了策略模式， 根据不同的策略 使用不同的实现类，下面将会详细讲解
triggerResult = executorRouteStrategyEnum.getRouter().routeRun(triggerParam, addressList);
}
JobFailMonitorHelper.monitor(jobLog.getId());
logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId());
}
}
上面的代码主要讲了分片广播这个策略的实现以及xxl-job的其他路由策略的调用位置在哪里。
ExecutorRouteStrategyEnum枚举类
这个是xxl-job路由策略非常重要的一个类， 该类通过枚举的方式，把路由key， 和策略实现类进行了一个聚合、
//ExecutorRouteStrategyEnum
public enum ExecutorRouteStrategyEnum {
FIRST("第一个", new ExecutorRouteFirst()),
LAST("最后一个", new ExecutorRouteLast()),
ROUND("轮循", new ExecutorRouteRound()),
RANDOM("随机", new ExecutorRouteRandom()),
CONSISTENT_HASH("一致性哈希", new ExecutorRouteConsistentHash()),
LEAST_FREQUENTLY_USED("最不经常使用", new ExecutorRouteLFU()),
LEAST_RECENTLY_USED("最近最久未使用", new ExecutorRouteLRU()),
FAILOVER("故障转移", new ExecutorRouteFailover()),
BUSYOVER("忙碌转移", new ExecutorRouteBusyover()),
SHARDING_BROADCAST("分片广播", null);
ExecutorRouteStrategyEnum(String title, ExecutorRouter router) {
this.title = title;
this.router = router;
}
private String title;
private ExecutorRouter router;
public String getTitle() {
return title;
}
public ExecutorRouter getRouter() {
return router;
}
// 数据库中存的是枚举的名称，此处通过名称的对比，找到路由策略对应的枚举信息
public static ExecutorRouteStrategyEnum match(String name, ExecutorRouteStrategyEnum defaultItem){
if (name != null) {
for (ExecutorRouteStrategyEnum item: ExecutorRouteStrategyEnum.values()) {
if (item.name().equals(name)) {
return item;
}
}
}
return defaultItem;
}
}
分片广播
通过源码回顾，我们可以清晰的看到，当系统判断当前任务的路由策略是分片广播时， 就会遍历执行器的集群机器列表，
给每一台机器都发送执行消息，分片总数为集群机器数量，分片标记从0开始，上面的代码已经非常清楚了，此处不再赘述。
第一个
由上面对ExecutorRouteStrategyEnum的分析，我们可以看到，该策略对应的是 这个ExecutorRouteFirst执行策略类。 主要看
routeRun 这个方法
//ExecutorRouteFirst
public String route(int jobId, ArrayList<String> addressList) {
return addressList.get(0);
}
@Override
public ReturnT<String> routeRun(TriggerParam triggerParam, ArrayList<String> addressList) {
// 直接取集群地址列表里面的第一台机器来进行执行
String address = route(triggerParam.getJobId(), addressList);
// run executor
ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
// 将执行该任务的执行器地址，放入到结果里面返回，最后会记录到日志里面取
runResult.setContent(address);
return runResult;
}
最后一个
直接 从执行机集群列表的list里面取最后一个，源码如下
//ExecutorRouteLast
public String route(int jobId, ArrayList<String> addressList) {
return addressList.get(addressList.size()-1);
}
@Override
public ReturnT<String> routeRun(TriggerParam triggerParam, ArrayList<String> addressList) {
// 通过看上面的route方法，可以看到直接取得是list最后一个数据
String address = route(triggerParam.getJobId(), addressList);
// run executor
ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address);
return runResult;
}
轮循
主要看ExecutorRouteRound这个类里面的代码
//ExecutorRouteRound
private static ConcurrentHashMap<Integer, Integer> routeCountEachJob = new ConcurrentHashMap<Integer, Integer>();
// 缓存过期时间戳
private static long CACHE_VALID_TIME = 0;
private static int count(int jobId) {
// 如果当前的时间，大于缓存的时间，那么说明需要刷新了
if (System.currentTimeMillis() > CACHE_VALID_TIME) {
routeCountEachJob.clear();
// 设置缓存时间戳，默认缓存一天，一天之后会从新开始
CACHE_VALID_TIME = System.currentTimeMillis() + 1000*60*60*24;
}
// count++
Integer count = routeCountEachJob.get(jobId);
// 当第一次执行轮循这个策略的时候，routeCountEachJob这个Map里面肯定是没有这个地址的， count==null ,
// 当 count==null或者count大于100万的时候，系统会默认在100之间随机一个数字 ， 放入hashMap, 然后返回该数字
// 当系统第二次进来的时候，count!=null 并且小于100万， 那么把count加1 之后返回出去。
count = (count==null || count>1000000)?(new Random().nextInt(100)):++count; // 初始化时主动Random一次，缓解首次压力
// 为啥首次需要随机一次，而不是指定第一台呢？
// 因为如果默认指定第一台的话，那么所有任务的首次加载全部会到第一台执行器上面去，这样会导致第一台机器刚开始的时候压力很大。
routeCountEachJob.put(jobId, count);
return count;
}
public String route(int jobId, ArrayList<String> addressList) {
// 在执行器地址列表，获取相应的地址， 通过count(jobid) 这个方法来实现，主要逻辑在这个方法
// 通过count（jobId）拿到数字之后， 通过求于的方式，拿到执行器地址
// 例： count=2 , addresslist.size = 3
// 2%3 = 2 , 则拿list中下表为2的地址
return addressList.get(count(jobId)%addressList.size());
}
@Override
public ReturnT<String> routeRun(TriggerParam triggerParam, ArrayList<String> addressList) {
// 通过route方法获取执行器地址
String address = route(triggerParam.getJobId(), addressList);
// run executor
ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address);
return runResult;
}
随机
随机这个策略比较简单，通过在集群列表的大小内随机拿出一台机器来执行，比较简单，此处不再赘述
//ExecutorRouteRandom
private static Random localRandom = new Random();
public String route(int jobId, ArrayList<String> addressList) {
// Collections.shuffle(addressList);
return addressList.get(localRandom.nextInt(addressList.size()));
}
@Override
public ReturnT<String> routeRun(TriggerParam triggerParam, ArrayList<String> addressList) {
// address
String address = route(triggerParam.getJobId(), addressList);
// run executor
ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address);
return runResult;
}
一致性Hash
在讲这个策略之前，先说一下一致性Hash算法 ，
先构造一个长度为2^32的整数环（这个环被称为一致性Hash环），根据节点名称的Hash值（其分布为[0, 2^32-1]）
将服务器节点放置在这个Hash环上，然后根据数据的Key值计算得到其Hash值（其分布也为[0, 2^32-1]），接着
在Hash环上顺时针查找距离这个Key值的Hash值最近的服务器节点，完成Key到服务器的映射查找。
详细介绍： http://blog.csdn.net/u010412301/article/details/52441400
分组下机器地址相同，不同JOB均匀散列在不同机器上，保证分组下机器分配JOB平均；且每个JOB固定调度其中一台机器；
这个地方使用的Hash方法是作者自己写的，因为String的hashCode可能重复，需要进一步扩大hashCode的取值范围
//ExecutorRouteConsistentHash
private static int VIRTUAL_NODE_NUM = 5;
/**
* get hash code on 2^32 ring (md5散列的方式计算hash值)
* @param key
* @return
*/
private static long hash(String key) {
// md5 byte
MessageDigest md5;
try {
md5 = MessageDigest.getInstance("MD5");
} catch (NoSuchAlgorithmException e) {
throw new RuntimeException("MD5 not supported", e);
}
md5.reset();
byte[] keyBytes = null;
try {
keyBytes = key.getBytes("UTF-8");
} catch (UnsupportedEncodingException e) {
throw new RuntimeException("Unknown string :" + key, e);
}
md5.update(keyBytes);
byte[] digest = md5.digest();
// hash code, Truncate to 32-bits
long hashCode = ((long) (digest[3] & 0xFF) << 24)
| ((long) (digest[2] & 0xFF) << 16)
| ((long) (digest[1] & 0xFF) << 8)
| (digest[0] & 0xFF);
long truncateHashCode = hashCode & 0xffffffffL;
return truncateHashCode;
}
public String route(int jobId, ArrayList<String> addressList) {
//
//
TreeMap<Long, String> addressRing = new TreeMap<Long, String>();
for (String address: addressList) {
for (int i = 0; i < VIRTUAL_NODE_NUM; i++) {
// 通过自定义的Hash方法，得到服务节点的Hash值，同时放入treeMap
long addressHash = hash("SHARD-" + address + "-NODE-" + i);
addressRing.put(addressHash, address);
}
}
// 得到JobId的Hash值
long jobHash = hash(String.valueOf(jobId));
// 调用treeMap的tailMap方法，拿到map中键大于jobHash的值列表
SortedMap<Long, String> lastRing = addressRing.tailMap(jobHash);
// 如果addressRing中有比jobHash的那么直接取lastRing 的第一个
if (!lastRing.isEmpty()) {
return lastRing.get(lastRing.firstKey());
}
// 如果没有，则直接取addresRing的第一个
// 反正最终的效果是在Hash环上，顺时针拿离jobHash最近的一个值
return addressRing.firstEntry().getValue();
}
@Override
public ReturnT<String> routeRun(TriggerParam triggerParam, ArrayList<String> addressList) {
// address
String address = route(triggerParam.getJobId(), addressList);
// run executor
ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address);
return runResult;
}
最不经常使用
单个JOB对应的每个执行器，使用频率最低的优先被选举
//ExecutorRouteLFU
// 定义个静态的MAP， 用来存储任务ID对应的执行信息
private static ConcurrentHashMap<Integer, HashMap<String, Integer>> jobLfuMap = new ConcurrentHashMap<Integer, HashMap<String, Integer>>();
// 定义过期时间戳
private static long CACHE_VALID_TIME = 0;
public String route(int jobId, ArrayList<String> addressList) {
// 如果当前系统时间大于过期时间
if (System.currentTimeMillis() > CACHE_VALID_TIME) {
jobLfuMap.clear(); //清空
//重新设置过期时间，默认为一天
CACHE_VALID_TIME = System.currentTimeMillis() + 1000*60*60*24;
}
// 从MAP中获取执行信息
//lfuItemMap中放的是执行器地址以及执行次数
HashMap<String, Integer> lfuItemMap = jobLfuMap.get(jobId); // Key排序可以用TreeMap+构造入参Compare；Value排序暂时只能通过ArrayList；
if (lfuItemMap == null) {
lfuItemMap = new HashMap<String, Integer>();
jobLfuMap.put(jobId, lfuItemMap);
}
for (String address: addressList) {
// map中不包含，并且值大于一万的时候，需要重新初始化执行器地址对应的执行次数
// 初始化的规则是在机器地址列表size里面进行随机
// 当运行一段时间后，有新机器加入的时候，此时，新机器初始化的执行次数较小，所以一开始，新机器的压力会比较大，后期慢慢趋于平衡
if (!lfuItemMap.containsKey(address) || lfuItemMap.get(address) >1000000 ) {
lfuItemMap.put(address, new Random().nextInt(addressList.size())); // 初始化时主动Random一次，缓解首次压力
}
}
// 将lfuItemMap中的key.value, 取出来，然后使用Comparator进行排序，value小的靠前。
List<Map.Entry<String, Integer>> lfuItemList = new ArrayList<Map.Entry<String, Integer>>(lfuItemMap.entrySet());
Collections.sort(lfuItemList, new Comparator<Map.Entry<String, Integer>>() {
@Override
public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
return o1.getValue().compareTo(o2.getValue());
}
});
//取第一个，也就是最小的一个，将address返回，同时对该address对应的值加1 。
Map.Entry<String, Integer> addressItem = lfuItemList.get(0);
String minAddress = addressItem.getKey();
addressItem.setValue(addressItem.getValue() + 1);
return addressItem.getKey();
}
@Override
public ReturnT<String> routeRun(TriggerParam triggerParam, ArrayList<String> addressList) {
// address
String address = route(triggerParam.getJobId(), addressList);
// run executor
ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
runResult.setContent(address);
return runResult;
}
最近最久未使用
单个JOB对应的每个执行器，最久为使用的优先被选举 ， 此处使用的是linkHashMap来实现LRU算法的。
通过linkHashMap的每次get/put的时候会进行排序，最新操作的数据会在最后面。 从而取第一个数据就
代表是最久没有被使用的
//ExecutorRouteLRU
// 定义个静态的MAP， 用来存储任务ID对应的执行信息
private static ConcurrentHashMap<Integer, LinkedHashMap<String, String>> jobLRUMap = new ConcurrentHashMap<Integer, LinkedHashMap<String, String>>();
// 定义过期时间戳
private static long CACHE_VALID_TIME = 0;
public String route(int jobId, ArrayList<String> addressList) {
// cache clear
if (System.currentTimeMillis() > CACHE_VALID_TIME) {
jobLRUMap.clear();
//重新设置过期时间，默认为一天
CACHE_VALID_TIME = System.currentTimeMillis() + 1000*60*60*24;
}
// init lru
LinkedHashMap<String, String> lruItem = jobLRUMap.get(jobId);
if (lruItem == null) {
/**
* LinkedHashMap
* a、accessOrder：ture=访问顺序排序（get/put时排序）；false=插入顺序排期；
* b、removeEldestEntry：新增元素时将会调用，返回true时会删除最老元素；可封装LinkedHashMap并重写该方法，比如定义最大容量，超出是返回true即可实现固定长度的LRU算法；
*/
lruItem = new LinkedHashMap<>(16, 0.75f, true);
jobLRUMap.put(jobId, lruItem);
}
// 如果地址列表里面有地址不在map中，此处是可以再次放入，防止添加机器的问题
for (String address: addressList) {
if (!lruItem.containsKey(address)) {
lruItem.put(address, address);
}
}
// 取头部的一个元素，也就是最久操作过的数据
String eldestKey = lruItem.entrySet().iterator().next().getKey();
String eldestValue = lruItem.get(eldestKey);
return eldestValue;
}
故障转移
这个策略比较简单，遍历集群地址列表，如果失败，则继续调用下一台机器，成功则跳出循环，返回成功信息
//ExecutorRouteFailover
public ReturnT<String> routeRun(TriggerParam triggerParam, ArrayList<String> addressList) {
StringBuffer beatResultSB = new StringBuffer();
//循环集群地址
for (String address : addressList) {
// beat
ReturnT<String> beatResult = null;
try {
// 向执行器发送 执行beat信息 ， 试探该机器是否可以正常工作
ExecutorBiz executorBiz = XxlJobDynamicScheduler.getExecutorBiz(address);
beatResult = executorBiz.beat();
} catch (Exception e) {
logger.error(e.getMessage(), e);
beatResult = new ReturnT<String>(ReturnT.FAIL_CODE, ""+e );
}
// 拼接日志 ， 收集日志信息，后期一起返回
beatResultSB.append( (beatResultSB.length()>0)?"<br><br>":"")
.append(I18nUtil.getString("jobconf_beat") + "：")
.append("<br>address：").append(address)
.append("<br>code：").append(beatResult.getCode())
.append("<br>msg：").append(beatResult.getMsg());
// 返回状态为成功
if (beatResult.getCode() == ReturnT.SUCCESS_CODE) {
// 执行任务
ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
beatResultSB.append("<br><br>").append(runResult.getMsg());
// result
runResult.setMsg(beatResultSB.toString());
runResult.setContent(address);
return runResult;
}
}
return new ReturnT<String>(ReturnT.FAIL_CODE, beatResultSB.toString());
}
忙碌转移
这个策略更上面那个故障转移的原理一致，只不过不同的是，故障转移是判断机器是否存活， 二忙碌转移是想执行器发送消息判断该任务
对应的线程是否处于执行状态。
//ExecutorRouteFailover
@Override
public ReturnT<String> routeRun(TriggerParam triggerParam, ArrayList<String> addressList) {
StringBuffer idleBeatResultSB = new StringBuffer();
// 循环集群地址
for (String address : addressList) {
// beat
ReturnT<String> idleBeatResult = null;
try {
// 向执行服务器发送消息，判断当前jobId对应的线程是否忙碌，接下来可以看一下idleBeat这个方法
ExecutorBiz executorBiz = XxlJobDynamicScheduler.getExecutorBiz(address);
idleBeatResult = executorBiz.idleBeat(triggerParam.getJobId());
} catch (Exception e) {
logger.error(e.getMessage(), e);
idleBeatResult = new ReturnT<String>(ReturnT.FAIL_CODE, ""+e );
}
idleBeatResultSB.append( (idleBeatResultSB.length()>0)?"<br><br>":"")
.append(I18nUtil.getString("jobconf_idleBeat") + "：")
.append("<br>address：").append(address)
.append("<br>code：").append(idleBeatResult.getCode())
.append("<br>msg：").append(idleBeatResult.getMsg());
// 返回成功，代表这台执行服务器对应的线程处于空闲状态
if (idleBeatResult.getCode() == ReturnT.SUCCESS_CODE) {
// 执行人呢无
ReturnT<String> runResult = XxlJobTrigger.runExecutor(triggerParam, address);
idleBeatResultSB.append("<br><br>").append(runResult.getMsg());
// result
runResult.setMsg(idleBeatResultSB.toString());
runResult.setContent(address);
return runResult;
}
}
return new ReturnT<String>(ReturnT.FAIL_CODE, idleBeatResultSB.toString());
}
看一下执行器那边的idleBeat代码实现
//ExecutorBizImpl
@Override
public ReturnT<String> idleBeat(int jobId) {
// isRunningOrHasQueue
boolean isRunningOrHasQueue = false;
// 从线程池里面获取当前任务对应的线程
JobThread jobThread = XxlJobExecutor.loadJobThread(jobId);
if (jobThread != null && jobThread.isRunningOrHasQueue()) {
// 线程处于运行中
isRunningOrHasQueue = true;
}
if (isRunningOrHasQueue) {
// 线程运行中，则返回fasle
return new ReturnT<String>(ReturnT.FAIL_CODE, "job thread is running or has trigger queue.");
}
// 线程空闲，返回success
return ReturnT.SUCCESS;
}
网友评论