elastic-job详解（一）：数据分片

数据分片的目的在于把一个任务分散到不一样的机器上运行，既能够解决单机计算能力上限的问题，也能下降部分任务失败对总体系统的影响。elastic-job并不直接提供数据处理的功能，框架只会将分片项分配至各个运行中的做业服务器（实际上是Job实例，部署在一台机器上的多个Job实例也能分片），开发者须要自行处理分片项与真实数据的对应关系。框架也预置了一些分片策略：平均分配算法策略，做业名哈希值奇偶数算法策略，轮转分片策略。同时也提供了自定义分片策略的接口。

分片原理

elastic-job的分片是经过zookeeper来实现的。分片的分片由主节点分配，以下三种状况都会触发主节点上的分片算法执行：

• 新的Job实例加入集群
• 现有的Job实例下线（若是下线的是leader节点，那么先选举而后触发分片算法的执行）
• 主节点选举

上述三种状况，会让zookeeper上leader节点的sharding节点上多出来一个necessary的临时节点，主节点每次执行Job前，都会去看一下这个节点，若是有则执行分片算法。

分片的执行结果会存储在zookeeper上，以下图，5个分片，每一个分片应该由哪一个Job实例来运行都已经分配好。分配的过程就是上面触发分片算法以后的操做。分配完成以后，各个Job实例就会在下次执行的时候使用上这个分配结果。

 

每一个job实例任务触发前都会获取本任务在本实例上的分片状况（直接和上图zookeeper上instance节点比对某一个分片是否该有这个Job实例执行），而后封装成shardingContext，传递给调用任务的实际执行方法：

/**
     * 执行做业.
     *
     * @param shardingContext 分片上下文
     */
void execute(ShardingContext shardingContext);

 

分片算法

全部的分片策略都继承JobShardingStrategy接口。根据当前注册到ZK的实例列表和在客户端配置的分片数量来进行数据分片。最终将每一个Job实例应该得到的分片数字返回出去。 方法签名以下：

/**
     * 做业分片.
     * 
     * @param jobInstances 全部参与分片的单元列表
     * @param jobName 做业名称
     * @param shardingTotalCount 分片总数
     * @return 分片结果
     */
    Map<JobInstance, List<Integer>> sharding(List<JobInstance> jobInstances, String jobName, int shardingTotalCount);

分片函数的触发，只会在leader选举的时候触发，也就是说只会在刚启动和leader节点离开的时候触发，而且是在leader节点上触发，而其余节点不会触发。

 

1. 基于平均分配算法的分片策略

基于平均分配算法的分片策略对应的类是：AverageAllocationJobShardingStrategy。它是默认的分片策略。它的分片效果以下：

• 若是有3个Job实例, 分红9片, 则每一个Job实例分到的分片是: 1=[0,1,2], 2=[3,4,5], 3=[6,7,8].
• 若是有3个Job实例, 分红8片, 则每一个Job实例分到的分片是: 1=[0,1,6], 2=[2,3,7], 3=[4,5].
• 若是有3个Job实例, 分红10片, 则个Job实例分到的分片是: 1=[0,1,2,9], 2=[3,4,5], 3=[6,7,8].

 

2. 做业名的哈希值奇偶数决定IP升降序算法的分片策略

这个策略的对应的类是：OdevitySortByNameJobShardingStrategy，它内部其实也是使用AverageAllocationJobShardingStrategy实现，只是在传入的节点实例顺序不同，也就是上面接口参数的List<JobInstance>。AverageAllocationJobShardingStrategy的缺点是一旦分片数小于Job实例数，做业将永远分配至IP地址靠前的Job实例上，致使IP地址靠后的Job实例空闲。而OdevitySortByNameJobShardingStrategy则能够根据做业名称从新分配Job实例负载。如：

• 若是有3个Job实例，分红2片，做业名称的哈希值为奇数，则每一个Job实例分到的分片是：1=[0], 2=[1], 3=[]
• 若是有3个Job实例，分红2片，做业名称的哈希值为偶数，则每一个Job实例分到的分片是：3=[0], 2=[1], 1=[]

实现比较简单：

long jobNameHash = jobName.hashCode();
if (0 == jobNameHash % 2) {
    Collections.reverse(jobInstances);
}
return averageAllocationJobShardingStrategy.sharding(jobInstances, jobName, shardingTotalCount);

 

3. 根据做业名的哈希值对Job实例列表进行轮转的分片策略

这个策略的对应的类是：RotateServerByNameJobShardingStrategy，和上面介绍的策略同样，内部一样是用AverageAllocationJobShardingStrategy实现，也是在传入的List<JobInstance>列表顺序上作文章。

 

4. 自定义分片策略

除了可使用上述分片策略以外，elastic-job还容许自定义分片策略。咱们能够本身实现JobShardingStrategy接口，而且配置到分片方法上去，整个过程比较简单，下面仅仅列出经过配置spring来切换自定义的分片算法的例子：

<job:simple id="MyShardingJob1" class="nick.test.elasticjob.MyShardingJob1" registry-center-ref="regCenter" cron="0/10 * * * * ?" sharding-total-count="5" sharding-item-parameters="0=A,1=B,2=C,3=D,4=E" job-sharding-strategy-class="nick.test.elasticjob.MyJobShardingStrategy"/>