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.javapublic 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框架会自动去数据库读取相关的配置信息,载入相关定时器信息