【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。

相关文章

Java图片压缩三种高效压缩方案详细解析

Java图片压缩三种高效压缩方案详细解析

《Java图片压缩三种高效压缩方案详细解析》图片压缩通常涉及减少图片的尺寸缩放、调整图片的质量(针对JPEG、PNG等)、使用特定的算法来减少图片的数据量等,：本文主要介绍Java图片压缩三种高效... 目录一、基于OpenCV的智能尺寸压缩技术亮点：适用场景：二、JPEG质量参数压缩关键技术：压缩效果对比
阅读更多...
Java调用C++动态库超详细步骤讲解(附源码)

Java调用C++动态库超详细步骤讲解(附源码)

《Java调用C++动态库超详细步骤讲解(附源码)》C语言因其高效和接近硬件的特性,时常会被用在性能要求较高或者需要直接操作硬件的场合,：本文主要介绍Java调用C++动态库的相关资料,文中通过代... 目录一、直接调用C++库第一步：动态库生成（vs2017+qt5.12.10）第二步：Java调用C++
阅读更多...
关于WebSocket协议状态码解析

关于WebSocket协议状态码解析

《关于WebSocket协议状态码解析》：本文主要介绍关于WebSocket协议状态码的使用方式,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录WebSocket协议状态码解析1. 引言2. WebSocket协议状态码概述3. WebSocket协议状态码详解3
阅读更多...
CSS Padding 和 Margin 区别全解析

CSS Padding 和 Margin 区别全解析

《CSSPadding和Margin区别全解析》CSS中的padding和margin是两个非常基础且重要的属性，它们用于控制元素周围的空白区域，本文将详细介绍padding和... 目录css Padding 和 Margin 全解析1. Padding: 内边距2. Margin: 外边距3. Padd
阅读更多...
Oracle数据库常见字段类型大全以及超详细解析

Oracle数据库常见字段类型大全以及超详细解析

《Oracle数据库常见字段类型大全以及超详细解析》在Oracle数据库中查询特定表的字段个数通常需要使用SQL语句来完成,：本文主要介绍Oracle数据库常见字段类型大全以及超详细解析,文中通过... 目录前言一、字符类型（Character）1、CHAR：定长字符数据类型2、VARCHAR2：变长字符数
阅读更多...
使用Jackson进行JSON生成与解析的新手指南

使用Jackson进行JSON生成与解析的新手指南

《使用Jackson进行JSON生成与解析的新手指南》这篇文章主要为大家详细介绍了如何使用Jackson进行JSON生成与解析处理,文中的示例代码讲解详细,感兴趣的小伙伴可以跟随小编一起学习一下... 目录1. 核心依赖2. 基础用法2.1 对象转 jsON（序列化）2.2 JSON 转对象（反序列化）3.
阅读更多...
Springboot @Autowired和@Resource的区别解析

Springboot @Autowired和@Resource的区别解析

《Springboot@Autowired和@Resource的区别解析》@Resource是JDK提供的注解,只是Spring在实现上提供了这个注解的功能支持,本文给大家介绍Springboot@... 目录【一】定义【1】@Autowired【2】@Resource【二】区别【1】包含的属性不同【2】@
阅读更多...
SpringCloud动态配置注解@RefreshScope与@Component的深度解析

SpringCloud动态配置注解@RefreshScope与@Component的深度解析

《SpringCloud动态配置注解@RefreshScope与@Component的深度解析》在现代微服务架构中,动态配置管理是一个关键需求,本文将为大家介绍SpringCloud中相关的注解@Re... 目录引言1. @RefreshScope 的作用与原理1.1 什么是 @RefreshScope1.
阅读更多...
Java并发编程必备之Synchronized关键字深入解析

Java并发编程必备之Synchronized关键字深入解析

《Java并发编程必备之Synchronized关键字深入解析》本文我们深入探索了Java中的Synchronized关键字,包括其互斥性和可重入性的特性,文章详细介绍了Synchronized的三种... 目录一、前言二、Synchronized关键字2.1 Synchronized的特性1. 互斥2.
阅读更多...
Python实现无痛修改第三方库源码的方法详解

Python实现无痛修改第三方库源码的方法详解

《Python实现无痛修改第三方库源码的方法详解》很多时候,我们下载的第三方库是不会有需求不满足的情况,但也有极少的情况,第三方库没有兼顾到需求,本文将介绍几个修改源码的操作,大家可以根据需求进行选择... 目录需求不符合模拟示例 1. 修改源文件2. 继承修改3. 猴子补丁4. 追踪局部变量需求不符合很
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2