摘要：之調(diào)度中心啟動(dòng)源碼分析一首先從的配置看起，從以下配置可以看出，內(nèi)部使用的是配置自動(dòng)啟動(dòng)延時(shí)啟動(dòng)，應(yīng)用啟動(dòng)成功后在啟動(dòng)覆蓋中以數(shù)據(jù)庫(kù)中已經(jīng)存在的為準(zhǔn)這個(gè)調(diào)度中心，在啟動(dòng)的時(shí)候，會(huì)做很多初始化的工作，比如執(zhí)行器信

首先從spring的配置看起， 從以下配置可以看出，xxl內(nèi)部使用的是quartz

com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler 在啟動(dòng)的時(shí)候會(huì)做如下工作：

public void init() throws Exception {
    // 啟動(dòng)自動(dòng)注冊(cè)線程， 獲取類型為自動(dòng)注冊(cè)的執(zhí)行器信息，完成機(jī)器的自動(dòng)注冊(cè)與發(fā)現(xiàn)
    JobRegistryMonitorHelper.getInstance().start();
     
    // 啟動(dòng)失敗日志監(jiān)控線程
    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.getInstance().start() 詳細(xì)代碼如下：

public void start(){
   //創(chuàng)建一個(gè)線程
   registryThread = new Thread(new Runnable() {
      @Override
      public void run() {
         // 當(dāng)toStop 為false時(shí)進(jìn)入該循環(huán)。
         while (!toStop) {
            try {
               // 獲取類型為自動(dòng)注冊(cè)的執(zhí)行器地址列表
               List groupList = XxlJobDynamicScheduler.xxlJobGroupDao.findByAddressType(0);
               if (CollectionUtils.isNotEmpty(groupList)) {
     
                  // 刪除 90秒之內(nèi)沒(méi)有更新信息的注冊(cè)機(jī)器， 90秒沒(méi)有心跳信息返回，代表機(jī)器已經(jīng)出現(xiàn)問(wèn)題，故移除
                  XxlJobDynamicScheduler.xxlJobRegistryDao.removeDead(RegistryConfig.DEAD_TIMEOUT);
     
                  // fresh online address (admin/executor)
                  HashMap> appAddressMap = new HashMap>();
                  // 查詢?cè)?0秒之內(nèi)有過(guò)更新的機(jī)器列表
                  List list = XxlJobDynamicScheduler.xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT);
                  if (list != null) {
                     //循環(huán)注冊(cè)機(jī)器列表，  根據(jù)執(zhí)行器不同，將這些機(jī)器列表區(qū)分拿出來(lái)
                     for (XxlJobRegistry item: list) {
                        // 判斷該機(jī)器注冊(cè)信息RegistryGroup ，RegistType 是否是EXECUTOR , EXECUTOR 代表該機(jī)器是注冊(cè)到執(zhí)行器上面的
                        // RegistType  分為兩種， ADMIN 和EXECUTOR
                        if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
                           // 獲取注冊(cè)的執(zhí)行器 KEY  （也就是執(zhí)行器）
                           String appName = item.getRegistryKey();
                           List registryList = appAddressMap.get(appName);
                           if (registryList == null) {
                              registryList = new ArrayList();
                           }
     
                           if (!registryList.contains(item.getRegistryValue())) {
                              registryList.add(item.getRegistryValue());
                           }
                           // 收集 機(jī)器信息，根據(jù)執(zhí)行器做區(qū)分
                           appAddressMap.put(appName, registryList);
                        }
                     }
                  }
     
                  //  遍歷執(zhí)行器列表
                  for (XxlJobGroup group: groupList) {
                     // 通過(guò)執(zhí)行器的APP_NAME  拿出他下面的集群機(jī)器地址
                     List registryList = appAddressMap.get(group.getAppName());
                     String addressListStr = null;
                     if (CollectionUtils.isNotEmpty(registryList)) {
                        Collections.sort(registryList);
                        // 轉(zhuǎn)為為String，　通過(guò)逗號(hào)分隔
                        addressListStr = StringUtils.join(registryList, ",");
                     }
                     group.setAddressList(addressListStr);
                     // 將 這個(gè)執(zhí)行器的 集群機(jī)器地址列表，寫入到數(shù)據(jù)庫(kù)
                     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);
   //啟動(dòng)線程
   registryThread.start();
}

JobFailMonitorHelper.getInstance().start(); 詳細(xì)代碼如下：

//JobFailMonitorHelper.java
public void start(){
   // 啟動(dòng)線程
 monitorThread = new Thread(new Runnable() {

 @Override
 public void run() {
         // monitor
 while (!toStop) {
            try {
               List jobLogIdList = new ArrayList();
               // 從隊(duì)列中拿出所有可用的 jobLogIds
                int drainToNum = JobFailMonitorHelper.instance.queue.drainTo(jobLogIdList);
               if (CollectionUtils.isNotEmpty(jobLogIdList)) {
                  for (Integer jobLogId : jobLogIdList) {
                     if (jobLogId==null || jobLogId==0) {
                        continue;
                     }
                     //從數(shù)據(jù)庫(kù)跟以前有日志信息
                    XxlJobLog log = XxlJobDynamicScheduler.xxlJobLogDao.load(jobLogId);
                     if (log == null) {
                        continue;
                     }
                     //任務(wù)觸發(fā)成功， 但是JobHandle 還沒(méi)有返回結(jié)果
                    if (IJobHandler.SUCCESS.getCode() == log.getTriggerCode() && log.getHandleCode() == 0) {
                        //將 JobLogId 放入隊(duì)列 ， 繼續(xù)監(jiān)控
                         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() ) {
                        // 任務(wù)執(zhí)行失敗， 執(zhí)行發(fā)送郵件等預(yù)警措施
                        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 在啟動(dòng)的時(shí)候做的操作， 主要是啟動(dòng)兩個(gè)線程，

用來(lái)監(jiān)控自動(dòng)注冊(cè)上來(lái)的機(jī)器，達(dá)到自動(dòng)注冊(cè)的目的
監(jiān)控任務(wù)的執(zhí)行狀態(tài)， 如若失敗，則發(fā)送郵件預(yù)警

xxl-job 是基于quartz 進(jìn)行的二次開(kāi)發(fā)，在系統(tǒng)啟動(dòng)的時(shí)候，quartz框架會(huì)自動(dòng)去數(shù)據(jù)庫(kù)讀取相關(guān)的配置信息,載入相關(guān)定時(shí)器信息