Elastic-Job2.1.5源码-图解分片算法动画

JobShardingStrategy jobShardingStrategy = JobShardingStrategyFactory.getStrategy(liteJobConfig.getJobShardingStrategyClass());
jobNodeStorage.executeInTransaction(new PersistShardingInfoTransactionExecutionCallback(jobShardingStrategy.sharding(availableJobInstances, jobName, shardingTotalCount)));

jobShardingStrategy.sharding(availableJobInstances, jobName, shardingTotalCount))

public final class AverageAllocationJobShardingStrategy implements JobShardingStrategy {
    
    @Override
    public Map<JobInstance, List<Integer>> sharding(final List<JobInstance> jobInstances, final String jobName, final int shardingTotalCount) {
        if (jobInstances.isEmpty()) {
            return Collections.emptyMap();
        }
        Map<JobInstance, List<Integer>> result = shardingAliquot(jobInstances, shardingTotalCount);
        addAliquant(jobInstances, shardingTotalCount, result);
        return result;
    }
/**
*加入3台机器 分8片 这个方法是对平均数量的分配，不包含余数那部分
*/
    private Map<JobInstance, List<Integer>> shardingAliquot(final List<JobInstance> shardingUnits, final int shardingTotalCount) {
        Map<JobInstance, List<Integer>> result = new LinkedHashMap<>(shardingTotalCount, 1);
        //计算出平均每个作业实例应该获取到的分片数量    2=8/3  这个计算结果不会包含余数，在这里余数将得不到分配，这个例子中会有两个分片得不到分配
        int itemCountPerSharding = shardingTotalCount / shardingUnits.size();
          int count = 0;
        //遍历作业实例为每个作业实例分配对应分片，这个循环过程中只会分配平均应得分片数量，余数对应的分片后面会有方法处理
        for (JobInstance each : shardingUnits) {
            List<Integer> shardingItems = new ArrayList<>(itemCountPerSharding + 1);
            for (int i = count * itemCountPerSharding; i < (count + 1) * itemCountPerSharding; i++) {
                shardingItems.add(i);
            }
            result.put(each, shardingItems);
            count++;
        }
        return result;
    }
   //这里是对余数分配，余数小于实例总数，遍历实例逐个取余数
    private void addAliquant(final List<JobInstance> shardingUnits, final int shardingTotalCount, final Map<JobInstance, List<Integer>> shardingResults) {
        int aliquant = shardingTotalCount % shardingUnits.size();
        int count = 0;
        for (Map.Entry<JobInstance, List<Integer>> entry : shardingResults.entrySet()) {
            if (count < aliquant) {
                entry.getValue().add(shardingTotalCount / shardingUnits.size() * shardingUnits.size() + count);
            }
            count++;
        }
    }
}

/**
* 根据作业名的哈希值奇偶数决定IP升降序算法的分片策略.
* 
* <p>
* 作业名的哈希值为奇数则IP升序.
* 作业名的哈希值为偶数则IP降序.
* 用于不同的作业平均分配负载至不同的服务器.
* 如: 
* 1. 如果有3台服务器, 分成2片, 作业名称的哈希值为奇数, 则每台服务器分到的分片是: 1=[0], 2=[1], 3=[].
* 2. 如果有3台服务器, 分成2片, 作业名称的哈希值为偶数, 则每台服务器分到的分片是: 3=[0], 2=[1], 1=[].
* </p>
* 
* @author zhangliang
*/
public final class OdevitySortByNameJobShardingStrategy implements JobShardingStrategy {
    
    private AverageAllocationJobShardingStrategy averageAllocationJobShardingStrategy = new AverageAllocationJobShardingStrategy();
    
    @Override
    public Map<JobInstance, List<Integer>> sharding(final List<JobInstance> jobInstances, final String jobName, final int shardingTotalCount) {
        long jobNameHash = jobName.hashCode();
        if (0 == jobNameHash % 2) { 
           //如果计算得到作业名字hash值是偶数则将作业实例顺序反转
            Collections.reverse(jobInstances);
        }
    //处理过顺序的作业实例列表进行平均分配
        return averageAllocationJobShardingStrategy.sharding(jobInstances, jobName, shardingTotalCount);
    }
}

/**
* 根据作业名的哈希值对服务器列表进行轮转的分片策略.
* 
* @author weishubin
*/
public final class RotateServerByNameJobShardingStrategy implements JobShardingStrategy {
    
    private AverageAllocationJobShardingStrategy averageAllocationJobShardingStrategy = new AverageAllocationJobShardingStrategy();
    
    @Override
    public Map<JobInstance, List<Integer>> sharding(final List<JobInstance> jobInstances, final String jobName, final int shardingTotalCount) {
        return averageAllocationJobShardingStrategy.sharding(rotateServerList(jobInstances, jobName), jobName, shardingTotalCount);
    }
   //对作业实例根据hash反转
    private List<JobInstance> rotateServerList(final List<JobInstance> shardingUnits, final String jobName) {
        int shardingUnitsSize = shardingUnits.size();
        //取余数 计算一个反转偏移量
        int offset = Math.abs(jobName.hashCode()) % shardingUnitsSize;
        //没有余数则可以均分
        if (0 == offset) {
            return shardingUnits;
        }
        //通过偏移量 将元素重新插入集合 如果便宜量为1则每个元素向前移动1个 最前面的元素向尾部移动
        List<JobInstance> result = new ArrayList<>(shardingUnitsSize);
        for (int i = 0; i < shardingUnitsSize; i++) {
            
            int index = (i + offset) % shardingUnitsSize;
            result.add(shardingUnits.get(index));
        }
        return result;
    }

/**
* 在事务中执行操作.
* 
* @param callback 执行操作的回调
*/
public void executeInTransaction(final TransactionExecutionCallback callback) {
    try {
        CuratorTransactionFinal curatorTransactionFinal = getClient().inTransaction().check().forPath("/").and();
        callback.execute(curatorTransactionFinal);
        curatorTransactionFinal.commit();
    //CHECKSTYLE:OFF
    } catch (final Exception ex) {
    //CHECKSTYLE:ON
        RegExceptionHandler.handleException(ex);
    }
}



class PersistShardingInfoTransactionExecutionCallback implements TransactionExecutionCallback {
    
    private final Map<JobInstance, List<Integer>> shardingResults;
    
    @Override
    public void execute(final CuratorTransactionFinal curatorTransactionFinal) throws Exception {
//这里我们遍历实例分片列表
        for (Map.Entry<JobInstance, List<Integer>> entry : shardingResults.entrySet()) {
    //写入分片节点值就是当前的key（作业实例）的id，对应节点/作业名字/sharding/分片项变量/instance 
            for (int shardingItem : entry.getValue()) {
                curatorTransactionFinal.create().forPath(jobNodePath.getFullPath(ShardingNode.getInstanceNode(shardingItem)), entry.getKey().getJobInstanceId().getBytes()).and();
            }
        }
        //所有写入操作处理完毕开始删除 需要分片标示和分片中的标示
curatorTransactionFinal.delete().forPath(jobNodePath.getFullPath(ShardingNode.NECESSARY)).and();
        curatorTransactionFinal.delete().forPath(jobNodePath.getFullPath(ShardingNode.PROCESSING)).and();
    }
}