xxl-job源碼分析

xxl-job

系統說明

安裝

安裝部署參考文檔：分佈式任務調度平臺xxl-job

功能

定時調度、服務解耦、靈活控制跑批時間（中止、開啓、從新設定時間、手動觸發）

XXL-JOB是一個輕量級分佈式任務調度平臺，其核心設計目標是開發迅速、學習簡單、輕量級、易擴展。現已開放源代碼並接入多家公司線上產品線，開箱即用

概念

執行器列表：一個執行器是一個項目

任務：一個任務是一個項目中的 JobHandler

一個xxl-job服務能夠有多個執行器（項目），一個項目下能夠有多個任務（JobHandler），他們是如何關聯的？

頁面操做：

1. 在管理平臺能夠新增執行器（項目）
2. 在任務列表能夠指定執行器（項目）下新增多個任務（JobHandler）

代碼操做：

1. 項目配置中增長 xxl.job.executor.appname = "執行器名稱"
2. 在實現類中增長 @JobHandler(value="xxl-job-demo") 註解，並繼承 IJobHandler

架構圖

拋出疑問

1. 調度中心啓動過程？
2. 執行器啓動過程？
3. 執行器如何註冊到調度中心？
4. 調度中心怎麼調用執行器？
5. 集羣調度時如何控制一個任務在該時刻不會重複執行
6. 集羣部署應該注意什麼？

系統分析

執行器依賴jar包

com.xuxueli:xxl-job-core:2.1.0

com.xuxueli:xxl-registry-client:1.0.2

com.xuxueli:xxl-rpc-core:1.4.1

調度中心啓動過程

// 1. 加載 XxlJobAdminConfig，adminConfig = this
XxlJobAdminConfig.java

// 啓動過程代碼
@Component
public class XxlJobScheduler implements InitializingBean, DisposableBean {
	private static final Logger logger = LoggerFactory.getLogger(XxlJobScheduler.class);

	@Override
	public void afterPropertiesSet() throws Exception {
		// init i18n
		initI18n();

		// admin registry monitor run
        // 2. 啓動註冊監控器（將註冊到register表中的IP加載到group表）/ 30執行一次
		JobRegistryMonitorHelper.getInstance().start();

		// admin monitor run
        // 3. 啓動失敗日誌監控器（失敗重試，失敗郵件發送）
		JobFailMonitorHelper.getInstance().start();

		// admin-server
        // 4. 初始化RPC服務
		initRpcProvider();

		// start-schedule
        // 5. 啓動定時任務調度器（執行任務，緩存任務）
		JobScheduleHelper.getInstance().start();

		logger.info(">>>>>>>>> init xxl-job admin success.");
	}
	
	......
}

執行器啓動過程

@Override
public void start() throws Exception {

	// init JobHandler Repository
    // 將執行 JobHandler 註冊到緩存中 jobHandlerRepository（ConcurrentMap）
	initJobHandlerRepository(applicationContext);

	// refresh GlueFactory
    // 刷新GLUE
	GlueFactory.refreshInstance(1);

	// super start
    // 核心啓動項
	super.start();
}

public void start() throws Exception {
    // 初始化日誌路徑 
    // private static String logBasePath = "/data/applogs/xxl-job/jobhandler";
	XxlJobFileAppender.initLogPath(this.logPath);
    // 初始化註冊中心列表 （把註冊地址放到 List）
	this.initAdminBizList(this.adminAddresses, this.accessToken);
    // 啓動日誌文件清理線程 （一天清理一次）
    // 天天清理一次過時日誌，配置參數必須大於3纔有效
	JobLogFileCleanThread.getInstance().start((long)this.logRetentionDays);
    // 開啓觸發器回調線程
	TriggerCallbackThread.getInstance().start();
    // 指定端口
	this.port = this.port > 0 ? this.port : NetUtil.findAvailablePort(9999);
    // 指定IP
	this.ip = this.ip != null && this.ip.trim().length() > 0 ? this.ip : IpUtil.getIp();
    // 初始化RPC 將執行器註冊到調度中心 30秒一次
	this.initRpcProvider(this.ip, this.port, this.appName, this.accessToken);
}

執行器註冊到調度中心

執行器

// 註冊執行器入口
XxlJobExecutor.java->initRpcProvider()->xxlRpcProviderFactory.start();

// 開啓註冊
XxlRpcProviderFactory.java->start();

// 執行註冊
ExecutorRegistryThread.java->start();
// RPC 註冊代碼
for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
	try {
		ReturnT<String> registryResult = adminBiz.registry(registryParam);
		if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
			registryResult = ReturnT.SUCCESS;
			logger.debug(">>>>>>>>>>> xxl-job registry success, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
			break;
		} else {
			logger.info(">>>>>>>>>>> xxl-job registry fail, registryParam:{}, registryResult:{}", new Object[]{registryParam, registryResult});
		}
	} catch (Exception e) {
		logger.info(">>>>>>>>>>> xxl-job registry error, registryParam:{}", registryParam, e);
	}

}

調度中心

// RPC 註冊服務
AdminBizImpl.java->registry();

數據庫

調度中心調用執行器

/* 調度中心執行步驟 */
// 1. 調用執行器
XxlJobTrigger.java->runExecutor();

// 2. 獲取執行器
XxlJobScheduler.java->getExecutorBiz();

// 3. 調用
ExecutorBizImpl.java->run();

/* 執行器執行步驟 */
// 1. 執行器接口
ExecutorBiz.java->run();

// 2. 執行器實現
ExecutorBizImpl.java->run();

// 3. 把jobInfo 從 jobThreadRepository (ConcurrentMap) 中獲取一個新線程，並開啓新線程
XxlJobExecutor.java->registJobThread();

// 4. 保存到當前線程隊列
JobThread.java->pushTriggerQueue();

// 5. 執行
JobThread.java->handler.execute(triggerParam.getExecutorParams());

調度中心（Admin）

實現 org.springframework.beans.factory.InitializingBean類，重寫 afterPropertiesSet 方法，在初始化bean的時候都會執行該方法

DisposableBean spring中止時執行

結束加載項

1. 中止定時任務調度器（中斷scheduleThread，中斷ringThread）
2. 中止觸發線程池（JobTriggerPoolHelper）
3. 中止註冊監控器（registryThread）
4. 中止失敗日誌監控器（monitorThread）
5. 中止RPC服務（stopRpcProvider）

手動執行方式

JobInfoController.java

@RequestMapping("/trigger")
@ResponseBody
//@PermissionLimit(limit = false)
public ReturnT<String> triggerJob(int id, String executorParam) {
    // force cover job param
    if (executorParam == null) {
        executorParam = "";
    }

    JobTriggerPoolHelper.trigger(id, TriggerTypeEnum.MANUAL, -1, null, executorParam);
    return ReturnT.SUCCESS;
}

定時調度策略

調度策略執行圖

調度策略源碼

JobScheduleHelper.java->start();

路由策略

第一個

固定選擇第一個機器

ExecutorRouteFirst.java->route();

最後一個

固定選擇最後一個機器

ExecutorRouteLast.java->route();

輪詢

隨機選擇在線的機器

ExecutorRouteRound.java->route();

private static int count(int jobId) {
	// cache clear
	if (System.currentTimeMillis() > CACHE_VALID_TIME) {
		routeCountEachJob.clear();
		CACHE_VALID_TIME = System.currentTimeMillis() + 1000*60*60*24;
	}

	// count++
	Integer count = routeCountEachJob.get(jobId);
	count = (count==null || count>1000000)?(new Random().nextInt(100)):++count;  // 初始化時主動Random一次，緩解首次壓力
	routeCountEachJob.put(jobId, count);
	return count;
}

隨機

隨機獲取地址列表中的一個

ExecutorRouteRandom.java->route();

一致性HASH

一個job經過hash算法固定使用一臺機器，且全部任務均勻散列在不一樣機器

ExecutorRouteConsistentHash.java->route();

public String hashJob(int jobId, List<String> addressList) {

	// ------A1------A2-------A3------
	// -----------J1------------------
	TreeMap<Long, String> addressRing = new TreeMap<Long, String>();
	for (String address: addressList) {
		for (int i = 0; i < VIRTUAL_NODE_NUM; i++) {
			long addressHash = hash("SHARD-" + address + "-NODE-" + i);
			addressRing.put(addressHash, address);
		}
	}

	long jobHash = hash(String.valueOf(jobId));
    // 取出鍵值 >= jobHash
	SortedMap<Long, String> lastRing = addressRing.tailMap(jobHash);
	if (!lastRing.isEmpty()) {
		return lastRing.get(lastRing.firstKey());
	}
	return addressRing.firstEntry().getValue();
}

最不常常使用

使用頻率最低的機器優先被選舉 把地址列表加入到內存中，等下次執行時剔除無效的地址，判斷地址列表中執行次數最少的地址取出 頻率、次數

ExecutorRouteLFU.java->route();

public String route(int jobId, List<String> addressList) {

	// cache clear
	if (System.currentTimeMillis() > CACHE_VALID_TIME) {
		jobLfuMap.clear();
		CACHE_VALID_TIME = System.currentTimeMillis() + 1000*60*60*24;
	}

	// lfu item init
	HashMap<String, Integer> lfuItemMap = jobLfuMap.get(jobId);     // Key排序能夠用TreeMap+構造入參Compare；Value排序暫時只能經過ArrayList；
	if (lfuItemMap == null) {
		lfuItemMap = new HashMap<String, Integer>();
		jobLfuMap.putIfAbsent(jobId, lfuItemMap);   // 避免重複覆蓋
	}

	// put new
	for (String address: addressList) {
		if (!lfuItemMap.containsKey(address) || lfuItemMap.get(address) >1000000 ) {
            // 0-n隨機數，包括0不包括n
			lfuItemMap.put(address, new Random().nextInt(addressList.size()));  // 初始化時主動Random一次，緩解首次壓力
		}
	}
	// remove old
	List<String> delKeys = new ArrayList<>();
	for (String existKey: lfuItemMap.keySet()) {
		if (!addressList.contains(existKey)) {
			delKeys.add(existKey);
		}
	}
	if (delKeys.size() > 0) {
		for (String delKey: delKeys) {
			lfuItemMap.remove(delKey);
		}
	}
    
    /*********************** 優化 START ***********************/
    // 優化  remove old部分
    Iterator<String> iterable = lfuItemMap.keySet().iterator();
    while (iterable.hasNext()) {
        String address = iterable.next();
        if (!addressList.contains(address)) {
            iterable.remove();
        }
    }
    /*********************** 優化 START ***********************/

	// load least userd count address
    // 從小到大排序
	List<Map.Entry<String, Integer>> lfuItemList = new ArrayList<Map.Entry<String, Integer>>(lfuItemMap.entrySet());
	Collections.sort(lfuItemList, new Comparator<Map.Entry<String, Integer>>() {
		@Override
		public int compare(Map.Entry<String, Integer> o1, Map.Entry<String, Integer> o2) {
			return o1.getValue().compareTo(o2.getValue());
		}
	});

	Map.Entry<String, Integer> addressItem = lfuItemList.get(0);
	String minAddress = addressItem.getKey();
	addressItem.setValue(addressItem.getValue() + 1);

	return addressItem.getKey();
}

最近最久未使用

最久未使用的機器優先被選舉 用鏈表的方式存儲地址，第一個地址使用後下次該任務過來使用第二個地址，依次類推（PS：有點相似輪詢策略） 與輪詢策略的區別：

1. 輪詢策略是第一次隨機找一臺機器執行，後續執行會將索引加1取餘
2. 輪詢策略依賴 addressList 的順序，若是這個順序變了，索引到下一次的機器可能不是指望的順序
3. LRU算法第一次執行會把全部地址加載進來並緩存，從第一個地址開始執行，即便 addressList 地址順序變了也不影響 次數

ExecutorRouteLRU.java->route();

public String route(int jobId, List<String> addressList) {

	// cache clear
	if (System.currentTimeMillis() > CACHE_VALID_TIME) {
		jobLRUMap.clear();
		CACHE_VALID_TIME = System.currentTimeMillis() + 1000*60*60*24;
	}

	// init lru
	LinkedHashMap<String, String> lruItem = jobLRUMap.get(jobId);
	if (lruItem == null) {
		/**
		 * LinkedHashMap
		 *      a、accessOrder：ture=訪問順序排序（get/put時排序）；false=插入順序排期；
		 *      b、removeEldestEntry：新增元素時將會調用，返回true時會刪除最老元素；可封裝LinkedHashMap並重寫該方法，好比定義最大容量，超出是返回true便可實現固定長度的LRU算法；
		 */
		lruItem = new LinkedHashMap<String, String>(16, 0.75f, true);
		jobLRUMap.putIfAbsent(jobId, lruItem);
	}

    /*********************** 舉個例子 START ***********************/
    // 若是accessOrder爲true的話，則會把訪問過的元素放在鏈表後面，放置順序是訪問的順序 
	// 若是accessOrder爲flase的話，則按插入順序來遍歷
    LinkedHashMap<String, String> lruItem = new LinkedHashMap<String, String>(16, 0.75f, true);
        jobLRUMap.putIfAbsent(1, lruItem);
        lruItem.put("192.168.0.1", "192.168.0.1");
        lruItem.put("192.168.0.2", "192.168.0.2");
        lruItem.put("192.168.0.3", "192.168.0.3");
        String eldestKey = lruItem.entrySet().iterator().next().getKey();
        String eldestValue = lruItem.get(eldestKey);
        System.out.println(eldestValue + ": " + lruItem);
        eldestKey = lruItem.entrySet().iterator().next().getKey();
        eldestValue = lruItem.get(eldestKey);
        System.out.println(eldestValue + ": " + lruItem);
    
    // 輸出結果：
    192.168.0.1: {192.168.0.2=192.168.0.2, 192.168.0.3=192.168.0.3, 192.168.0.1=192.168.0.1}
192.168.0.2: {192.168.0.3=192.168.0.3, 192.168.0.1=192.168.0.1, 192.168.0.2=192.168.0.2}
    /*********************** 舉個例子 END ***********************/
    
	// put new
	for (String address: addressList) {
		if (!lruItem.containsKey(address)) {
			lruItem.put(address, address);
		}
	}
	// remove old
	List<String> delKeys = new ArrayList<>();
	for (String existKey: lruItem.keySet()) {
		if (!addressList.contains(existKey)) {
			delKeys.add(existKey);
		}
	}
	if (delKeys.size() > 0) {
		for (String delKey: delKeys) {
			lruItem.remove(delKey);
		}
	}

	// load
	String eldestKey = lruItem.entrySet().iterator().next().getKey();
	String eldestValue = lruItem.get(eldestKey);
	return eldestValue;
}

故障轉移

按照順序依次進行心跳檢測，第一個心跳檢測成功的機器選定爲目標執行器併發起調度

ExecutorRouteFailover.java->route();

忙碌轉移

按照順序依次進行空閒檢測，第一個空閒檢測成功的機器選定爲目標執行器併發起調度

ExecutorRouteBusyover.java->route();

分片廣播

廣播觸發對應集羣中全部機器執行一次任務，同時傳遞分片參數；可根據分片參數開發分片任務

阻塞處理策略

爲了解決執行線程因併發問題、執行效率慢、任務多等緣由而作的一種線程處理機制，主要包括 串行、丟棄後續調度、覆蓋以前調度，通常經常使用策略是串行機制

ExecutorBlockStrategyEnum.java

SERIAL_EXECUTION("Serial execution"), // 串行
DISCARD_LATER("Discard Later"), // 丟棄後續調度
COVER_EARLY("Cover Early"); // 覆蓋以前調度

ExecutorBizImpl.java->run();

// executor block strategy
if (jobThread != null) {
	ExecutorBlockStrategyEnum blockStrategy = ExecutorBlockStrategyEnum.match(triggerParam.getExecutorBlockStrategy(), null);
	if (ExecutorBlockStrategyEnum.DISCARD_LATER == blockStrategy) {
		// discard when running
		if (jobThread.isRunningOrHasQueue()) {
			return new ReturnT<String>(ReturnT.FAIL_CODE, "block strategy effect："+ExecutorBlockStrategyEnum.DISCARD_LATER.getTitle());
		}
	} else if (ExecutorBlockStrategyEnum.COVER_EARLY == blockStrategy) {
		// kill running jobThread
		if (jobThread.isRunningOrHasQueue()) {
			removeOldReason = "block strategy effect：" + ExecutorBlockStrategyEnum.COVER_EARLY.getTitle();

			jobThread = null;
		}
	} else {
		// just queue trigger
	}
}

單機串行

對當前線程不作任何處理，並在當前線程的隊列裏增長一個執行任務

丟棄後續調度

若是當前線程阻塞，後續任務再也不執行，直接返回失敗

覆蓋以前調度

建立一個移除緣由，新建一個線程去執行後續任務

運行模式

ExecutorBizImpl.java->run();

BEAN

java裏的bean對象

GLUE(Java)

利用java的反射機制，經過代碼字符串生成實體類

IJobHandler originJobHandler = GlueFactory.getInstance().loadNewInstance(triggerParam.getGlueSource());

GroovyClassLoader

GLUE(Shell Python PHP Nodejs PowerShell)

按照文件命名規則建立一個執行腳本文件和一個日誌輸出文件，經過腳本執行器執行

失敗重試次數

任務失敗後記錄到 xxl_job_log 中，由失敗監控線程查詢處理失敗的任務且失敗次數大於0，繼續執行

任務超時時間

把超時時間給 triggerParam 觸發參數，在調用執行器的任務時超時時間，有點相似HttpClient的超時時間

執行器（Exector）

1. 註冊本身的機器地址

2. 註冊項目中的 JobHandler

3. 提供被調度中心調用的接口

   public interface ExecutorBiz {
   
       /**
        * 供調度中心檢測機器是否存活
        *
        * beat
        * @return
        */
       public ReturnT<String> beat();
   
       /**
        * 供調度中心檢測機器是否空閒
        *
        * @param jobId
        * @return
        */
       public ReturnT<String> idleBeat(int jobId);
   
       /**
        * kill
        * @param jobId
        * @return
        */
       public ReturnT<String> kill(int jobId);
   
       /**
        * log
        * @param logDateTim
        * @param logId
        * @param fromLineNum
        * @return
        */
       public ReturnT<LogResult> log(long logDateTim, long logId, int fromLineNum);
   
       /**
        * 執行觸發器
        * 
        * @param triggerParam
        * @return
        */
       public ReturnT<String> run(TriggerParam triggerParam);
   
   }

總結

學到了什麼

1. 算法（LFU、LRU、輪詢等）
2. JDK動態代理對象（詳細研究）
3. 用到了Netty（詳細研究）
4. FutureTask
5. GroovyClassLoader