xxl-job分布式框架源码路由/负载均衡策略算法_xxljob负载均衡

概述

xxl-job就是因为内涵丰富的调度策略，使得框架的多样性，灵活性更高。现在就开始讲解xxl-job的核心路由策略算法，总共有10种路由策略，对于以后想从事分布式微服务开发，任务调度的学习是很有必要的。

个人认为这个路由算法可以作为标准样例应用在其他分布式路由策略中，后面有时间的话我会整理为工具类
牧竹子注：2023/6/4

路由策略种类

• 第一个
• 最后一个
• 随机选取
• 轮询选取
• 一致性hash
• 最不经常使用 (LFU)
• 最近最久未使用（LRU）
• 故障转移
• 忙碌转移
• 分片广播

以上就是xxl-job内部封装的路由策略算法，也是很常见的路由算法，学习掌握之后对自己设计分布式架构很有帮助。

路由策略枚举类型所有的分片

拓展其他负载均衡算法

我们来看看一些常见中间件使用的负载均衡算法

Nginx

RoundRobin：轮询。
WeightedRoundRobin：加权轮询。
IPHash：按访问IP的Hash选择服务器。
URLHash：按请求URL的Hash选择服务器。
Fair：根据后端服务器的响应时间判断负载情况，从中选出负载最轻的机器进行分流。

Dubbo

RandomLoadBalance：加权随机。
RoundRobinLoadBalance：加权轮询。
LeastActionLoadBalance：最少链接数。
ShortestResponseLoadBalance：最短响应时间。
ConsistentHashLoadBalance：一致性Hash。

Ribbon

RoundRobinRule：轮询。
RandomRule：随机。
WeightedResponseTimeRule：根据响应时间来分配权重的方式，响应的越快，分配的值越大。
BestAvailableRule：会先过滤掉由于多次访问故障而处于断路器跳闸状态的服务，然后选择一个并发量最小的服务。
RetryRule：先按照轮询策略获取服务，如果获取服务失败则在指定时间内进行重试，获取可用的服务。
ZoneAvoidanceRule：根据性能和可用性选择服务。
AvailabilityFilteringRule：会先过滤掉由于多次访问故障而处于断路器状态的服务，还有并发的连接数量超过阈值的服务，然后对剩余的服务列表按照轮询策略进行访问。

本节原文链接：https://blog.csdn.net/Anenan/article/details/116203518

路由策略讲解

第一个

获取注册地址集合的第一个

package com.xxl.job.admin.core.route.strategy;
 
import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
 
import java.util.List;
 
/**
 * Created by xuxueli on 17/3/10.
 */
public class ExecutorRouteFirst extends ExecutorRouter {
 
    @Override
    public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList){
        return new ReturnT<String>(addressList.get(0));
    }
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

最后一个

获取注册地址集合的最后一个

package com.xxl.job.admin.core.route.strategy;
 
import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
 
import java.util.List;
 
/**
 * Created by xuxueli on 17/3/10.
 */
public class ExecutorRouteLast extends ExecutorRouter {
 
    @Override
    public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) {
        return new ReturnT<String>(addressList.get(addressList.size()-1));
    }
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

随机选取

通过一个Random对象的nextInt方法在求出[0,addressList.size()）区间内的任意一个index

package com.xxl.job.admin.core.route.strategy;
 
import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
 
import java.util.List;
import java.util.Random;
 
/**
 * Created by xuxueli on 17/3/10.
 */
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);
    }
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

轮询选取

一个静态的ConcurrentMap对象，这个routeCountEachJob就是用来存放路由任务的，而且还设置了缓存时间，有效期为24小时，当超过24小时的时候，延迟调用时清空当前的缓存。

其中ConcurrentMap的key为jobId，value为当前jobId所对应的计数器，每访问一次就自增一，最大增到100000，然后又从[0，100)的随机数开始重新自增。这里100以内的次数随机获取，因此100以内特别是50前的数字可能要比100以后的数字选中机会少很多

这个算法的思想就是取余数，每次先计算出当前jobId所对应的计数器的值，然后 计数器的值 % addressList.size() 求得这一次轮询的数组下标index地址。

根据上面轮询的算法原理，前50的被选中概率要远小于100以后的值，因此可以把随机算法改成从0开始，那么基本都会轮询到

package com.xxl.job.admin.core.route.strategy;
 
import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
 
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.IntStream;
 
/**
 * Created by xuxueli on 17/3/10.
 */
public class ExecutorRouteRound extends ExecutorRouter {


	//缓存jobId和对应被轮的序号 
    private static ConcurrentMap<Integer, AtomicInteger> routeCountEachJob = new ConcurrentHashMap<>();
    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;
        }
 
        AtomicInteger count = routeCountEachJob.get(jobId);
        if (count == null || count.get() > 1000000) {//初次和到达1000000了就从新开始初始化
            // 初始化时主动Random一次，缓解首次压力
            count = new AtomicInteger(new Random().nextInt(100));
        } else {
            // count++
            count.addAndGet(1);
        }
        routeCountEachJob.put(jobId, count);//放入jobId和对应轮询的序号
        return count.get();
    }
 
	//路由方法入口
    @Override
    public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList) {
	//调用count(JobId)获取轮询到的次数，根据计数器值获取数组下标=count%list.size()
        String address = addressList.get(count(triggerParam.getJobId())%addressList.size());
        return new ReturnT<String>(address);
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

一致性hash

一致哈希 是一种特殊的哈希算法。在使用一致哈希算法后，哈希表槽位数2^32是固定的，只需要对 K/n 个关键字重新映射，其中K是关键字的数量， n是槽位数量。然而在传统的哈希表中，添加或删除一个槽位的几乎需要对所有关键字进行重新映射。引入了一致性hash算法，即服务节点的变更不会造成大量的哈希重定位。

一致性hash引入之后，若服务器节点数量过少，有几率出现数据倾斜的情况，既大量的数据映射到某一区间，其它区间没有数据映射，造成了资源分配不均匀，为了解决这个问题，xxl-job源码引入了虚拟节点,既将每台服务器的节点都生成所对应的100个虚拟节点，这应少量的服务器节点通过引入虚拟节点，就会加大节点的数量，这样大量的节点分配到hash环上是比较均匀的，从而很容易的解决数据分配不均匀问题。

package com.xxl.job.admin.core.route.strategy;
 
import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
 
import java.io.UnsupportedEncodingException;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.util.List;
import java.util.SortedMap;
import java.util.TreeMap;
 
/**
 * 分组下机器地址相同，不同JOB均匀散列在不同机器上，保证分组下机器分配JOB平均；且每个JOB固定调度其中一台机器；
 *      a、virtual node：解决不均衡问题
 *      b、hash method replace hashCode：String的hashCode可能重复，需要进一步扩大hashCode的取值范围
 * Created by xuxueli on 17/3/10.
 */
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);
 
        //通过md5算出的hashcode % 2^32 余数，将hash值散列在一致性hash环上  这个环分了2^32个位置
        long truncateHashCode = hashCode & 0xffffffffL;
        return truncateHashCode;
    }
 
    /**
     * 哈希槽定位
     * @return
     */
    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++) {
                //为每一个注册的节点分配100个虚拟节点，并算出这些节点的一致性hash值，存放到TreeMap中
                long addressHash = hash("SHARD-" + address + "-NODE-" + i);
                addressRing.put(addressHash, address);
            }
        }
        //第二步求出job的hash值 通过jobId计算
        long jobHash = hash(String.valueOf(jobId));
        //通过treeMap性质，所有的key都按照从小到大的排序，即按照hash值从小到大排序,通过tailMap 求出>=hash(jobId)的剩余一部分map，
        SortedMap<Long, String> lastRing = addressRing.tailMap(jobHash);
        if (!lastRing.isEmpty()) {
            //若找到则取第一个key，为带路由的地址
            return lastRing.get(lastRing.firstKey());
        }
        //若本身hash(jobId)为treeMap的最后一个key，则找当前treeMap的第一个key
        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);
    }
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

最不经常使用 (LFU)

Least Recently Used：最近最近少使用频次，最少使用的作为首选

使用ConcurrentMap维护一个jobId为key的cachMap，vue为Map<String,int>对应url，使用次数的lfuItemMap

• 第一次url地址依次放入lfuItemMap ,随机生成它的次数
• 遍历url的list，把新加入的url和次数到达1000000 的url重置次数
• 淘汰不存在的url：遍历lfuItemMap 的key，如果urlList中不存在则说明该地址已经无效需要移除
• 对lfuItemMap进行升序排序，选第一个作为最少使用的值返回

package com.xxl.job.admin.core.route.strategy;
 
import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
 
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
 
/**
 * 单个JOB对应的每个执行器，使用频率最低的优先被选举
 *      a(*)、LFU(Least Frequently Used)：最不经常使用，频率/次数
 *      b、LRU(Least Recently Used)：最近最久未使用，时间
 *
 *      算法思想：
 *      构建一个作业和地址map    jobid -> addressList
 *      第一次随机的将任务所对应的执行器的注册地址编一个序列号
 *      然后将执行器的注册地址按照从小到大进行排序
 *      筛选过程找第一个序列号最小的作为下一次的路由地址
 *      随后将当前选中的地址编号值+1
 *      这样最终我们都会挑选编号最小的注册器地址作为下一个路由地址，既最不常使用的
 *
 * 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) {
            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
        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());
            }
        });
 
        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);
    }
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

最近最久未使用（LRU）

LRU(Least Recently Used)：最近最久未使用
这里关注的是最久未使用，不关注次数

因此LinkedHashMap的accessOrder可以直接满足这个要求，accessOrder默认值是false

• false：按插入顺序排序，谁最先插入put就放入头部，最后进入放入尾部，FIFO特性

• true:按访问顺序排序（get/put时排序）排序，谁被访问就放到头部，没有的则会排到尾部

• 依次遍历urlList，把对应的url放入LinkedHashMap

• 淘汰不存在的url：遍历LinkedHashMap的key，如果服务urlList中不存在该url则直接移除

• 获取LinkedHashMap的第一个url作为服务地址返回

package com.xxl.job.admin.core.route.strategy;
 
import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
 
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
 
/**
 * 单个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算法；
             */
            lruItem = new LinkedHashMap<String, String>(16, 0.75f, true);
            jobLRUMap.putIfAbsent(jobId, lruItem);
        }
 
        // put new
        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);
    }
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76

故障转移

遍历所有的该组下的所有注册节点地址集合，然后分别进行心跳处理，直到找到一个发送心跳成功的节点作为下一次路由的节点。每次调用当前路由时都会促发一次执行寻找下一个路由节点提前备用

package com.xxl.job.admin.core.route.strategy;
 
import com.xxl.job.admin.core.scheduler.XxlJobScheduler;
import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.util.I18nUtil;
import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
 
import java.util.List;
 
/**
 * Created by xuxueli on 17/3/10.
 * 故障转移路由策略
 * 思想：遍历所有的该组下的所有注册节点地址集合，然后分别进行心跳处理，直到找到一个发送心跳成功的节点作为下一次路由的节点
 */
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);
                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());
 
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

忙碌转移

忙碌转移也很容易理解，原理遍历所有的执行器，对所有执行器发送空闲心跳数据包
收集所有的返回信息，若当前机器繁忙则响应getCode500 否则空闲则getCode200
直到找到空闲的机器则返回该空闲机器的地址返回作为下次服务调用的节点

package com.xxl.job.admin.core.route.strategy;
 
import com.xxl.job.admin.core.scheduler.XxlJobScheduler;
import com.xxl.job.admin.core.route.ExecutorRouter;
import com.xxl.job.admin.core.util.I18nUtil;
import com.xxl.job.core.biz.ExecutorBiz;
import com.xxl.job.core.biz.model.IdleBeatParam;
import com.xxl.job.core.biz.model.ReturnT;
import com.xxl.job.core.biz.model.TriggerParam;
 
import java.util.List;
 
/**
 * Created by xuxueli on 17/3/10.
 * 忙碌转移
 * 原理遍历所有的执行器，对所有执行器发送空闲心跳数据包
 * 收集所有的返回信息，若当前机器繁忙则响应getCode==500 否则空闲则getCode==200
 * 找到空闲的机器则返回该空闲机器的地址
 */
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());
    }
 
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52

EmbedServer来处理这个请求，判断当前执行器节点是否执行当前任务或者当前执行器节点的任务队列是否为空，若既不是执行当前任务的节点或者任务队列为空则返回SUCCESS，以下代码就是上述所说。

else if ("/idleBeat".equals(uri)){
 IdleBeatParam idleBeatParam = GsonTool.fromJson(reguestData, IdleBeatParam.class);
return executorBiz.idleBeat(idleBeatParam);
}

//空闲心跳处理 查询当前处理器是否为执行状态
@Override
public ReturnT<string> idleBeat(IdleBeatParam idleBeatParam) {
	isRunningOrHasOueue
	boolean isRunningOrHasOueue = false:
	JobThread jobthread = XxlJobExecutor.loadJobThread(idleBeatParamgetJobid());
	if (jobThread != null && jobThread.isRunningorHasOueue()) {
		isRunningOrHasOueue = true;
	}
	if (isRunningOrHasQueue) {
		return new ReturnT<String>(ReturnT.FAIL CODE, msg: "job thread is running or has trigger gueue.");
	return ReturnT.sUCCEss;
	}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

分片路由

这个实现还是有点意外的，没想到它就只是把注册地址遍历出来，依次给对应的路由地址发送参数(index,RegistryList().size()),这样每个任务执行器收到的是对应的index和分片总数size。是不是很意外，它比你想象中实现的要简单许多

 如果是分片路由，走的是这段逻辑
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++) {
	            // 最后两个参数，i是当前机器在执行器集群当中的index，group.getRegistryList().size()为执行器总数=分片总数
                processTrigger(group, jobInfo, finalFailRetryCount, triggerType, i, group.getRegistryList().size());
            }
        } 
1
2
3
4
5
6
7
8
9

以上便是xxl-job所有的路由策略，除了分片广播，其他都是单例模式运行执行器，本章节完更。

更新时间2023/06/09

原文链接

https://blog.csdn.net/s6056826a/article/details/113447358