揪出XXL-JOB中的细节

时间 2019-11-30

标签揪出 xxl job 细节繁體版

原文原文链接

前言

国庆节快到了，要开始休假了，笔者仍是很开心的，国庆快乐！java

废话少说，直接进入正题。算法

相信你们对XXL-JOB都很了解，故本文对源码不进行过多介绍，侧重的是看源码过程当中想到的几个知识点，不必定都对，请大神们批评指正。服务器

XXL-JOB简介

XXL-JOB是一个轻量级分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用。
XXL-JOB分为调度中心、执行器、数据中心，调度中心负责任务管理及调度、执行器管理、日志管理等，执行器负责任务执行及执行结果回调。

任务调度 - “类时间轮”的实现

时间轮

时间轮出自Netty中的HashedWheelTimer，是一个环形结构，能够用时钟来类比，钟面上有不少bucket，每个bucket上能够存放多个任务，使用一个List保存该时刻到期的全部任务，同时一个指针随着时间流逝一格一格转动，并执行对应bucket上全部到期的任务。任务经过取模决定应该放入哪一个bucket。和HashMap的原理相似，newTask对应put，使用List来解决 Hash 冲突。负载均衡

以上图为例，假设一个bucket是1秒，则指针转动一轮表示的时间段为8s，假设当前指针指向 0，此时须要调度一个3s后执行的任务，显然应该加入到(0+3=3)的方格中，指针再走3s次就能够执行了；若是任务要在10s后执行，应该等指针走完一轮零2格再执行，所以应放入2，同时将round(1)保存到任务中。检查到期任务时只执行round为0的，bucket上其余任务的round减1。框架

固然，还有优化的“分层时间轮”的实现，请参考https://cnkirito.moe/timer/。dom

XXL-JOB中的“时间轮”

XXL-JOB中的调度方式从Quartz变成了自研调度的方式，很像时间轮，能够理解为有60个bucket且每一个bucket为1秒，可是没有了round的概念。async
具体能够看下图。分布式

XXL-JOB中负责任务调度的有两个线程，分别为ringThread和scheduleThread，其做用以下。

一、scheduleThread：对任务信息进行读取，预读将来5s即将触发的任务，放入时间轮。二、ringThread：对当前bucket和前一个bucket中的任务取出并执行。ide

下面结合源代码看下，为何说是“类时间轮”，关键代码附上了注解，请你们留意观看。

// 环状结构
private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();

// 任务下次启动时间（单位为秒） % 60
int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);

// 任务放进时间轮
private void pushTimeRing(int ringSecond, int jobId){
        // push async ring
        List<Integer> ringItemData = ringData.get(ringSecond);
        if (ringItemData == null) {
            ringItemData = new ArrayList<Integer>();
            ringData.put(ringSecond, ringItemData);
        }
        ringItemData.add(jobId);
    }
复制代码

// 同时取两个时间刻度的任务
List<Integer> ringItemData = new ArrayList<>();
int nowSecond = Calendar.getInstance().get(Calendar.SECOND);  
// 避免处理耗时太长，跨过刻度，向前校验一个刻度；
for (int i = 0; i < 2; i++) {
	List<Integer> tmpData = ringData.remove( (nowSecond+60-i)%60 );
	if (tmpData != null) {
		ringItemData.addAll(tmpData);
	}
}
// 运行
for (int jobId: ringItemData) {
	JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null);
}
复制代码

一致性Hash路由中的Hash算法

你们也知道，XXL-JOB在执行任务时，任务具体在哪一个执行器上运行是根据路由策略来决定的，其中有一个策略是一致性Hash策略（源码在ExecutorRouteConsistentHash.java），天然而然想到了一致性Hash算法。
一致性Hash算法是为了解决分布式系统中负载均衡的问题时候可使用Hash算法让固定的一部分请求落到同一台服务器上，这样每台服务器固定处理一部分请求（并维护这些请求的信息），起到负载均衡的做用。
普通的余数hash（hash(好比用户id)%服务器机器数）算法伸缩性不好，当新增或者下线服务器机器时候，用户id与服务器的映射关系会大量失效。一致性hash则利用hash环对其进行了改进。
一致性Hash算法在实践中，当服务器节点比较少的时候会出现上节所说的一致性hash倾斜的问题，一个解决方法是多加机器，可是加机器是有成本的，那么就加虚拟节点。
具体原理请参考https://www.jianshu.com/p/e968c081f563。
下图为带有虚拟节点的Hash环，其中ip1-1是ip1的虚拟节点，ip2-1是ip2的虚拟节点，ip3-1是ip3的虚拟节点。

可见，一致性Hash算法的关键在于Hash算法，保证虚拟节点及Hash结果的均匀性，而均匀性能够理解为减小Hash冲突，Hash冲突的知识点请参考从HashMap，Redis 字典看【Hash】。。。。函数

XXL-JOB中的一致性Hash的Hash函数以下。

// jobId转换为md5
// 不直接用hashCode() 是由于扩大hash取值范围，减小冲突
byte[] digest = md5.digest();

// 32位hashCode
long hashCode = ((long) (digest[3] & 0xFF) << 24)
	| ((long) (digest[2] & 0xFF) << 16)
	| ((long) (digest[1] & 0xFF) << 8)
	| (digest[0] & 0xFF);

long truncateHashCode = hashCode & 0xffffffffL;
复制代码

看到上图的Hash函数，让我想到了HashMap的Hash函数

f(key) = hash(key) & (table.length - 1) 
// 使用>>> 16的缘由，hashCode()的高位和低位都对f(key)有了必定影响力，使得分布更加均匀，散列冲突的概率就小了。
hash(key) = (h = key.hashCode()) ^ (h >>> 16)
复制代码

同理，将jobId的md5编码的高低位都对Hash结果有影响，使得Hash冲突的几率减少。

分片任务的实现 - 维护线程上下文

XXL-JOB的分片任务实现了任务的分布式执行，实际上是笔者调研的重点，平常开发中不少定时任务都是单机执行，对于后续数据量大的任务最好有一个分布式的解决方案。
分片任务的路由策略，源代码做者提出了分片广播的概念，刚开始还有点摸不清头脑，看了源码逐渐清晰了起来。
想必看过源码的也遇到过这么一个小插曲，路由策略咋没实现？以下图所示。

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()),
    // 说好的实现呢？？？居然是null
    SHARDING_BROADCAST(I18nUtil.getString("jobconf_route_shard"), null);
复制代码

再继续追查获得告终论，待我慢慢道来，首先分片任务执行参数传递的是什么？看XxlJobTrigger.trigger函数中的一段代码。

...
// 若是是分片路由，走的是这段逻辑
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());
            }
        } 
...
复制代码

参数通过自研RPC传递到执行器，在执行器中具体负责任务执行的JobThread.run中，看到了以下代码。

// 分片广播的参数比set进了ShardingUtil
ShardingUtil.setShardingVo(new ShardingUtil.ShardingVO(triggerParam.getBroadcastIndex(), triggerParam.getBroadcastTotal()));
...
// 将执行参数传递给jobHandler执行
handler.execute(triggerParamTmp.getExecutorParams())
复制代码

接着看ShardingUtil，才发现了其中的奥秘，请看代码。

public class ShardingUtil {
	// 线程上下文
    private static InheritableThreadLocal<ShardingVO> contextHolder = new InheritableThreadLocal<ShardingVO>();
	// 分片参数对象
    public static class ShardingVO {

        private int index;  // sharding index
        private int total;  // sharding total
		// 次数省略 get/set
    }
	// 参数对象注入上下文
    public static void setShardingVo(ShardingVO shardingVo){
        contextHolder.set(shardingVo);
    }
	// 从上下文中取出参数对象
    public static ShardingVO getShardingVo(){
        return contextHolder.get();
    }

}
复制代码

显而易见，在负责分片任务的ShardingJobHandler里取出了线程上下文中的分片参数，这里也给个代码把~

@JobHandler(value="shardingJobHandler")
@Service
public class ShardingJobHandler extends IJobHandler {

	@Override
	public ReturnT<String> execute(String param) throws Exception {

		// 分片参数
		ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
		XxlJobLogger.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardingVO.getIndex(), shardingVO.getTotal());

		// 业务逻辑
		for (int i = 0; i < shardingVO.getTotal(); i++) {
			if (i == shardingVO.getIndex()) {
				XxlJobLogger.log("第 {} 片, 命中分片开始处理", i);
			} else {
				XxlJobLogger.log("第 {} 片, 忽略", i);
			}
		}

		return SUCCESS;
	}

}
复制代码

由此得出，分布式实现是根据分片参数index及total来作的，简单来说，就是给出了当前执行器的标识，根据这个标识将任务的数据或者逻辑进行区分，便可实现分布式运行。
题外话：至于为何用外部注入分片参数的方式，不直接execute传递？

一、多是由于只有分片任务才用到这两个参数二、IJobHandler只有String类型参数

看完源码后的思考

一、通过这次看源代码，XXL-JOB的设计目标确实符合开发迅速、学习简单、轻量级、易扩展。
二、至于自研RPC尚未具体考量，具体接入应该会考虑公司的RPC框架。
三、做者给出的Quartz调度的不足，笔者得继续深刻了解。
四、框架中不少对宕机、故障、超时等异常情况的兼容值得学习。
五、Rolling日志以及日志系统实现须要继续了解。