xxl-job-admin初始化工作

1.1 啟動時, 載入applicationcontext-xxl-job-admin.xml, 初始化spring容器

<!-- 這個排程中心,在啟動的時候,會做很多初始化的工作 ,比如:執行器資訊,註冊機器列表等資訊 -->
	<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>

Spring容器在初始化xxlJobDynamicScheduler時, 會呼叫inti-method指定的init()方法.

1.2 xxlJobDynamicScheduler類init()分析

 public void init() throws Exception {
         // admin排程中心專門開啟一個自動註冊執行緒registryThread, 獲取型別為自動註冊的執行器資訊,完成executor執行器的自動註冊與發現
        // 1 用來監控自動註冊上來的機器,達到自動註冊的目的
        // 2 admin排程中心的自動註冊就是, 將executor端每30秒心跳傳送過來的address寫入registry表,
        // 然後就是開啟一個自動registry執行緒, 每隔30s, 將registry表資料寫入到group表.
        // admin registry monitor run
        JobRegistryMonitorHelper.getInstance().start();

        // 啟動失敗日誌監控執行緒
        // 2 監控任務的執行狀態, 如若失敗,則傳送郵件預警
        // admin monitor run
        JobFailMonitorHelper.getInstance().start();

        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(); // registryThread執行緒: 用於發現註冊上來的executor執行器

JobFailMonitorHelper.getInstance().start(); // monitorThread執行緒: 查詢資料庫log表, 根據trigger_code欄位值(200成功, 500失敗), 用於監控executor執行器本次job執行是否成功.

NetComServerFactory.putService(AdminBiz.class, XxlJobDynamicScheduler.adminBiz); // 將Spring容器初始化的adminBizImpl物件放入本地快取
NetComServerFactory.setAccessToken(accessToken);// 設定訪問AccessToken

1.3 具體分析admin排程中心是如何實現對executor執行器暴露服務的自動發現 -- registry執行緒

public void start() {
        registryThread = new Thread(new Runnable() {

            @Override
            public void run() {
                while (!toStop) {
                    try {

                        // 查詢自動註冊的,address_type欄位型別為0的即為自動註冊的
                        // auto registry group
                        List<XxlJobGroup> groupList = XxlJobDynamicScheduler.xxlJobGroupDao.findByAddressType(0);
                        // groupList = [{"addressType":0,"appName":"xiongxianze-test1","id":2,"order":2,"title":"新分組,
                        // 註冊方式:自動"}]>
                        logger.info("==>groupList : {}", JSON.toJSONString(groupList));
                        if (CollectionUtils.isNotEmpty(groupList)) {

                            // 刪除 90秒之內沒有更新資訊的註冊機器, 90秒沒有心跳資訊返回,代表機器已經出現問題,故移除
                            // remove dead address (admin/executor)
                            XxlJobDynamicScheduler.xxlJobRegistryDao.removeDead(RegistryConfig.DEAD_TIMEOUT);

                            // fresh online address (admin/executor)
                            HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
                            // 正常心跳的executor端
                            List<XxlJobRegistry> list = XxlJobDynamicScheduler.xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT);
                            // list =
                            // [{"id":7,"registryGroup":"EXECUTOR","registryKey":"local-execute","registryValue":"10.0.254.5:9999","updateTime":1539711248000}]>
                            logger.info("==>fresh online address list:{}", JSON.toJSONString(list));

                            if (list != null) {
                                for (XxlJobRegistry item : list) {
                                    // 判斷該機器註冊資訊RegistryGroup ,RegistType 是否是EXECUTOR 
                                   
                                    if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
                                        // appName = local-execute
                                        String appName = item.getRegistryKey();
                                        // 獲取註冊的執行器 KEY (也就是執行器)
                                        List<String> registryList = appAddressMap.get(appName);
                                        // registryList1 = null
                                        logger.info("==>registryList1 :{}", JSON.toJSONString(registryList));
                                        if (registryList == null) {
                                            registryList = new ArrayList<String>();
                                        }

                                        if (!registryList.contains(item.getRegistryValue())) {
                                            registryList.add(item.getRegistryValue());
                                        }
                                        // registryList2 = ["10.0.254.5:9999"]
                                        logger.info("==>registryList2 :{}", JSON.toJSONString(registryList));
                                        // local-execute : ["10.0.254.5:9999"]
                                        appAddressMap.put(appName, registryList);
                                    }
                                }
                            }

                            // fresh group address
                            for (XxlJobGroup group : groupList) {
                                // xiongxianze-test1
                                // 從這裡就可以看出來, xxl.job.executor.appname properties檔案中該項配置,
                                // 必須與admin端管理執行器時AppName一致
                                List<String> registryList = appAddressMap.get(group.getAppName());
                                String addressListStr = null;
                                // 判斷是否已經有了address, 有了的話, 使用"," 拼接addressList
                                if (CollectionUtils.isNotEmpty(registryList)) {
                                    Collections.sort(registryList);
                                    addressListStr = StringUtils.join(registryList, ",");
                                }
                                // 所以addressListStr = null
                                group.setAddressList(addressListStr);
                                // group = {"addressType":0,"appName":"xiongxianze-test1","id":2,"order":2,"title":"新分組,
                                // 註冊方式:自動"}>
                                logger.info("==> group : {}", JSON.toJSONString(group));
                                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();
    }

上述程式碼片段主要的流程就是:

①XxlJobDynamicScheduler.xxlJobGroupDao.findByAddressType(0);//從資料庫中查詢address_type=0(即自動註冊)的全部分組資訊

②XxlJobDynamicScheduler.xxlJobRegistryDao.removeDead(RegistryConfig.DEAD_TIMEOUT);//刪除資料庫中該xxl_job_qrtz_trigger_registry表中已經過期, 沒有進行心跳註冊的executor執行器

③XxlJobDynamicScheduler.xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT);//查詢全部正常心跳註冊上來的executor執行器

④if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) //判斷是否是executor執行器

appAddressMap.put(appName, registryList);//這裡的appname指的是executor執行器在xxl-job-executor.properties檔案中配置的xxl.job.executor.appname=xxl-job-executor-sample, 而registryList即使每個executor執行器的ip和executor啟動jetty server埠號組成, 埠號由配置檔案中的xxl.job.executor.port=9999決定.

⑤ 接著的這段程式碼就是, 迴圈判斷admin排程中心group分組中是否有與executor執行器appname相同的名稱.     

// fresh group address
                            for (XxlJobGroup group : groupList) {
                                // xiongxianze-test1
                                // 從這裡就可以看出來, xxl.job.executor.appname properties檔案中該項配置,
                                // 必須與admin端管理執行器時AppName一致, 因為這裡是根據從資料庫的group表的appname作為key,
                                // 去取executor執行器端的放入快取中appname
                                List<String> registryList = appAddressMap.get(group.getAppName());
                                String addressListStr = null;
                                // 判斷是否已經有了address, 有了的話, 使用"," 拼接addressList
                                if (CollectionUtils.isNotEmpty(registryList)) {
                                    Collections.sort(registryList);
                                    addressListStr = StringUtils.join(registryList, ",");
                                }
                                // 所以addressListStr = null
                                group.setAddressList(addressListStr);
                                // group = {"addressType":0,"appName":"xiongxianze-test1","id":2,"order":2,"title":"新分組,
                                // 註冊方式:自動"}>
                                logger.info("==> group : {}", JSON.toJSONString(group));
                                XxlJobDynamicScheduler.xxlJobGroupDao.update(group);
                            }

這裡的關鍵程式碼就是, XxlJobDynamicScheduler.xxlJobGroupDao.update(group); 將executor執行器端註冊上來相同的xxl.job.executor.appname值, 放入了同一個組, 這樣admin排程中心, 就能從這個group組中選擇不同的排程策略排程本次執行的job. 同時也說明了這幾個executor執行器, 對應的是排程同一個job, 這樣admin排程中心就能選擇相應的排程策略排程job, 比如:first第一個executor執行本次job, last最後一個executor執行器執行本次排程, random隨機選擇一個executor執行進行本次排程, 或者是LFU最不經常, LRU最近最久未使用的執行器執行本次排程.

到這裡admin排程中心就實現了executor執行器註冊上來的服務暴露地址的自動發現.

1.4 具體分析admin排程中心是如何實現監控executor執行器執行結果的 -- monitorThread執行緒

@Override
            public void run() {

                // monitor
                while (!toStop) {
                    try {
                        // 從快取佇列中獲取資料, jogLogId
                        Integer jobLogId = JobFailMonitorHelper.instance.queue.take();
                        if (jobLogId != null && jobLogId > 0) {
                            // 根據joblogid從xxl_job_qrtz_trgger_log表中獲取job執行的log
                            XxlJobLog log = XxlJobDynamicScheduler.xxlJobLogDao.load(jobLogId);
                            if (log != null) {

                                // log.getHandleCode()預設值為0, 而且返回值200, job正在執行中
                                if (ReturnT.SUCCESS_CODE == log.getTriggerCode() && log.getHandleCode() == 0) {
                                    // job running, wait + again monitor
                                    TimeUnit.SECONDS.sleep(10);

                                    JobFailMonitorHelper.monitor(jobLogId);
                                    logger.info(">>>>>>>>>>> job monitor, job running, JobLogId:{}", jobLogId);
                                }

                                // job成功執行
                                if (ReturnT.SUCCESS_CODE == log.getTriggerCode()
                                    && ReturnT.SUCCESS_CODE == log.getHandleCode()) {
                                    // job success, pass
                                    logger.info(">>>>>>>>>>> job monitor, job success, JobLogId:{}", jobLogId);
                                }

                                // job執行失敗
                                if (ReturnT.FAIL_CODE == log.getTriggerCode()
                                    || ReturnT.FAIL_CODE == log.getHandleCode()) {
                                    // job fail,
                                    // 執行失敗, 傳送告警郵件
                                    sendMonitorEmail(log);
                                    logger.info(">>>>>>>>>>> job monitor, job fail, JobLogId:{}", jobLogId);
                                }
                            }
                        }
                    } catch (Exception e) {
                        logger.error("job monitor error:{}", e);
                    }
                }

                // monitor all clear
                List<Integer> jobLogIdList = new ArrayList<Integer>();
                int drainToNum = getInstance().queue.drainTo(jobLogIdList);
                if (jobLogIdList != null && jobLogIdList.size() > 0) {
                    for (Integer jobLogId : jobLogIdList) {
                        XxlJobLog log = XxlJobDynamicScheduler.xxlJobLogDao.load(jobLogId);
                        if (ReturnT.FAIL_CODE == log.getTriggerCode() || ReturnT.FAIL_CODE == log.getHandleCode()) {
                            // job fail,
                            sendMonitorEmail(log);
                            logger.info(">>>>>>>>>>> job monitor last, job fail, JobLogId:{}", jobLogId);
                        }
                    }
                }

            }

①JobFailMonitorHelper.instance.queue.take(); // 首先monitor執行緒會從queue佇列中取出已經logJobId, 而這個jobLogId是什麼時候放入queue佇列中的呢? 是XxlJobTrigger類在執行trigger()進行一次排程時, 寫入的, 如下:

當執行本次排程時, 其實是走了一遍rpc遠端呼叫, 得到執行結果200或者500狀態, 而monitor監控執行緒就是根據該狀態判斷本次排程是否成功了.

②monitor執行緒再根據jobLogId查詢資料庫, 判斷本次排程job執行結果, trigger_code=200成功, 則在控制檯列印本次排程執行成功.

否則就傳送告警郵件.

以上分析的只是admin排程中心初始化時, 所做的一些初始化工作, 一是, 開啟一個自動發現executor執行器服務的執行緒, 並寫入xxl-job-qrtz-trigger-group表, 二是, 再開啟一個monitor監控執行緒, 用於監控executor執行器執行本次job的執行結果.

下一篇將解析admin排程中心具體的排程過程, 排程策略以及rpc通訊的實現.