XXL-JOB-01

xxx-job 源碼解讀（一）

2018年03月07日 14:43:56

閱讀數：52

1.  調度中心啓動源碼分析

 首先從spring的配置看起， 從如下配置能夠看出，xxl內部使用的是quartz

spring配置

<bean id="quartzScheduler" lazy-init="false" class="org.springframework.scheduling.quartz.SchedulerFactoryBean">    <property name="dataSource" ref="dataSource" />    <property name="autoStartup" value="true" />         <!--自動啓動 -->    <property name="startupDelay" value="20" />             <!--延時啓動，應用啓動成功後在啓動 -->    <property name="overwriteExistingJobs" value="true" /> <!--覆蓋DB中JOB：true、以數據庫中已經存在的爲準：false -->    <property name="applicationContextSchedulerContextKey"  value="applicationContextKey" />    <property name="configLocation" value="classpath:quartz.properties"/> </bean> <!-- 這個調度中心，在啓動的時候，會作不少初始化的工做 ，好比：執行器信息，註冊機器列表等信息 --> <bean id="xxlJobDynamicScheduler" class="com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler" init-method="init" destroy-method="destroy" >     <!-- 配置調度中心的名稱 -->    <property name="scheduler" ref="quartzScheduler"/>     <!-- 用於調度中心和執行器之間通訊的時候作數據加密 -->    <property name="accessToken" value="${xxl.job.accessToken}" /> </bean>

com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler  在啓動的時候會作以下工做：

 

XxlJobDynamicScheduler

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15

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.getInstance().start()   詳細代碼以下：

 

 

JobRegistryMonitorHelper

public void start(){    //建立一個線程    registryThread = new Thread(new Runnable() {       @Override       public void run() {          // 當toStop 爲false時進入該循環。          while (!toStop) {             try {                // 獲取類型爲自動註冊的執行器地址列表                List<XxlJobGroup> groupList = XxlJobDynamicScheduler.xxlJobGroupDao.findByAddressType(0);                if (CollectionUtils.isNotEmpty(groupList)) {                     // 刪除 90秒以內沒有更新信息的註冊機器， 90秒沒有心跳信息返回，表明機器已經出現問題，故移除                   XxlJobDynamicScheduler.xxlJobRegistryDao.removeDead(RegistryConfig.DEAD_TIMEOUT);                     // fresh online address (admin/executor)                   HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();                   // 查詢在90秒以內有過更新的機器列表                   List<XxlJobRegistry> 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<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);                         }                      }                   }                     //  遍歷執行器列表                   for (XxlJobGroup group: groupList) {                      // 經過執行器的APP_NAME  拿出他下面的集羣機器地址                      List<String> 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.getInstance().start(); 詳細代碼以下：

JobFailMonitorHelper

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53

public void start(){    // 啓動線程  monitorThread = new Thread(new Runnable() {    @Override  public void run() {          // monitor  while (!toStop) {             try {                List<Integer> jobLogIdList = new ArrayList<Integer>();                // 從隊列中拿出全部可用的 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 在啓動的時候作的操做，  主要是啓動兩個線程，  

1. 用來監控自動註冊上來的機器，達到自動註冊的目的
2. 監控任務的執行狀態， 如若失敗，則發送郵件預警
    

xxl-job 是基於quartz 進行的二次開發，在系統啓動的時候，quartz框架會自動去數據庫讀取相關的配置信息,載入相關定時器信息