上一篇博客Elastic-Job原理--服务器初始化、节点选举与通知(二)介绍了Elastic-Job的启动流程，这篇博客我们了解学习一下Elastic-Job的任务分片策略，目前提供了三种任务分片策略，分片策略的实现最终是在注册中心zk中在分片的instance中写入实例信息。



目前Elastic-Job提供分片接口JobShardingStrategy：

/**
 * 作业分片策略.
 * 
 * @author zhangliang
 */
public interface JobShardingStrategy {
    
    /**
     * 作业分片.
     * 
     * @param jobInstances 所有参与分片的单元列表
     * @param jobName 作业名称
     * @param shardingTotalCount 分片总数
     * @return 分片结果
     */
    Map<JobInstance, List<Integer>> sharding(List<JobInstance> jobInstances, String jobName, int shardingTotalCount);
}

目前有如下实现类：



（1）AverageAllocationJobShardingStrategy：基于平均分配算法的分片策略.

 如果分片不能整除, 则不能整除的多余分片将依次追加到序号小的服务器.

 如: 

 1. 如果有3台服务器, 分成9片, 则每台服务器分到的分片是: 1=[0,1,2], 2=[3,4,5], 3=[6,7,8].

  2. 如果有3台服务器, 分成8片, 则每台服务器分到的分片是: 1=[0,1,6], 2=[2,3,7], 3=[4,5].

  3. 如果有3台服务器, 分成10片, 则每台服务器分到的分片是: 1=[0,1,2,9], 2=[3,4,5], 3=[6,7,8].

/**
 * 基于平均分配算法的分片策略.
 * 
 * <p>
 * 如果分片不能整除, 则不能整除的多余分片将依次追加到序号小的服务器.
 * 如: 
 * 1. 如果有3台服务器, 分成9片, 则每台服务器分到的分片是: 1=[0,1,2], 2=[3,4,5], 3=[6,7,8].
 * 2. 如果有3台服务器, 分成8片, 则每台服务器分到的分片是: 1=[0,1,6], 2=[2,3,7], 3=[4,5].
 * 3. 如果有3台服务器, 分成10片, 则每台服务器分到的分片是: 1=[0,1,2,9], 2=[3,4,5], 3=[6,7,8].
 * </p>
 * 
 * @author zhangliang
 */
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;
    }
    //根据整除规则，将整除后的数据进行分配
    private Map<JobInstance, List<Integer>> shardingAliquot(final List<JobInstance> shardingUnits, final int shardingTotalCount) {
        Map<JobInstance, List<Integer>> result = new LinkedHashMap<>(shardingTotalCount, 1);
        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++;
        }
    }
}

（2）OdevitySortByNameJobShardingStrategy：根据作业名的哈希值奇偶数决定IP升降序算法的分片策略.

首先 作业名的哈希值为奇数则IP升序. 作业名的哈希值为偶数则IP降序.然后再调用AverageAllocationJobShardingStrategy的平均分片算法进行分片。

/**
 * 根据作业名的哈希值奇偶数决定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) {
            Collections.reverse(jobInstances);
        }
        return averageAllocationJobShardingStrategy.sharding(jobInstances, jobName, shardingTotalCount);
    }
}

（3）RotateServerByNameJobShardingStrategy：根据作业名的哈希值对服务器列表进行轮转的分片策略.

/**
 * 根据作业名的哈希值对服务器列表进行轮转的分片策略.
 * 
 * @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);
    }
    
    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;
        }
        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;
    }
}

总结：总体上使用的还是平均分片算法，不过是将实例进行了不同的排序操作。

背景

假设有k个地市，每个地市有x个订单执行，总共kx个订单，而每个订单中又有一个字段体现出地市信息。
@Component
@Slf4j
public class AhOrdersXxlJob {

    //城市编号
    private static final List<Integer> CITY_ID_LIST = Arrays.asList(550, 551, 552, 553, 554, 555, 556, 557, 558, 559, 561, 562, 563, 564, 566);

	//每个城市的任务数
    private static final int PER_LATN_TASK_NUM = 30; 
	
	// 任务数据库
    private static final Map<Integer, List<String>> singleMachineMultiTasks; 

     static {
        singleMachineMultiTasks = new HashMap<>();
        CITY_ID_LIST.forEach(city -> {
            List<String> tasks = new ArrayList<>(PER_LATN_TASK_NUM);
            IntStream.rangeClosed(1, PER_LATN_TASK_NUM).forEach(index -> {
                String orderInfo = city + "------NO." + index ;
                tasks.add(orderInfo);
            });

            singleMachineMultiTasks.put(city, tasks);
        });
    }
}

现使用xxljob进行分片任务执行，有两种解决思路。
单机多任务
自定义业务规则，配置多个xxl任务，来实现分片功能。

配置多个任务



每个任务指定不同的参数,但使用相同的jobhanlder:


JobHandler实现
    @XxlJob(value = "singleMachineMultiTasks", init = "init", destroy = "destroy")
    public ReturnT<String> singleMachineMultiTasks(String cities) throws Exception {

        if (StringUtils.isEmpty(cities)) {
            return new ReturnT(FAIL_CODE, "latnIds不能为空");
        }

        Arrays.stream(cities.split(",")).map(String::trim).filter(StringUtils::isNotBlank).map(Integer::parseInt).forEach(latnId -> {
            List<String> tasks = singleMachineMultiTasks.get(latnId);
            Optional.ofNullable(tasks).ifPresent(todoTasks -> {
                todoTasks.forEach(task -> {
                    XxlJobLogger.log("【{}】执行【{}】，任务内容为：{}", Thread.currentThread().getName(), latnId, task);
                });
            });
        });
        return ReturnT.SUCCESS;
    }

执行结果

分别启动两个任务，查看执行日志：




分两个线程，分别执行各自的任务清单;

多机分片
采用多机器取模的方式，来为不同的机器指定各自服务的城市列表。

一致性hash ?

启动多个执行管理器实例


xxl执行管理器


配置多实例分片任务


JobHandler实现
 @XxlJob(value = "multiMachineMultiTasks", init = "init", destroy = "destroy")
    public ReturnT<String> multiMachineMultiTasks(String params) throws Exception {
        ShardingUtil.ShardingVO shardingVO = ShardingUtil.getShardingVo();
        int n = shardingVO.getTotal(); // n 个实例  
        int i = shardingVO.getIndex(); // 当前为第i个

        IntStream.range(0, CITY_ID_LIST.size()).forEach(cityIndex -> {
            if (cityIndex % n == i) {
                int city = CITY_ID_LIST.get(cityIndex);
                List<String> tasks = singleMachineMultiTasks.get(city);
                Optional.ofNullable(tasks).ifPresent(todoTasks -> {
                    todoTasks.forEach(task -> {
                        XxlJobLogger.log("实例【{}】执行【{}】，任务内容为：{}", i, city, task);
                    });

                });
            }
        });
        return ReturnT.SUCCESS;
    }

执行结果





生命周期函数
@XxlJob(value = "singleMachineMultiTasks", init = "init", destroy = "destroy")
public ReturnT<String> singleMachineMultiTasks(String cities) throws Exception {
	//.....
}

@XxlJob(value = "multiMachineMultiTasks", init = "init", destroy = "destroy")
public ReturnT<String> multiMachineMultiTasks(String params) throws Exception {
	//.....
}

 public void init() {
     log.info("init");
 }

 public void destroy() {
     log.info("destory");
 }

如果不显示的指明生命周期函数，在方法执行完之后，会被销毁。
todo - - 思考
如果执行器管理某一个实例挂掉了？ 如何保证那一部分数据不会丢失，不会重复消费？

文章目录
简单任务
分片任务

简单任务
@Component
public class SimpleJobHandler {

    @XxlJob(value ="simpleJobHandler" )
    public ReturnT<String> execute(String param) throws InterruptedException {

        IntStream.rangeClosed(1,20).forEach(index->{
            XxlJobLogger.log("simpleJobHandler>>"+index);
        });
        //官方文档说 如果任务超时 是采用interrupt机制打断子线程的，因此需要将InterruptedException 向上抛出
        //不能catch，否则任务超时后 任务还会被正常执行完
        Thread.sleep(ThreadLocalRandom.current().nextInt(10000));

        //任务超时后 这句日志不会被打印出来
        XxlJobLogger.log("执行完毕");

        return ReturnT.SUCCESS;
    }

}

注意：xxl-job任务需用XxlJobLogger输出日志

其中页面上JobHandler对应于代码上@XxlJob的value值，超时时间设置2s

由于本地测试时 运行了多个执行器，所以路由策略那里选用了轮询。也可根据实际情况自行调整

查看下运行日志

发现当任务执行超时后，程序即被中断，后续的代码逻辑不再被执行。
分片任务
使用分片是为了更大效率地利用执行器集群资源，比如有一个任务需要处理20W条数据，每条数据的业务逻辑处理要0.1s。对于普通任务来说，只有一个线程来处理 可能需要5~6个小时才能处理完。
如果将20W数据均匀分给集群里的3台机器同时处理，那么耗时会大大缩短，也能充分利用集群资源。
在xxl-job里，如果有一个执行器集群有3个机器，那么分片总数是3，分片序号0，1，2 分别对应那三台机器。
举例：
有一个执行器集群，有三台机器。对一批用户信息进行处理，处理规则如下

id % 分片总数 余数是0 的，在第1个执行器上执行
id % 分片总数 余数是1 的，在第2个执行器上执行
id % 分片总数 余数是2 的，在第3个执行器上执行

@Component
public class ShardingJob {

    private List<User> userList;

    @PostConstruct
    public void init() {

        userList = LongStream.rangeClosed(1, 10)
                .mapToObj(index -> new User(index, "wojiushiwo" + index))
                .collect(Collectors.toList());

    }

    @XxlJob(value = "shardingJobHandler")
    public ReturnT<String> execute(String param) {

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

        for (int i = 0; i < userList.size(); i++) {
            //将数据均匀地分布到各分片上执行
            if (i % shardingVo.getTotal() == shardingVo.getIndex()) {
                XxlJobLogger.log("第 {} 片, 命中分片开始处理{}", shardingVo.getIndex(), userList.get(i).toString());
            } else {
                XxlJobLogger.log("{},忽略", shardingVo.getIndex());
            }

        }
        return ReturnT.SUCCESS;
    }


}

如果是数据库数据的话，可以进行如下处理，让每个分片执行一部分数据
-- 其中count是分片总数，index是当前分片数
select id,name
from user
where status=1 and mod(id,#{count})=#{index};

下面查看下三个执行器的执行日志：

Transer

是一个传输框架,目前支持： 

- 支持 HTTP/HTTPS 断点续传下载 

- 支持 HTTP/HTTPS 大文件分片上传 

- 支持 类EventBus的task状态变更通知，支持三种线程的订阅模式 

- 支持 任务分组，分用户



设计





简单的下载或上传:

下载: 

““ java  

   mHandler = new DefaultHttpDownloadHandler(); 

        //创建一个任务 

        ITask task = new TaskBuilder() 

                .setName(“test.zip”) //设置任务名称 

                .setDataSource(URL)  //设置数据源 

                .setDestSource(FILE_PATH) //设置目标路径 

                .build(); 

        mHandler.setTask(task);

    //设置请求参数
    Map<String,String> params = new HashMap<>();
    params.put("path","test.zip");
    mHandler.setParams(params);
    mHandler.setHandlerListenner(new SimpleTaskHandlerListenner());

    //设置一个线程池去下载文件，如果不设置，则会在当前线程进行下载。
    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3,3,
            6000, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(10000));
    mHandler.setThreadPool(threadPool);


 

上传: 
 java  

   mHandler = new DefaultHttpUploadHandler(); 

        //创建一个任务 

        ITask task = new TaskBuilder() 

                .setName(“test.zip”) //设置任务名称 

                .setDataSource(FILE_PATH)  //设置数据源 

                .setDestSource(URL) //设置目标路径 

                .build(); 

        mHandler.setTask(task);

    //设置请求参数
    Map<String,String> params = new HashMap<>();
    params.put("path","test.zip");
    mHandler.setParams(params);
    mHandler.setHandlerListenner(new SimpleTaskHandlerListenner());

    //设置一个线程池去上传文件，如果不设置，则会在当前线程进行上传。
    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3,3,
            6000, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(10000));
    mHandler.setThreadPool(threadPool);


““



使用任务管理:

1.创建任务

““ java 

ITask task = new TaskBuilder() 

                .setTaskType(task_type)  //任务类型 

                .setDataSource(source)   //任务数据源 (下在任务为要下载的服务文件链接，上传任务为要上传的本地文件路径) 

                .setDestSource(dest)     //任务目标源 (下载任务为保存的本地路径，上传任务为服务器地址) 

                .setName(NAME)           //任务名称 

                .build();

    ITaskCmd cmd = new TaskCmdBuilder()
            .setTaskType(task_type) //任务类型
            .setProcessType(ProcessType.TYPE_ADD_TASK) //操作类型(添加任务)
            .setTask(task) //任务信息
            .build();

    TaskEventBus.getDefault().execute(cmd); //执行命令


““ 

2.开始任务

““ java 

        ITaskCmd cmd = new TaskCmdBuilder() 

                .setTaskType(task_type) //任务类型 

                .setState(TaskState.STATE_START) 

                .setProcessType(ProcessType.TYPE_CHANGE_TASK) //操作类型(添加任务) 

                .setTask(task) //任务信息 

                .build();

    TaskEventBus.getDefault().execute(cmd); //执行命令


 

3.结束/暂停 任务 
 java 

        ITaskCmd cmd = new TaskCmdBuilder() 

                .setTaskType(task_type) //任务类型 

                .setState(TaskState.STATE_START) 

                .setProcessType(ProcessType.TYPE_CHANGE_TASK) //操作类型(添加任务) 

                .setTask(task) //任务信息 

                .build();

    TaskEventBus.getDefault().execute(cmd); //执行命令


““ 

4.接收任务变更通知 

- 在Activity,Fragement,Service,Dialog 等 onResume 或 onStart 中:

java 

TaskEventBus.getDefault().regesit(this); 
 

- 在 onPause , onStop 中使用 

java 

TaskEventBus.getDefault().unregesit(this); 
 

- 添加一个方法，参数为List tasks,并且添加注解TaskSubscriber

java  

@TaskSubscriber 

public void onTasksChanged(List<ITask> tasks) { 

       //TODO update ui on any processtype 

} 



TaskScriber 



  
默认情况下TaskScriber 会接受所有任务变更的消息，也可以指定只接受某个操作的消息例如：
   
java 

  @TaskSubscriber(taskType = TYPE_DOWNLOAD, processType = TYPE_CHANGE_TASK) 

  public void onTasksChanged(List<ITask> tasks) { 

         //TODO update ui in posting thread 

  } 
 

  也可以指定消息接收的线程，默认为发送消息的线程，例如:


java 

@TaskSubscriber(taskType = TYPE_DOWNLOAD, processType = TYPE_CHANGE_TASK,threadMode = ThreadMode.MODE_MAIN) 

public void onTasksChanged(List<ITask> tasks) { 

       //TODO update ui in main thread 

} 




配置传输服务(TranferService的onCreate 中)

““ java 

        TaskEventBus.init(getApplicationContext()); //初始化消息管理器 

        DaoHelper.init(getApplicationContext());    //初始化数据库 

        mContext = getApplicationContext();

    mTaskManagerProxy = new TaskManagerProxy();  
    mTaskManagerProxy.setProcessCallback(this);
    //Taskprocessor 为在内存中处理任务，TaskdDbProcessor 为数据库操作， 可以 实现ITaskProcessor 替换默认处理器
    mTaskManagerProxy.setTaskProcessor(new TaskProcessorProxy(new TaskProcessor(),new TaskDbProcessor())); 

    mTaskManagerProxy.setTaskManager(new TaskManager()); 
    //可以继承BaseTaskHandler 实现新的传输处理器
    mTaskManagerProxy.setTaskHandler(TaskType.TYPE_HTTP_DOWNLOAD, DefaultHttpDownloadHandler.class); //设置下载器
    mTaskManagerProxy.setTaskHandler(TaskType.TYPE_HTTP_UPLOAD, DefaultHttpUploadHandler.class); //设置上传器

    ThreadPoolExecutor threadPool = new ThreadPoolExecutor(3,3,
            6000, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(10000));
    mTaskManagerProxy.setThreadPool(TaskType.TYPE_HTTP_UPLOAD, threadPool);//设置上传线程池

    threadPool = new ThreadPoolExecutor(3,3,
            6000, TimeUnit.MILLISECONDS,new ArrayBlockingQueue<Runnable>(10000));
    mTaskManagerProxy.setThreadPool(TaskType.TYPE_HTTP_DOWNLOAD,threadPool); //下载线程池

    mTaskManagerProxy.setHeaders(new HashMap<String, String>()); //设置请求头
    mTaskManagerProxy.setParams(new HashMap<String, String>()); //设置url参数


““

项目地址: 

- https://github.com/shilec/Transer

服务端测试Demo详见： 

- https://github.com/shilec/WebDemo