接下來看下服務端代碼
服務端源碼
- 服務端通過管理quartz定時任務組件，分發任務
- 先從入口看起，由web.xml進入，可以看出，自己編寫的代碼從applicationcontext-xxl-job-admin.xml文件開始
<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>
這就是調度器的主要方法了，由init方法進入，可以看到和客戶端很類似的結構，我添點注釋
public void init() throws Exception {
// admin registry monitor run 服務端注冊監聽
JobRegistryMonitorHelper.getInstance().start();
// admin monitor run 服務端監聽運行
JobFailMonitorHelper.getInstance().start();
// admin-server(spring-mvc) 工廠方法管理服務 設置令牌等
NetComServerFactory.putService(AdminBiz.class, XxlJobDynamicScheduler.adminBiz);
NetComServerFactory.setAccessToken(accessToken);
// init i18n 國際化實現
initI18n();
// valid
Assert.notNull(scheduler, "quartz scheduler is null");
logger.info(">>>>>>>>> init xxl-job admin success.");
}
然后進入第一個start，這里主要做的就是清除已經死亡的注冊信息，添加新的注冊信息，存入自己定義的數據庫
public void start(){
registryThread = new Thread(new Runnable() {
@Override
public void run() {
while (!toStop) {
try {
// auto registry group
List<XxlJobGroup> groupList = XxlJobDynamicScheduler.xxlJobGroupDao.findByAddressType(0);
if (CollectionUtils.isNotEmpty(groupList)) {
// 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>>();
List<XxlJobRegistry> list = XxlJobDynamicScheduler.xxlJobRegistryDao.findAll(RegistryConfig.DEAD_TIMEOUT);
if (list != null) {
for (XxlJobRegistry item: list) {
if (RegistryConfig.RegistType.EXECUTOR.name().equals(item.getRegistryGroup())) {
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);
}
}
}
// fresh group address
for (XxlJobGroup group: groupList) {
List<String> registryList = appAddressMap.get(group.getAppName());
String addressListStr = null;
if (CollectionUtils.isNotEmpty(registryList)) {
Collections.sort(registryList);
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();
}
第二個start，將工作隊列啟動
public void start(){
monitorThread = new Thread(new Runnable() {
@Override
public void run() {
// monitor
while (!toStop) {
try {
List<Integer> jobLogIdList = new ArrayList<Integer>();
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;
}
if (IJobHandler.SUCCESS.getCode() == log.getTriggerCode() && log.getHandleCode() == 0) {
// job running
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() )*/ {
// job fail,
// 1、fail retry
XxlJobInfo info = XxlJobDynamicScheduler.xxlJobInfoDao.loadById(log.getJobId());
if (log.getExecutorFailRetryCount() > 0) {
// TODO，分片任務失敗重試優化，僅重試失敗分片
JobTriggerPoolHelper.trigger(log.getJobId(), (log.getExecutorFailRetryCount()-1), TriggerTypeEnum.RETRY);
String retryMsg = "<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_type_retry") +"<<<<<<<<<<< </span><br>";
log.setTriggerMsg(log.getTriggerMsg() + retryMsg);
XxlJobDynamicScheduler.xxlJobLogDao.updateTriggerInfo(log);
}
// 2、fail alarm
failAlarm(info, 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);
}
}
// 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,
XxlJobInfo info = XxlJobDynamicScheduler.xxlJobInfoDao.loadById(log.getJobId());
failAlarm(info, log);
logger.info(">>>>>>>>>>> job monitor last, job fail, JobLogId:{}", jobLogId);
}
}
}
}
});
monitorThread.setDaemon(true);
monitorThread.start();
}
可以看到此類中具體方法
// producer
public static void monitor(int jobLogId){
getInstance().queue.offer(jobLogId);
}
通過該方法添加的隊列，由此向上追蹤幾層后發現
public static void trigger(int jobId, int failRetryCount, TriggerTypeEnum triggerType) {
helper.addTrigger(jobId, failRetryCount, triggerType);
}
調用此方法的地方很多，找個比較有代表的
@RequestMapping("/add")
@ResponseBody
public ReturnT<String> add(XxlJobInfo jobInfo) {
return xxlJobService.add(jobInfo);
}
@RequestMapping("/update")
@ResponseBody
public ReturnT<String> update(XxlJobInfo jobInfo) {
return xxlJobService.update(jobInfo);
}
@RequestMapping("/remove")
@ResponseBody
public ReturnT<String> remove(int id) {
return xxlJobService.remove(id);
}
@RequestMapping("/pause")
@ResponseBody
public ReturnT<String> pause(int id) {
return xxlJobService.pause(id);
}
@RequestMapping("/resume")
@ResponseBody
public ReturnT<String> resume(int id) {
return xxlJobService.resume(id);
}
@RequestMapping("/trigger")
@ResponseBody
public ReturnT<String> triggerJob(int id) {
JobTriggerPoolHelper.trigger(id, -1, TriggerTypeEnum.MANUAL);
return ReturnT.SUCCESS;
}
看到這個，就知道是頁面點擊的執行方法trigger，這樣又從底層追溯到了controller層，差不多套了一遍。