分布式任务调度系统xxl-job源码探究（二、服务中心）

接下来看下服务端代码

服务端源码

1. 服务端通过管理quartz定时任务组件，分发任务
2. 先从入口看起，由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层，差不多套了一遍。