摘要:之調度中心啟動源碼分析一首先從的配置看起,從以下配置可以看出,內部使用的是配置自動啟動延時啟動,應用啟動成功后在啟動覆蓋中以數據庫中已經存在的為準這個調度中心,在啟動的時候,會做很多初始化的工作,比如執行器信
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 { // 獲取類型為自動注冊的執行器地址列表 ListJobFailMonitorHelpergroupList = 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.getInstance().start(); 詳細代碼如下:
//JobFailMonitorHelper.java public void start(){ // 啟動線程 monitorThread = new Thread(new Runnable() { @Override public void run() { // monitor while (!toStop) { try { ListjobLogIdList = 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框架會自動去數據庫讀取相關的配置信息,載入相關定時器信息
文章版權歸作者所有,未經允許請勿轉載,若此文章存在違規行為,您可以聯系管理員刪除。
轉載請注明本文地址:http://specialneedsforspecialkids.com/yun/71473.html
摘要:分布式任務調度平臺,美團點評員工許雪里開發,其核心設計目標是開發迅速學習簡單輕量級易擴展。源碼地址二本文簡介主要是利用成熟的部署工具,結合開源的分布式任務調度框架作為工程,快速搭建自己的持續集成項目,其他項目可以類似構建。 一、概述 1、自動化部署能簡化開發過程的代碼管理,讓開發人員把更多的時間專注于業務實現, 簡化繁瑣的上線流程和操作步驟,做到項目的快速打包和部署,減少人...
摘要:分布式任務調度平臺一簡介概述是一個輕量級分布式任務調度框架,其核心設計目標是開發迅速學習簡單輕量級易擴展。 《分布式任務調度平臺XXL-JOB》 showImg(https://segmentfault.com/img/remote/1460000010402214);showImg(https://segmentfault.com/img/remote/146000001040221...
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...
摘要:自研調度組件,移除依賴一方面是為了精簡系統降低冗余依賴,另一方面是為了提供系統的可控度與穩定性觸發單節點周期性觸發,運行事件如調度集群競爭,負載方式協同處理,鎖競爭更新觸發信息推送時間輪鎖釋放鎖競爭底層表結構重構移除張相關表,并對現 v2.0.2 Release Notes 1、自研調度組件,移除quartz依賴:一方面是為了精簡系統降低冗余依賴,另一方面是為了提供系統的可控度與穩...
閱讀 1294·2021-10-08 10:05
閱讀 4107·2021-09-22 15:54
閱讀 3105·2021-08-27 16:18
閱讀 3107·2019-08-30 15:55
閱讀 1436·2019-08-29 12:54
閱讀 2748·2019-08-26 11:42
閱讀 543·2019-08-26 11:39
閱讀 2129·2019-08-26 10:11