【XXL-JOB】执行器架构设计和源码解析

2024-03-29 01:20
文章标签 源码 解析 架构设计 xxl job 执行器

本文主要是介绍【XXL-JOB】执行器架构设计和源码解析，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

简介

XXL-JOB是一个分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用。
在这里插入图片描述
XXL-JOB分为B/S架构，调用中心是XXL-JOB服务端，执行器是客户端。

  • 调度中心核心功能：执行器管理、任务管理、任务调度、监控告警和故障转移
  • 执行器核心功能：负责业务任务处理，不关心任务调度

XXL-JOB将任务调度和任务执行隔离，将任务调度和执行进行解耦，让研发人员只关注业务部分，提高搞开发效率和系统扩展性。

集成XXL-JOB

添加依赖

<dependency><groupId>com.xuxueli</groupId><artifactId>xxl-job-core</artifactId><version>${最新稳定版}</version>
</dependency>

添加配置

### 调度中心部署根地址 [选填]：如调度中心集群部署存在多个地址则用逗号分隔。执行器将会使用该地址进行"执行器心跳注册"和"任务结果回调"；为空则关闭自动注册；
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.admin.addresses：调度中心地址，用于自动注册和心跳检测
  • xxl.job.executor.appname：指定执行器名称，每个服务都应该有不同的执行器名称，同一个服务的不同集群节点的执行器名称应该相同
  • xxl.job.accessToken：需要和调度中心配置保持一致，默认是default_token
  • xxl.job.executor.port=9999：如果在单机部署多个执行器时，注意要配置不同执行器端口，否则服务启动时会报端口冲突

启动类配置

执行器启动配置支持两种种方式：

  • XxlJobSpringExecutor：集成Spring框架【推荐】，不会有类爆炸问题，集成方便
  • XxlJobSimpleExecutor：无框架模式，可以参考源码示例xxl-job-executor-sample-frameless中的FrameLessXxlJobConfig类，优点不限制环境和框架，缺点每个任务就是一个类

这里以XxlJobSpringExecutor为例：

@Bean
public XxlJobSpringExecutor xxlJobExecutor() {logger.info(">>>>>>>>>>> xxl-job config init.");XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();xxlJobSpringExecutor.setAdminAddresses(adminAddresses);xxlJobSpringExecutor.setAppname(appname);xxlJobSpringExecutor.setAddress(address);xxlJobSpringExecutor.setIp(ip);xxlJobSpringExecutor.setPort(port);xxlJobSpringExecutor.setAccessToken(accessToken);xxlJobSpringExecutor.setLogPath(logPath);xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);return xxlJobSpringExecutor;
}

到这里我们的执行器就集成完成啦。

自定义任务处理器

结合上面的配置一个执行器服务就配置好了，现在我们添加一个自定义的任务：

@Component
public class SampleXxlJob {private static Logger logger = LoggerFactory.getLogger(SampleXxlJob.class);@XxlJob("demoJobHandler")public void demoJobHandler() throws Exception {XxlJobHelper.log("XXL-JOB, Hello World.");for (int i = 0; i < 5; i++) {XxlJobHelper.log("beat at:" + i);TimeUnit.SECONDS.sleep(2);}// default success}
}

开发步骤：

  1. 任务开发：在Spring Bean实例中，开发Job方法；
  2. 注解配置：为Job方法添加注解 “@XxlJob(value=“自定义jobhandler名称”, init = “JobHandler初始化方法”, destroy = “JobHandler销毁方法”)”，注解value值对应的是调度中心新建任务的JobHandler属性的值。
  3. 执行日志：需要通过 “XxlJobHelper.log” 打印执行日志，这个日志可以在管理端的执行日志被查看；
  4. 任务结果：默认任务结果为 “成功” 状态，不需要主动设置；如有需求，比如设置任务结果为失败，可以通过 XxlJobHelper.handleFail/handleSuccess自主设置任务结果；
  5. 在调度中心配置任务调度

在这里插入图片描述

任务类型

这列演示的是最简单的任务，XXL_JOB还支持更为复杂的任务，任务分类：

  • 简单任务示例（Bean模式）：定义一个Spring Bean，其中包含要执行的任务方法。
  • 分片广播任务：允许将一个大任务拆分成多个小任务，并在多个执行器实例上并行执行。这通常用于大数据处理或并行计算。
  • 命令行任务：允许直接执行系统命令或脚本。
  • 跨平台Http任务：通过HTTP请求来触发任务。在XXL-JOB调度中心添加任务时，选择HTTP模式，并配置相应的URL、请求方法和参数。
  • 生命周期任务示例：任务初始化与销毁时，支持自定义相关逻辑，常用于资源准备和清理工作。；

命令行任务和跨平台Http任务都是通过传入指定的参数在JOB中实现的任务操作，具体实现给可以看源码。

运行模式

在这里插入图片描述
这里的运行模式大致可以分为两种BEAN和````GLUE```。

  • BEAN：就是刚刚提到的简单示例的模式，也是最常用的模式。
  • GLUE：允许你在线编写任务的执行代码。你可以使用Groovy语言编写代码，并在XXL-JOB的Web界面上直接运行和调试。

GLUE示例

在这里插入图片描述
然后在线编辑任务，实现任务调度。
在这里插入图片描述

11:06:56.418 logback [xxl-job, EmbedServer bizThreadPool-1281151494] INFO  c.x.job.core.executor.XxlJobExecutor - >>>>>>>>>>> xxl-job regist JobThread success, jobId:3, handler:com.xxl.job.core.handler.impl.GlueJobHandler@7ae04e7a
GLUE任务测试

任务执行成功。

执行器生命周期

当服务启动后会通过XxlJobSpringExecutor去集成执行器，集成过程中就会完成执行器自动注册。

public class XxlJobExecutor  {public void start() throws Exception {// init logpathXxlJobFileAppender.initLogPath(logPath);// init invoker, admin-clientinitAdminBizList(adminAddresses, accessToken);// init JobLogFileCleanThreadJobLogFileCleanThread.getInstance().start(logRetentionDays);// init TriggerCallbackThreadTriggerCallbackThread.getInstance().start();// init executor-serverinitEmbedServer(address, ip, port, appname, accessToken);}
}
  • XxlJobFileAppender.initLogPath(logPath);：执行日志路和GULE源码路径初始化
  • initAdminBizList(adminAddresses, accessToken);：初始化调用调度中心的RPC工具
  • JobLogFileCleanThread.getInstance().start(logRetentionDays);：初始化日志定期清楚守护线程
  • TriggerCallbackThread.getInstance().start();：初始化任务执行结果通知调度中心的回调守护线程
  • initEmbedServer(address, ip, port, appname, accessToken);：初始化执行器服务端线程

启动内嵌服务 EmbedServer

在上诉执行器完成初始化后会启动嵌入式服务。

embedServer = new EmbedServer();
embedServer.start(address, port, appname, accessToken);

start方法中第一步会启动一个内嵌的netty服务器。

// start server
ServerBootstrap bootstrap = new ServerBootstrap();
bootstrap.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overridepublic void initChannel(SocketChannel channel) throws Exception {channel.pipeline().addLast(new IdleStateHandler(0, 0, 30 * 3, TimeUnit.SECONDS))  // beat 3N, close if idle.addLast(new HttpServerCodec()).addLast(new HttpObjectAggregator(5 * 1024 * 1024))  // merge request & reponse to FULL.addLast(new EmbedHttpServerHandler(executorBiz, accessToken, bizThreadPool));}}).childOption(ChannelOption.SO_KEEPALIVE, true);// bind
ChannelFuture future = bootstrap.bind(port).sync();logger.info(">>>>>>>>>>> xxl-job remoting server start success, nettype = {}, port = {}", EmbedServer.class, port);

内嵌服务启动后会对外暴露接口给调度中心使用，完成执行器与调度中心的通讯。

  • /beat：调度中心故障探测接口，当任路由策略务配置为故障转移时是调用这个接口探活
  • /idleBeat：忙碌检测接口，当任务路由策略配置为忙碌转移时调用这个接口
  • /run：任务触发接口
  • /kill：任务终止接口
  • /log：在调用中心查看任务执行日志时调用这个接口

执行器注册&心跳检测 registryThread

启动内嵌服务后就会开始注册执行器到调度中心，执行器注册和心跳检测调用的是调度中心的同一个接口/registry。

ExecutorRegistryThread.getInstance().start(appname, address);

while (!toStop) {try {RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appname, address);for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {try {// 调用接口registry接口完成上报ReturnT<String> registryResult = adminBiz.registry(registryParam);if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {registryResult = ReturnT.SUCCESS;logger.debug(">>>>>>>>>>> xxl-job registry success, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});break;} else {logger.info(">>>>>>>>>>> xxl-job registry fail, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});}} catch (Exception e) {logger.info(">>>>>>>>>>> xxl-job registry error, registryParam:{}", registryParam, e);}}} catch (Exception e) {...}try {if (!toStop) {// 每次循环休眠30秒TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);}} catch (InterruptedException e) {...}
}
  • ExecutorRegistryThread.getInstance().start(appname, address);这里的start方法会启动一个守护线程，一直循环调用调度中心/registry接口，以此完成对注册器的注册和心跳检测
  • 心跳检测频率为30S一次
  • 调用中心有个守护线程每30秒执行一次，每次检查并删除超过90秒没有上报的执行器

执行器销毁 registryThread

当服务停用时会主动调用/registryRemove接口去调度中心销毁执行器

ExecutorRegistryThread.getInstance().toStop();public void start(final String appname, final String address){// validif (appname==null || appname.trim().length()==0) {logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, appname is null.");return;}if (XxlJobExecutor.getAdminBizList() == null) {logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, adminAddresses is null.");return;}registryThread = new Thread(new Runnable() {@Overridepublic void run() {...// registry removetry {RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appname, address);for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {try {// 调用`/registryRemove`接口去调度中心销毁执行器ReturnT<String> registryResult = adminBiz.registryRemove(registryParam);if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {registryResult = ReturnT.SUCCESS;logger.info(">>>>>>>>>>> xxl-job registry-remove success, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});break;} else {logger.info(">>>>>>>>>>> xxl-job registry-remove fail, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});}} catch (Exception e) {...}}} catch (Exception e) {...}}});registryThread.setDaemon(true);registryThread.setName("xxl-job, executor ExecutorRegistryThread");registryThread.start();}

执行器生命周期核心流程

在这里插入图片描述

执行定时任务

我们执行器初始化完成后执行器就正常注册到调度中心，当任务在时间轮被拿出来后通过调度策略和阻塞策略，最终通过EmbedServer暴露出来的/run接口触发任务执行。

case "/run":TriggerParam triggerParam = GsonTool.fromJson(requestData, TriggerParam.class);return executorBiz.run(triggerParam);

这里会带上几个核心参数：

  • jobId：任务ID
  • executorHandler：任务执行器名称
  • executorParams：执行参数

任务执行线程管理 JobThread

// 执行线程容器
private static ConcurrentMap<Integer, JobThread> jobThreadRepository = new ConcurrentHashMap<Integer, JobThread>();
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});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 static JobThread loadJobThread(int jobId){return jobThreadRepository.get(jobId);
}

XXL-JOB会给每个任务创建一个执行线程 JobThread，每个线程会有一个任务队列triggerQueue。每当有任务提交过来会先获取到对应的执行线程，并将任务放到这个线程下的队列中。这个线程如果在30个周期内都没有新任务需要执行，那么这个线程将会被回收。

任务阻塞处理策略 ExecutorBlockStrategyEnum

...
// 如果存在 jobThread 说明可能存在有任务在执行，则需要进行阻塞处理策略
if (jobThread != null) {ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(triggerParam.getExecutorBlockStrategy(), null);if (ExecutorBlockStrategyEnum.DISCARD_LATER == blockStrategy) {// 判断是否有任务在执行，如果有则直接丢弃任务if (jobThread.isRunningOrHasQueue()) {return new ReturnT<String>(ReturnT.FAIL_CODE, "block strategy effect："+ExecutorBlockStrategyEnum.DISCARD_LATER.getTitle());}} else if (ExecutorBlockStrategyEnum.COVER_EARLY == blockStrategy) {// 判断是否有任务在执行，如果有则覆盖之前的调度if (jobThread.isRunningOrHasQueue()) {removeOldReason = "block strategy effect：" + ExecutorBlockStrategyEnum.COVER_EARLY.getTitle();jobThread = null;}} else {}
}
...
if (jobThread == null) {jobThread = XxlJobExecutor.registJobThread(triggerParam.getJobId(), jobHandler, removeOldReason);
}
...
// 单机串行或者没有任务执行则直接放到线程队列中
ReturnT<String> pushResult = jobThread.pushTriggerQueue(triggerParam);

任务执行JobThread.run()

// execute
while(!toStop){running = false;idleTimes++;TriggerParam triggerParam = null;try {// 获取需要执行的任务triggerParam = triggerQueue.poll(3L, TimeUnit.SECONDS);if (triggerParam!=null) {// 生成对应的日志文件, 格式为 "logPath/yyyy-MM-dd/9999.log" 999为调度日志的IDString logFileName = XxlJobFileAppender.makeLogFileName(new Date(triggerParam.getLogDateTime()), triggerParam.getLogId());...if (triggerParam.getExecutorTimeout() > 0) {// limit timeoutThread futureThread = null;try {// 超时任务的执行FutureTask<Boolean> futureTask = new FutureTask<Boolean>(new Callable<Boolean>() {@Overridepublic Boolean call() throws Exception {// init job contextXxlJobContext.setXxlJobContext(xxlJobContext);handler.execute();return true;}});futureThread = new Thread(futureTask);futureThread.start();Boolean tempResult = futureTask.get(triggerParam.getExecutorTimeout(), TimeUnit.SECONDS);} catch (TimeoutException e) {...} finally {futureThread.interrupt();}} else {// 普通任务的执行handler.execute();}// 任务回调if (XxlJobContext.getXxlJobContext().getHandleCode() <= 0) {XxlJobHelper.handleFail("job handle result lost.");} } else {if (idleTimes > 30) {if(triggerQueue.size() == 0) {	// avoid concurrent trigger causes jobId-lostXxlJobExecutor.removeJobThread(jobId, "excutor idel times over limit.");}}}} catch (Throwable e) {...} finally {...}
}

这里首先需要在对象获取需要执行的任务，如果获取到任务需要直接修改线程运行状态为运行中，再判断是否是超时任务，如果是需要采用FutureTask来执行任务，否则直接执行。任务的执行直接调用Spring注册过来的hander。

执行器核心流程

在这里插入图片描述

总结

  1. 大量采用异步线程来任务调度的性能问题
  2. 执行器通过心跳机制来保证执行器的可用性
  3. 通过线程隔离、阻塞策略的方式来解决任务调度的可靠性
  4. 通过提调度中心dashboard的来解决系统可维护性和可观测性
  5. 通过accessToken来解决远程通讯的安全性

XXL-JOB完成流程

在这里插入图片描述

源码

https://github.com/xuxueli/xxl-job

引流

GitFlowPlus

GitFlowPlus分支管理IDEA插件

layering-cache

layering-cache 多级缓存开发框架

这篇关于【XXL-JOB】执行器架构设计和源码解析的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/857227。

相关文章

C语言中自动与强制转换全解析

C语言中自动与强制转换全解析

《C语言中自动与强制转换全解析》在编写C程序时,类型转换是确保数据正确性和一致性的关键环节,无论是隐式转换还是显式转换,都各有特点和应用场景,本文将详细探讨C语言中的类型转换机制,帮助您更好地理解并在... 目录类型转换的重要性自动类型转换（隐式转换）强制类型转换（显式转换）常见错误与注意事项总结与建议类型
阅读更多...
MySQL 缓存机制与架构解析(最新推荐)

MySQL 缓存机制与架构解析(最新推荐)

《MySQL缓存机制与架构解析(最新推荐)》本文详细介绍了MySQL的缓存机制和整体架构,包括一级缓存（InnoDBBufferPool）和二级缓存（QueryCache）,文章还探讨了SQL... 目录一、mysql缓存机制概述二、MySQL整体架构三、SQL查询执行全流程四、MySQL 8.0为何移除查
阅读更多...
在Rust中要用Struct和Enum组织数据的原因解析

在Rust中要用Struct和Enum组织数据的原因解析

《在Rust中要用Struct和Enum组织数据的原因解析》在Rust中,Struct和Enum是组织数据的核心工具,Struct用于将相关字段封装为单一实体,便于管理和扩展,Enum用于明确定义所有... 目录为什么在Rust中要用Struct和Enum组织数据？一、使用struct组织数据：将相关字段绑
阅读更多...
使用Java实现一个解析CURL脚本小工具

使用Java实现一个解析CURL脚本小工具

《使用Java实现一个解析CURL脚本小工具》文章介绍了如何使用Java实现一个解析CURL脚本的工具,该工具可以将CURL脚本中的Header解析为KVMap结构,获取URL路径、请求类型,解析UR... 目录使用示例实现原理具体实现CurlParserUtilCurlEntityICurlHandler
阅读更多...
深入解析Spring TransactionTemplate 高级用法(示例代码)

深入解析Spring TransactionTemplate 高级用法(示例代码)

《深入解析SpringTransactionTemplate高级用法(示例代码)》TransactionTemplate是Spring框架中一个强大的工具,它允许开发者以编程方式控制事务,通过... 目录1. TransactionTemplate 的核心概念2. 核心接口和类3. TransactionT
阅读更多...
数据库使用之union、union all、各种join的用法区别解析

数据库使用之union、union all、各种join的用法区别解析

《数据库使用之union、unionall、各种join的用法区别解析》：本文主要介绍SQL中的Union和UnionAll的区别,包括去重与否以及使用时的注意事项,还详细解释了Join关键字,... 目录一、Union 和Union All1、区别：2、注意点：3、具体举例二、Join关键字的区别&php
阅读更多...
Spring IOC控制反转的实现解析

Spring IOC控制反转的实现解析

《SpringIOC控制反转的实现解析》：本文主要介绍SpringIOC控制反转的实现,IOC是Spring的核心思想之一,它通过将对象的创建、依赖注入和生命周期管理交给容器来实现解耦,使开发者... 目录1. IOC的基本概念1.1 什么是IOC1.2 IOC与DI的关系2. IOC的设计目标3. IOC
阅读更多...
java中的HashSet与 == 和 equals的区别示例解析

java中的HashSet与 == 和 equals的区别示例解析

《java中的HashSet与==和equals的区别示例解析》HashSet是Java中基于哈希表实现的集合类,特点包括：元素唯一、无序和可包含null,本文给大家介绍java中的HashSe... 目录什么是HashSetHashSet 的主要特点是HashSet 的常用方法hasSet存储为啥是无序的
阅读更多...
Go中sync.Once源码的深度讲解

Go中sync.Once源码的深度讲解

《Go中sync.Once源码的深度讲解》sync.Once是Go语言标准库中的一个同步原语,用于确保某个操作只执行一次,本文将从源码出发为大家详细介绍一下sync.Once的具体使用,x希望对大家有... 目录概念简单示例源码解读总结概念sync.Once是Go语言标准库中的一个同步原语，用于确保某个操
阅读更多...
Linux中shell解析脚本的通配符、元字符、转义符说明

Linux中shell解析脚本的通配符、元字符、转义符说明

《Linux中shell解析脚本的通配符、元字符、转义符说明》：本文主要介绍shell通配符、元字符、转义符以及shell解析脚本的过程,通配符用于路径扩展,元字符用于多命令分割,转义符用于将特殊... 目录一、linux shell通配符(wildcard)二、shell元字符（特殊字符 Meta）三、s
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2