分布式任务调度框架XXL-JOB入门教程

本文涉及的产品
日志服务 SLS,月写入数据量 50GB 1个月
简介: XXL-JOB是一个分布式任务调度平台,其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线,开箱即用。

1、概述

XXL-JOB是一个分布式任务调度平台,其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线,开箱即用。

官方文档:https://www.xuxueli.com/xxl-job/#%E4%BA%8C%E3%80%81%E5%BF%AB%E9%80%9F%E5%85%A5%E9%97%A8

  • 官网地址:https://www.xuxueli.com/xxl-job/
  • GitHub地址:https://github.com/xuxueli/xxl-job/
  • xxl-job的设计思想

    • 将调度行为抽象形成“调度中心”公共平台,而平台自身并不承担业务逻辑,“调度中心”负责发起调度请求。
    • 将任务抽象成分散的JobHandler,交由“执行器”统一管理
    • “执行器”负责接收调度请求并执行对应的JobHandler中业务逻辑。
    • 因此,“调度”和“任务”两部分可以相互解耦,提高系统整体稳定性和扩展性

    img

  • 架构图(图片来源是xxl-job官网)

    • 调度中心

      • 负责管理调度的信息,按照调度的配置来发出调度请求
      • 支持可视化、简单的动态管理调度信息,包括新建、删除、更新等,这些操作都会实时生效,同时也支持监控调度结果以及执行日志。
    • 执行器

      • 负责接收请求并且执行任务的逻辑。任务模块专注于任务的执行操作等等,使得开发和维护更加的简单与高效
  • XXL-Job具有哪些特性

    • 调度中心HA(中心式):调度采用了中心式进行设计,“调度中心”支持集群部署,可保证调度中心HA
    • 执行器HA(分布式):任务分布式的执行,任务执行器支持集群部署,可保证任务执行HA
    • 触发策略:有Cron触发、固定间隔触发、固定延时触发、API事件触发、人工触发、父子任务触发
    • 路由策略:执行器在集群部署的时候提供了丰富的路由策略,如:第一个、最后一个、轮询、随机、一致性HASH、最不经常使用LFU、最久未使用LRU、故障转移等等
    • 故障转移:如果执行器集群的一台机器发生故障,会自动切换到一台正常的执行器发送任务调度
    • Rolling实时日志的监控:支持rolling方式查看输入的完整执行日志
    • 脚本任务:支持GLUE模式开发和运行脚本任务,包括Shell、python、node.js、php等等类型脚本

2、搭建调度中心

1、下载源码

下载源码导入idea,源码地址:https://gitee.com/xuxueli0323/xxl-job.git

  • doc:xxl-job的文档资料,包括了数据库的脚本(后面要用到)
  • xxl-job-core:公共jar包依赖
  • xxl-job-admin:调度中心,项目源码,是Springboot项目,可以直接启动
  • xxl-job-executor-samples:执行器,是Sample实例项目,里面的Springboot工程可以直接启动,也可以在该项目的基础上进行开发,也可以将现有的项目改造成为执行器项目

2、数据库

数据库文件在源码doc/db目录下

  • xxl_job的数据库里有如下几个表

    • xxl_job_group:执行器信息表,用于维护任务执行器的信息
    • xxl_job_info:调度扩展信息表,主要是用于保存xxl-job的调度任务的扩展信息,比如说像任务分组、任务名、机器的地址等等
    • xxl_job_lock:任务调度锁表
    • xxl_job_log:日志表,主要是用在保存xxl-job任务调度历史信息,像调度结果、执行结果、调度入参等等
    • xxl_job_log_report:日志报表,会存储xxl-job任务调度的日志报表,会在调度中心里的报表功能里使用到
    • xxl_job_logglue:任务的GLUE日志,用于保存GLUE日志的更新历史变化,支持GLUE版本的回溯功能
    • xxl_job_registry:执行器的注册表,用在维护在线的执行器与调度中心的地址信息
    • xxl_job_user:系统的用户表

3、调度中心配置:

调度中心配置文件地址:

/xxl-job/xxl-job-admin/src/main/resources/application.properties
  • 配置数据库连接
## xxl-job, datasource
spring.datasource.url=jdbc:mysql://127.0.0.1:3306/xxl_job?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&serverTimezone=Asia/Shanghai
spring.datasource.username=root
spring.datasource.password=xdclass.net
spring.datasource.driver-class-name=com.mysql.cj.jdbc.Driver
  • 配置 xxl.job.accessToken(后续要配置客户端接入配置token)
xxl.job.accessToken=javaxiaobear.cn

4、启动项目

调度中心访问地址:http: //localhost:8080/xxl-job-admin (该地址执行器将会使用到,作为回调地址)

默认登录账号 “admin/123456”, 登录后运行界面如下图所示。

image-20230630144053744

5、UI界面介绍

1、运行报表:以图形化来展示了整体的任务执行情况

  • 任务数量:能够看到调度中心运行的任务数量
  • 调度次数:调度中心所触发的调度次数
  • 执行器数量:在整个调度中心中,在线的执行器数量有多少

2、任务管理(配置执行任务)

image-20230630144304198

  • 示例执行器:所用到的执行器
  • 任务描述:概述该任务是做什么的
  • 路由策略:

    • 第一个:选择第一个机器
    • 最后一个:选择最后一个机器
    • 轮询:依次选择执行
    • 随机:随机选择在线的机器
    • 一致性HASH:每个任务按照Hash算法固定选择某一台机器,并且所有的任务均匀散列在不同的机器上
    • 最不经常使用:使用频率最低的机器优先被使用
    • 最近最久未使用:最久未使用的机器优先被选举
    • 故障转移:按照顺序依次进行心跳检测,第一个心跳检测成功的机器选定为目标的执行器并且会发起任务调度
    • 忙碌转移:按照顺序来依次进行空闲检测,第一个空闲检测成功的机器会被选定为目标群机器,并且会发起任务调度
    • 分片广播:广播触发对于集群中的所有机器执行任务,同时会系统会自动传递分片的参数
  • Cron:执行规则
  • 调度过期策略:调度中心错过调度时间的补偿处理策略,包括:忽略、立即补偿触发一次等
  • JobHandler:定义执行器的名字
  • 阻塞处理策略:

    • 单机串行:新的调度任务在进入到执行器之后,该调度任务进入FIFO队列,并以串行的方式去进行
    • 丢弃后续调度:新的调度任务在进入到执行器之后,如果存在相同的且正在运行的调度任务,本次的调度任务请求就会被丢弃掉,并且标记为失败
    • 覆盖之前的调度:新的调度任务在进入到执行器之后,如果存在相同的且正在运行的调度任务,就会终止掉当前正在运行的调度任务,并且清空队列,运行新的调度任务。
  • 子任务ID:输入子任务的任务id,可填写多个
  • 任务超时时间:添加任务超时的时候,单位s,设置时间大于0的时候就会生效
  • 失败重试次数:设置失败重试的次数,设置时间大于0的时候就会生效
  • 负责人:填写该任务调度的负责人
  • 报警邮件:出现报警,则发送邮件

3、调度日志

  • 这里是查看调度的日志,根据日志来查看任务具体的执行情况是怎样的

4、执行器管理

  • 这里是配置执行器,等待执行器启动的时候都会被调度中心监听加入到地址列表

5、用户管理

  • 可以对用户的一些操作

3、整合xxl_job

1、项目搭建

1、引入xxl_job依赖

<!-- http://repo1.maven.org/maven2/com/xuxueli/xxl-job-core/ -->
<dependency>
    <groupId>com.xuxueli</groupId>
    <artifactId>xxl-job-core</artifactId>
    <version>2.3.1</version>
</dependency>

2、配置yaml

xxl:
  job:
    admin:
      addresses: http://127.0.0.1:8080/xxl-job-admin
      # 执行器的名字
    executor:
      appname: javaxiaobear-xxl-job-test
    accessToken: default_token
server:
  port: 8081

3、编写配置类

@Configuration
@Slf4j
public class XxlJobConfig {

    @Value("${xxl.job.admin.addresses}")
    private String adminAddresses;

    @Value("${xxl.job.executor.appname}")
    private String appName;

    @Value("${xxl.job.accessToken}")
    private String accessToken;


    //旧版的有bug
    //@Bean(initMethod = "start", destroyMethod = "destroy")
    @Bean
    public XxlJobSpringExecutor xxlJobExecutor() {
        log.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
        xxlJobSpringExecutor.setAppname(appName);
//        xxlJobSpringExecutor.setIp(ip);
//        xxlJobSpringExecutor.setPort(port);
        xxlJobSpringExecutor.setAccessToken(accessToken);
//        xxlJobSpringExecutor.setLogPath(logPath);
//        xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);

        return xxlJobSpringExecutor;
    }
}

2、第一个XXL-Job分布式调度任务

1、界面新增一个任务

image-20230630152704891

2、新增一个执行器

这里是因为在配置文件里面用的是自定义的执行器,所以我们需要新增,当然,你可以用默认的执行器

image-20230630154909704

3、代码配置handler

@Slf4j
@Component
public class MyXxlJobHandler {

    @XxlJob("myXxlJobHandler")
    public ReturnT<String> execute(String param){
        log.info("小熊学Java 任务方法触发成功");
        return ReturnT.SUCCESS;
    }
}

4、重新启动项目

image-20230630155107647

image-20230630155113771

4、路由策略

  • 第一个:选择第一个机器
  • 最后一个:选择最后一个机器
  • 轮询:依次选择执行,流量均摊(推荐)
  • 随机:随机选择在线的机器
  • 一致性HASH:每个任务按照Hash算法固定选择某一台机器,并且所有的任务均匀散列在不同的机器上
  • 最不经常使用:使用频率最低的机器优先被使用
  • 最近最久未使用:最久未使用的机器优先被选举
  • 故障转移:按照顺序依次进行心跳检测,第一个心跳检测成功的机器选定为目标的执行器并且会发起任务调度
  • 忙碌转移:按照顺序来依次进行空闲检测,第一个空闲检测成功的机器会被选定为目标群机器,并且会发起任务调度
  • 分片广播:广播触发对于集群中的所有机器执行任务,同时会系统会自动传递分片的参数

5、分片广播

1、场景

需求

  • 有一个任务需要处理100W条数据,每条数据的业务逻辑处理要0.1s
  • 对于普通任务来说,只有一个线程来处理 可能需要10万秒才能处理完,业务则严重受影响
  • 案例:双十一大促,给1000万用户发营销短信

什么是分片任务

  • 执行器集群部署,如果任务的路由策略选择【分片广播】,一次任务调度将会【广播触发】对应集群中所有执行器执行一次任务,同时系统自动传递分片参数,执行器可根据分片参数开发分片任务
  • 需要处理的海量数据,以执行器为划分,每个执行器分配一定的任务数,并行执行
  • XXL-Job支持动态扩容执行器集群,从而动态增加分片数量,到达更快处理任务
  • 分片的值是调度中心分配的
// 当前分片数,从0开始,即执行器的序号
int shardIndex = XxlJobHelper.getShardIndex();
//总分片数,执行器集群总机器数量
int shardTotal = XxlJobHelper.getShardTotal();

image-20230703140222054

解决思路

  1. 分片广播
  2. 也可以启动多个job,使用同个jobHandler,通过命令行参数控制
  • 如果将100W数据均匀分给集群里的10台机器同时处理,
  • 每台机器耗时,1万秒即可,耗时会大大缩短,也能充分利用集群资源
  • 在xxl-job里,可以配置执行器集群有10个机器,那么分片总数是10,分片序号0~9 分别对应那10台机器。
  • 分片方式

    • id % 分片总数 余数是0 的,在第1个执行器上执行
    • id % 分片总数 余数是1 的,在第2个执行器上执行
    • id % 分片总数 余数是2 的,在第3个执行器上执行
    • ...
    • id % 分片总数 余数是9 的,在第10个执行器上执行

2、代码编写

/**
     * 100个用户,分片处理
     */
    @XxlJob("myShardingJobHandler")
    public void shardingJobHandler(){
        // 当前分片数,从0开始,即执行器的序号
        int shardIndex = XxlJobHelper.getShardIndex();
        //总分片数,执行器集群总机器数量
        int shardTotal = XxlJobHelper.getShardTotal();

        XxlJobHelper.log("分片参数:当前分片序号 = {}, 总分片数 = {}", shardIndex, shardTotal);


        List<Integer> allUserIds = getAllUserIds();
        allUserIds.forEach(obj -> {
            if (obj % shardTotal == shardIndex) {
                log.info("第 {} 片, 命中分片开始处理用户id={}",shardIndex,obj);
            }
        });
    }

    private List<Integer> getAllUserIds() {
        List<Integer> ids = new ArrayList<>();
        for (int i = 0; i < 100; i++) {
            ids.add(i);
        }
        return ids;
    }

3、测试

  1. 新建一个分片任务

image-20230703141009035

  1. 新建一个服务器实例,直接copy一份

    image-20230703141109854

  2. 启动系统,点击运行一次

    image-20230703141221278

6、阻塞策略

  • 单机串行:新的调度任务在进入到执行器之后,该调度任务进入FIFO队列,并以串行的方式去进行
  • 丢弃后续调度:新的调度任务在进入到执行器之后,如果存在相同的且正在运行的调度任务,本次的调度任务请求就会被丢弃掉,并且标记为失败
  • 覆盖之前的调度:新的调度任务在进入到执行器之后,如果存在相同的且正在运行的调度任务,就会终止掉当前正在运行的调度任务,并且清空队列,运行新的调度任务。

好了,本文就到这里了!如果觉得内容不错的话,希望大家可以帮忙点赞转发一波,这是对我最大的鼓励,感谢🙏🏻

资料获取👇 最后面就是领取暗号,公众号回复即可!

相关实践学习
日志服务之使用Nginx模式采集日志
本文介绍如何通过日志服务控制台创建Nginx模式的Logtail配置快速采集Nginx日志并进行多维度分析。
相关文章
|
22天前
|
Java 数据库
在Java中使用Seata框架实现分布式事务的详细步骤
通过以上步骤,利用 Seata 框架可以实现较为简单的分布式事务处理。在实际应用中,还需要根据具体业务需求进行更详细的配置和处理。同时,要注意处理各种异常情况,以确保分布式事务的正确执行。
|
22天前
|
消息中间件 Java Kafka
在Java中实现分布式事务的常用框架和方法
总之,选择合适的分布式事务框架和方法需要综合考虑业务需求、性能、复杂度等因素。不同的框架和方法都有其特点和适用场景,需要根据具体情况进行评估和选择。同时,随着技术的不断发展,分布式事务的解决方案也在不断更新和完善,以更好地满足业务的需求。你还可以进一步深入研究和了解这些框架和方法,以便在实际应用中更好地实现分布式事务管理。
|
21天前
|
存储 Java 关系型数据库
在Spring Boot中整合Seata框架实现分布式事务
可以在 Spring Boot 中成功整合 Seata 框架,实现分布式事务的管理和处理。在实际应用中,还需要根据具体的业务需求和技术架构进行进一步的优化和调整。同时,要注意处理各种可能出现的问题,以保障分布式事务的顺利执行。
34 6
|
21天前
|
数据库
如何在Seata框架中配置分布式事务的隔离级别?
总的来说,配置分布式事务的隔离级别是实现分布式事务管理的重要环节之一,需要认真对待和仔细调整,以满足业务的需求和性能要求。你还可以进一步深入研究和实践 Seata 框架的配置和使用,以更好地应对各种分布式事务场景的挑战。
27 6
|
19天前
|
消息中间件 运维 数据库
Seata框架和其他分布式事务框架有什么区别
Seata框架和其他分布式事务框架有什么区别
22 1
|
1月前
|
机器学习/深度学习 自然语言处理 并行计算
DeepSpeed分布式训练框架深度学习指南
【11月更文挑战第6天】随着深度学习模型规模的日益增大,训练这些模型所需的计算资源和时间成本也随之增加。传统的单机训练方式已难以应对大规模模型的训练需求。
114 3
|
1月前
|
机器学习/深度学习 并行计算 Java
谈谈分布式训练框架DeepSpeed与Megatron
【11月更文挑战第3天】随着深度学习技术的不断发展,大规模模型的训练需求日益增长。为了应对这种需求,分布式训练框架应运而生,其中DeepSpeed和Megatron是两个备受瞩目的框架。本文将深入探讨这两个框架的背景、业务场景、优缺点、主要功能及底层实现逻辑,并提供一个基于Java语言的简单demo例子,帮助读者更好地理解这些技术。
74 2
|
2月前
|
NoSQL Java Redis
太惨痛: Redis 分布式锁 5个大坑,又大又深, 如何才能 避开 ?
Redis分布式锁在高并发场景下是重要的技术手段,但其实现过程中常遇到五大深坑:**原子性问题**、**连接耗尽问题**、**锁过期问题**、**锁失效问题**以及**锁分段问题**。这些问题不仅影响系统的稳定性和性能,还可能导致数据不一致。尼恩在实际项目中总结了这些坑,并提供了详细的解决方案,包括使用Lua脚本保证原子性、设置合理的锁过期时间和使用看门狗机制、以及通过锁分段提升性能。这些经验和技巧对面试和实际开发都有很大帮助,值得深入学习和实践。
太惨痛: Redis 分布式锁 5个大坑,又大又深, 如何才能 避开 ?
|
14天前
|
存储 NoSQL Java
使用lock4j-redis-template-spring-boot-starter实现redis分布式锁
通过使用 `lock4j-redis-template-spring-boot-starter`,我们可以轻松实现 Redis 分布式锁,从而解决分布式系统中多个实例并发访问共享资源的问题。合理配置和使用分布式锁,可以有效提高系统的稳定性和数据的一致性。希望本文对你在实际项目中使用 Redis 分布式锁有所帮助。
40 5
|
17天前
|
NoSQL Java 数据处理
基于Redis海量数据场景分布式ID架构实践
【11月更文挑战第30天】在现代分布式系统中,生成全局唯一的ID是一个常见且重要的需求。在微服务架构中,各个服务可能需要生成唯一标识符,如用户ID、订单ID等。传统的自增ID已经无法满足在集群环境下保持唯一性的要求,而分布式ID解决方案能够确保即使在多个实例间也能生成全局唯一的标识符。本文将深入探讨如何利用Redis实现分布式ID生成,并通过Java语言展示多个示例,同时分析每个实践方案的优缺点。
34 8