背景
假设有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};

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

分片

ElasticJob 中任务分片项的概念，使得任务可以在分布式的环境下运行，每台任务服务器只运行分配给该服务器的分片。 随着服务器的增加或宕机，ElasticJob 会近乎实时的感知服务器数量的变更，从而重新为分布式的任务服务器分配更加合理的任务分片项，使得任务可以随着资源的增加而提升效率。

任务的分布式执行，需要将一个任务拆分为多个独立的任务项，然后由分布式的服务器分别执行某一个或几个分片项。

举例说明，如果作业分为 4 片，用两台服务器执行，则每个服务器分到 2 片，分别负责作业的 50% 的负载，如下图所示。

分片项
ElasticJob 并不直接提供数据处理的功能，而是将分片项分配至各个运行中的作业服务器，开发者需要自行处理分片项与业务的对应关系。 分片项为数字，始于 0 而终于分片总数减 1。

个性化分片参数
个性化参数可以和分片项匹配对应关系，用于将分片项的数字转换为更加可读的业务代码。

例如：按照地区水平拆分数据库，数据库 A 是北京的数据；数据库 B 是上海的数据；数据库 C 是广州的数据。 如果仅按照分片项配置，开发者需要了解 0 表示北京；1 表示上海；2 表示广州。 合理使用个性化参数可以让代码更可读，如果配置为 0=北京,1=上海,2=广州，那么代码中直接使用北京，上海，广州的枚举值即可完成分片项和业务逻辑的对应关系。

资源最大限度利用
ElasticJob 提供最灵活的方式，最大限度的提高执行作业的吞吐量。 当新增加作业服务器时，ElasticJob 会通过注册中心的临时节点的变化感知到新服务器的存在，并在下次任务调度的时候重新分片，新的服务器会承载一部分作业分片，如下图所示。

将分片项设置为大于服务器的数量，最好是大于服务器倍数的数量，作业将会合理的利用分布式资源，动态的分配分片项。

例如：3 台服务器，分成 10 片，则分片项分配结果为服务器 A = 0,1,2,9；服务器 B = 3,4,5；服务器 C = 6,7,8。 如果服务器 C 崩溃，则分片项分配结果为服务器 A = 0,1,2,3,4; 服务器 B = 5,6,7,8,9。 在不丢失分片项的情况下，最大限度的利用现有资源提高吞吐量。

高可用
当作业服务器在运行中宕机时，注册中心同样会通过临时节点感知，并将在下次运行时将分片转移至仍存活的服务器，以达到作业高可用的效果。 本次由于服务器宕机而未执行完的作业，则可以通过失效转移的方式继续执行。如下图所示。

将分片总数设置为 1，并使用多于 1 台的服务器执行作业，作业将会以 1 主 n 从的方式执行。 一旦执行作业的服务器宕机，等待执行的服务器将会在下次作业启动时替补执行。开启失效转移功能效果更好，如果本次作业在执行过程中宕机，备机会立即替补执行。

• ElasticJob-Lite
  ElasticJob-Lite 定位为轻量级无中心化解决方案，使用 jar 的形式提供分布式任务的协调服务。

对比

简单作业
意为简单实现，未经任何封装的类型。需实现 SimpleJob 接口。 该接口仅提供单一方法用于覆盖，此方法将定时执行。 与Quartz原生接口相似，但提供了弹性扩缩容和分片等功能。

public class MyElasticJob implements SimpleJob {
    
    @Override
    public void execute(ShardingContext context) {
        switch (context.getShardingItem()) {
            case 0: 
                // do something by sharding item 0
                break;
            case 1: 
                // do something by sharding item 1
                break;
            case 2: 
                // do something by sharding item 2
                break;
            // case n: ...
        }
    }
}

数据流作业
用于处理数据流，需实现 DataflowJob 接口。 该接口提供2个方法可供覆盖，分别用于抓取 (fetchData) 和处理 (processData) 数据。

public class MyElasticJob implements DataflowJob<Foo> {
    
    @Override
    public List<Foo> fetchData(ShardingContext context) {
        switch (context.getShardingItem()) {
            case 0: 
                List<Foo> data = // get data from database by sharding item 0
                return data;
            case 1: 
                List<Foo> data = // get data from database by sharding item 1
                return data;
            case 2: 
                List<Foo> data = // get data from database by sharding item 2
                return data;
            // case n: ...
        }
    }
    
    @Override
    public void processData(ShardingContext shardingContext, List<Foo> data) {
        // process data
        // ...
    }
}

流式处理

可通过属性配置 streaming.process 开启或关闭流式处理。

如果开启流式处理，则作业只有在 fetchData 方法的返回值为 null 或集合容量为空时，才停止抓取，否则作业将一直运行下去； 如果关闭流式处理，则作业只会在每次作业执行过程中执行一次 fetchData 和 processData 方法，随即完成本次作业。

如果采用流式作业处理方式，建议 processData 在处理数据后更新其状态，避免 fetchData 再次抓取到，从而使得作业永不停止。

简单使用

引入 Maven 依赖

<dependency>
    <groupId>org.apache.shardingsphere.elasticjob</groupId>
    <artifactId>elasticjob-lite-core</artifactId>
    <version>${latest.release.version}</version>
</dependency>

作业开发

public class MyJob implements SimpleJob {
    
    @Override
    public void execute(ShardingContext context) {
        switch (context.getShardingItem()) {
            case 0: 
                // do something by sharding item 0
                break;
            case 1: 
                // do something by sharding item 1
                break;
            case 2: 
                // do something by sharding item 2
                break;
            // case n: ...
        }
    }
}

作业配置

 JobConfiguration jobConfig = JobConfiguration.newBuilder("MyJob", 3).cron("0/5 * * * * ?").build();
作业调度
public class MyJobDemo {
    
    public static void main(String[] args) {
        new ScheduleJobBootstrap(createRegistryCenter(), new MyJob(), createJobConfiguration()).schedule();
    }
    
    private static CoordinatorRegistryCenter createRegistryCenter() {
        CoordinatorRegistryCenter regCenter = new ZookeeperRegistryCenter(new ZookeeperConfiguration("zk_host:2181", "my-job"));
        regCenter.init();
        return regCenter;
    }
    
    private static JobConfiguration createJobConfiguration() {
        // 创建作业配置
        // ...
    }
}