资讯专栏INFORMATION COLUMN

xxl-job之调度中心启动源码分析(一)

Xufc / 693人阅读

摘要:之调度中心启动源码分析一首先从的配置看起,从以下配置可以看出,内部使用的是配置自动启动延时启动,应用启动成功后在启动覆盖中以数据库中已经存在的为准这个调度中心,在启动的时候,会做很多初始化的工作,比如执行器信

xxl-job之调度中心启动源码分析(一)

首先从spring的配置看起, 从以下配置可以看出,xxl内部使用的是quartz

spring配置

   
            
                
    
   
   



    
   
    
   
XxlJobDynamicScheduler

com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler 在启动的时候会做如下工作:

public void init() throws Exception {
    // 启动自动注册线程, 获取类型为自动注册的执行器信息,完成机器的自动注册与发现
    JobRegistryMonitorHelper.getInstance().start();
     
    // 启动失败日志监控线程
    JobFailMonitorHelper.getInstance().start();
     
    // admin-server(spring-mvc)
    NetComServerFactory.putService(AdminBiz.class, XxlJobDynamicScheduler.adminBiz);
    NetComServerFactory.setAccessToken(accessToken);
     
    // valid
    Assert.notNull(scheduler, "quartz scheduler is null");
    logger.info(">>>>>>>>> init xxl-job admin success.");
}
JobRegistryMonitorHelper

JobRegistryMonitorHelper.getInstance().start() 详细代码如下:

public void start(){
   //创建一个线程
   registryThread = new Thread(new Runnable() {
      @Override
      public void run() {
         // 当toStop 为false时进入该循环。
         while (!toStop) {
            try {
               // 获取类型为自动注册的执行器地址列表
               List groupList = XxlJobDynamicScheduler.xxlJobGroupDao.findByAddressType(0);
               if (CollectionUtils.isNotEmpty(groupList)) {
     
                  // 删除 90秒之内没有更新信息的注册机器, 90秒没有心跳信息返回,代表机器已经出现问题,故移除
                  XxlJobDynamicScheduler.xxlJobRegistryDao.removeDead(RegistryConfig.DEAD_TIMEOUT);
     
                  // fresh online address (admin/executor)
                  HashMap> appAddressMap = new HashMap>();
                  // 查询在90秒之内有过更新的机器列表
                  List list = XxlJobDynamicScheduler.xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT);
                  if (list != null) {
                     //循环注册机器列表,  根据执行器不同,将这些机器列表区分拿出来
                     for (XxlJobRegistry item: list) {
                        // 判断该机器注册信息RegistryGroup ,RegistType 是否是EXECUTOR , EXECUTOR 代表该机器是注册到执行器上面的
                        // RegistType  分为两种, ADMIN 和EXECUTOR
                        if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
                           // 获取注册的执行器 KEY  (也就是执行器)
                           String appName = item.getRegistryKey();
                           List registryList = appAddressMap.get(appName);
                           if (registryList == null) {
                              registryList = new ArrayList();
                           }
     
                           if (!registryList.contains(item.getRegistryValue())) {
                              registryList.add(item.getRegistryValue());
                           }
                           // 收集 机器信息,根据执行器做区分
                           appAddressMap.put(appName, registryList);
                        }
                     }
                  }
     
                  //  遍历执行器列表
                  for (XxlJobGroup group: groupList) {
                     // 通过执行器的APP_NAME  拿出他下面的集群机器地址
                     List registryList = appAddressMap.get(group.getAppName());
                     String addressListStr = null;
                     if (CollectionUtils.isNotEmpty(registryList)) {
                        Collections.sort(registryList);
                        // 转为为String, 通过逗号分隔
                        addressListStr = StringUtils.join(registryList, ",");
                     }
                     group.setAddressList(addressListStr);
                     // 将 这个执行器的 集群机器地址列表,写入到数据库
                     XxlJobDynamicScheduler.xxlJobGroupDao.update(group);
                  }
               }
            } catch (Exception e) {
               logger.error("job registry instance error:{}", e);
            }
            try {
               TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
            } catch (InterruptedException e) {
               logger.error("job registry instance error:{}", e);
            }
         }
      }
   });
   registryThread.setDaemon(true);
   //启动线程
   registryThread.start();
}
JobFailMonitorHelper

JobFailMonitorHelper.getInstance().start(); 详细代码如下:

//JobFailMonitorHelper.java
public void start(){
   // 启动线程
 monitorThread = new Thread(new Runnable() {

 @Override
 public void run() {
         // monitor
 while (!toStop) {
            try {
               List jobLogIdList = new ArrayList();
               // 从队列中拿出所有可用的 jobLogIds
                int drainToNum = JobFailMonitorHelper.instance.queue.drainTo(jobLogIdList);
               if (CollectionUtils.isNotEmpty(jobLogIdList)) {
                  for (Integer jobLogId : jobLogIdList) {
                     if (jobLogId==null || jobLogId==0) {
                        continue;
                     }
                     //从数据库跟以前有日志信息
                    XxlJobLog log = XxlJobDynamicScheduler.xxlJobLogDao.load(jobLogId);
                     if (log == null) {
                        continue;
                     }
                     //任务触发成功, 但是JobHandle 还没有返回结果
                    if (IJobHandler.SUCCESS.getCode() == log.getTriggerCode() && log.getHandleCode() == 0) {
                        //将 JobLogId 放入队列 , 继续监控
                         JobFailMonitorHelper.monitor(jobLogId);
                        logger.info(">>>>>>>>>>> job monitor, job running, JobLogId:{}", jobLogId);
                     } else if (IJobHandler.SUCCESS.getCode() == log.getHandleCode()) {
                        // job success, pass
                        logger.info(">>>>>>>>>>> job monitor, job success, JobLogId:{}", jobLogId);
                     } else if (IJobHandler.FAIL.getCode() == log.getTriggerCode()
                           || IJobHandler.FAIL.getCode() == log.getHandleCode()
                           || IJobHandler.FAIL_RETRY.getCode() == log.getHandleCode() ) {
                        // 任务执行失败, 执行发送邮件等预警措施
                        failAlarm(log);
                        logger.info(">>>>>>>>>>> job monitor, job fail, JobLogId:{}", jobLogId);
                     } else {
                        JobFailMonitorHelper.monitor(jobLogId);
                        logger.info(">>>>>>>>>>> job monitor, job status unknown, JobLogId:{}", jobLogId);
                     }
                  }
               }
               // 停顿一下
                TimeUnit.SECONDS.sleep(10);
            } catch (Exception e) {
               logger.error("job monitor error:{}", e);
            }
         }

   });
   monitorThread.setDaemon(true);
   monitorThread.start();
}

以上 是xxl-job 在启动的时候做的操作, 主要是启动两个线程,

用来监控自动注册上来的机器,达到自动注册的目的
监控任务的执行状态, 如若失败,则发送邮件预警

xxl-job 是基于quartz 进行的二次开发,在系统启动的时候,quartz框架会自动去数据库读取相关的配置信息,载入相关定时器信息

文章版权归作者所有,未经允许请勿转载,若此文章存在违规行为,您可以联系管理员删除。

转载请注明本文地址:https://www.ucloud.cn/yun/71473.html

相关文章

  • 分布式任务调度平台的自动化部署

    摘要:分布式任务调度平台,美团点评员工许雪里开发,其核心设计目标是开发迅速学习简单轻量级易扩展。源码地址二本文简介主要是利用成熟的部署工具,结合开源的分布式任务调度框架作为工程,快速搭建自己的持续集成项目,其他项目可以类似构建。 一、概述 1、自动化部署能简化开发过程的代码管理,让开发人员把更多的时间专注于业务实现, 简化繁琐的上线流程和操作步骤,做到项目的快速打包和部署,减少人...

    tomato 评论0 收藏0
  • 分布式任务调度平台XXL-JOB

    摘要:分布式任务调度平台一简介概述是一个轻量级分布式任务调度框架,其核心设计目标是开发迅速学习简单轻量级易扩展。 《分布式任务调度平台XXL-JOB》 showImg(https://segmentfault.com/img/remote/1460000010402214);showImg(https://segmentfault.com/img/remote/146000001040221...

    huangjinnan 评论0 收藏0
  • XXL-JOB v2.0.0 发布,分布式任务调度平台

    v2.0.0 Release Notes 1、调度中心迁移到 springboot; 2、底层通讯组件迁移至 xxl-rpc; 3、容器化:提供官方docker镜像,并实时更新推送dockerhub(docker pull xuxueli/xxl-job-admin),进一步实现产品开箱即用; 4、新增无框架执行器Sample示例项目 xxl-job-executor-sample-framel...

    Edison 评论0 收藏0
  • XXL-JOB v2.1.0,分布式任务调度平台

    摘要:自研调度组件,移除依赖一方面是为了精简系统降低冗余依赖,另一方面是为了提供系统的可控度与稳定性触发单节点周期性触发,运行事件如调度集群竞争,负载方式协同处理,锁竞争更新触发信息推送时间轮锁释放锁竞争底层表结构重构移除张相关表,并对现 v2.0.2 Release Notes 1、自研调度组件,移除quartz依赖:一方面是为了精简系统降低冗余依赖,另一方面是为了提供系统的可控度与稳...

    GitChat 评论0 收藏0

发表评论

0条评论

最新活动
阅读需要支付1元查看
<