从0到1，百亿级任务调度平台的架构与实现
尼恩：百亿级海量任务调度平台起源
本文目录
- 尼恩：百亿级海量任务调度平台起源
- 百亿级海量任务调度平台
- 百亿级海量任务调度平台业务诉求：
- 海量百亿级任务调度平台主要特点和功能：
- 海量亿级任务调度平台解决方案：
- 海量百亿级任务调度平台的功能性诉求
- 海量百亿级任务调度平台的非功能性诉求
- 百亿级任务量和百万TPM 并发的架构方案
- 海量百亿级任务调度平台的实操
- XXL-JOB 入门学习
- 单体任务调度框架
- 分布式任务调度框架
- 方式一：手工部署XXL-Job 的5大步骤
-步骤一：部署XXL-Job
-安装要点：初始化“调度数据库”
-步骤二：访问管理界面
-步骤三：创建任务
-步骤四：监控任务执行情况
-步骤五：扩展其他功能
- 方式二：通过编译源码部署 XXL-Job 的2大步骤
- 步骤1：解压的编译源码
- 步骤2：配置部署“调度中心”
- 方式三：通过Docker 镜像方式搭建调度中心
- 开发、配置、部署XXl-JOB“执行器项目”
- 步骤2：maven依赖
- 步骤3：执行器配置
- 步骤4：编写任务执行逻辑
- 步骤5：部署执行器项目：
- 步骤6：执行器集群（可选）：
- 开发第一个任务“Hello World”
- XXL-Job 几种不同的任务模式
- “GLUE模式(Java)” 运行模式的开发第一个调度任务
- 步骤一：新建任务：
- 步骤二：“GLUE模式(Java)” 任务开发：
- 步骤三：触发执行：
- 步骤四：查看日志：
- 大规模任务的分片调度
- XXL-JOB的配置属性说明
- XXL-JOB实现原理
- 1. 调度中心（JobAdmin）：
- 2. 执行器（JobExecutor）：
- 工作流程：
- XXL-JOB 系统总体架构
- JobAdmin服务器启动流程
- JobAdmin服务器初始化总体流程
- 初始化JobTriggerPool触发器线程池
- 初始化registryOrRemoveThreadPool 线程池，剔除超时主机，更新主机列表
- 执行失败重试和告警（发邮件）
-告警(这里可向spring注入JobAlarm)
- 初始化callbackThreadPool回调线程池，更改丢失主机的任务状态为失败
- 启动统计日志线程
- 启动调度器
-1 添加任务到时间轮
-2 时间轮线程执行任务
-3 时间轮算法
- 执行器Executor 启动流程
- 1 遍历XxlJob注解查找方法
- 2 启动客户端执行器
-2.1 初始化日志文件
-2.2 封装调度中心请求路径，用于访问调度中心
-2.3 清除过期日志
-2.4 回调调度中心，汇报执行状态
-2.5 执行内嵌服务
- XXL-JOB的服务注册
- 客户端
- 服务端
- XXL-JOB的 执行任务
- 服务端下发任务
-XxlJobTrigger#trigger
-路由策略
- 客户端处理任务
-接下来两个工作
- 客户端回调服务端
- 调度中心处理回调
- 执行子任务
- XXL-JOB的mysql分布式锁
- 百亿级海量任务调度平台的架构与实现
- 说在最后：有问题找老架构取经
百亿级海量任务调度平台
百亿级海量任务调度平台业务诉求：
超时未支付订单关单 每隔 2h 更新好友排行榜 3.22 日 17 点《xx》剧上线等。
海量百亿级任务调度平台主要特点和功能：
分布式架构： 任务调度平台采用分布式架构，可以横向扩展以应对海量任务的调度需求。通过将任务分散到多个节点上执行，提高了系统的吞吐量和稳定性。 高可靠性： 平台具有高可靠性，能够保证任务的准确执行。采用主备节点、任务重试机制、任务监控和告警等方式确保任务不会丢失和错过。 低延迟： 任务调度平台能够保证低延迟的任务执行。通过优化任务调度算法、减少任务执行的等待时间、提高任务执行的并发度等方式，降低了任务执行的延迟。 统一管理： 提供统一的任务管理界面和接口，方便业务方统一管理和监控各类任务。通过可视化界面展示任务执行情况、任务调度状态等信息，提供给用户更直观的操作体验。 灵活扩展： 平台具有良好的扩展性，可以根据业务需求灵活定制任务调度策略和调度算法。支持定制化的任务调度器、触发器、任务处理器等组件，满足不同业务场景的需求。 多样化任务类型： 平台支持多样化的任务类型，包括定时任务、周期任务、延时任务、异步任务等。能够满足不同业务场景下的任务调度需求。
海量亿级任务调度平台解决方案：
统一调度平台： 搭建统一的任务调度平台，为各业务提供统一的任务调度服务，避免各业务重复开发调度功能，提高开发效率。 高可靠性设计： 在平台设计中考虑高可靠性，采用集群部署、主备机制、任务重试、监控告警等方式确保任务执行的稳定性和可靠性。 低延迟优化： 优化任务调度算法和执行流程，减少任务调度的等待时间和执行时间，提高任务执行的效率和响应速度。 解耦业务和调度： 对业务和调度进行解耦，采用消息队列等方式将任务调度数据存储与业务数据存储分离，降低了业务和调度的耦合性，减少了大 key、慢 SQL 的风险。 技术选型： 综合考虑开源解决方案如XXL-Job、Elastic-Job、Quartz等，并结合业务需求和现有技术栈选择合适的调度平台，以满足海量任务调度的需求。
海量百亿级任务调度平台的功能性诉求
海量百亿级任务调度平台的非功能性诉求
平台化：支持多业务接入、百亿级任务注册 易用性：自助化接入、运维，使用成本远低自建 高可靠：全年 3 个 9 可用性、p99(时延)<1s 高性能：支持 100w+TPM 的任务触发
注册\触发可用性>99.95% 任务触达率>99.99% p99(触达延时)<1s
百亿级任务量和百万 TPM 并发的架构方案
MTTR 是指 Mean Time to Repair，即平均修复时间。 在系统运维和故障处理领域，MTTR 是一个重要的性能指标，用于衡量系统故障发生后，从故障发生到恢复正常运行所花费的平均时间。 MTTR 的计算方式为： 𝑀𝑇𝑇𝑅=总的修复时间/发生故障次数 MTTR 的值越小越好，因为它反映了系统修复故障的效率和速度。 较低的 MTTR 意味着系统能够更快地从故障中恢复，减少了系统的停机时间，提高了系统的可用性和稳定性。 在实际运维工作中，降低 MTTR 可以通过以下方式实现：
自动化故障检测和修复： 引入自动化工具和脚本，实现对常见故障的自动检测和修复，减少人工干预的时间和错误。 监控和告警系统： 配置有效的监控系统，及时发现系统异常和故障，通过告警通知运维人员进行快速响应和处理。 故障排查和诊断： 提前准备好故障排查的工具和流程，快速定位故障原因，并采取有效的措施进行修复。
海量百亿级任务调度平台的实操
XXL-JOB 入门学习 XXL-JOB 架构和源码学习 XXL-JOB-Massive 海量任务调度平台的架构和实现
XXL-JOB 入门学习
单体任务调度框架
怎么做到对任务的控制（如何避免任务重复执行，单体项目集群部署，会重复执行）。 如果某台机器宕机了，会不会存在任务丢失。 如果要增加服务实例，怎么做到弹性扩容。 如何做到对任务调度的执行情况统一监测。 如何做到高可用 如何做到动态配置 如何实现任务分片执行
分布式任务调度框架
分布式定时任务调度的框架可以选择的有：quartz、elastic-job、xxl-job
| 功能 | quartz | elastic-job | xxl-job |
quartz采用api的方式调用任务，quartz不方便，但是xxl-job使用的是管理界面。 quartz比xxl-job代码侵入更强 quartz调度逻辑和QuartzJobBean耦合在一个项目中，当任务增多，逻辑复杂的时候，性能会受到影响 quartz底层以抢占式获取db锁并且由抢占成功的节点运行，导致节点负载悬殊非常大；xxl-job通过执行器实现协同分配式运行任务，各个节点比较均衡。
方式一：手工部署XXL-Job 的5大步骤
步骤一：部署 XXL-Job
下载 XXL-Job： 访问 XXL-Job 的官方网站（https://github.com/xuxueli/xxl-job/）下载最新版本的XXL-Job。 解压文件： 将下载的压缩包解压到你选择的目录中。 配置数据库： 进入解压后的目录，找到 conf/application.properties文件，配置数据库连接信息，包括数据库地址、用户名、密码等。创建数据库表： 运行 docs/sql/tables_xxl_job.sql文件中的 SQL 脚本，在数据库中创建 XXL-Job 所需的表结构。启动服务： 运行 bin/startup.sh（Linux/Mac）或bin/startup.cmd（Windows）启动 XXL-Job 服务。
安装要点：初始化“调度数据库”
/xxl-job/doc/db/tables_xxl_job.sql
步骤二：访问管理界面
打开浏览器，输入 XXL-Job 的访问地址，默认为 http://localhost:8080/xxl-job-admin。使用默认的管理员账号和密码登录，默认账号密码均为 admin。登录成功后，可以在管理界面中进行任务的创建、管理、监控等操作。
步骤三：创建任务
在管理界面中，点击左侧菜单栏的 "任务管理"，然后点击 "新增" 按钮。 填写任务信息，包括任务描述、任务执行器、任务参数等。 点击 "保存" 按钮完成任务的创建。
步骤四：监控任务执行情况
在管理界面中，点击左侧菜单栏的 "任务监控"，可以查看任务的执行情况、调度情况、日志等信息。 可以手动触发任务执行，也可以查看任务的执行历史记录。
步骤五：扩展其他功能
方式二：通过 编译源码部署 XXL-Job 的2大步骤
步骤1：解压的编译源码
xxl-job-admin：调度中心
xxl-job-core：公共依赖
xxl-job-executor-samples：执行器Sample示例（选择合适的版本执行器，可直接使用，也可以参考其并将现有项目改造成执行器）
xxl-job-executor-sample-springboot：Springboot版本，通过Springboot管理执行器，推荐这种方式；
xxl-job-executor-sample-frameless：无框架版本；
步骤2：配置部署“调度中心”
/xxl-job/xxl-job-admin/src/main/resources/application.properties
### 调度中心JDBC链接：链接地址请保持和 2.1章节 所创建的调度数据库的地址一致
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=root_pwd
spring.datasource.driver-class-name=com.mysql.jdbc.Driver
### 报警邮箱
spring.mail.host=smtp.qq.com
spring.mail.port=25
spring.mail.username=xxx@qq.com
spring.mail.password=xxx
spring.mail.properties.mail.smtp.auth=true
spring.mail.properties.mail.smtp.starttls.enable=true
spring.mail.properties.mail.smtp.starttls.required=true
spring.mail.properties.mail.smtp.socketFactory.class=javax.net.ssl.SSLSocketFactory
### 调度中心通讯TOKEN [选填]：非空时启用；
xxl.job.accessToken=
### 调度中心国际化配置 [必填]： 默认为 "zh_CN"/中文简体, 可选范围为 "zh_CN"/中文简体, "zh_TC"/中文繁体 and "en"/英文；
xxl.job.i18n=zh_CN
## 调度线程池最大线程配置【必填】
xxl.job.triggerpool.fast.max=200
xxl.job.triggerpool.slow.max=100
### 调度中心日志表数据保存天数 [必填]：过期日志自动清理；限制大于等于7时生效，否则, 如-1，关闭自动清理功能；
xxl.job.logretentiondays=30
DB配置保持一致； 集群机器时钟保持一致（单机集群忽视）；
推荐通过nginx为调度中心集群做负载均衡。 通过nginx代理的地址，去进行 调度中心访问、执行器回调配置、调用API服务等操作。
方式三：通过Docker 镜像方式搭建调度中心
下载镜像
// Docker地址：https://hub.docker.com/r/xuxueli/xxl-job-admin/ (建议指定版本号)
docker pull xuxueli/xxl-job-admin
创建容器并运行
/**
* 如需自定义 mysql 等配置，可通过 "-e PARAMS" 指定，参数格式 PARAMS="--key=value --key2=value2" ；
* 配置项参考文件：/xxl-job/xxl-job-admin/src/main/resources/application.properties
* 如需自定义 JVM内存参数 等配置，可通过 "-e JAVA_OPTS" 指定，参数格式 JAVA_OPTS="-Xmx512m" ；
*/
docker run -e PARAMS="--spring.datasource.url=jdbc:mysql://127.0.0.1:3306/xxl_job?useUnicode=true&characterEncoding=UTF-8&autoReconnect=true&serverTimezone=Asia/Shanghai" -p 8080:8080 -v /tmp:/data/applogs --name xxl-job-admin -d xuxueli/xxl-job-admin:{指定版本}
开发、配置、部署 XXl-JOB“执行器项目”
任务执行： 接收调度中心下发的任务执行请求，并执行相应的任务逻辑，例如调用接口、执行业务逻辑、发送消息等。 任务注册： 将执行器项目注册到 XXL-JOB 的调度中心，以便调度中心能够向执行器发送任务执行请求。 执行器管理： 提供执行器项目的管理功能，包括启动、停止、监控等。
创建执行器项目： 创建一个新的 Java 项目，作为执行器项目。 引入 XXL-JOB 客户端库： 在项目中引入 XXL-JOB 客户端库，可以通过 Maven 或 Gradle 等依赖管理工具进行引入。 配置执行器参数： 在项目中配置执行器的参数，包括执行器名称、注册地址、执行器端口等。 编写任务执行逻辑： 编写任务执行器的逻辑代码，接收任务执行请求并执行相应的任务逻辑。你可以根据任务类型和具体业务需求来编写不同的任务执行逻辑。 注册执行器： 在执行器项目中注册执行器到 XXL-JOB 的调度中心，以便调度中心能够向执行器发送任务执行请求。 启动执行器： 启动执行器项目，让执行器能够正常接收和执行任务。
步骤2：maven依赖
步骤3：执行器配置
/xxl-job/xxl-job-executor-samples/xxl-job-executor-sample-springboot/src/main/resources/application.properties
### 调度中心部署根地址 [选填]：如调度中心集群部署存在多个地址则用逗号分隔。执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调"；为空则关闭自动注册；
xxl.job.admin.addresses=http://127.0.0.1:8080/xxl-job-admin
### 执行器通讯TOKEN [选填]：非空时启用；
xxl.job.accessToken=
### 执行器AppName [选填]：执行器心跳注册分组依据；为空则关闭自动注册
xxl.job.executor.appname=xxl-job-executor-sample
### 执行器注册 [选填]：优先使用该配置作为注册地址，为空时使用内嵌服务 ”IP:PORT“ 作为注册地址。从而更灵活的支持容器类型执行器动态IP和动态映射端口问题。
xxl.job.executor.address=
### 执行器IP [选填]：默认为空表示自动获取IP，多网卡时可手动设置指定IP，该IP不会绑定Host仅作为通讯实用；地址信息用于 "执行器注册" 和 "调度中心请求并触发任务"；
xxl.job.executor.ip=
### 执行器端口号 [选填]：小于等于0则自动获取；默认端口为9999，单机部署多个执行器时，注意要配置不同执行器端口；
xxl.job.executor.port=9999
### 执行器运行日志文件存储磁盘路径 [选填] ：需要对该路径拥有读写权限；为空则使用默认路径；
xxl.job.executor.logpath=/data/applogs/xxl-job/jobhandler
### 执行器日志文件保存天数 [选填] ： 过期日志自动清理, 限制值大于等于3时生效; 否则, 如-1, 关闭自动清理功能；
xxl.job.executor.logretentiondays=30
/xxl-job/xxl-job-executor-samples/xxl-job-executor-sample-springboot/src/main/java/com/xxl/job/executor/core/config/XxlJobConfig.java
@Bean
public XxlJobSpringExecutor xxlJobExecutor() {
logger.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;
}
步骤4：编写任务执行逻辑
步骤5：部署执行器项目：
xxl-job-executor-sample-springboot：项目编译打包成springboot类型的可执行JAR包，命令启动即可；
xxl-job-executor-sample-frameless：项目编译打包成JAR包，命令启动即可；
步骤6：执行器集群（可选）：
执行器回调地址（xxl.job.admin.addresses）需要保持一致；执行器根据该配置进行执行器自动注册等操作。 同一个执行器集群内AppName（xxl.job.executor.appname）需要保持一致；调度中心根据该配置动态发现不同集群的在线执行器列表。
开发第一个任务“Hello World”
XXL-Job 几种不同的任务模式
Bean模式（Spring Bean模式）： Bean 模式是 XXL-Job 最简单的任务模式之一，适用于基于 Spring 框架开发的应用。在 Bean 模式下，任务执行逻辑由一个 Spring Bean 实现，并通过 Spring 的依赖注入机制来管理任务实例。在 XXL-Job 的任务配置中，需要指定任务执行的 Bean 名称和方法名称。 Script模式（GLUE脚本模式）： Script 模式是 XXL-Job 提供的另一种灵活的任务模式，允许用户通过脚本来定义任务执行逻辑。在 Script 模式下，任务执行逻辑可以使用多种脚本语言编写，如 Java、Python、Shell 等。用户可以在 XXL-Job 的任务配置中直接编写脚本代码，而无需提前编译和打包。 Remote模式（远程调用模式）： Remote 模式允许用户将任务执行逻辑部署在独立的远程服务上，并通过远程调用的方式来执行任务。在 Remote 模式下，XXL-Job 调度中心发送任务执行请求到远程服务，远程服务接收请求并执行任务逻辑，然后返回执行结果给调度中心。这种模式适用于任务逻辑较为复杂或需要与其他系统进行交互的情况。 Shell模式（Shell脚本模式）： Shell 模式是一种简单的任务执行模式，适用于需要执行 Shell 脚本的场景。在 Shell 模式下，XXL-Job 的任务配置中直接指定要执行的 Shell 命令或脚本文件路径，XXL-Job 调度中心将执行任务的请求发送给执行器，并在执行器上执行相应的 Shell 命令。
GLUE模式(Java)：任务以源码方式维护在调度中心；该模式的任务实际上是一段继承自IJobHandler的Java类代码并"groovy"源码方式维护，它在执行器项目中运行，可使用@Resource/@Autowire注入执行器里中的其他服务； GLUE模式(Shell)：任务以源码方式维护在调度中心；该模式的任务实际上是一段"shell"脚本； GLUE模式(Python)：任务以源码方式维护在调度中心；该模式的任务实际上是一段"python"脚本； GLUE模式(PHP)：任务以源码方式维护在调度中心；该模式的任务实际上是一段"php"脚本； GLUE模式(NodeJS)：任务以源码方式维护在调度中心；该模式的任务实际上是一段"nodejs"脚本； GLUE模式(PowerShell)：任务以源码方式维护在调度中心；该模式的任务实际上是一段"PowerShell"脚本；
“GLUE模式(Java)” 运行模式的开发第一个调度任务
步骤一：新建任务：
步骤二：“GLUE模式(Java)” 任务开发：
步骤三：触发执行：
步骤四：查看日志：
大规模任务的分片调度
static Map<Integer, String> singleMachineMultiTasks = new HashMap<>();
static {
singleMachineMultiTasks.put(1, "武汉");
singleMachineMultiTasks.put(2, "222");
singleMachineMultiTasks.put(3, "北京");
singleMachineMultiTasks.put(4, "444");
singleMachineMultiTasks.put(5, "上海");
singleMachineMultiTasks.put(6, "666");
}
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
XXL-JOB的配置属性说明
基础配置：
执行器：任务的绑定的执行器，任务触发调度时将会自动发现注册成功的执行器, 实现任务自动发现功能; 另一方面也可以方便的进行任务分组。每个任务必须绑定一个执行器, 可在 "执行器管理" 进行设置; 任务描述：任务的描述信息，便于任务管理； 负责人：任务的负责人； 报警邮件：任务调度失败时邮件通知的邮箱地址，支持配置多邮箱地址，配置多个邮箱地址时用逗号分隔；
触发配置：
调度类型： 无：该类型不会主动触发调度； CRON：该类型将会通过CRON，触发任务调度； 固定速度：该类型将会以固定速度，触发任务调度；按照固定的间隔时间，周期性触发； 固定延迟：该类型将会以固定延迟，触发任务调度；按照固定的延迟时间，从上次调度结束后开始计算延迟时间，到达延迟时间后触发下次调度； CRON：触发任务执行的Cron表达式； 固定速度：固定速度的时间间隔，单位为秒； 固定延迟：固定延迟的时间间隔，单位为秒；
任务配置：
运行模式： BEAN模式：以JobHandler方式维护在执行器端；需要结合 "JobHandler" 属性匹配执行器中任务； GLUE模式(Java)：任务以源码方式维护在调度中心；该模式的任务实际上是一段继承自IJobHandler的Java类代码并 "groovy" 源码方式维护，它在执行器项目中运行，可使用@Resource/@Autowire注入执行器里中的其他服务； GLUE模式(Shell)：任务以源码方式维护在调度中心；该模式的任务实际上是一段 "shell" 脚本； GLUE模式(Python)：任务以源码方式维护在调度中心；该模式的任务实际上是一段 "python" 脚本； GLUE模式(PHP)：任务以源码方式维护在调度中心；该模式的任务实际上是一段 "php" 脚本； GLUE模式(NodeJS)：任务以源码方式维护在调度中心；该模式的任务实际上是一段 "nodejs" 脚本； GLUE模式(PowerShell)：任务以源码方式维护在调度中心；该模式的任务实际上是一段 "PowerShell" 脚本； JobHandler：运行模式为 "BEAN模式" 时生效，对应执行器中新开发的JobHandler类“@JobHandler”注解自定义的value值； 执行参数：任务执行所需的参数；
高级配置：
路由策略：当执行器集群部署时，提供丰富的路由策略，包括； FIRST（第一个）：固定选择第一个机器； LAST（最后一个）：固定选择最后一个机器； ROUND（轮询）：； RANDOM（随机）：随机选择在线的机器； CONSISTENT_HASH（一致性HASH）：每个任务按照Hash算法固定选择某一台机器，且所有任务均匀散列在不同机器上。 LEAST_FREQUENTLY_USED（最不经常使用）：使用频率最低的机器优先被选举； LEAST_RECENTLY_USED（最近最久未使用）：最久未使用的机器优先被选举； FAILOVER（故障转移）：按照顺序依次进行心跳检测，第一个心跳检测成功的机器选定为目标执行器并发起调度； BUSYOVER（忙碌转移）：按照顺序依次进行空闲检测，第一个空闲检测成功的机器选定为目标执行器并发起调度； SHARDING_BROADCAST(分片广播)：广播触发对应集群中所有机器执行一次任务，同时系统自动传递分片参数；可根据分片参数开发分片任务； 子任务：每个任务都拥有一个唯一的任务ID(任务ID可以从任务列表获取)，当本任务执行结束并且执行成功时，将会触发子任务ID所对应的任务的一次主动调度。 调度过期策略： 忽略：调度过期后，忽略过期的任务，从当前时间开始重新计算下次触发时间； 立即执行一次：调度过期后，立即执行一次，并从当前时间开始重新计算下次触发时间； 阻塞处理策略：调度过于密集执行器来不及处理时的处理策略； 单机串行（默认）：调度请求进入单机执行器后，调度请求进入FIFO队列并以串行方式运行； 丢弃后续调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，本次请求将会被丢弃并标记为失败； 覆盖之前调度：调度请求进入单机执行器后，发现执行器存在运行的调度任务，将会终止运行中的调度任务并清空队列，然后运行本地调度任务； 任务超时时间：支持自定义任务超时时间，任务运行超时将会主动中断任务； 失败重试次数；支持自定义任务失败重试次数，当任务失败时将会按照预设的失败重试次数主动进行重试；
XXL-JOB实现原理
1. 调度中心（JobAdmin）：
任务管理： 调度中心负责管理所有注册的任务信息，包括任务的配置、调度策略、执行器信息等。 任务调度： 调度中心根据配置的调度策略，定时触发任务的执行请求，并将执行请求发送给相应的执行器。 任务监控： 调度中心实时监控任务的执行情况，包括任务的执行状态、执行日志、执行结果等，提供给用户进行查看和管理。
2. 执行器（JobExecutor）：
任务接收： 执行器接收调度中心下发的任务执行请求，并根据任务配置执行相应的任务逻辑。 任务执行： 执行器执行任务逻辑，包括调用业务接口、执行脚本、执行 Shell 命令等，根据任务执行结果将执行结果返回给调度中心。 任务线程池： 执行器维护一个任务线程池，用于执行任务逻辑。任务线程池的大小、任务执行策略等可以根据实际情况进行配置。
工作流程：
任务注册： 执行器在启动时向调度中心注册自己的信息，包括执行器名称、IP地址、端口号等。 任务调度： 调度中心根据任务的调度策略，定时触发任务的执行请求，并将执行请求发送给相应的执行器。 任务执行： 执行器接收到任务执行请求后，根据任务配置执行相应的任务逻辑，并将执行结果返回给调度中心。 任务监控： 调度中心实时监控任务的执行情况，记录任务的执行日志和执行结果，提供给用户进行查看和管理。
XXL-JOB 系统总体架构
JobAdmin服务器启动流程
加载配置文件： JobAdmin 服务器启动时，首先加载配置文件，包括数据库连接配置、任务调度配置等。 初始化数据库： JobAdmin 服务器会连接数据库，并初始化数据库表结构。如果数据库中已经存在 XXL-JOB 相关的表结构，则会检查表结构是否正确，如果表结构不正确，则会进行修复或升级。 启动任务调度： JobAdmin 服务器启动时，会启动任务调度器，定时触发任务的执行请求。任务调度器根据配置的调度策略，定时触发任务的执行请求，并将执行请求发送给相应的执行器。 注册执行器： JobAdmin 服务器会尝试连接注册在注册中心的执行器，并获取执行器的信息，包括执行器的名称、IP地址、端口号等。如果注册中心中不存在任何执行器，则 JobAdmin 服务器会等待执行器的注册。 启动Web服务： JobAdmin 服务器启动后，会提供 Web 服务，包括任务管理、任务监控等功能。用户可以通过浏览器访问 JobAdmin 的 Web 界面，进行任务的配置、管理和监控。 启动日志服务： JobAdmin 服务器启动后，会启动日志服务，负责记录任务的执行日志和执行结果。用户可以通过 Web 界面查看任务的执行日志和执行结果。 定时检查任务状态： JobAdmin 服务器会定时检查任务的状态，包括任务的执行状态、执行结果等。如果发现任务执行异常或超时，则会记录相应的日志，并发送告警通知给相关人员。 提供服务： JobAdmin 服务器启动完成后，开始提供服务，用户可以通过 Web 界面进行任务的配置、管理和监控。
JobAdmin服务器初始化总体流程
@Override
public void afterPropertiesSet() throws Exception {
//利用静态声明的只会加载一次的特性，初始化一个单例对象。
adminConfig = this;
//初始化xxjob调度器
xxlJobScheduler = new XxlJobScheduler();
xxlJobScheduler.init();
}
public void init() throws Exception {
//1. init i18n
initI18n();
//2. admin trigger pool start 初始化下发任务（触发器）线程池
JobTriggerPoolHelper.toStart();
//3. admin registry monitor run 初始化注册中心线程池，心跳30秒一次
JobRegistryHelper.getInstance().start();
//4. admin lose-monitor run ( depend on JobTriggerPoolHelper ) 初始化失败重试和告警（发邮件）线程，间隔10s执行一次
JobFailMonitorHelper.getInstance().start();
//5. admin lose-monitor run ( depend on JobTriggerPoolHelper ) 初始化回调线程池，初始化监控线程，记录运行了10分钟，执行器不在线的任务日志
JobCompleteHelper.getInstance().start();
//6. admin log report start 6. 初始化统计日志线程，间隔1分钟执行一次，统计3天日志，应该是管控台要用，如果设置了日志保留天数，清除日志
JobLogReportHelper.getInstance().start();
// start-schedule ( depend on JobTriggerPoolHelper ) 执行调度器
JobScheduleHelper.getInstance().start();
logger.info(">>>>>>>>> init xxl-job admin success.");
}
初始化 i18n 初始化下发任务（触发器）线程池 初始化注册中心线程池，心跳30秒一次 初始化失败重试和告警（发邮件）线程，间隔10s执行一次 初始化回调线程池，初始化监控线程，记录运行了10分钟，执行器不在线的任务日志 初始化统计日志线程，间隔1分钟执行一次，统计3天日志，应该是管控台要用，如果设置了日志保留天数，清除日志 执行调度器（下发任务）
初始化JobTriggerPool触发器线程池
任务调度触发： JobTriggerPool 定期触发任务的执行请求，根据任务的调度策略，按照设定的触发频率或触发时间点触发任务执行。 灵活配置： JobTriggerPool 的参数可以通过配置文件进行灵活配置，包括线程池大小、调度频率、调度策略等，以满足不同场景下的任务调度需求。 并发执行： JobTriggerPool 支持并发执行多个任务，根据任务的调度情况，同时触发多个任务的执行请求，提高任务执行效率。 可靠性保障： JobTriggerPool 实现了任务调度的可靠性保障机制，保证任务的准时触发和执行，同时支持任务调度的容错和重试功能，确保任务的可靠执行。 监控和管理： JobTriggerPool 提供了监控和管理功能，可以实时监控触发器线程池的运行状态和任务执行情况，包括线程池的活跃线程数、任务调度情况、任务执行结果等。 异常处理： JobTriggerPool 实现了异常处理机制，当触发器线程池发生异常时，可以进行相应的异常处理，如记录日志、发送告警等，保证系统的稳定运行。
public static void toStart() {
helper.start();
}
public void start(){
//最大200线程，最多处理1000任务
fastTriggerPool = new ThreadPoolExecutor(
10,
XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax(),
60L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(1000),
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "xxl-job, admin JobTriggerPoolHelper-fastTriggerPool-" + r.hashCode());
}
});
//最大100线程，最多处理2000任务
//一分钟内超时10次，则采用慢触发器执行
slowTriggerPool = new ThreadPoolExecutor(
10,
XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax(),
60L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(2000),
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "xxl-job, admin JobTriggerPoolHelper-slowTriggerPool-" + r.hashCode());
}
});
}
初始化registryOrRemoveThreadPool 线程池，剔除超时主机，更新主机列表
registryOrRemoveThreadPool 线程池主要用于执行器的注册和注销操作。下面是它的主要功能和特点：执行器注册： 当执行器启动时，会向调度中心注册自己的信息，包括执行器的名称、IP 地址、端口号等。 registryOrRemoveThreadPool线程池负责处理执行器注册的请求，确保注册操作的可靠性和及时性。执行器注销： 当执行器停止或下线时，需要向调度中心注销自己的信息，以便调度中心不再向该执行器发送任务执行请求。 registryOrRemoveThreadPool线程池也负责处理执行器注销的请求，确保注销操作的及时性和有效性。并发处理： registryOrRemoveThreadPool线程池支持并发处理多个注册和注销请求，根据系统的负载情况和注册/注销请求的数量，动态调整线程池的大小，以提高系统的并发处理能力。任务调度保障： 由于执行器的注册和注销操作是任务调度的关键步骤之一， registryOrRemoveThreadPool线程池的稳定运行对于保障任务调度的正常进行非常重要。它保证了注册和注销操作的及时处理，确保任务调度系统的可靠性和稳定性。异常处理： registryOrRemoveThreadPool线程池实现了异常处理机制，当注册或注销操作发生异常时，会进行相应的异常处理，如记录日志、发送告警等，保证系统的稳定运行。
registryOrRemoveThreadPool 线程池在 XXL-JOB 中起着至关重要的作用，它保证了执行器注册和注销操作的可靠性和及时性，是任务调度系统的重要组成部分。初始化注册或者删除线程池，主要负责客户端注册或者销毁到xxl_job_registry表 剔除超时注册机器 更新xxl_job_group执行器地址列表
public void start(){
// for registry or remove
//初始化注册或者删除线程池
registryOrRemoveThreadPool = new ThreadPoolExecutor(
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
执行失败重试和告警（发邮件）
失败重试 告警，发送邮件
失败重试
trigger_code：服务器下发任务的状态，200表示成功，500失败 handle_code：执行器执行任务的状态，200表示成功，500失败
第一种：trigger_code!=(0,200) 且 handle_code!=0 第二种：handle_code!=200
告警(这里可向spring注入JobAlarm)
public void start(){
monitorThread = new Thread(new Runnable() {
@Override
public void run() {
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
boolean alarmResult = XxlJobAdminConfig.getAdminConfig().getJobAlarmer().alarm(info, log);
public boolean alarm(XxlJobInfo info, XxlJobLog jobLog) {
boolean result = false;
if (jobAlarmList!=null && jobAlarmList.size()>0) {
result = true; // success means all-success
for (JobAlarm alarm: jobAlarmList) {
boolean resultItem = false;
try {
resultItem = alarm.doAlarm(info, jobLog);
} catch (Exception e) {
logger.error(e.getMessage(), e);
}
if (!resultItem) {
result = false;
}
}
}
return result;
}
初始化callbackThreadPool回调线程池，更改丢失主机的任务状态为失败
初始化回调线程池 启动线程，每个1分钟执行一次，将丢失主机信息的调度日志更改状态
public void start(){
// for callback 针对回调函数处理的线程池
callbackThreadPool = new ThreadPoolExecutor(
2,
20,
30L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(3000),
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
任务结果丢失处理：调度记录停留在 "运行中" 状态超过10min，且对应执行器心跳注册失败不在线，则将本地调度主动标记失败；
public static int updateHandleInfoAndFinish(XxlJobLog xxlJobLog) {
// finish 若父任务正常结束，则终止子任务,以及设置Childmsg
finishJob(xxlJobLog);
// text最大64kb 避免长度过长 截断超过长度限制字符
if (xxlJobLog.getHandleMsg().length() > 15000) {
xxlJobLog.setHandleMsg( xxlJobLog.getHandleMsg().substring(0, 15000) );
}
// fresh handle 更新超时joblog
return XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateHandleInfo(xxlJobLog);
}
启动统计日志线程
启动线程，间隔1分钟执行一次 统计“前三天”失败和成功的报表数据，管控台要用 如果设置了日志保留天数，比如1天，清除超时日志
public void start(){
logrThread = new Thread(new Runnable() {
//每分钟刷新一次
@Override
public void run() {
// last clean log time 记录上次清除日志时间
long lastCleanLogTime = 0;
while (!toStop) {
// 1、log-report refresh: refresh log report in 3 days
try {
for (int i = 0; i < 3; i++) {
// today 分别统计今天,昨天,前天0~24点的数据
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
启动调度器
scheduleThread：添加任务到时间轮 ringThread：执行任务的时间轮线程
1 添加任务到时间轮
触发器下次执行时间过期时间 > 5S 触发器下次执行时间过期时间 < 5S 触发器下次执行时间在未来5S以内。
public void start(){
// schedule thread
scheduleThread = new Thread(new Runnable() {
@Override
public void run() {
try {
//下5秒之后执行一次，等待服务器启动。
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
2 时间轮线程执行任务
// ring thread
ringThread = new Thread(new Runnable() {
@Override
public void run() {
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
手动触发：管控台手动启动一次任务 自动触发：根据触发条件触发任务，有时间轮线程处理
3 时间轮算法
方案1：每个任务对应一个调度线程，然后让线程等待间隔时间，然后执行，但是xxl-job任务太多的话，会创建太多线程，所以采用时间轮，也就是让一个线程执行多个任务 方案2：采用DelayQueue延迟队列，大量入队和出队对性能影响应该比较大。
执行器Executor 启动流程
@Bean
public XxlJobSpringExecutor xxlJobExecutor() {
logger.info(">>>>>>>>>>> xxl-job config init.");
XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
//xxjob管理地址
xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
//注册应用名称
xxlJobSpringExecutor.setAppname(appname);
//xxl作业执行器注册表地址
xxlJobSpringExecutor.setAddress(address);
xxlJobSpringExecutor.setIp(ip);
xxlJobSpringExecutor.setPort(port);
//注册地址的token
xxlJobSpringExecutor.setAccessToken(accessToken);
//日志路径
xxlJobSpringExecutor.setLogPath(logPath);
//日志保存天数
xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);
return xxlJobSpringExecutor;
}
// start
@Override
public void afterSingletonsInstantiated() {
// init JobHandler Repository
/*initJobHandlerRepository(applicationContext);*/
// init JobHandler Repository (for method) 初始化调度器资源管理器
initJobHandlerMethodRepository(applicationContext);
// refresh GlueFactory 刷新GlueFactory
GlueFactory.refreshInstance(1);
// super start
try {
super.start();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
初始化调度器资源管理器（从spring容器中将标记了XxlJob注解的方法，将其封装并添加到map中。） 刷新GlueFactory 启动服务，接收服务器请求。
// start
@Override
public void afterSingletonsInstantiated() {
// init JobHandler Repository
/*initJobHandlerRepository(applicationContext);*/
// init JobHandler Repository (for method) 初始化调度器资源管理器（从spring容器中将标记了XxlJob注解的方法，将其封装并添加到map中。）
initJobHandlerMethodRepository(applicationContext);
// refresh GlueFactory 刷新GlueFactory
GlueFactory.refreshInstance(1);
// super start
try {
super.start();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
1 遍历XxlJob注解查找方法
从spring容器获取所有对象，并遍历查找方法上标记XxlJob注解的方法 将xxljob配置的jobname作为key，对象,反射的执行,初始,销毁方法作为value注册jobHandlerRepository 中
private void initJobHandlerMethodRepository(ApplicationContext applicationContext) {
if (applicationContext == null) {
return;
}
// init job handler from method
String[] beanDefinitionNames = applicationContext.getBeanNamesForType(Object.class, false, true);
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
2 启动客户端执行器
执行器启动，主要做了如下事情：
初始化日志文件 封装调度中心请求路径，用于访问调度中心 清除过期日志 回调调度中心任务，汇报执行状态 执行内嵌服务
public void start() throws Exception {
// init logpath
XxlJobFileAppender.initLogPath(logPath);//初始化日志文件
// init invoker, admin-client
initAdminBizList(adminAddresses, accessToken); //初始化admin链接路径存储集合
// init JobLogFileCleanThread 清除过期日志
JobLogFileCleanThread.getInstance().start(logRetentionDays);
// init TriggerCallbackThread
TriggerCallbackThread.getInstance().start();
// init executor-server 执行内嵌服务
initEmbedServer(address, ip, port, appname, accessToken);
}
2.1 初始化日志文件
public static void initLogPath(String logPath){
// init
if (logPath!=null && logPath.trim().length()>0) {
logBasePath = logPath;
}
// mk base dir 日志文件不存在则创建
File logPathDir = new File(logBasePath);
if (!logPathDir.exists()) {
logPathDir.mkdirs();
}
logBasePath = logPathDir.getPath();
// mk glue dir 创建glue目录
File glueBaseDir = new File(logPathDir, "gluesource");
if (!glueBaseDir.exists()) {
glueBaseDir.mkdirs();
}
glueSrcPath = glueBaseDir.getPath();
}
2.2 封装调度中心请求路径，用于访问调度中心
private void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
if (adminAddresses!=null && adminAddresses.trim().length()>0) {
for (String address: adminAddresses.trim().split(",")) { //多个admin地址以,分隔
if (address!=null && address.trim().length()>0) {
AdminBiz adminBiz = new AdminBizClient(address.trim(), accessToken);
if (adminBizList == null) {
adminBizList = new ArrayList<AdminBiz>();
} //将admin地址以及token添加adminBiz中
adminBizList.add(adminBiz);
}
}
}
}
2.3 清除过期日志
public void start(final long logRetentionDays){
// limit min value 日志最大保存天数<3天,直接退出
if (logRetentionDays < 3 ) {
return;
}
//一天执行一次
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
2.4 回调调度中心，汇报执行状态
public void start() {
// valid 是否有配置admin路径
if (XxlJobExecutor.getAdminBizList() == null) {
logger.warn(">>>>>>>>>>> xxl-job, executor callback config fail, adminAddresses is null.");
return;
}
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
2.5 执行内嵌服务
使用netty开放端口，等待服务端调用 注册到服务端(心跳30S) 向服务端申请剔除服务
private void initEmbedServer(String address, String ip, int port, String appname, String accessToken) throws Exception {
// fill ip port 若没设置端口,则寻找可用端口
port = port>0?port: NetUtil.findAvailablePort(9999);
//若没设置IP，则获取本机Ip
ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp();
// generate address 构造地址,若没设置地址,则将ip,port拼接成地址
if (address==null || address.trim().length()==0) {
String ip_port_address = IpUtil.getIpPort(ip, port); // registry-address：default use address to registry , otherwise use ip:port if address is null
address = "http://{ip_port}/".replace("{ip_port}", ip_port_address);
}
// accessToken
if (accessToken==null || accessToken.trim().length()==0) {
logger.warn(">>>>>>>>>>> xxl-job accessToken is empty. To ensure system security, please set the accessToken.");
}
// start 启动嵌入服务器 ,向服务端注册,以及监听端口,主要服务于服务端调用。
embedServer = new EmbedServer();
embedServer.start(address, port, appname, accessToken);
}
public void start(final String address, final int port, final String appname, final String accessToken) {
executorBiz = new ExecutorBizImpl();
thread = new Thread(new Runnable() {
@Override
public void run() {
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
public void startRegistry(final String appname, final String address) {
// start registry
ExecutorRegistryThread.getInstance().start(appname, address);
}
public void start(final String appname, final String address){
// valid appname不允许为null
if (appname==null || appname.trim().length()==0) {
logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, appname is null.");
return;
}//服务端地址不能为null
if (XxlJobExecutor.getAdminBizList() == null) {
logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, adminAddresses is null.");
return;
}
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
XXL-JOB的服务注册
客户端
@Override
public ReturnT<String> registry(RegistryParam registryParam) {
return XxlJobRemotingUtil.postBody(addressUrl + "api/registry", accessToken, timeout, registryParam, String.class);
}
服务端
@Override
public ReturnT<String> registry(RegistryParam registryParam) {
return JobRegistryHelper.getInstance().registry(registryParam);
}
public ReturnT<String> registry(RegistryParam registryParam) {
// valid 校验
if (!StringUtils.hasText(registryParam.getRegistryGroup())
|| !StringUtils.hasText(registryParam.getRegistryKey())
|| !StringUtils.hasText(registryParam.getRegistryValue())) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "Illegal Argument.");
}
// async execute 异步注册
registryOrRemoveThreadPool.execute(new Runnable() {
@Override
public void run() { //更新修改时间
int ret = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().registryUpdate(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue(), new Date());
if (ret < 1) {//说明暂未数据,才新增
XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().registrySave(registryParam.getRegistryGroup(), registryParam.getRegistryKey(), registryParam.getRegistryValue(), new Date());
// fresh 空实现
freshGroupRegistryInfo(registryParam);
}
}
});
return ReturnT.SUCCESS;
}
客户端接收响应
XXL-JOB的 执行任务
调度中心向客户端发起post请求 client 通过内嵌服务netty接收，异步线程处理 找到job绑定的线程，将任务丢到阻塞队列中。然后返回结果给调度中心。 调度中心更改任务状态。 客户端执行任务后，将执行结果丢到回调线程的阻塞队列处理。 回调线程通过post请求访问调度中心，调度中心更改job最终结果。 倘若超过10分钟调度中心没收到回调线程的请求，则设置job最终结果失败。
服务端下发任务
@RequestMapping("/trigger")
@ResponseBody
//@PermissionLimit(limit = false)
public ReturnT<String> triggerJob(int id, String executorParam, String addressList) {
// force cover job param 设置默认值
if (executorParam == null) {
executorParam = "";
}
//触发器类型，手动 ，重试次数，'执行器任务分片参数，格式如 1/2'，任务参数，机器地址
JobTriggerPoolHelper.trigger(id, TriggerTypeEnum.MANUAL, -1, null, executorParam, addressList);
return ReturnT.SUCCESS;
}
public static void trigger(int jobId, TriggerTypeEnum triggerType, int failRetryCount, String executorShardingParam, String executorParam, String addressList) {
helper.addTrigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);
}
public void addTrigger(final int jobId,
final TriggerTypeEnum triggerType,
final int failRetryCount,
final String executorShardingParam,
final String executorParam,
final String addressList) {
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
XxlJobTrigger#trigger
public static void trigger(int jobId,
TriggerTypeEnum triggerType,
int failRetryCount,
String executorShardingParam,
String executorParam,
String addressList) {
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
private static void processTrigger(XxlJobGroup group, XxlJobInfo jobInfo, int finalFailRetryCount, TriggerTypeEnum triggerType, int index, int total){
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
路由策略
FIRST（第一个）：固定选择第一个机器； LAST（最后一个）：固定选择最后一个机器； ROUND（轮询）：； RANDOM（随机）：随机选择在线的机器； CONSISTENT_HASH（一致性HASH）：每个任务按照Hash算法固定选择某一台机器，且所有任务均匀散列在不同机器上。 LEAST_FREQUENTLY_USED（最不经常使用）：使用频率最低的机器优先被选举； LEAST_RECENTLY_USED（最近最久未使用）：最久未使用的机器优先被选举； FAILOVER（故障转移）：按照顺序依次进行心跳检测，第一个心跳检测成功的机器选定为目标执行器并发起调度； BUSYOVER（忙碌转移）：按照顺序依次进行空闲检测，第一个空闲检测成功的机器选定为目标执行器并发起调度； SHARDING_BROADCAST(分片广播)：广播触发对应集群中所有机器执行一次任务，同时系统自动传递分片参数；可根据分片参数开发分片任务；
@Override
public ReturnT<String> route(TriggerParam triggerParam, List<String> addressList){
return new ReturnT<String>(addressList.get(0));
}
public static ReturnT<String> runExecutor(TriggerParam triggerParam, String address){
ReturnT<String> runResult = null;
try {
ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address);//获取业务执行器地址,就执行器地址后面拼接token
//通过post请求客户端执行job
runResult = executorBiz.run(triggerParam);
} catch (Exception e) {
logger.error(">>>>>>>>>>> xxl-job trigger error, please check if the executor[{}] is running.", address, e);
runResult = new ReturnT<String>(ReturnT.FAIL_CODE, ThrowableUtil.toString(e));
}
//返回结果设置msg
StringBuffer runResultSB = new StringBuffer(I18nUtil.getString("jobconf_trigger_run") + "：");
runResultSB.append("<br>address：").append(address);
runResultSB.append("<br>code：").append(runResult.getCode());
runResultSB.append("<br>msg：").append(runResult.getMsg());
runResult.setMsg(runResultSB.toString());
return runResult;
}
public static ExecutorBiz getExecutorBiz(String address) throws Exception {
// valid
if (address==null || address.trim().length()==0) {
return null;
}
// load-cache
//从缓冲中通过地址获取ExecutorBiz
address = address.trim();
ExecutorBiz executorBiz = executorBizRepository.get(address);
if (executorBiz != null) {
return executorBiz;
}
// set-cache 找不到就新建
executorBiz = new ExecutorBizClient(address, XxlJobAdminConfig.getAdminConfig().getAccessToken());
//添加缓存
executorBizRepository.put(address, executorBiz);
return executorBiz;
}
@Override
public ReturnT<String> run(TriggerParam triggerParam) {
return XxlJobRemotingUtil.postBody(addressUrl + "run", accessToken, timeout, triggerParam, String.class);
}
客户端处理任务
执行器执行任务时序图：
@Override
protected void channelRead0(final ChannelHandlerContext ctx, FullHttpRequest msg) throws Exception {
// request parse
//final byte[] requestBytes = ByteBufUtil.getBytes(msg.content()); // byteBuf.toString(io.netty.util.CharsetUtil.UTF_8);
String requestData = msg.content().toString(CharsetUtil.UTF_8);//解析请求数据
String uri = msg.uri();//获取uri,后面通过uri来处理不同的请求
HttpMethod httpMethod = msg.method();//获取请求方式,Post/Get
boolean keepAlive = HttpUtil.isKeepAlive(msg);//保持长连接
String accessTokenReq = msg.headers().get(XxlJobRemotingUtil.XXL_JOB_ACCESS_TOKEN);
// invoke
bizThreadPool.execute(new Runnable() {
@Override
public void run() {
// do invoke
Object responseObj = process(httpMethod, uri, requestData, accessTokenReq);
// to json 响应结果转json
String responseJson = GsonTool.toJson(responseObj);
// write response
writeResponse(ctx, keepAlive, responseJson);
}
});
}
private Object process(HttpMethod httpMethod, String uri, String requestData, String accessTokenReq) {
// valid 不是POST直接返回异常
if (HttpMethod.POST != httpMethod) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, HttpMethod not support.");
}
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
接下来两个工作
绑定作业到具体线程JobThread，启动线程 任务丢入线程处理JobThread#triggerQueue
@Override
public ReturnT<String> run(TriggerParam triggerParam) {
// load old：jobHandler + jobThread
//获取job绑定线程
JobThread jobThread = XxlJobExecutor.loadJobThread(triggerParam.getJobId());
//获取作业处理器
IJobHandler jobHandler = jobThread!=null?jobThread.getHandler():null;
String removeOldReason = null;
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
public static JobThread registJobThread(int jobId, IJobHandler handler, String removeOldReason){
JobThread newJobThread = new JobThread(jobId, handler);//启动新线程处理工作任务
newJobThread.start();
logger.info(">>>>>>>>>>> xxl-job regist JobThread success, jobId:{}, handler:{}", new Object[]{jobId, handler});
//存储jobId与绑定工作的线程
JobThread oldJobThread = jobThreadRepository.put(jobId, newJobThread); // putIfAbsent | oh my god, map's put method return the old value!!!
if (oldJobThread != null) {//中断并删除旧线程
oldJobThread.toStop(removeOldReason);
oldJobThread.interrupt();
}
return newJobThread;
}
public ReturnT<String> pushTriggerQueue(TriggerParam triggerParam) {
// avoid repeat 若包含，则说明重复执行
if (triggerLogIdSet.contains(triggerParam.getLogId())) {
logger.info(">>>>>>>>>>> repeate trigger job, logId:{}", triggerParam.getLogId());
return new ReturnT<String>(ReturnT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId());
}
triggerLogIdSet.add(triggerParam.getLogId());
triggerQueue.add(triggerParam);
return ReturnT.SUCCESS;
}
public void execute() throws Exception {
//利用反射调用方法
Class<?>[] paramTypes = method.getParameterTypes();
if (paramTypes.length > 0) {
method.invoke(target, new Object[paramTypes.length]); // method-param can not be primitive-types
} else {
method.invoke(target);
}
}
@Override
public void execute() throws Exception {
Class<?>[] paramTypes = method.getParameterTypes();
if (paramTypes.length > 0) {
method.invoke(target, new Object[paramTypes.length]); // method-param can not be primitive-types
} else {
method.invoke(target);
}
}
客户端回调服务端
public ReturnT<String> pushTriggerQueue(TriggerParam triggerParam) {
// avoid repeat 若包含，则说明重复执行
if (triggerLogIdSet.contains(triggerParam.getLogId())) {
logger.info(">>>>>>>>>>> repeate trigger job, logId:{}", triggerParam.getLogId());
return new ReturnT<String>(ReturnT.FAIL_CODE, "repeate trigger job, logId:" + triggerParam.getLogId());
}
triggerLogIdSet.add(triggerParam.getLogId());
triggerQueue.add(triggerParam);
return ReturnT.SUCCESS;
}
private void doCallback(List<HandleCallbackParam> callbackParamList){
boolean callbackRet = false;
// callback, will retry if error 获取admin地址
for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
try { //回调admin,返回执行结果
ReturnT<String> callbackResult = adminBiz.callback(callbackParamList);
if (callbackResult!=null && ReturnT.SUCCESS_CODE == callbackResult.getCode()) {
callbackLog(callbackParamList, "<br>----------- xxl-job job callback finish.");
callbackRet = true;
break;
} else {
callbackLog(callbackParamList, "<br>----------- xxl-job job callback fail, callbackResult:" + callbackResult);
}
} catch (Exception e) {
callbackLog(callbackParamList, "<br>----------- xxl-job job callback error, errorMsg:" + e.getMessage());
}
}
if (!callbackRet) {
appendFailCallbackFile(callbackParamList);
}
}
public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList) {
return XxlJobRemotingUtil.postBody(addressUrl+"api/callback", accessToken, timeout, callbackParamList, String.class);
}
调度中心处理回调
public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList) {
callbackThreadPool.execute(new Runnable() {
@Override
public void run() {
for (HandleCallbackParam handleCallbackParam: callbackParamList) {
ReturnT<String> callbackResult = callback(handleCallbackParam);
logger.debug(">>>>>>>>> JobApiController.callback {}, handleCallbackParam={}, callbackResult={}",
(callbackResult.getCode()== ReturnT.SUCCESS_CODE?"success":"fail"), handleCallbackParam, callbackResult);
}
}
});
return ReturnT.SUCCESS;
}
private ReturnT<String> callback(HandleCallbackParam handleCallbackParam) {
// valid log item 加载log
XxlJobLog log = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().load(handleCallbackParam.getLogId());
if (log == null) { //日志没找到返回异常
return new ReturnT<String>(ReturnT.FAIL_CODE, "log item not found.");
}
if (log.getHandleCode() > 0) { //说明重复执行
return new ReturnT<String>(ReturnT.FAIL_CODE, "log repeate callback."); // avoid repeat callback, trigger child job etc
}
// handle msg
StringBuffer handleMsg = new StringBuffer();
if (log.getHandleMsg()!=null) {
handleMsg.append(log.getHandleMsg()).append("<br>");
}
if (handleCallbackParam.getHandleMsg() != null) {
handleMsg.append(handleCallbackParam.getHandleMsg());
}
// success, save log 更改
log.setHandleTime(new Date());
log.setHandleCode(handleCallbackParam.getHandleCode()); //更改处理状态,200正常,500错误
log.setHandleMsg(handleMsg.toString());
XxlJobCompleter.updateHandleInfoAndFinish(log);
return ReturnT.SUCCESS;
}
执行子任务
public static int updateHandleInfoAndFinish(XxlJobLog xxlJobLog) {
// finish 若父任务正常结束，则终止子任务,以及设置Childmsg
finishJob(xxlJobLog);
// text最大64kb 避免长度过长 截断超过长度限制字符
if (xxlJobLog.getHandleMsg().length() > 15000) {
xxlJobLog.setHandleMsg( xxlJobLog.getHandleMsg().substring(0, 15000) );
}
// fresh handle 更新joblog
return XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateHandleInfo(xxlJobLog);
}
private static void finishJob(XxlJobLog xxlJobLog){
// 1、handle success, to trigger child job
String triggerChildMsg = null;
if (XxlJobContext.HANDLE_COCE_SUCCESS == xxlJobLog.getHandleCode()) {
XxlJobInfo xxlJobInfo = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(xxlJobLog.getJobId());
if (xxlJobInfo!=null && xxlJobInfo.getChildJobId()!=null && xxlJobInfo.getChildJobId().trim().length()>0) {
triggerChildMsg = "<br><br><span style=\"color:#00c0ef;\" > >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_child_run") +"<<<<<<<<<<< </span><br>";
//获取子任务id
String[] childJobIds = xxlJobInfo.getChildJobId().split(",");
for (int i = 0; i < childJobIds.length; i++) {
int childJobId = (childJobIds[i]!=null && childJobIds[i].trim().length()>0 && isNumeric(childJobIds[i]))?Integer.valueOf(childJobIds[i]):-1;
if (childJobId > 0) {
//触发子任务
JobTriggerPoolHelper.trigger(childJobId, TriggerTypeEnum.PARENT, -1, null, null, null);
ReturnT<String> triggerChildResult = ReturnT.SUCCESS;
// add msg
triggerChildMsg += MessageFormat.format(I18nUtil.getString("jobconf_callback_child_msg1"),
(i+1),
childJobIds.length,
childJobIds[i],
(triggerChildResult.getCode()==ReturnT.SUCCESS_CODE?I18nUtil.getString("system_success"):I18nUtil.getString("system_fail")),
triggerChildResult.getMsg());
} else {
triggerChildMsg += MessageFormat.format(I18nUtil.getString("jobconf_callback_child_msg2"),
(i+1),
childJobIds.length,
childJobIds[i]);
}
}
}
}
if (triggerChildMsg != null) {
xxlJobLog.setHandleMsg( xxlJobLog.getHandleMsg() + triggerChildMsg );
}
// 2、fix_delay trigger next
// on the way
}
XXL-JOB的mysql分布式锁
基于数据库的集群方案，数据库选用Mysql；集群分布式并发环境中进行定时任务调度时，会在各个节点会上报任务，存到数据库中，执行时会从数据库中取出触发器来执行，如果触发器的名称和执行时间相同，则只有一个节点去执行此任务。
CREATE TABLE `xxl_job_lock` (
`lock_name` varchar(50) NOT NULL COMMENT '锁名称',
PRIMARY KEY (`lock_name`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
public class JobScheduleHelper {
private static Logger logger = LoggerFactory.getLogger(JobScheduleHelper.class);
private static JobScheduleHelper instance = new JobScheduleHelper();
public static JobScheduleHelper getInstance(){
return instance;
}
……
特别说明，由于平台限制50000字，这里删掉了2000字，具体内容请参见 本文pdf版本，找尼恩获取。
百亿级海量任务调度平台的架构与实现
说在最后：有问题找老架构取经
大龄男的最佳出路是 架构+ 管理 大龄女的最佳出路是 DPM，
部分历史案例
被裁太爽：每月大涨1W，10年CRUD小伙，靠转架构一个月上岸，年薪50W爽歪了！ 小白转架构，逆涨30%：CRUD小伙7年，狠卷3个月升级为架构师，爽爆了 绝地逢生：9年经验自考小伙伴，跟着尼恩狠卷3月硬核技术，面试机会爆表，2周后收3个offer ，满血复活 硬转架构：35岁被裁6个月 职业绝望，转架构急救上岸，DDD和3高项目太重要了 逆天改命：3年经验2本，卷3个月涨薪60%，进准大厂(得物、乐信、58集团等)，年薪36W 女35岁CRUD程序媛，12月被裁后，拿30个机会2个offer上岸，年底大裁员太不容易了 38还有救吗？有。42岁被裁2年天快塌了，急救1个月上岸，成开发经理offer，起死回生 被裁不慌，9年小伙1个月喜提年薪60W offer，做中间件架构，爽歪了 12年小伙转架构，1小时蜕变收年薪55万offer，秘诀：GO+Java双栖架构 年薪百万：40岁小伙被裁6个月，跟着尼恩猛卷3月硬核技术，100W年薪逆袭 ，上岸秘诀：首席架构/总架构 不怕裁员，8年小伙去中年危机，秘诀：换架构师赛道，上得厅堂，下得厨房，越早越好 被裁员后因祸得福，拿N+1后，逆涨30%，如何实现？ 逆天啦：4年crud小伙收shein+银行两优质offer，狠卷1月收年薪43万大涨30% 原来，拿大厂offer 有捷径...... 太劲爆.... 被裁4个月，38岁Android转Java，2个月提架构offer 起死回生：8年小伙高中毕业+频繁跳槽，狠卷2月，提Java高级offer 爽翻了：指导3轮，5年小伙收5大offer，涨50%，领路模式太牛 降维攻击，37年大龄老伙喜提60W年薪offer，1个月顺利上岸 架构速成：从一面杀到一次过，记7年女架构的速成经历 极速拿offer：阿里P6被裁后极速上岸，1个月内喜提2优质offer(含滴滴) 大逆袭：做化工12年，35岁转Java，被裁4个月后，1个月收架构Offer，大龄跨行，超级牛 被裁后炸爽：10年小伙12天火速上岸，反涨20%，爽暴了 裁就裁，6年小伙60W极速上岸，白拿20W还游一圈拉萨，真香 极速拿offer：被毕业3个月，11年经验小伙0.5个月极速拿offer，领路模式的巨大威力 惊天大逆袭：8年小伙20天时间提75W年薪offer，逆涨50%，秘诀在这
实现职业转型，极速上岸
助您实现职业转型、职业升级、极速上岸
---------------------------------
实现架构转型，再无中年危机
一起成为牛逼的未来超级架构师
请加尼恩个人微信 免费拿走