在上文中通过一个简单的例子,给大家简单的介绍了xxl-job的快速入门过程。xxl-job作为分布式调度任务框架,底层是如何进行调度和管理的呢,本文将带着这些问题深入了解xxl-job的运行原理。
1. xxl-job初识
1.1 xxl-job介绍
xxl-job 是大众点评大佬徐雪里开源的一款分布式任务调度框架,具有简单易用、轻量级、可扩展的特点。相比于Spring Task, Quartz,xxl-job有记录执行日志和运行大盘,方便开发人员和运维人员更好的管理任务。
1.2 名词解释
- 调度中心:进行执行器的自动注册,任务的调度管理,调度日志的记录等操作。
- 执行器:执行器相当于一个应用服务,通过appName唯一标识。
- 任务:一个任务即是最小的调度单元,任务必须隶属于某个执行器,任务的调度支持cron和固定速度配置。
- jobHandler:任务处理器,在调度任务时会回调开发定义的接口,这个接口就是jobHandler,在Spring以bean形式存在。
1.3 系统架构
xxl-job 2.1版本的架构图如下
- 调度中心包含任务管理,执行器管理,日志管理登等几大模块。
- 执行器包含注册线程自动注册,回调线程执行任务,执行器将调度添加到调度队列中
这个架构图涉及的东西还是很多的,一眼看下来还是不太清楚流程处理。比如:
- 执行器是如何自动注册到调度中心的?
- 调度中心是如何管理执行器的?
- 调度中心是如何触发任务的?
- 任务是怎么回调jobHandler的?
- 任务执行超时会有什么应对策略吗?
...
本文将带着这些疑问来深入了解xxl-job的工作原理,逐个击破。
2. 执行器注册到调度中心
在上面介绍过,执行器相当于一个应用服务,需要注册到调度中心这样才能由调度中心统一管理。xxl-job没有选择zookeeper、Eureka或者Nacos作为注册中心,而是直接用调度中心作为注册中心。
2.1 XxlJobConfig 执行器配置
执行器自动注册应当Spring在初始化相关bean的时候完成自动注册,因此从XxlJobConfig
这个配置类开始了解。
在此新增XxlJobSpringExecutor
执行器,这个执行器会在Spring Bean容器中进行管理,和普通bean的生命周期一致。
@Configuration
public class XxlJobConfig {
private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);
// 调度中心地址
@Value("${xxl.job.admin.addresses}")
private String adminAddresses;
// 访问token
@Value("${xxl.job.accessToken}")
private String accessToken;
// 执行器应用名
@Value("${xxl.job.executor.appname}")
private String appname;
@Value("${xxl.job.executor.address}")
private String address;
@Value("${xxl.job.executor.ip}")
private String ip;
@Value("${xxl.job.executor.port}")
private int port;
// 日志路径
@Value("${xxl.job.executor.logpath}")
private String logPath;
// 日志保留天数
@Value("${xxl.job.executor.logretentiondays}")
private int logRetentionDays;
@Bean
public XxlJobSpringExecutor xxlJobExecutor() {
logger.info(">>>>>>>>>>> xxl-job config init.");
logger.info(">>>>>>>>>>> xxl-job adminAddresses {}. ", adminAddresses);
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;
}
/**
* 针对多网卡、容器内部署等情况,可借助 "spring-cloud-commons" 提供的 "InetUtils" 组件灵活定制注册IP;
*
* 1、引入依赖:
* <dependency>
* <groupId>org.springframework.cloud</groupId>
* <artifactId>spring-cloud-commons</artifactId>
* <version>${version}</version>
* </dependency>
*
* 2、配置文件,或者容器启动变量
* spring.cloud.inetutils.preferred-networks: 'xxx.xxx.xxx.'
*
* 3、获取IP
* String ip_ = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress();
*/
}
复制代码
2.2 XxlJobSpringExecutor 执行器
XxlJobSpringExecutor
继承了 XxlJobExecutor
基础类,实现了ApplicationContextAware
,SmartInitializingSingleton
,DisposableBean
等接口。
- XxlJobExecutor : 执行器的基本实现,定义了执行器的启动和停止方法,包含生成调度中心代理集合,初始化回调线程,初始化执行器内置容器等。
- ApplicationContextAware:在Spring生命周期中,
ApplicationContextAware
是容器刷新的钩子函数,通过此接口可以获取ApplicationContext
的应用上下文 - SmartInitializingSingleton:这个接口可以在初始化单例bean进行一些后置操作,因此将bean注册到Spring容器时,会调用
afterSingletonsInstantiated
方法进行一些初始化操作 - DisposableBean:在bean被容器销毁时会调用的钩子函数,进行一些销毁操作
在旧版本中,执行器需要继承IJobhandler,实现execute方法,在新版本中只需要在方法中添加@xxljob
注解即可,因此一个类可以注册多个任务,具体实现是在initJobHandlerMethodRepository
方法中执行的。initJobHandlerMethodRepository
的执行流程如下:
- 从applicationContext中获取所有有注解的bean
- 获取每个bean的所有的带有XxlJob注解的方法,被XxlJob注解修饰的的方法是一个任务
- 遍历所有的任务
- 注册任务处理器,
registJobHandler
调用的是父类XxlJobExecutor
的registJobHandler
/**
* xxl-job executor (for spring)
*
* @author xuxueli 2018-11-01 09:24:52
*/
public class XxlJobSpringExecutor extends XxlJobExecutor implements ApplicationContextAware, SmartInitializingSingleton, DisposableBean {
private static final Logger logger = LoggerFactory.getLogger(XxlJobSpringExecutor.class);
// start
@Override
public void afterSingletonsInstantiated() {
// init JobHandler Repository
/*initJobHandlerRepository(applicationContext);*/
// init JobHandler Repository (for method)
// 初始化任务处理器仓库 (针对于方法处理器)
initJobHandlerMethodRepository(applicationContext);
// refresh GlueFactory
GlueFactory.refreshInstance(1);
// super start
try {
// 调用 XxlJobExecutor 的start方法
super.start();
} catch (Exception e) {
throw new RuntimeException(e);
}
}
// destroy
@Override
public void destroy() {
super.destroy();
}
private void initJobHandlerMethodRepository(ApplicationContext applicationContext) {
if (applicationContext == null) {
return;
}
// init job handler from method
// 1. 从applicationContext中获取所有有注解的bean
String[] beanDefinitionNames = applicationContext.getBeanNamesForType(Object.class, false, true);
for (String beanDefinitionName : beanDefinitionNames) {
Object bean = applicationContext.getBean(beanDefinitionName);
// 2. 获取每个bean的所有的带有XxlJob注解的方法,被XxlJob注解修饰的的方法是一个任务
Map<Method, XxlJob> annotatedMethods = null; // referred to :org.springframework.context.event.EventListenerMethodProcessor.processBean
try {
annotatedMethods = MethodIntrospector.selectMethods(bean.getClass(),
new MethodIntrospector.MetadataLookup<XxlJob>() {
@Override
public 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;
}
// 3.遍历所有的任务
for (Map.Entry<Method, XxlJob> methodXxlJobEntry : annotatedMethods.entrySet()) {
Method executeMethod = methodXxlJobEntry.getKey();
XxlJob xxlJob = methodXxlJobEntry.getValue();
if (xxlJob == null) {
continue;
}
String name = xxlJob.value();
if (name.trim().length() == 0) {
throw new RuntimeException("xxl-job method-jobhandler name invalid, for[" + bean.getClass() + "#" + executeMethod.getName() + "] .");
}
if (loadJobHandler(name) != null) {
throw new RuntimeException("xxl-job jobhandler[" + name + "] naming conflicts.");
}
// execute method
/*if (!(method.getParameterTypes().length == 1 && method.getParameterTypes()[0].isAssignableFrom(String.class))) {
throw new RuntimeException("xxl-job method-jobhandler param-classtype invalid, for[" + bean.getClass() + "#" + method.getName() + "] , " +
"The correct method format like " public ReturnT<String> execute(String param) " .");
}
if (!method.getReturnType().isAssignableFrom(ReturnT.class)) {
throw new RuntimeException("xxl-job method-jobhandler return-classtype invalid, for[" + bean.getClass() + "#" + method.getName() + "] , " +
"The correct method format like " public ReturnT<String> execute(String param) " .");
}*/
executeMethod.setAccessible(true);
// init and destory
Method initMethod = null;
Method destroyMethod = null;
if (xxlJob.init().trim().length() > 0) {
try {
initMethod = bean.getClass().getDeclaredMethod(xxlJob.init());
initMethod.setAccessible(true);
} catch (NoSuchMethodException e) {
throw new RuntimeException("xxl-job method-jobhandler initMethod invalid, for[" + bean.getClass() + "#" + executeMethod.getName() + "] .");
}
}
if (xxlJob.destroy().trim().length() > 0) {
try {
destroyMethod = bean.getClass().getDeclaredMethod(xxlJob.destroy());
destroyMethod.setAccessible(true);
} catch (NoSuchMethodException e) {
throw new RuntimeException("xxl-job method-jobhandler destroyMethod invalid, for[" + bean.getClass() + "#" + executeMethod.getName() + "] .");
}
}
// 4. 注册任务处理器
// registry jobhandler
registJobHandler(name, new MethodJobHandler(bean, executeMethod, initMethod, destroyMethod));
}
}
}
// ---------------------- applicationContext ----------------------
private static ApplicationContext applicationContext;
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
this.applicationContext = applicationContext;
}
public static ApplicationContext getApplicationContext() {
return applicationContext;
}
}
复制代码
- MethodJobHandler 方法代理
MethodJobHandler
继承了 IJobHandler
,通过反射的方式调用目标对象的execute
、initMethod
、destroyMethod
操作。
/**
* @author xuxueli 2019-12-11 21:12:18
*/
public class MethodJobHandler extends IJobHandler {
// 被代理的目标对象
private final Object target;
// 任务执行方法
private final Method method;
// 初始化方法
private Method initMethod;
// 销毁方法
private Method destroyMethod;
public MethodJobHandler(Object target, Method method, Method initMethod, Method destroyMethod) {
this.target = target;
this.method = method;
this.initMethod = initMethod;
this.destroyMethod = destroyMethod;
}
@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);
}
}
@Override
public void init() throws Exception {
if(initMethod != null) {
initMethod.invoke(target);
}
}
@Override
public void destroy() throws Exception {
if(destroyMethod != null) {
destroyMethod.invoke(target);
}
}
@Override
public String toString() {
return super.toString()+"["+ target.getClass() + "#" + method.getName() +"]";
}
}
复制代码
2.3 XxlJobExecutor注册
XxlJobExecutor
的 registJobHandler
方法就是将 每个jobHandler的名字和MethodJobHandler
以key-value的形式放在一个线程安全的map里面,缓存在内存中。
// ---------------------- job handler repository ----------------------
private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
public static IJobHandler loadJobHandler(String name){
return jobHandlerRepository.get(name);
}
public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){
logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
return jobHandlerRepository.put(name, jobHandler);
}
复制代码
2.4 XxlJobExecutor 初始化
XxlJobSpringExecutor
中会调用 XxlJobExecutor
中的start
方法,start
方法主要做了以下内容:
- 初始化日志路径
- 初始化admin调度中心代理,通过配置的adminAdress,为每个调度中心生成一个AdminBiz代理类,并放到adminBizList中作为调度中心集合
- 初始化日志文件清理线程,这个清理线程会在后台定期清理那些过期的日志文件
- 初始化触发回调线程
- 初始化执行器内置容器
关于注册相关代码如下:
AdminBiz
定义了回调,注册,移除注册等方法,定义了执行器和调度中心的交互方式
- 执行器启动调用注册接口注册到调度中心
- 调度中心回调执行器的任务方法
- 执行器退出,通知调度中心移除注册信息
initAdminBizList
方法就是识别配置的调度中心地址,生成AdminBizClient
代理类,并用一个List集合进行缓存。
// ---------------------- admin-client (rpc invoker) ----------------------
private static List<AdminBiz> adminBizList;
private void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
if (adminAddresses!=null && adminAddresses.trim().length()>0) {
for (String address: adminAddresses.trim().split(",")) {
if (address!=null && address.trim().length()>0) {
AdminBiz adminBiz = new AdminBizClient(address.trim(), accessToken);
if (adminBizList == null) {
adminBizList = new ArrayList<AdminBiz>();
}
adminBizList.add(adminBiz);
}
}
}
}
public static List<AdminBiz> getAdminBizList(){
return adminBizList;
}
复制代码
重点来了,什么时候调用adminBiz
的registry
方法呢。在XxlJobExecutor``的initEmbedServer
方法中创建了内置容器,EmbedServer
的start
方法又会启动容器,并且会开始注册执行器。
// ---------------------- executor-server (rpc provider) ----------------------
private EmbedServer embedServer = null;
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!=null&&ip.trim().length()>0)?ip: IpUtil.getIp();
// generate address
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);
}
复制代码
在这里只贴出 EmbedServer
中和注册相关的部分代码,ExecutorRegistryThread
是一个执行器注册线程,
// ---------------------- registry ----------------------
public void startRegistry(final String appname, final String address) {
// start registry
ExecutorRegistryThread.getInstance().start(appname, address);
}
public void stopRegistry() {
// stop registry
ExecutorRegistryThread.getInstance().toStop();
}
复制代码
2.5 ExecutorRegistryThread 执行器注册线程
ExecutorRegistryThread
采用懒加载的方式实现单利模式,并且启动一个后台线程,每隔30s的心跳时间,不断地将执行器注册到调度中心。并在内置容器停止后,会调用取消注册方法
/**
* Created by xuxueli on 17/3/2.
*/
public class ExecutorRegistryThread {
private static Logger logger = LoggerFactory.getLogger(ExecutorRegistryThread.class);
private static ExecutorRegistryThread instance = new ExecutorRegistryThread();
public static ExecutorRegistryThread getInstance(){
return instance;
}
private Thread registryThread;
private volatile boolean toStop = false;
public void start(final String appname, final String address){
// valid
if (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() {
@Override
public void run() {
// 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 {
if (!toStop) {
TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
}
} catch (InterruptedException e) {
if (!toStop) {
logger.warn(">>>>>>>>>>> xxl-job, executor registry thread interrupted, error msg:{}", e.getMessage());
}
}
}
// registry remove
try {
RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appname, address);
for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
try {
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) {
if (!toStop) {
logger.info(">>>>>>>>>>> xxl-job registry-remove error, registryParam:{}", registryParam, e);
}
}
}
} catch (Exception e) {
if (!toStop) {
logger.error(e.getMessage(), e);
}
}
logger.info(">>>>>>>>>>> xxl-job, executor registry thread destory.");
}
});
registryThread.setDaemon(true);
registryThread.setName("xxl-job, executor ExecutorRegistryThread");
registryThread.start();
}
public void toStop() {
toStop = true;
// interrupt and wait
if (registryThread != null) {
registryThread.interrupt();
try {
registryThread.join();
} catch (InterruptedException e) {
logger.error(e.getMessage(), e);
}
}
}
}
复制代码
2.6 总结
执行器注册到调度中心的流程如下:
XxlJobConfig
配置类向Spring bean容器中注册了XxlJobSpringExecutor
,XxlJobSpringExecutor
这个bean实现了SmartInitializingSingleton
接口,会在实例化bean后进行初始化操作。XxlJobSpringExecutor
会调用XxlJobExecutor
的start
方法,进行一些初始化操作。其中包括将配置的调度中心包装成一个AdminBiz
代理类,这个代理类包含回调、注册、取消注册等方法,执行器就是通过这个代理类进行注册的。XxlJobExecutor
的start
方法还会初始化一个内置容器,这个内置容器在启动时会启动一个ExecutorRegistryThread
后台注册线程,以默认30s的心跳时间,不断的通过AdminBiz
代理向调度中心注册当前执行器的信息。
3. 调度中心管理注册信息
在上一节中讲述了执行器注册到调度中心的流程,最终是通过AdminBiz
代理来向调度中心注册。xxl-job-core中采用内部的xxl-rpc来实现远程方法调用,执行器作为一个客户端,通过XxlJobRemotingUtil
向调度中心发起远程调用,接口路由是api/registry
com.xxl.job.core.biz.client.AdminBizClient#registry
@Override
public ReturnT<String> registry(RegistryParam registryParam) {
return XxlJobRemotingUtil.postBody(addressUrl + "api/registry", accessToken, timeout, registryParam, String.class);
}
复制代码
3.1 JobApiController 远程方法
执行器注册最终会调用到xxl-job-admin
的JobApiController
暴露的Http接口,如果是回调、注册、注册移除等操作则会调用AdminBiz
的调度中心端实现
AdminBizImpl
。
/**
* Created by xuxueli on 17/5/10.
*/
@Controller
@RequestMapping("/api")
public class JobApiController {
@Resource
private AdminBiz adminBiz;
/**
* api
*
* @param uri
* @param data
* @return
*/
@RequestMapping("/{uri}")
@ResponseBody
@PermissionLimit(limit=false)
public ReturnT<String> api(HttpServletRequest request, @PathVariable("uri") String uri, @RequestBody(required = false) String data) {
// valid
if (!"POST".equalsIgnoreCase(request.getMethod())) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, HttpMethod not support.");
}
if (uri==null || uri.trim().length()==0) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "invalid request, uri-mapping empty.");
}
if (XxlJobAdminConfig.getAdminConfig().getAccessToken()!=null
&& XxlJobAdminConfig.getAdminConfig().getAccessToken().trim().length()>0
&& !XxlJobAdminConfig.getAdminConfig().getAccessToken().equals(request.getHeader(XxlJobRemotingUtil.XXL_JOB_ACCESS_TOKEN))) {
return new ReturnT<String>(ReturnT.FAIL_CODE, "The access token is wrong.");
}
// 如果是调用对应的回调,注册或者注册移除
// 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.");
}
}
}
复制代码
3.2 AdminBizImpl
AdminBizImpl
的registry
会使用JobRegistryHelper来实现注册信息的管理。
@Service
public class AdminBizImpl implements AdminBiz {
@Override
public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList) {
return JobCompleteHelper.getInstance().callback(callbackParamList);
}
@Override
public ReturnT<String> registry(RegistryParam registryParam) {
// 注册
return JobRegistryHelper.getInstance().registry(registryParam);
}
@Override
public ReturnT<String> registryRemove(RegistryParam registryParam) {
return JobRegistryHelper.getInstance().registryRemove(registryParam);
}
}
复制代码
3.3 JobRegistryHelper 注册
com.xxl.job.admin.core.thread.JobRegistryHelper#registry
方法会将注册信息通过异步的方式保存在xxl_job_registry表中
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() {
// 将注册信息保存在xxl_job_registry表中
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_registry表结构如下,主要存储注册的执行器以及对应的执行器地址,以及最新的心跳续期时间。
CREATE TABLE `xxl_job_registry` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`registry_group` varchar(50) NOT NULL,
`registry_key` varchar(255) NOT NULL,
`registry_value` varchar(255) NOT NULL,
`update_time` datetime DEFAULT NULL,
PRIMARY KEY (`id`),
KEY `i_g_k_v` (`registry_group`,`registry_key`,`registry_value`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
复制代码
3.4 JobRegistryHelper 注册信息更新
com.xxl.job.admin.core.thread.JobRegistryHelper#start
会启动注册线程池,并会开启一个监控线程,扫描那些已经失效的注册信息并进行清理。start方法中的监控线程主要做了以下内容:
- 获取自动注册的执行器
- 根据注册信息的更新时间判断是否超过了续期超时时间3*30s,清除已经失效的注册信息
- 更新每个执行器的注册信息列表,修改xxl_job_group的注册列表字段
public void start(){
// for registry or remove
registryOrRemoveThreadPool = new ThreadPoolExecutor(
2,
10,
30L,
TimeUnit.SECONDS,
new LinkedBlockingQueue<Runnable>(2000),
new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
return new Thread(r, "xxl-job, admin JobRegistryMonitorHelper-registryOrRemoveThreadPool-" + r.hashCode());
}
},
new RejectedExecutionHandler() {
@Override
public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
r.run();
logger.warn(">>>>>>>>>>> xxl-job, registry or remove too fast, match threadpool rejected handler(run now).");
}
});
// for monitor
registryMonitorThread = new Thread(new Runnable() {
@Override
public void run() {
while (!toStop) {
try {
// 1. 获取自动注册的执行器
// auto registry group
List<XxlJobGroup> groupList = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().findByAddressType(0);
if (groupList!=null && !groupList.isEmpty()) {
// 2. 根据注册信息的更新时间判断是否超过了续期超时时间3*30s,清除已经失效的注册信息
// remove dead address (admin/executor)
List<Integer> ids = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findDead(RegistryConfig.DEAD_TIMEOUT, new Date());
if (ids!=null && ids.size()>0) {
XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(ids);
}
// 3. 更新每个应用注册的信息,以appName为key,List为value缓存起来
// fresh online address (admin/executor)
HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
List<XxlJobRegistry> list = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findAll(RegistryConfig.DEAD_TIMEOUT, new Date());
if (list != null) {
for (XxlJobRegistry item: list) {
if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
String appname = item.getRegistryKey();
List<String> registryList = appAddressMap.get(appname);
if (registryList == null) {
registryList = new ArrayList<String>();
}
if (!registryList.contains(item.getRegistryValue())) {
registryList.add(item.getRegistryValue());
}
appAddressMap.put(appname, registryList);
}
}
}
// 4. 更新每个app的注册信息,修改xxl_job_group的注册列表字段
// fresh group address
for (XxlJobGroup group: groupList) {
List<String> registryList = appAddressMap.get(group.getAppname());
String addressListStr = null;
if (registryList!=null && !registryList.isEmpty()) {
Collections.sort(registryList);
StringBuilder addressListSB = new StringBuilder();
for (String item:registryList) {
addressListSB.append(item).append(",");
}
addressListStr = addressListSB.toString();
addressListStr = addressListStr.substring(0, addressListStr.length()-1);
}
group.setAddressList(addressListStr);
group.setUpdateTime(new Date());
XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().update(group);
}
}
} catch (Exception e) {
if (!toStop) {
logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
}
}
try {
TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
} catch (InterruptedException e) {
if (!toStop) {
logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
}
}
}
logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop");
}
});
registryMonitorThread.setDaemon(true);
registryMonitorThread.setName("xxl-job, admin JobRegistryMonitorHelper-registryMonitorThread");
registryMonitorThread.start();
}
public void toStop(){
toStop = true;
// stop registryOrRemoveThreadPool
registryOrRemoveThreadPool.shutdownNow();
// stop monitir (interrupt and wait)
registryMonitorThread.interrupt();
try {
registryMonitorThread.join();
} catch (InterruptedException e) {
logger.error(e.getMessage(), e);
}
}
复制代码
3.5 总结
- 调度中心对外暴露一个api HTTP接口,执行器在注册时候通过调用这个接口进行信息的注册
- 调度中心将注册信息保存在
xxl_job_registry
表中,该表主要存储执行器以及对应的地址信息 - 调度中心
JobRegistryHelper
会启动一个后台监控线程,及时从注册表中清理那些已经失效的注册信息,并同步到执行器表中的注册列表里面
4. 总结
限于文章篇幅原因,本文从源码角度了解xxl-job的执行器注册到调度中心以及调度中心管理注册信息的基本流程。
执行器注册到调度中心的流程如下:
XxlJobConfig
配置类向Spring bean容器中注册了XxlJobSpringExecutor
,XxlJobSpringExecutor
这个bean实现了SmartInitializingSingleton
接口,会在实例化bean后进行初始化操作。XxlJobSpringExecutor
会调用XxlJobExecutor
的start
方法,进行一些初始化操作。其中包括将配置的调度中心包装成一个AdminBiz
代理类,这个代理类包含回调、注册、取消注册等方法,执行器就是通过这个代理类进行注册的。XxlJobExecutor
的start
方法还会初始化一个内置容器,这个内置容器在启动时会启动一个ExecutorRegistryThread
后台注册线程,以默认30s的心跳时间,不断的通过AdminBiz
代理向调度中心注册当前执行器的信息。
调度中心管理注册信息的流程如下:
- 调度中心对外暴露一个api HTTP接口,执行器在注册时候通过调用这个接口进行信息的注册
- 调度中心将注册信息保存在
xxl_job_registry
表中,该表主要存储执行器以及对应的地址信息 - 调度中心
JobRegistryHelper
会启动一个后台监控线程,及时从注册表中清理那些已经失效的注册信息,并同步到执行器表中的注册列表里面
本文参考:xxl-job任务触发流程
近期评论