利用 XXL-JOB 实现灵活控制的分片处理

简介: 本文讲述了一种利用 XXL-JOB 来进行分片任务处理的方法,另外加入对执行节点数的灵活控制。

本文讲述了一种利用 XXL-JOB 来进行分片任务处理的方法,另外加入对执行节点数的灵活控制。

场景

现在一张数据表里有大量数据需要某个服务端应用来处理,要求:

  1. 能够并行处理;

  2. 能够较灵活地控制并行任务数量。

  3. 压力较均衡地分散到不同的服务器节点;

思路

因为需要并行处理同一张数据表里的数据,所以比较自然地想到了分片查询数据,可以利用对 id 取模的方法进行分片,避免同一条数据被重复处理。

根据第 1、2 点要求,本来想通过对线程池的动态配置来实现,但结合第 3 点来考虑,服务器节点数量有可能会变化,节点之间相互无感知无通信,自己在应用内实现一套调度机制可能会很复杂。

如果有现成的独立于这些服务器节点之外的调度器就好了——顺着这个思路,就想到了已经接入的分布式任务调度平台 XXL-JOB,而在阅读其 官方文档 后发现「分片广播 & 动态分片」很贴合这种场景。

图片

方案

  1. 利用 XXL-JOB 的路由策略「分片广播」来调度定时任务;

  2. 通过任务参数传入执行任务节点数量;

  3. 定时任务逻辑里,根据获取到的分片参数、执行任务节点数量,决策当前节点是否需要执行,分片查询数据并处理:

  • 如果 分片序号 > (执行任务节点数量 - 1),则当前节点不执行任务,直接返回;

  • 否则,取 分片序号 和 执行任务节点数量 作为分片参数,查询数据并处理。

这样,我们可以实现灵活调度 [1, N] 个节点并行执行任务处理数据。

主要代码示例

JobHandler 示例:

@XxlJob("demoJobHandler")
public void execute() {
    String param = XxlJobHelper.getJobParam();
    if (StringUtils.isBlank(param)) {
        XxlJobHelper.log("任务参数为空");
        XxlJobHelper.handleFail();
        return;
    }

    // 执行任务节点数量
    int executeNodeNum = Integer.valueOf(param);
    // 分片序号
    int shardIndex = XxlJobHelper.getShardIndex();
    // 分片总数
    int shardTotal = XxlJobHelper.getShardTotal();

    if (executeNodeNum <= 0 || executeNodeNum > shardTotal) {
        XxlJobHelper.log("执行任务节点数量取值范围[1,节点总数]");
        XxlJobHelper.handleFail();
        return;
    }

    if (shardIndex > (executeNodeNum - 1)) {
        XxlJobHelper.log("当前分片 {} 无需执行", shardIndex);
        XxlJobHelper.handleSuccess();
        return;
    }

    shardTotal = executeNodeNum;

    // 分片查询数据并处理
    process(shardIndex, shardTotal);

    XxlJobHelper.handleSuccess();
}

分片查询数据示例:

select field1, field2 
from table_name 
where ... 
    and mod(id, #{shardTotal}) = #{shardIndex} 
order by id limit #{rows};

进一步思考

  1. 如果需要更大的并发量,需要有大于应用节点数量的任务并行,如何处理?

    两种思路:

  • 通过任务参数传入一个并发数,单个节点在处理任务时,将查询到的数据按这个数字进行再分片,交由线程池并行处理;

  • 配置 M 个定时任务,指定相同的 JobHandler,给它们编号 0、1、2…M,并将定时任务编号和 M 这两个数,由任务参数传入,定时任务逻辑里,先根据分片参数、定时任务编号、M,重新计算出新的分片参数,如 分片序号 = (分片序号 * M) + 定时任务编号,分片总数 = 分片总数 * M,再查询数据并处理。

  1. 如果有可能频繁调整任务执行逻辑,包括可能要新增任务参数等,而不想重启服务器,如何解决?

    可以考虑使用 XXL-JOB 的「GLUE模式」任务,能够在线编辑和更新定时任务执行逻辑。

参考

  • 分布式任务调度平台XXL-JOB
目录
相关文章
|
消息中间件 数据可视化 NoSQL
XXL-Job搭建(传统方式&Docker方式)与使用(Linux环境下)
XXL-Job搭建(传统方式&Docker方式)与使用(Linux环境下)
8141 0
XXL-Job搭建(传统方式&Docker方式)与使用(Linux环境下)
|
10月前
|
Java 调度
什么是分片广播任务
本文介绍XXL-JOB的分片广播机制,通过集群执行器动态分片处理任务。调度中心为每个执行器分配分片参数,实现任务并行处理,提升效率。适用于大数据量分布式场景,支持动态扩容,每台机器处理部分数据,显著降低耗时。开发时可通过`getShardIndex()`和`getShardTotal()`获取分片信息,灵活控制业务逻辑。
|
Java 调度 数据库
SpringBoot整合XXL-JOB【05】- 任务分片
在实际业务中,批量定时任务可能因上一批任务未完成而影响业务。为解决此问题,本文介绍如何使用Xxl-job对批量任务进行分片处理,通过分片广播形式调度集群机器并行执行任务,大幅提升执行效率。具体步骤包括环境准备、添加依赖和配置、声明实体类与查询类,以及改造业务逻辑实现分片查询。测试结果显示,分片处理将两千条数据的执行时间从30秒缩短至15秒,性能提升显著。
2703 13
SpringBoot整合XXL-JOB【05】-  任务分片
|
10月前
|
Java 调度 数据库
搭建XXL-JOB
XXL-JOB由调度中心和执行器两部分组成。调度中心负责任务调度与管理,支持动态配置、监控告警;执行器部署在微服务中,接收调度请求并执行任务。通过SpringBoot集成xxl-job-core,配置注册地址后可自动注册到调度中心,实现分布式任务调度。
|
12月前
|
Java 测试技术 数据库连接
【SpringBoot(四)】还不懂文件上传?JUnit使用?本文带你了解SpringBoot的文件上传、异常处理、组件注入等知识!并且带你领悟JUnit单元测试的使用!
Spring专栏第四章,本文带你上手 SpringBoot 的文件上传、异常处理、组件注入等功能 并且为你演示Junit5的基础上手体验
1252 3
|
人工智能 自然语言处理 Java
Spring AI,搭建个人AI助手
本期主要是实操性内容,聊聊AI大模型,并使用Spring AI搭建属于自己的AI助手、知识库。本期所需的演示源码笔者托管在Gitee上(https://gitee.com/catoncloud/spring-ai-demo),读者朋友可自行查阅。
4954 44
Spring AI,搭建个人AI助手
|
人工智能 分布式计算 Java
说说XXLJob分片任务实现原理?
说说XXLJob分片任务实现原理?
2868 0
说说XXLJob分片任务实现原理?
|
算法 调度 微服务
【黑马头条】day20—xxl-job(三)
【黑马头条】day20—xxl-job(三)
867 0
|
Arthas Kubernetes 调度
一次线上Xxl-Job定时任务调度失败的排查与解决
在XXL-JOB系统中,每分钟执行一次的任务出现调度异常:首次调度成功,第二次调度失败,但第三次调度显示的是第二次任务的执行时间。问题源于调度中心发送的心跳请求未得到执行器响应,导致连接被Envoy代理关闭。Envoy在等待心跳响应时,后续调度请求被阻塞,直至连接因超时关闭。调度中心收到503响应后也会关闭连接,从而影响第三次调度。解决方案是更新执行器以处理心跳请求或配置Istio Sidecar以绕过Envoy代理特定流量。
4873 1
|
消息中间件 监控 Linux
Centos7安装RabbitMQ
Centos7安装RabbitMQ

热门文章

最新文章