说说XXLJob分片任务实现原理?

xxljob · 浏览次数 : 12

正文

XXL Job 是一个开源的分布式任务调度平台,其核心设计目标是开发迅速、学习简单、轻量级、易扩展的分布式任务调度框架。

这两天咱们开发的 AI Cloud 项目中,也使用到了 XXL Job 来执行分布式任务的调度,可以看出它的部署和使用虽然步骤很多,但用起来还是很简单的。

因为其本身为 Spring Boot 项目,所有对于 Java 程序员来说很友好,而且它还提供中文控制台,所以这也是他能在国内分布式任务调度系统这块一直流行的原因,如下图所示:
image.png

那么接下来咱们就来聊聊,XXL Job 的路由策略,以及路由策略中分片任务的执行原理。

1.路由策略

XXL Job 的路由策略主要作用是在任务执行器集群环境中,决定如何选择合适的执行器来执行任务。

XXL Job 路由策略包含以下几个:
image.png
其中:

  1. 第一个:选取执行器管理的注册地址列表中的第一个执行器来执行任务;
  2. 最后一个:选取执行器管理的注册地址列表中的最后一个执行器来执行任务;
  3. 轮询:依次选取执行器管理的注册地址列表中的执行器,周而复始。为了应对多个定时任务同时触发带来的数据一致性问题,XXL-JOB 使用一个静态的同步 Map 来存储每个任务的 jobId 和其对应的计数。每次计数增加后,对执行器地址列表的数量取余,将结果作为索引来获取对应的执行器地址。如果超过 24 小时没有触发调用该任务,会清空 Map 以释放一定空间;
  4. 随机:从执行器管理的注册地址列表中随机选取一个执行器来执行任务;
  5. 一致性 HASH:实现一致性 HASH 负载均衡算法;
  6. 最不经常使用:选择最近最少被调度的执行器执行任务(通过次数维度选取任务);
  7. 最近最久未使用:选择距离上次被调度时间最长的执行器执行任务(通过时间维度选取任务),有助于平衡各执行器的工作负载;
  8. 故障转移:在任务路由策略选择“故障转移”的情况下,如果执行器集群中的某一台机器出现故障,将会自动 Failover 切换到一台正常的执行器发送调度请求;
  9. 忙碌转移:当任务分配到某个执行器时,如果该执行器正处于忙碌状态(可能正在执行其他任务或资源紧张),则会尝试将任务转移到其他相对空闲的执行器上执行;
  10. 分片广播:选取执行器管理的注册地址列表中的所有地址,每个地址都执行一次任务。这种方式类似于 MQ 的广播模式,可以将任务广播到集群中的所有执行器上执行。此策略适用于需要在多个执行器上同时执行相同任务的场景,例如数据同步或分布式计算等。

也就是说在这些路由策略中,最复杂的就是分片广播了。

2.分片任务实现

所谓的分片广播也就是分片(执行)任务,它是将一个大任务划分为多个子任务并行执行,以提高效率。

假设,我们现在要使用分片任务执行一个大数据的查询与处理,此时的实现代码如下:

import com.xxl.job.core.context.XxlJobHelper;
import com.xxl.job.core.log.XxlJobLogger;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class ShardingBroadcastJob {

    @XxlJob("shardingBroadcastTask") 
    public void execute(String param) {
        // 获取分片参数:分片总数和分片序列号
        int shardIndex = XxlJobHelper.getShardIndex(); 
        int shardTotal = XxlJobHelper.getShardTotal(); 

        XxlJobLogger.log("当前节点的 index={}, 总结点数={}, 参数={}", shardIndex, shardTotal, param);

        // 模拟获取数据列表
        List<String> dataList = getDataList(); 

        // 执行分片逻辑
        shardingExecute(dataList, shardIndex, shardTotal);
    }

    public List<String> getDataList() {
        // 这里可以根据实际情况从数据库或其他数据源获取数据列表
        // 为了示例简单,直接返回一个固定的列表
        return List.of("data1", "data2", "data3", "data4", "data5", "data6", "data7", "data8", "data9", "data10");
    }

    public void shardingExecute(List<String> dataList, int shardIndex, int shardTotal) {
        XxlJobLogger.log("开始执行分片任务,当前分片={}, 总分片数={}", shardIndex, shardTotal);

        // 计算当前分片应处理的数据范围
        int start = (shardIndex * dataList.size()) / shardTotal;
        int end = ((shardIndex + 1) * dataList.size()) / shardTotal;

        // 处理当前分片的数据
        for (int i = start; i < end; i++) {
            String data = dataList.get(i);
            XxlJobLogger.log("处理数据: {}", data);
            // 在此处添加具体的数据处理逻辑
        }

        XxlJobLogger.log("分片任务执行完成");
    }
}

在上述代码中,在execute方法中,通过 XxlJobHelper.getShardIndex() 获取当前分片序号,通过 XxlJobHelper.getShardTotal() 获取总分片数。然后模拟获取了一个数据列表 dataList,接下来使用 shardingExecute 方法执行分片逻辑。

在 shardingExecute 方法中,根据分片序号和总分片数计算出当前分片应处理的数据范围,然后遍历该范围内的数据并进行处理(此处仅打印数据,实际应用中可添加具体的数据处理逻辑)。

在实际使用时,需要将任务部署到 XXL Job 执行器集群中,并在调度中心配置相应的任务,选择分片广播的路由策略。这样,当调度中心触发任务时,所有执行器都会执行该任务,并根据分片参数处理相应的数据分片,这样就能提升程序整体的执行效率了。

3.执行原理

了解了 XXL Job 的代码实现就能明白其运行原理,它的实现原理如下:

  1. 任务配置与分发:在 XXL Job 的调度中心,用户通过 Web 界面创建一个分片广播类型的任务,并设置相应的参数,如分片总数(shardingTotalCount)。当调度触发时,调度中心会将此任务广播至所有注册的执行器。
  2. 分片参数传递:每个执行器在接收到广播的任务时,会自动获得分片参数,包括分片总数和当前执行器应该处理的分片序号(shardingItem)。这些参数由 XXL Job 框架自动注入,使得执行器能够知道它应当处理哪个数据分片。
  3. 分片逻辑执行:实际的分片逻辑需要在执行器的任务处理器代码中实现,开发者需根据分片序号和总数,决定处理哪些数据。这通常涉及对数据源的分片访问,如数据库查询时使用分页查询或者 ID 取模等方法来确定每个执行器处理的数据范围。然后各个执行器并行处理各自分片的数据,互不影响。
  4. 结果汇总:由于是广播任务,每个执行器处理的是全量数据的一个子集,因此不存在汇总操作,每个执行器独立完成自己的处理逻辑。如果需要最终汇总结果,需要额外的逻辑来收集和整合各个执行器的输出。

课后思考

在分片任务时,如果其中某台机器掉电了导致结果一直未能正常返回,XXL Job 会如何处理?XXL Job 怎么保证任务只会被执行一次的?

本文已收录到我的面试小站 www.javacn.site,其中包含的内容有:Redis、JVM、并发、并发、MySQL、Spring、Spring MVC、Spring Boot、Spring Cloud、MyBatis、设计模式、消息队列等模块。

与说说XXLJob分片任务实现原理?相似的内容:

说说XXLJob分片任务实现原理?

XXL Job 是一个开源的分布式任务调度平台,其核心设计目标是开发迅速、学习简单、轻量级、易扩展的分布式任务调度框架。 这两天咱们开发的 AI Cloud 项目中,也使用到了 XXL Job 来执行分布式任务的调度,可以看出它的部署和使用虽然步骤很多,但用起来还是很简单的。 因为其本身为 Spri

说说RabbitMQ延迟队列实现原理?

使用 RabbitMQ 和 RocketMQ 的人是幸运的,因为这两个 MQ 自身提供了延迟队列的实现,不像用 Kafka 的同学那么苦逼,还要自己实现延迟队列。当然,这都是题外话,今天咱们重点来聊聊 RabbitMQ 延迟队列的实现原理,以及 RabbitMQ 实现延迟队列的优缺点有哪些? 很多人

使用libzip压缩文件和文件夹

简单说说自己遇到的坑: 分清楚三个组件:zlib、minizip和libzip。zlib是底层和最基础的C库,用于使用Deflate算法压缩和解压缩文件流或者单个文件,但是如果要压缩文件夹就很麻烦,主要是不知道如何归档,在zip内部形成对应的目录。这时就需要用更高级别的库,也就是minizip或li

面试官:说说Netty对象池的实现原理?

Netty 作为一个高性能的网络通讯框架,它内置了很多恰夺天工的设计,目的都是为了将网络通讯的性能做到极致,其中「对象池技术」也是实现这一目标的重要技术。 1.什么是对象池技术? 对象池技术是一种重用对象以减少对象创建和销毁带来的开销的方法。在对象池中,只有第一次访问时会创建对象,并将其维护在内存中

面试官:说说Netty的核心组件?

Netty 核心组件是指 Netty 在执行过程中所涉及到的重要概念,这些核心组件共同组成了 Netty 框架,使 Netty 框架能够正常的运行。 Netty 核心组件包含以下内容: 启动器 Bootstrap/ServerBootstrap 事件循环器 EventLoopGroup/EventL

都说DevOps落地难,到底难在哪里?也许你还没找到套路

当你打开这篇文章的时候,也许你也在为DevOps的落地而苦恼,也许你的组织正在尝试DevOps转型,作为一线的实践者,说说我对这个“落地难”的看法,欢迎交流不同看法~ DevOps是实践摸索出来的,别人的终究是别人的 如下图所示,你可能在不同企业研发效能的分享都看到过,各种关于DevOps的书上有会

哈啰面试:说说Dubbo运行原理?

Dubbo 是一款高性能、轻量级的开源 RPC(远程过程调用)框架,主要用于构建分布式服务和微服务架构。那 Dubbo 又是如何运行的呢?让我们一起来看。 1.核心组件 要说 Dubbo 运行流程就不得不先来了解一下 Dubbo 的核心组件了,因为 Dubbo 的交互流程是和核心组件息息相关的。 D

阿里面试:说说自适应限流?

限流想必大家都不陌生,它是一种控制资源访问速率的策略,用于保护系统免受过载和崩溃的风险。限流可以控制某个服务、接口或系统在一段时间内能够处理的请求或数据量,以防止系统资源耗尽、性能下降或服务不可用。 常见的限流策略有以下几种: 令牌桶算法:基于令牌桶的方式,限制每个单位时间内允许通过的请求量,请求量

字节面试:说说Java中的锁机制?

Java 中的锁(Locking)机制主要是为了解决多线程环境下,对共享资源并发访问时的同步和互斥控制,以确保共享资源的安全访问。 锁的作用主要体现在以下几个方面: 互斥访问:确保在任何时刻,只有一个线程能够访问特定的资源或执行特定的代码段。这防止了多个线程同时修改同一资源导致的数据不一致问题。 内

腾讯音乐:说说Redis脑裂问题?

Redis 脑裂问题是指,在 Redis 哨兵模式或集群模式中,由于网络原因,导致主节点(Master)与哨兵(Sentinel)和从节点(Slave)的通讯中断,此时哨兵就会误以为主节点已宕机,就会在从节点中选举出一个新的主节点,此时 Redis 的集群中就出现了两个主节点的问题,就是 Redis