xxl-job路由策略源码分析

版本：2.2.1-SNAPSHOT

源码入口：XxlJobAdminConfig->XxlJobScheduler->JobScheduleHelper(start方法)->JobTriggerPoolHelper(addTrigger方法）-> XxlJobTrigger(trigger方法[processTrigger方法]) 具体代码如下： /** * @param group job group, registry list may be empty * @param jobInfo * @param finalFailRetryCount * @param triggerType * @param index sharding index * @param total sharding index */ private static void processTrigger(XxlJobGroup group, XxlJobInfo jobInfo, int finalFailRetryCount, TriggerTypeEnum triggerType, int index, int total){ // param ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION); // block strategy ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null); // route strategy String shardingParam = (ExecutorRouteStrategyEnum.SHARDING_BROADCAST==executorRouteStrategyEnum)?String.valueOf(index).concat("/").concat(String.valueOf(total)):null; // 1、save log-id XxlJobLog jobLog = new XxlJobLog(); jobLog.setJobGroup(jobInfo.getJobGroup()); jobLog.setJobId(jobInfo.getId()); jobLog.setTriggerTime(new Date()); XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().save(jobLog); logger.debug(">>>>>>>>>>> xxl-job trigger start, jobId:{}", jobLog.getId()); // 2、init trigger-param TriggerParam triggerParam = new TriggerParam(); triggerParam.setJobId(jobInfo.getId()); triggerParam.setExecutorHandler(jobInfo.getExecutorHandler()); triggerParam.setExecutorParams(jobInfo.getExecutorParam()); triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy()); triggerParam.setExecutorTimeout(jobInfo.getExecutorTimeout()); triggerParam.setLogId(jobLog.getId()); triggerParam.setLogDateTime(jobLog.getTriggerTime().getTime()); triggerParam.setGlueType(jobInfo.getGlueType()); triggerParam.setGlueSource(jobInfo.getGlueSource()); triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime()); triggerParam.setBroadcastIndex(index); triggerParam.setBroadcastTotal(total); // 3、init address String address = null; ReturnT<String> routeAddressResult = null; if (group.getRegistryList()!=null && !group.getRegistryList().isEmpty()) { if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum) { if (index < group.getRegistryList().size()) { address = group.getRegistryList().get(index); } else { address = group.getRegistryList().get(0); } } else { routeAddressResult = executorRouteStrategyEnum.getRouter().route(triggerParam, group.getRegistryList()); if (routeAddressResult.getCode() == ReturnT.SUCCESS_CODE) { address = routeAddressResult.getContent(); } } } else { routeAddressResult = new ReturnT<String>(ReturnT.FAIL_CODE, I18nUtil.getString("jobconf_trigger_address_empty")); } // 4、trigger remote executor ReturnT<String> triggerResult = null; if (address != null) { triggerResult = runExecutor(triggerParam, address); } else { triggerResult = new ReturnT<String>(ReturnT.FAIL_CODE, null); } // 5、collection trigger info StringBuffer triggerMsgSb = new StringBuffer(); triggerMsgSb.append(I18nUtil.getString("jobconf_trigger_type")).append("：").append(triggerType.getTitle()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobconf_trigger_admin_adress")).append("：").append(IpUtil.getIp()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobconf_trigger_exe_regtype")).append("：") .append( (group.getAddressType() == 0)?I18nUtil.getString("jobgroup_field_addressType_0"):I18nUtil.getString("jobgroup_field_addressType_1") ); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobconf_trigger_exe_regaddress")).append("：").append(group.getRegistryList()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorRouteStrategy")).append("：").append(executorRouteStrategyEnum.getTitle()); if (shardingParam != null) { triggerMsgSb.append("("+shardingParam+")"); } triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorBlockStrategy")).append("：").append(blockStrategy.getTitle()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_timeout")).append("：").append(jobInfo.getExecutorTimeout()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorFailRetryCount")).append("：").append(finalFailRetryCount); triggerMsgSb.append("<br><br><span style=\"color:#00c0ef;\" > >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_run") +"<<<<<<<<<<< </span><br>") .append((routeAddressResult!=null&&routeAddressResult.getMsg()!=null)?routeAddressResult.getMsg()+"<br><br>":"").append(triggerResult.getMsg()!=null?triggerResult.getMsg():""); // 6、save log trigger-info jobLog.setExecutorAddress(address); jobLog.setExecutorHandler(jobInfo.getExecutorHandler()); jobLog.setExecutorParam(jobInfo.getExecutorParam()); jobLog.setExecutorShardingParam(shardingParam); jobLog.setExecutorFailRetryCount(finalFailRetryCount); //jobLog.setTriggerTime(); jobLog.setTriggerCode(triggerResult.getCode()); jobLog.setTriggerMsg(triggerMsgSb.toString()); XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateTriggerInfo(jobLog); logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId()); }

 

上面的代码主要讲了分片广播这个策略的实现以及xxl-job的其他路由策略的调用位置在哪里。

ExecutorRouteStrategyEnum枚举类 这个是xxl-job路由策略非常重要的一个类， 该类通过枚举的方式，把路由key， 和策略实现类进行了一个聚合、

/** * Created by xuxueli on 17/3/10. */ public enum ExecutorRouteStrategyEnum { //第一个 FIRST(I18nUtil.getString("jobconf_route_first"), new ExecutorRouteFirst()), //最后一个 LAST(I18nUtil.getString("jobconf_route_last"), new ExecutorRouteLast()), //轮循 ROUND(I18nUtil.getString("jobconf_route_round"), new ExecutorRouteRound()), //随机 RANDOM(I18nUtil.getString("jobconf_route_random"), new ExecutorRouteRandom()), //一致性哈希 CONSISTENT_HASH(I18nUtil.getString("jobconf_route_consistenthash"), new ExecutorRouteConsistentHash()), //最不经常使用 LEAST_FREQUENTLY_USED(I18nUtil.getString("jobconf_route_lfu"), new ExecutorRouteLFU()), //最近最久未使用 LEAST_RECENTLY_USED(I18nUtil.getString("jobconf_route_lru"), new ExecutorRouteLRU()), //故障转移 FAILOVER(I18nUtil.getString("jobconf_route_failover"), new ExecutorRouteFailover()), //忙碌转移 BUSYOVER(I18nUtil.getString("jobconf_route_busyover"), new ExecutorRouteBusyover()), //分片广播 SHARDING_BROADCAST(I18nUtil.getString("jobconf_route_shard"), 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; } }

1.FIRST 第一个

public class ExecutorRouteFirst extends ExecutorRouter { @Override public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList){ return new ReturnT<String>(addressList.get(0)); } }

2.LAST 最后一个

public class ExecutorRouteLast extends ExecutorRouter { @Override public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) { return new ReturnT<String>(addressList.get(addressList.size()-1)); } }

 3.ROUND 轮询

public class ExecutorRouteRound extends ExecutorRouter { private static ConcurrentMap<Integer, Integer> routeCountEachJob = new ConcurrentHashMap<Integer, Integer>(); private static long CACHE_VALID_TIME = 0; private static int count(int jobId) { // cache clear // 如果当前的时间，大于缓存的时间，那么说明需要刷新了 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; } @Override public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) { // 在执行器地址列表，获取相应的地址，  通过count(jobid) 这个方法来实现，主要逻辑在这个方法 // 通过count（jobId）拿到数字之后， 通过取模的方式，拿到执行器地址 // 例： count=2 , addresslist.size = 3 // 2%3 = 2 ,  则拿list中下表为2的地址 String address = addressList.get(count(triggerParam.getJobId())%addressList.size()); return new ReturnT<String>(address); }

4. RANDOM 随机

public class ExecutorRouteRandom extends ExecutorRouter { private static Random localRandom = new Random(); @Override public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) { //集群列表的大小内随机拿出一台机器来执行 String address = addressList.get(localRandom.nextInt(addressList.size())); return new ReturnT<String>(address); } }

5.CONSISTENT_HASH 一致性hash

一致性Hash算法是对 2^32 取模，什么意思呢？ 简单来说，一致性Hash算法将整个哈希值空间组织成一个虚拟的圆环，如假设某哈希函数H的值空间为0-2^32-1（即哈希值是一个32位无符号整形） 整个哈希环如下，从 0 ~ 2^32-1 代表的分别是一个个的节点,这个环也叫哈希环（即所有的节点分布在一个哈希值为0-2^32-1的哈希环上），接着在Hash环上顺时针查找距离这个Key值的Hash值最近的服务器节点，完成Key到服务器的映射查找。 /** * 分组下机器地址相同，不同JOB均匀散列在不同机器上，保证分组下机器分配JOB平均；且每个JOB固定调度其中一台机器； * a、virtual node：解决不均衡问题 * b、hash method replace hashCode：String的hashCode可能重复，需要进一步扩大hashCode的取值范围 * Created by xuxueli on 17/3/10. * 说明：一致性 Hash 算法也是使用取模的思想，只是，刚才描述的取模法是对节点数量进行取模，而一致性Hash算法是对 2^32 取模，什么意思呢？ * 简单来说，一致性Hash算法将整个哈希值空间组织成一个虚拟的圆环，如假设某哈希函数H的值空间为0-2^32-1（即哈希值是一个32位无符号整形）， * 整个哈希环如下，从 0 ~ 2^32-1 代表的分别是一个个的节点,这个环也叫哈希环（即所有的节点分布在一个哈希值为0-2^32-1的哈希环上） * */ public class ExecutorRouteConsistentHash extends ExecutorRouter { private static int VIRTUAL_NODE_NUM = 100; /** * 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 hashJob(int jobId, List<String> addressList) { // ------A1------A2-------A3------ // -----------J1------------------ 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); if (!lastRing.isEmpty()) { // 如果addressRing中有比jobHash的那么直接取lastRing 的第一个 return lastRing.get(lastRing.firstKey()); } // 如果没有，则直接取addresRing的第一个 // 反正最终的效果是在Hash环上，顺时针拿离jobHash最近的一个值 return addressRing.firstEntry().getValue(); } @Override public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) { String address = hashJob(triggerParam.getJobId(), addressList); return new ReturnT<String>(address); } }

 6.LEAST_FREQUENTLY_USED 最不经常使用

/** * 单个JOB对应的每个执行器，使用频率最低的优先被选举 * a(*)、LFU(Least Frequently Used)：最不经常使用，频率/次数 * b、LRU(Least Recently Used)：最近最久未使用，时间 * * Created by xuxueli on 17/3/10. */ public class ExecutorRouteLFU extends ExecutorRouter { private static ConcurrentMap<Integer, HashMap<String, Integer>> jobLfuMap = new ConcurrentHashMap<Integer, HashMap<String, Integer>>(); private static long CACHE_VALID_TIME = 0; public String route(int jobId, List<String> addressList) { // cache clear // 如果当前系统时间大于过期时间 if (System.currentTimeMillis() > CACHE_VALID_TIME) { jobLfuMap.clear(); //重新设置过期时间，默认为一天 CACHE_VALID_TIME = System.currentTimeMillis() + 1000*60*60*24; } // lfu item init //存放的是执行器地址以及执行次数 HashMap<String, Integer> lfuItemMap = jobLfuMap.get(jobId); // Key排序可以用TreeMap+构造入参Compare；Value排序暂时只能通过ArrayList； if (lfuItemMap == null) { lfuItemMap = new HashMap<String, Integer>(); jobLfuMap.putIfAbsent(jobId, lfuItemMap); // 避免重复覆盖 } // put new for (String address: addressList) { // map中不包含，并且值大于一万的时候，需要重新初始化执行器地址对应的执行次数 // 初始化的规则是在机器地址列表size里面进行随机 // 当运行一段时间后，有新机器加入的时候，此时，新机器初始化的执行次数较小，所以一开始，新机器的压力会比较大，后期慢慢趋于平衡 if (!lfuItemMap.containsKey(address) || lfuItemMap.get(address) >1000000 ) { lfuItemMap.put(address, new Random().nextInt(addressList.size())); // 初始化时主动Random一次，缓解首次压力 } } // remove old List<String> delKeys = new ArrayList<>(); for (String existKey: lfuItemMap.keySet()) { if (!addressList.contains(existKey)) { delKeys.add(existKey); } } if (delKeys.size() > 0) { for (String delKey: delKeys) { lfuItemMap.remove(delKey); } } // load least userd count address //将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> route(TriggerParam triggerParam, List<String> addressList) { String address = route(triggerParam.getJobId(), addressList); return new ReturnT<String>(address); } }

7.LEAST_RECENTLY_USED 最近最久未使用

此处使用的是linkHashMap来实现LRU算法的。

通过linkHashMap的每次get/put的时候会进行排序，最新操作的数据会在最后面。 从而取第一个数据就是最久未使用的。

/** * 单个JOB对应的每个执行器，最久为使用的优先被选举 * a、LFU(Least Frequently Used)：最不经常使用，频率/次数 * b(*)、LRU(Least Recently Used)：最近最久未使用，时间 * * Created by xuxueli on 17/3/10. */ public class ExecutorRouteLRU extends ExecutorRouter { private static ConcurrentMap<Integer, LinkedHashMap<String, String>> jobLRUMap = new ConcurrentHashMap<Integer, LinkedHashMap<String, String>>(); private static long CACHE_VALID_TIME = 0; public String route(int jobId, List<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：true=访问顺序排序（get/put时排序）；false=插入顺序排期； * b、removeEldestEntry：新增元素时将会调用，返回true时会删除最老元素；可封装LinkedHashMap并重写该方法，比如定义最大容量，超出是返回true即可实现固定长度的LRU算法； */ //通过linkHashMap的每次get/put的时候会进行排序，最新操作的数据会在最后面。 从而取第一个数据就 代表是最久没有被使用的 lruItem = new LinkedHashMap<String, String>(16, 0.75f, true); jobLRUMap.putIfAbsent(jobId, lruItem); } // put new // 如果地址列表里面有地址不在map中，此处是可以再次放入，防止添加机器的问题 for (String address: addressList) { if (!lruItem.containsKey(address)) { lruItem.put(address, address); } } // remove old List<String> delKeys = new ArrayList<>(); for (String existKey: lruItem.keySet()) { if (!addressList.contains(existKey)) { delKeys.add(existKey); } } if (delKeys.size() > 0) { for (String delKey: delKeys) { lruItem.remove(delKey); } } // load // 取头部的一个元素，也就是最久操作过的数据 String eldestKey = lruItem.entrySet().iterator().next().getKey(); String eldestValue = lruItem.get(eldestKey); return eldestValue; } @Override public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) { String address = route(triggerParam.getJobId(), addressList); return new ReturnT<String>(address); } }

 8.FAILOVER 故障转移

遍历集群地址列表，如果失败，则继续调用下一台机器，成功则跳出循环。

public class ExecutorRouteFailover extends ExecutorRouter { @Override public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) { StringBuffer beatResultSB = new StringBuffer(); for (String address : addressList) { // beat ReturnT<String> beatResult = null; try { ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address); 向执行器发送 执行beat信息 ， 试探该机器是否可以正常工作 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()); // beat success if (beatResult.getCode() == ReturnT.SUCCESS_CODE) { beatResult.setMsg(beatResultSB.toString()); beatResult.setContent(address); return beatResult; } } return new ReturnT<String>(ReturnT.FAIL_CODE, beatResultSB.toString()); } } public class ExecutorBizClient implements ExecutorBiz { public ExecutorBizClient() { } public ExecutorBizClient(String addressUrl, String accessToken) { this.addressUrl = addressUrl; this.accessToken = accessToken; // valid if (!this.addressUrl.endsWith("/")) { this.addressUrl = this.addressUrl + "/"; } } private String addressUrl ; private String accessToken; private int timeout = 3; @Override public ReturnT<String> beat() { return XxlJobRemotingUtil.postBody(addressUrl+"beat", accessToken, timeout, null, String.class); } @Override public ReturnT<String> idleBeat(IdleBeatParam idleBeatParam){ return XxlJobRemotingUtil.postBody(addressUrl+"idleBeat", accessToken, timeout, idleBeatParam, String.class); } @Override public ReturnT<String> run(TriggerParam triggerParam) { return XxlJobRemotingUtil.postBody(addressUrl + "run", accessToken, timeout, triggerParam, String.class); } @Override public ReturnT<String> kill(KillParam killParam) { return XxlJobRemotingUtil.postBody(addressUrl + "kill", accessToken, timeout, killParam, String.class); } @Override public ReturnT<LogResult> log(LogParam logParam) { return XxlJobRemotingUtil.postBody(addressUrl + "log", accessToken, timeout, logParam, LogResult.class); } } @Override public ReturnT<String> beat() { return ReturnT.SUCCESS; }

 

 9. BUSYOVER 忙碌转移

这个策略更上面那个故障转移的原理一致，不同的是，故障转移是判断机器是否存活， 忙碌转移则是通过执行器发送消息判断该任务对应的线程是否处于执行状态。

public class ExecutorRouteBusyover extends ExecutorRouter { @Override public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) { StringBuffer idleBeatResultSB = new StringBuffer(); for (String address : addressList) { // beat ReturnT<String> idleBeatResult = null; try { ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address); idleBeatResult = executorBiz.idleBeat(new IdleBeatParam(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()); // beat success if (idleBeatResult.getCode() == ReturnT.SUCCESS_CODE) { idleBeatResult.setMsg(idleBeatResultSB.toString()); idleBeatResult.setContent(address); return idleBeatResult; } } return new ReturnT<String>(ReturnT.FAIL_CODE, idleBeatResultSB.toString()); } } @Override public ReturnT<String> idleBeat(IdleBeatParam idleBeatParam) { // isRunningOrHasQueue boolean isRunningOrHasQueue = false; JobThread jobThread = XxlJobExecutor.loadJobThread(idleBeatParam.getJobId()); if (jobThread != null && jobThread.isRunningOrHasQueue()) { isRunningOrHasQueue = true; } if (isRunningOrHasQueue) { return new ReturnT<String>(ReturnT.FAIL_CODE, "job thread is running or has trigger queue."); } return ReturnT.SUCCESS; }

 

 10.SHARDING_BROADCAST 分片广播

给每一台机器都发送执行消息，分片总数为集群机器数量，分片标记从0开始。

if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST==ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null) && group.getRegistryList()!=null && !group.getRegistryList().isEmpty() && shardingParam==null) { for (int i = 0; i < group.getRegistryList().size(); i++) { processTrigger(group, jobInfo, finalFailRetryCount, triggerType, i, group.getRegistryList().size()); } } private static void processTrigger(XxlJobGroup group, XxlJobInfo jobInfo, int finalFailRetryCount, TriggerTypeEnum triggerType, int index, int total){ // param ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(jobInfo.getExecutorBlockStrategy(), ExecutorBlockStrategyEnum.SERIAL_EXECUTION); // block strategy ExecutorRouteStrategyEnum executorRouteStrategyEnum = ExecutorRouteStrategyEnum.match(jobInfo.getExecutorRouteStrategy(), null); // route strategy String shardingParam = (ExecutorRouteStrategyEnum.SHARDING_BROADCAST==executorRouteStrategyEnum)?String.valueOf(index).concat("/").concat(String.valueOf(total)):null; // 1、save log-id XxlJobLog jobLog = new XxlJobLog(); jobLog.setJobGroup(jobInfo.getJobGroup()); jobLog.setJobId(jobInfo.getId()); jobLog.setTriggerTime(new Date()); XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().save(jobLog); logger.debug(">>>>>>>>>>> xxl-job trigger start, jobId:{}", jobLog.getId()); // 2、init trigger-param TriggerParam triggerParam = new TriggerParam(); triggerParam.setJobId(jobInfo.getId()); triggerParam.setExecutorHandler(jobInfo.getExecutorHandler()); triggerParam.setExecutorParams(jobInfo.getExecutorParam()); triggerParam.setExecutorBlockStrategy(jobInfo.getExecutorBlockStrategy()); triggerParam.setExecutorTimeout(jobInfo.getExecutorTimeout()); triggerParam.setLogId(jobLog.getId()); triggerParam.setLogDateTime(jobLog.getTriggerTime().getTime()); triggerParam.setGlueType(jobInfo.getGlueType()); triggerParam.setGlueSource(jobInfo.getGlueSource()); triggerParam.setGlueUpdatetime(jobInfo.getGlueUpdatetime().getTime()); triggerParam.setBroadcastIndex(index); triggerParam.setBroadcastTotal(total); // 3、init address String address = null; ReturnT<String> routeAddressResult = null; if (group.getRegistryList()!=null && !group.getRegistryList().isEmpty()) { if (ExecutorRouteStrategyEnum.SHARDING_BROADCAST == executorRouteStrategyEnum) { if (index < group.getRegistryList().size()) { address = group.getRegistryList().get(index); } else { address = group.getRegistryList().get(0); } } else { routeAddressResult = executorRouteStrategyEnum.getRouter().route(triggerParam, group.getRegistryList()); if (routeAddressResult.getCode() == ReturnT.SUCCESS_CODE) { address = routeAddressResult.getContent(); } } } else { routeAddressResult = new ReturnT<String>(ReturnT.FAIL_CODE, I18nUtil.getString("jobconf_trigger_address_empty")); } // 4、trigger remote executor ReturnT<String> triggerResult = null; if (address != null) { triggerResult = runExecutor(triggerParam, address); } else { triggerResult = new ReturnT<String>(ReturnT.FAIL_CODE, null); } // 5、collection trigger info StringBuffer triggerMsgSb = new StringBuffer(); triggerMsgSb.append(I18nUtil.getString("jobconf_trigger_type")).append("：").append(triggerType.getTitle()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobconf_trigger_admin_adress")).append("：").append(IpUtil.getIp()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobconf_trigger_exe_regtype")).append("：") .append( (group.getAddressType() == 0)?I18nUtil.getString("jobgroup_field_addressType_0"):I18nUtil.getString("jobgroup_field_addressType_1") ); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobconf_trigger_exe_regaddress")).append("：").append(group.getRegistryList()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorRouteStrategy")).append("：").append(executorRouteStrategyEnum.getTitle()); if (shardingParam != null) { triggerMsgSb.append("("+shardingParam+")"); } triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorBlockStrategy")).append("：").append(blockStrategy.getTitle()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_timeout")).append("：").append(jobInfo.getExecutorTimeout()); triggerMsgSb.append("<br>").append(I18nUtil.getString("jobinfo_field_executorFailRetryCount")).append("：").append(finalFailRetryCount); triggerMsgSb.append("<br><br><span style=\"color:#00c0ef;\" > >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_run") +"<<<<<<<<<<< </span><br>") .append((routeAddressResult!=null&&routeAddressResult.getMsg()!=null)?routeAddressResult.getMsg()+"<br><br>":"").append(triggerResult.getMsg()!=null?triggerResult.getMsg():""); // 6、save log trigger-info jobLog.setExecutorAddress(address); jobLog.setExecutorHandler(jobInfo.getExecutorHandler()); jobLog.setExecutorParam(jobInfo.getExecutorParam()); jobLog.setExecutorShardingParam(shardingParam); jobLog.setExecutorFailRetryCount(finalFailRetryCount); //jobLog.setTriggerTime(); jobLog.setTriggerCode(triggerResult.getCode()); jobLog.setTriggerMsg(triggerMsgSb.toString()); XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateTriggerInfo(jobLog); logger.debug(">>>>>>>>>>> xxl-job trigger end, jobId:{}", jobLog.getId()); }