【源码分析】XXL-JOB的执行器的注册流程

目的:分析xxl-job执行器的注册过程

流程:

  1. 获取执行器中所有被注解(@xxlJjob)修饰的handler
  2. 执行器注册过程
  3. 执行器中任务执行过程

版本:xxl-job 2.3.1

建议:下载xxl-job源码,按流程图debug调试,看堆栈信息并按文章内容理解执行流程

完整流程图:img

 

 

img

查找Handler任务

部分流程图:

 

img

首先启动管理台界面(服务XxlJobAdminApplication),然后启动项目中给的执行器实例(SpringBoot);

img

这个方法是扫描项目中使用@xxlJob注解的所有handler方法。接着往下走

private void initJobHandlerMethodRepository(ApplicationContext applicationContext) {if (applicationContext == null) {return;}//获取该项目中所有的bean,然后遍历String[] beanDefinitionNames = applicationContext.getBeanNamesForType(Object.class, false, true);for (String beanDefinitionName : beanDefinitionNames) {Object bean = applicationContext.getBean(beanDefinitionName);Map<Method, XxlJob> annotatedMethods = null;   // referred to :org.springframework.context.event.EventListenerMethodProcessor.processBeantry {annotatedMethods = MethodIntrospector.selectMethods(bean.getClass(),new MethodIntrospector.MetadataLookup<XxlJob>() {//注意点★@Overridepublic XxlJob inspect(Method method) {return AnnotatedElementUtils.findMergedAnnotation(method, XxlJob.class);}});} catch (Throwable ex) {logger.error("xxl-job method-jobhandler resolve error for bean[" + beanDefinitionName + "].", ex);}//没有跳过本次循环继续if (annotatedMethods==null || annotatedMethods.isEmpty()) {continue;}//获取了当前执行器中所有@xxl-job的方法,获取方法以及对应的初始化和销毁方法for (Map.Entry<Method, XxlJob> methodXxlJobEntry : annotatedMethods.entrySet()) {Method executeMethod = methodXxlJobEntry.getKey();XxlJob xxlJob = methodXxlJobEntry.getValue();// registregistJobHandler(xxlJob, bean, executeMethod);}}
}

Spring案例执行器中有5个handler:

img

XxlJobExecutor.registJobHandler()中部分源码

String name = xxlJob.value();
//make and simplify the variables since they'll be called several times later
Class<?> clazz = bean.getClass();
String methodName = executeMethod.getName();
if (name.trim().length() == 0) {throw new RuntimeException("xxl-job method-jobhandler name invalid, for[" + clazz + "#" + methodName + "] .");
}
if (loadJobHandler(name) != null) {throw new RuntimeException("xxl-job jobhandler[" + name + "] naming conflicts.");
}

然后进行遍历注册;开始进行名字判断:

  1. 判断bean名字是否为空
  2. 判断bean是否被注册了(存在了)

loadJobHandler校验方式会去该方法中查找:当bean注册完成后时存放到jobHandlerRepository一个map类型中;

private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
public static IJobHandler loadJobHandler(String name){return jobHandlerRepository.get(name);
}

executeMethod.setAccessible(true);它实现了修改对象访问权限的功能,参数为true,则表示允许调用方在使用反射时忽略Java语言的访问控制检查.

往后走会判断该注解的生命周期方法(init和destroy)

  1. 未设置生命周期,则直接开始注册
//注意MethodJobHandler,后面会用到
registJobHandler(name, new MethodJobHandler(bean, executeMethod, initMethod, destroyMethod));
//添加执行器名字及对应的hob方法信息(当前类、方法、init和destroy属性)
public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);return jobHandlerRepository.put(name, jobHandler);
}
  1. 有生命周期,设置init和destroy方法权限
if (xxlJob.init().trim().length() > 0) {try {initMethod = clazz.getDeclaredMethod(xxlJob.init());initMethod.setAccessible(true);} catch (NoSuchMethodException e) {throw new RuntimeException("xxl-job method-jobhandler initMethod invalid, for[" + clazz + "#" + methodName + "] .");}
}
if (xxlJob.destroy().trim().length() > 0) {try {destroyMethod = clazz.getDeclaredMethod(xxlJob.destroy());destroyMethod.setAccessible(true);} catch (NoSuchMethodException e) {throw new RuntimeException("xxl-job method-jobhandler destroyMethod invalid, for[" + clazz + "#" + methodName + "] .");}
}

首先检查@XxlJob注解中的init属性是否存在且不为空。如果存在,则尝试获取该类中名为init的方法,并将其设置为可访问状态,以便后续调用。

同理,代码接下来也检查了@XxlJob注解中的destroy属性是否存在且不为空,如果是,则获取该类中名为destroy的方法,并设置其为可访问状态

在这个过程中,如果某个方法不存在或者无法被访问,则会抛出NoSuchMethodException异常,并且使用throw new RuntimeException将其包装并抛出一个运行时异常。这样做的目的是为了提醒开发人员在任务处理器类中正确地设置init和destroy属性,并确保方法名称与属性值一致。

执行器的注册过程

部分流程图:

img

public void afterSingletonsInstantiated() {// init JobHandler Repository/*initJobHandlerRepository(applicationContext);*/// init JobHandler Repository (for method)initJobHandlerMethodRepository(applicationContext);// refresh GlueFactoryGlueFactory.refreshInstance(1);// super starttry {super.start();} catch (Exception e) {throw new RuntimeException(e);}
}

在扫描完执行器中所有的任务后,开始进行执行器注册XxlJobSpringExecutor中的super.start()方法。

在初始化执行服务器启动之前,进行了四种操作,初始化日志、初始化adminBizList地址(可视化管理台地址)、初始化日志清除、初始化回调线程等。

这里需要注意的是第二步初始化地址,在初始化服务器启动的时候需要用到。

private void initEmbedServer(String address, String ip, int port, String appname, String accessToken) throws Exception {// fill ip portport = port>0?port: NetUtil.findAvailablePort(9999);ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp();// generate addressif (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 nulladdress = "http://{ip_port}/".replace("{ip_port}", ip_port_address);}// accessTokenif (accessToken==null || accessToken.trim().length()==0) {logger.warn(">>>>>>>>>>> xxl-job accessToken is empty. To ensure system security, please set the accessToken.");}// startembedServer = new EmbedServer();embedServer.start(address, port, appname, accessToken);
}

继续到initEmbedServer,开始初始化ip地址和端口等,需要明白的是,这一步的参数获取方式其实是第一步读取**XxlJobConfig**获得的;进行ip的校验和拼接等操作,开始进行真正的注册。

创建一个嵌入式的HTTP服务器,将当前执行器信息(包含应用名称和IP地址端口等)注册到注册中心,注册方式的实现在ExecutorRegistryThread中实现。

校验名字和注册中心,如果注册中心不可用,则等待一段时间后重新尝试连接。

// registry
while (!toStop) {try {RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appname, address);for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {try {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) {if (!toStop) {logger.error(e.getMessage(), e);}}try {//心跳检测,默认30sif (!toStop) {TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);}} catch (InterruptedException e) {if (!toStop) {logger.warn(">>>>>>>>>>> xxl-job, executor registry thread interrupted, error msg:{}", e.getMessage());}}
}

开启一个新线程,首先构建注册参数(包含执行器分组、执行器名字、执行器本地地址及端口号),遍历注册中心地址,开始进行执行器注册,注册方式通过发送http的post请求。

@Override
public ReturnT<String> registry(RegistryParam registryParam) {return XxlJobRemotingUtil.postBody(addressUrl + "api/registry", accessToken, timeout, registryParam, String.class);
}

debug的过程中,XxlJobRemotingUtil 执行到int statusCode =  connection.getResponseCode();才会跳转到JobApiController.api中的注册地址.

// services mapping
if ("callback".equals(uri)) {List<HandleCallbackParam> callbackParamList = GsonTool.fromJson(data, List.class, HandleCallbackParam.class);return adminBiz.callback(callbackParamList);
} else if ("registry".equals(uri)) {RegistryParam registryParam = GsonTool.fromJson(data, RegistryParam.class);return adminBiz.registry(registryParam);
} else if ("registryRemove".equals(uri)) {RegistryParam registryParam = GsonTool.fromJson(data, RegistryParam.class);return adminBiz.registryRemove(registryParam);
} else {return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping("+ uri +") not found.");
}

最后进入到JobRegistryHelper.registry()方法中完成数据库的入库和更新操作。

通过更新语句判断该执行器是否注册,结果小于1,那么保存注册器信息,并向注册中心发送一个请求,更新当前执行器所属的应用名称、执行器名称和 IP 地址等信息,否则跳过。

public ReturnT<String> registry(RegistryParam registryParam) {//.......// async executeregistryOrRemoveThreadPool.execute(new Runnable() {@Overridepublic 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;
}

至此执行器的注册流程分析完成。

执行器中的任务执行过程

img

部分流程图:

img

执行器中的任务流程比较简单,如果执行器启动的话,那么每次执行任务是通过JobThread通过Cron表达式进行操作的。

通过handler.execute()进行执行,是在框架内部通过反射机制调用作业处理器对象 handler 中的 execute() 方法实现的。在这个过程中,handler 对象表示被加载的作业处理器,并且已经调用了init()方法进行初始化。

method.invoke() 方法使用反射机制调用指定对象target 中的方法 method。在这个方法中,target 表示作业处理器对象,method 表示作业处理器中的 execute() 方法。

通过上述方法,获取到SampleXxlJob.demoJobHandler的任务,然后开始进行任务逻辑操作。

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/88242.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

gcc/g++ 编译选项详解

gcc/g 编译选项详解 文章目录 gcc/g 编译选项详解编译步骤gcc 与 g 区别gcc 命令的常用选项编译优化选项-O 优化-O1优化-O2-O0-Os-Ofast-Og-Oz-O 选项控制特定的优化 WarningsReference>>>>> 欢迎关注公众号【三戒纪元】 <<<<< 编译步骤 gcc 、…

Shell编程之正则表达式(非常详细)

正则表达式 1.通配符和正则表达式的区别2.基本正则表达式2.1 元字符 &#xff08;字符匹配)2.2 表示匹配次数2.4 位置锚定2.5 分组 和 或者 3.扩展正则表达式4.部分文本处理工具4.1 tr 命令4.2 cut命令4.3 sort命令4.4 uniq命令 1.通配符和正则表达式的区别 通配符一般用于文件…

Ansible从入门到精通【六】

大家好&#xff0c;我是早九晚十二&#xff0c;目前是做运维相关的工作。写博客是为了积累&#xff0c;希望大家一起进步&#xff01; 我的主页&#xff1a;早九晚十二 专栏名称&#xff1a;Ansible从入门到精通 立志成为ansible大佬 ansible templates 模板&#xff08;templa…

闭环控制方法及其应用:优缺点、场景和未来发展

闭环控制是一种基本的控制方法&#xff0c;它通过对系统输出与期望值之间的误差进行反馈&#xff0c;从而调整系统输入&#xff0c;使系统输出更加接近期望值。闭环控制的主要目标是提高系统的稳定性、精确性和鲁棒性。在实际应用中&#xff0c;闭环控制有多种方法&#xff0c;…

释放AI创作潜能:从大模型训练到高产力应用

文章目录 每日一句正能量前言什么是人工智能生成内容&#xff08;AIGC&#xff09;人工智能生成内容&#xff08;AIGC&#xff09;能做什么为什么要用人工智能生成内容&#xff08;AIGC&#xff09;创作成果用Java实现冒泡排序算法学生信息收集系统学生请假管理系统需求分析教务…

苹果电脑图像元数据编辑器:MetaImage for Mac

MetaImage for Mac是一款功能强大的照片元数据编辑器&#xff0c;它可以帮助用户编辑并管理照片的元数据信息&#xff0c;包括基本信息和扩展信息。用户可以根据需要进行批量处理&#xff0c;方便快捷地管理大量照片。 MetaImage for Mac还提供了多种导入和导出格式&#xff0…

东南大学齿轮箱故障诊断(Python代码,MSCNN结合LSTM结合注意力机制模型,代码有注释)

运行代码要求&#xff1a; 代码运行环境要求&#xff1a;Keras版本>2.4.0&#xff0c;python版本>3.6.0 1.东南大学采集数据平台&#xff1a; 数据 该数据集包含2个子数据集&#xff0c;包括轴承数据和齿轮数据&#xff0c;这两个子数据集都是在传动系动力学模拟器&am…

基于Matlab实现心电信号小波特征提取和对应疾病识别仿真(附上源码+数据集)

本文基于Matlab平台&#xff0c;研究了心电信号的小波特征提取方法&#xff0c;并应用于心电信号疾病识别仿真实验中。首先&#xff0c;介绍了心电信号的基本特征和常见的心电疾病。然后&#xff0c;详细阐述了小波变换的原理和方法&#xff0c;并提出了一种基于小波分解和小波…

运维监控学习笔记3

DELL的IPMI页面的登录&#xff1a; 风扇的状态&#xff1a; 电源温度&#xff1a;超过70度就告警&#xff1a; 日志信息&#xff1a; 可以看到更换过磁盘。 iDRAC的设置 虚拟控制台&#xff1a;启动远程控制台&#xff1a; 可以进行远程控制。 机房工程师帮我们接远程控制&…

如何让ES低成本、高性能?滴滴落地ZSTD压缩算法的实践分享

前文分别介绍了滴滴自研的ES强一致性多活是如何实现的、以及如何提升ES的性能潜力。由于滴滴ES日志场景每天写入量在5PB-10PB量级&#xff0c;写入压力和业务成本压力大&#xff0c;为了提升ES的写入性能&#xff0c;我们让ES支持ZSTD压缩算法&#xff0c;本篇文章详细展开滴滴…

CCLINK IE 转MODBUS-RTU网关modbusrtu与485区别

远创智控YC-CCLKIE-RTU。这款产品的主要功能是将各种MODBUS-RTU、RS485、RS232设备接入到CCLINK IE FIELD BASIC网络中。 那么&#xff0c;这款通讯网关又有哪些特点呢&#xff1f;首先&#xff0c;它能够连接到CCLINK IE FIELD BASIC总线中作为从站使用&#xff0c;同时也能连…

Python Opencv实践 - 图像属性相关

import numpy as np import cv2 as cv import matplotlib.pyplot as pltimg cv.imread("../SampleImages/pomeranian.png", cv.IMREAD_COLOR) plt.imshow(img[:,:,::-1])#像素操作 pixel img[320,370] print(pixel)#只获取蓝色通道的值 pixel_blue img[320,370,0]…

JProfiler —CPU评测

当JProfiler测量方法调用的执行时间及其调用堆栈时&#xff0c;我们称之为“CPU评测”。这些数据以多种方式呈现。根据你试图解决的问题&#xff0c;其中一个或另一个演示将是最有帮助的。默认情况下不会记录CPU数据&#xff0c;您必须打开CPU记录才能捕获有趣的用例。 一、调…

Mac如何打开隐藏文件中Redis的配置文件redis.conf

Redis下载(通过⬇️博客下载的Redis默认路径为&#xff1a;/usr/local/etc) Redis下载 1.打开终端进入/usr文件夹 cd /usr 2.打开/local/文件夹 open local 3.找到redis.conf并打开,即可修改配置信息

《Zookeeper》源码分析(九)之选举通信网络

在上一篇文章中讲到QuorumCnxManager&#xff0c;它负责zookeeper服务器在选举期间最底层的网络通信&#xff0c;整个网络涉及到的类如下&#xff1a; 整个网络建立的过程如下&#xff1a; 选举前创建好QuorumCnxManager实例&#xff0c;并在QuorumCnxManager构造函数中创建好…

解决selenium的“can‘t access dead object”错误

目录 问题描述 原因 解决方法 示例代码 资料获取方法 问题描述 在python执行过程中&#xff0c;提示selenium.common.exceptions.WebDriverException: Message: TypeError: cant access dead object 原因 原因是代码中用到了frame,获取元素前需要切换到frame才能定位到…

21 | 朝阳医院数据分析

朝阳医院2018年销售数据为例,目的是了解朝阳医院在2018年里的销售情况,通过对朝阳区医院的药品销售数据的分析,了解朝阳医院的患者的月均消费次数,月均消费金额、客单价以及消费趋势、需求量前几位的药品等。 import numpy as np from pandas import Series,DataFrame impo…

C++ 虚继承

C棱形继承 在 C 中&#xff0c;在使用 多继承 时&#xff0c;如果发生了如果类 A 派生出类 B 和类 C&#xff0c;类 D 继承自类 B 和类 C&#xff0c;这时候就发生了菱形继承。 如果发生了菱形继承&#xff0c;这个时候类 A 中的 成员变量 和 成员函数 继承到类 D 中变成了两…

约束综合中的逻辑互斥时钟(Logically Exclusive Clocks)

注&#xff1a;本文翻译自Constraining Logically Exclusive Clocks in Synthesis 逻辑互斥时钟的定义 逻辑互斥时钟是指设计中活跃&#xff08;activate&#xff09;但不彼此影响的时钟。常见的情况是&#xff0c;两个时钟作为一个多路选择器的输入&#xff0c;并根据sel信号…

【Linux】程序地址空间

程序地址空间 首先引入地址空间的作用什么是地址空间为什么要有地址空间 首先引入地址空间的作用 1 #include <stdio.h>2 #include <unistd.h>3 #include <stdlib.h>4 int g_val 100;6 int main()7 {8 pid_t id fork();9 if(id 0)10 {11 int cn…