首先從spring的配置看起翔烁， 從以下配置可以看出蹬屹，xxl內(nèi)部使用的是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、以數(shù)據(jù)庫中已經(jīng)存在的為準：false -->
<property name="applicationContextSchedulerContextKey" value="applicationContextKey" />
<property name="configLocation" value="classpath:quartz.properties"/>
</bean>
<!-- 這個調(diào)度中心樊拓，在啟動的時候蛤织，會做很多初始化的工作 摊鸡，比如：執(zhí)行器信息囤热，注冊機器列表等信息 -->
<bean id="xxlJobDynamicScheduler" class="com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler" init-method="init" destroy-method="destroy" >
<!-- 配置調(diào)度中心的名稱 -->
<property name="scheduler" ref="quartzScheduler"/>
<!-- 用于調(diào)度中心和執(zhí)行器之間通信的時候做數(shù)據(jù)加密 -->
<property name="accessToken" value="${xxl.job.accessToken}" />
</bean>
XxlJobDynamicScheduler
com.xxl.job.admin.core.schedule.XxlJobDynamicScheduler 在啟動的時候會做如下工作：
public void init() throws Exception {
// 啟動自動注冊線程， 獲取類型為自動注冊的執(zhí)行器信息犀暑，完成機器的自動注冊與發(fā)現(xiàn)
JobRegistryMonitorHelper.getInstance().start();
// 啟動失敗日志監(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
JobRegistryMonitorHelper.getInstance().start() 詳細代碼如下：
public void start(){
//創(chuàng)建一個線程
registryThread = new Thread(new Runnable() {
@Override
public void run() {
// 當toStop 為false時進入該循環(huán)。
while (!toStop) {
try {
// 獲取類型為自動注冊的執(zhí)行器地址列表
List<XxlJobGroup> groupList = XxlJobDynamicScheduler.xxlJobGroupDao.findByAddressType(0);
if (CollectionUtils.isNotEmpty(groupList)) {
// 刪除 90秒之內(nèi)沒有更新信息的注冊機器， 90秒沒有心跳信息返回，代表機器已經(jīng)出現(xiàn)問題，故移除
XxlJobDynamicScheduler.xxlJobRegistryDao.removeDead(RegistryConfig.DEAD_TIMEOUT);
// fresh online address (admin/executor)
HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
// 查詢在90秒之內(nèi)有過更新的機器列表
List<XxlJobRegistry> list = XxlJobDynamicScheduler.xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT);
if (list != null) {
//循環(huán)注冊機器列表， 根據(jù)執(zhí)行器不同，將這些機器列表區(qū)分拿出來
for (XxlJobRegistry item: list) {
// 判斷該機器注冊信息RegistryGroup ，RegistType 是否是EXECUTOR , EXECUTOR 代表該機器是注冊到執(zhí)行器上面的
// RegistType 分為兩種帆焕， ADMIN 和EXECUTOR
if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
// 獲取注冊的執(zhí)行器 KEY （也就是執(zhí)行器）
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());
}
// 收集 機器信息，根據(jù)執(zhí)行器做區(qū)分
appAddressMap.put(appName, registryList);
}
}
}
// 遍歷執(zhí)行器列表
for (XxlJobGroup group: groupList) {
// 通過執(zhí)行器的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);
// 將 這個執(zhí)行器的 集群機器地址列表讼溺，寫入到數(shù)據(jù)庫
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
JobFailMonitorHelper.getInstance().start(); 詳細代碼如下：
//JobFailMonitorHelper.java
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;
}
//從數(shù)據(jù)庫跟以前有日志信息
XxlJobLog log = XxlJobDynamicScheduler.xxlJobLogDao.load(jobLogId);
if (log == null) {
continue;
}
//任務觸發(fā)成功， 但是JobHandle 還沒有返回結果
if (IJobHandler.SUCCESS.getCode() == log.getTriggerCode() && log.getHandleCode() == 0) {
//將 JobLogId 放入隊列 最易， 繼續(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() ) {
// 任務執(zhí)行失敗怒坯， 執(zhí)行發(fā)送郵件等預警措施
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 在啟動的時候做的操作， 主要是啟動兩個線程藻懒，
用來監(jiān)控自動注冊上來的機器剔猿，達到自動注冊的目的
監(jiān)控任務的執(zhí)行狀態(tài)， 如若失敗嬉荆，則發(fā)送郵件預警
xxl-job 是基于quartz 進行的二次開發(fā)归敬，在系統(tǒng)啟動的時候，quartz框架會自動去數(shù)據(jù)庫讀取相關的配置信息,載入相關定時器信息
