分布式任务处理：XXL-JOB分布式任务调度框架（三）

2023-07-05 843

版权

本文内容由阿里云实名注册用户自发贡献，版权归原作者所有，阿里云开发者社区不拥有其著作权，亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容，填写侵权投诉表单进行举报，一经查实，本社区将立刻删除涉嫌侵权内容。

简介： 分布式任务处理：XXL-JOB分布式任务调度框架

7.2 调度策略

执行器在集群部署下调度中心有哪些调度策略呢？查看xxl-job官方文档，阅读高级配置相关的内容：

路由策略：当执行器集群部署时，提供丰富的路由策略，包括：

FIRST（第一个）：每次调度选择集群中第一台执行器。
LAST（最后一个）：每次调度选择集群中最后一台执行器。
ROUND（轮询）：按照顺序每次调度选择一台执行器去调度。
RANDOM（随机）：每次调度随机选择一台执行器去调度。
CONSISTENT_HASH（一致性HASH）：每个任务按照Hash算法固定选择某一台机器，且所有任务均匀散列在不同机器上。
LEAST_FREQUENTLY_USED（最不经常使用）：使用频率最低的机器优先被选举。
LEAST_RECENTLY_USED（最近最久未使用）：最久未使用的机器优先被选举。
FAILOVER（故障转移）：按照顺序依次进行心跳检测，第一个心跳检测成功的机器选定为目标执行器并发起调度。
BUSYOVER（忙碌转移）：按照顺序依次进行空闲检测，第一个空闲检测成功的机器选定为目标执行器并发起调度。
SHARDING_BROADCAST(分片广播)：广播触发对应集群中所有机器执行一次任务，同时系统自动传递分片参数；可根据分片参数开发分片任务。

7.3 分片广播

我们思考一下如何进行分布式任务处理呢？如下图，我们会启动多个执行器组成一个集群，去执行任务。

分片广播策略：分片是指是调度中心将集群中的执行器标上序号：0，1，2，3…，广播是指每次调度会向集群中所有执行器发送调度请求，请求中携带分片参数。

每个执行器收到调度请求根据分片参数自行决定是否执行任务。

另外xxl-job还支持动态分片，当执行器数量有变更时，调度中心会动态修改分片的数量。

📍 作业分片适用哪些场景呢？

分片任务场景：10个执行器的集群来处理10w条数据，每台机器只需要处理1w条数据，耗时降低10倍；
广播任务场景：广播执行器同时运行shell脚本、广播集群节点进行缓存更新等。

所以，广播分片方式不仅可以充分发挥每个执行器的能力，并且根据分片参数可以控制任务是否执行，最终灵活控制了执行器集群分布式处理任务。

💬 “分片广播” 和普通任务开发流程一致，不同之处在于可以获取分片参数进行分片业务处理。

7.3.1 编写任务方法

/**
     * 分片广播任务
     */
    @XxlJob("shardingJobHandler")
    public void shardingJobHandler() throws Exception {
        /*
            分片参数：
                - shardIndex：分片序号
                - shardTotal：分片总数
         */
        int shardIndex = XxlJobHelper.getShardIndex();
        int shardTotal = XxlJobHelper.getShardTotal();
        System.out.printf("分片参数：当前分片序号 = %d, 总分片数 = %d\n", shardIndex, shardTotal);
        XxlJobHelper.log("分片参数：当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);
        // todo 业务逻辑
    }

7.3.2 增加一个节点服务

修改新节点的服务端口和执行器访问端口信息：-Dserver.port=10002 -Dxxl.job.executor.port=60001

启动这两个服务：

7.3.3 调度中心-执行器管理

上图说明在调度中心已经注册成功。

7.3.4 调度中心-新增与启用任务

7.3.5 校验任务

7.4 高级配置说明

7.4.1 子任务

每个任务都拥有一个唯一的任务ID(任务ID可以从任务列表获取)，当本任务执行结束并且执行成功时，将会触发子任务ID所对应的任务的一次主动调度，通过子任务可以实现一个任务执行完成去执行另一个任务。

7.4.2 调度过期策略

忽略：调度过期后，忽略过期的任务，从当前时间开始重新计算下次触发时间。
立即执行一次：调度过期后，立即执行一次，并从当前时间开始重新计算下次触发时间。

7.4.3 阻塞处理策略

调度过于密集执行器来不及处理时的处理策略

单机串行（默认）：调度请求进入单机执行器后，调度请求进入FIFO队列并以串行方式运行。
丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败。
覆盖之前调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务。

7.4.4 任务超时时间

支持自定义任务超时时间，任务运行超时将会主动中断任务。

7.4.5 失败重试次数

支持自定义任务失败重试次数，当任务失败时将会按照预设的失败重试次数主动进行重试。

8.作业分片方案

❓ 当一次分片广播到来，各执行器如何根据分片参数去分布式执行任务，保证执行器之间执行的任务不重复呢？

执行器收到调度请求后各自己查询属于自己的任务，这样就保证了执行器之间不会重复执行任务。

xxl-job设计作业分片就是为了分布式执行任务，XXL-JOB并不直接提供数据处理的功能，它只会给执行器分配好分片序号并向执行器传递分片总数、分片序号这些参数，开发者需要自行处理分片项与真实数据的对应关系。

每个执行器收到广播任务有两个参数：分片总数、分片序号。每个执行从数据表取任务时可以让任务id 模上分片总数，如果等于分片序号则执行此任务。

上边两个执行器实例那么分片总数为2，序号为0、1，从任务1开始，如下：

1 % 2 = 1 执行器2执行
2 % 2 = 0 执行器1执行
3 % 2 = 1 执行器2执行
以此类推

9.三个经典面试题

9.1 xxl-jobo是怎么工作的?

XXL-JOB分布式任务调度服务由调用中心和执行器组成，调用中心负责按任务调度策略向执行器下发任务，执行器负责接收任务执行任务。

首先部署并启动xxl-job调度中心。(一个java工程)
首先在微服务添加xxl-job依赖，在微服务中配置执行器
启动微服务，执行器向调度中心上报自己.
在微服务中写一个任务方法并用xxl-job的注解去标记执行任务的方法名称。
在调度中心配置任务调度策略，调度策略就是每隔多长时间执行还是在每天或每月的固定时间去执行，比如每天0点执行，或每隔1小时执行一次等
在调度中心启动任务。
调度中心根据任务调度策略，到达时间就开始下发任务给执行器。、
执行器收到任务就开始执行任务。

9.2 如何保证任务不重复执行？

调度中心按分片广播的方式去下发任务。
执行器收到作业分片广播的参数：分片总数和分月序号,计算任务id除以分片总数得到一个余数，如果余数等于分片序号这时就去执行这个任务，这里保证了不同的执行器执行不同的任务。
配置调度过期策略为"忽略”，避免同一个执行器多次重复执行同一个任务。

忽略：调度过期后，忽略过期的任务，从当前时间开始重新计算下次触发时间。

配置任务阻塞处理策略为“丢弃后续调度”，注意:弃也没事下一次调度就又可以执行了。

丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败。

另外还要保证任务处理的幂等性，执行过的任务可以打一个状态标记已完成，下次再调度执行该任务判断该任务已完成就不再执行。

9.3 如何保证任务处理的幂等性？

任务的幂等性是指：对于数据的操作不论多少次，操作的结果始终是一致的。执行器接收调度请求去执行任务，要有办法去判断该任务是否处理完成，如果处理完则不再处理，即使重复调度处理相同的任务也不能重复处理已经处理过的数据。

幂等性描述了一次和多次请求某一个资源对于资源本身应该具有同样的结果。

幂等性是为了解决重复提交问题，比如：恶意刷单，重复支付等。

📍 解决幂等性常用的方案：

数据库约束，比如：唯一索引，主键。
乐观锁，常用于数据库，更新数据时根据乐观锁状态去更新。
唯一序列号，请求前生成唯一的序列号，携带序列号去请求，操作时先判断与该序列号是否相等。不相等则说明已经执行过了就不再执行，否则执行并且修改序列号或删除。

例如在数据库中我们对于操作过的记录修改字段status的值来表示已经操作。

分布式任务处理：XXL-JOB分布式任务调度框架（三）

7.2 调度策略