一、设计思路
借用官网的话：
将调度行为抽象形成“调度中心”公共平台，而平台自身并不承担业务逻辑，“调度中心”负责发起调度请求。
将任务抽象成分散的JobHandler，交由“执行器”统一管理，“执行器”负责接收调度请求并执行对应的JobHandler中业务逻辑。因此，“调度”和“任务”两部分可以相互解耦，提高系统整体稳定性和扩展性；
借用官网的图：
xxl_job架构图.png
简化的架构图：
简化架构图.png
二、启动原理
1.xxl-job-admin服务启动原理
启动XxlJobAdminApplication类，在spring容器实例化之前，会执行实现了 InitializingBean 接口的 afterPropertiesSet() 的方法，这里是利用了springboot的拓展接口，来将xxl-job的相关bean给注册到IOC容器当中。然后执行最关键的 xxlJobScheduler.init()
XxlJobAdminConfig.png
调用 XxlJobScheduler 的 init() 方法
XxlJobScheduler.png
JobRegistryHelper，JobFailMonitorHelper，JobCompleteHelper，JobLogReportHelper，JobScheduleHelper这五个类都是使用了饿汉式的单例模式（个人觉得还需要将构造方法私有化），
image.png
a.调用 JobTriggerPoolHelper.toStart() 本质就是调用JobTriggerPoolHelper的start()方法，构造出两个线程池，如下图所示的，一个快触发线程池，一个慢触发线程池。
image.png
b.调用 JobRegistryHelper.getInstance().start()，方法内部主要做了两件事，一件事是初始化一个注册或者移除的线程池 registryMonitorThread，然后创建一个 registryMonitorThread 的守护线程。设置成守护线程。
JobRegistryHelper.start()方法.png
单独把线程构造拿出来分析。
// for monitor
registryMonitorThread = new Thread(new Runnable() {
@Override
public void run() {
// 死循环
while (!toStop) {
try {
// 从 xxl_job_group 表中查询出 自动注册的执行器 （address_type：0 自动注册，1：手动注册）
List<XxlJobGroup> groupList = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().findByAddressType(0);
if (groupList!=null && !groupList.isEmpty()) {
// 从 xxl_job_registry 表中找到更新时间 小于当前时间+死亡间隔时间（就是找到注册表中规定时间没有更新的记录）
List<Integer> ids = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findDead(RegistryConfig.DEAD_TIMEOUT, new Date());
if (ids!=null && ids.size()>0) {
// 如果找到在规定时间内没有更新的注册，就直接删除这些注册执行器
XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(ids);
}
// fresh online address (admin/executor)
HashMap<String, List<String>> appAddressMap = new HashMap<String, List<String>>();
// 从 xxl_job_registry 表中找到更新时间 小于当前时间+死亡间隔时间（就是找到注册表中规定时间没有更新的记录）
List<XxlJobRegistry> list = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findAll(RegistryConfig.DEAD_TIMEOUT, new Date());
if (list != null) {
// 遍历当前过期的注册器，将过期的注册器的 register_value 注册地址保存到临时变量 appAddressMap 中
for (XxlJobRegistry item: list) {
// 如果注册器类型（registry_group） 是 EXECUTOR
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);
}
}
}
// 刷新对应执行器地址和最新修改时间
for (XxlJobGroup group: groupList) {
List<String> registryList = appAddressMap.get(group.getAppname());
String addressListStr = null;
if (registryList!=null && !registryList.isEmpty()) {
Collections.sort(registryList);
StringBuilder addressListSB = new StringBuilder();
for (String item:registryList) {
addressListSB.append(item).append(",");
}
addressListStr = addressListSB.toString();
addressListStr = addressListStr.substring(0, addressListStr.length()-1);
}
group.setAddressList(addressListStr);
group.setUpdateTime(new Date());
XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().update(group);
}
}
} catch (Exception e) {
if (!toStop) {
logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
}
}
try {
// 休眠心跳的时间
TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
} catch (InterruptedException e) {
if (!toStop) {
logger.error(">>>>>>>>>>> xxl-job, job registry monitor thread error:{}", e);
}
}
}
logger.info(">>>>>>>>>>> xxl-job, job registry monitor thread stop");
}
});
C.调用 JobFailMonitorHelper.getInstance().start()，方法内部创建一个 monitorThread 的守护线程。设置成守护线程。
image.png
下面代码详细介绍 JobFailMonitorHelper.getInstance().start() 里面的 构造的线程主要做的是什么事。
public void start(){
monitorThread = new Thread(new Runnable() {
@Override
public void run() {
// monitor 死循环监控，每10秒钟（每次执行完休眠十秒）执行一遍监控的内容
while (!toStop) {
try {
// 从 xxl_job_log 查询出失败的日志记录
List<Long> failLogIds = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().findFailJobLogIds(1000);
if (failLogIds!=null && !failLogIds.isEmpty()) {
for (long failLogId: failLogIds) {
// 修改 xxl_job_log 将警报状态改成 （-1锁定状态） 告警状态：0-默认、1-无需告警、2-告警成功、3-告警失败
int lockRet = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateAlarmStatus(failLogId, 0, -1);
if (lockRet < 1) {
continue;
}
// 根据失败的日志id，查询出该条日志
XxlJobLog log = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().load(failLogId);
// 根据这条日志记录的 执行器Id查询对应的执行器
XxlJobInfo info = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(log.getJobId());
// 如果 日志的重试次数大于0，就直接触发JobTriggerPoolHelper.trigger()方法，这个方法就是admin远程调用执行器的方法。
if (log.getExecutorFailRetryCount() > 0) {
JobTriggerPoolHelper.trigger(log.getJobId(), TriggerTypeEnum.RETRY, (log.getExecutorFailRetryCount()-1), log.getExecutorShardingParam(), log.getExecutorParam(), null);
String retryMsg = "<br><br><span style=\"color:#F39C12;\" > >>>>>>>>>>>"+ I18nUtil.getString("jobconf_trigger_type_retry") +"<<<<<<<<<<< </span><br>";
log.setTriggerMsg(log.getTriggerMsg() + retryMsg);
// 触发完成，将失败重试的次数减一，更新
XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateTriggerInfo(log);
}
// 2、fail alarm monitor
int newAlarmStatus = 0; // 告警状态：0-默认、-1=锁定状态、1-无需告警、2-告警成功、3-告警失败
// 如果存在失败的日志，发送警报邮件
if (info != null) {
boolean alarmResult = XxlJobAdminConfig.getAdminConfig().getJobAlarmer().alarm(info, log);
newAlarmStatus = alarmResult?2:3;
} else {
newAlarmStatus = 1;
}
// 最后更新一下 xxl_job_log 表的 alarm_status 状态字段
XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateAlarmStatus(failLogId, -1, newAlarmStatus);
}
}
} catch (Exception e) {
if (!toStop) {
logger.error(">>>>>>>>>>> xxl-job, job fail monitor thread error:{}", e);
}
}
try {
// 休眠 10秒
TimeUnit.SECONDS.sleep(10);
} catch (Exception e) {
if (!toStop) {
logger.error(e.getMessage(), e);
}
}
}
logger.info(">>>>>>>>>>> xxl-job, job fail monitor thread stop");
}
});
monitorThread.setDaemon(true);
monitorThread.setName("xxl-job, admin JobFailMonitorHelper");
monitorThread.start();
}
调用 JobCompleteHelper.getInstance().start()，方法内部创建一个 monitorThread 的守护线程。设置成守护线程；以及一个callbackThreadPool线程池。
image.png
monitorThread 线程内部的工作内容
// for monitor
monitorThread = new Thread(new Runnable() {
@Override
public void run() {
// wait for JobTriggerPoolHelper-init
try {
// 上来休眠50毫秒，等待 JobTriggerPoolHelper 初始化完成
TimeUnit.MILLISECONDS.sleep(50);
} catch (InterruptedException e) {
if (!toStop) {
logger.error(e.getMessage(), e);
}
}
// monitor
while (!toStop) {
try {
// 任务结果丢失处理：调度记录停留在 "运行中" 状态超过10min，且对应执行器心跳注册失败不在线，则将本地调度主动标记失败；
Date losedTime = DateUtil.addMinutes(new Date(), -10);
List<Long> losedJobIds = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().findLostJobIds(losedTime);
if (losedJobIds!=null && losedJobIds.size()>0) {
for (Long logId: losedJobIds) {
XxlJobLog jobLog = new XxlJobLog();
jobLog.setId(logId);
jobLog.setHandleTime(new Date());
jobLog.setHandleCode(ReturnT.FAIL_CODE);
jobLog.setHandleMsg( I18nUtil.getString("joblog_lost_fail") );
// 处理日志，并更新执行器的完成结果
XxlJobCompleter.updateHandleInfoAndFinish(jobLog);
}
}
} catch (Exception e) {
if (!toStop) {
logger.error(">>>>>>>>>>> xxl-job, job fail monitor thread error:{}", e);
}
}
try {
// 休眠60秒，再执行一次
TimeUnit.SECONDS.sleep(60);
} catch (Exception e) {
if (!toStop) {
logger.error(e.getMessage(), e);
}
}
}
logger.info(">>>>>>>>>>> xxl-job, JobLosedMonitorHelper stop");
}
});
E.最后看一下 JobScheduleHelper.getInstance().start(); 方法。方法里面最主要的就是起两个线程，分别将两个线程设置成守护线程，ringThread是最干活的线程，scheduleThread是检测并调度执行器任务的线程。
image.png
看一下这ringThread和scheduleThread里面都是干啥的。这里先总结一下，scheduleThread线程主要就是将需要执行的定时器任务分个类，并维护每个定时器里面的下次执行时间，以及处理 调度过期的 执行器，要么立刻执行一次，要么直接忽略，等待下次执行，最后就是将需要在5秒内执行的定时器放进一个map里面，交给ringThread线程去执行定时器。而ringThread线程就是直接从map中拿到需要执行的执行器去执行，并且每轮执行只处理两个时间点（毫秒级）的所有执行器。具体的可以看代码里面的讲解。
public void start(){
// schedule thread
scheduleThread = new Thread(new Runnable() {
@Override
public void run() {
try {
TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis()%1000 );
} catch (InterruptedException e) {
if (!scheduleThreadToStop) {
logger.error(e.getMessage(), e);
}
}
logger.info(">>>>>>>>> init xxl-job admin scheduler success.");
// pre-read count: treadpool-size * trigger-qps (each trigger cost 50ms, qps = 1000/50 = 20)
// 这里默认么个定时任务线程执行耗时50毫秒，每秒1000毫秒，可以执行20个任务，快线程池默认是200个最大线程数，慢线程默认是100个最大线程数，
// 所以当线程数拉满的情况下，每秒钟可以处理任务数是：（100+200）*20 = 6000 ；所以这里的 preReadCount 表示预读数（默认最大：6000）
// 如果想要提高并发性，通过修改快慢线程池的最大线程数这个参数调节
int preReadCount = (XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax() + XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax()) * 20;
logger.info("==========================preReadCount ={}",preReadCount);
while (!scheduleThreadToStop) {
// Scan Job
long start = System.currentTimeMillis();
Connection conn = null;
Boolean connAutoCommit = null;
PreparedStatement preparedStatement = null;
boolean preReadSuc = true;
try {
conn = XxlJobAdminConfig.getAdminConfig().getDataSource().getConnection();
connAutoCommit = conn.getAutoCommit();
conn.setAutoCommit(false);
// 利用数据库的行锁
preparedStatement = conn.prepareStatement( "select * from xxl_job_lock where lock_name = 'schedule_lock' for update" );
preparedStatement.execute();
// tx start
// 1、预读数据
long nowTime = System.currentTimeMillis();
// 由于分析了最多可以执行6000个任务，所以这里在去查任务表的时候，最多去查出来6000条满足条件的
// 条件：下次执行时间 小于等于 （当前时间 + 5秒） 并且 执行状态是 正在运行的状态
List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);
if (scheduleList!=null && scheduleList.size()>0) {
// 2、push time-ring
for (XxlJobInfo jobInfo: scheduleList) {
// time-ring jump
//如果服务宕机了，或者重启等等，导致超过了调度周期(5秒的调度周期)，也就是本来由时间上的上一次或上很多次调度触发的数据被本次调度查到了，
// 这就可能代表着可能中间存在多次调度未触发，而按照周期性一次一次计算下次预期调度时间，那这次调度完了计算出来的下次调度还是在当前时间以前，
// 例如调度周期1分钟调度一次，宕机5分钟了，现在查到的预期调度时间为5分钟前，如果直接调度成功会重复调度5次当前时间以前的任务，这里直接pass并计算下一次调度时间，
// 但是计算下一次调度时间也是传入当前时间，直接修正预期下次调度时间为当前时间之后，因为调度时间周期为 5秒，所以会+ PRE_READ_MS 判断，如果是一次性的调度则会补偿这次调度
if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
// 2.1、trigger-expire > 5s：pass && make next-trigger-time
logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId());
// 拿到设置的 调度过期策略
MisfireStrategyEnum misfireStrategyEnum = MisfireStrategyEnum.match(jobInfo.getMisfireStrategy(), MisfireStrategyEnum.DO_NOTHING);
// 如果是忽略就不执行，如果是立刻执行一次，就立马调用执行一次该执行器
if (MisfireStrategyEnum.FIRE_ONCE_NOW == misfireStrategyEnum) {
// FIRE_ONCE_NOW 》 trigger
JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.MISFIRE, -1, null, null, null);
logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() );
}
// 2、从当前时间算起，算出这个定时器下次应该执行时间
refreshNextValidTime(jobInfo, new Date());
// 当前的时间大于执行器下次的执行的时间，说明上次执行器可能遗漏了这个执行器
} else if (nowTime > jobInfo.getTriggerNextTime()) {
// 2.2、trigger-expire < 5s：direct-trigger && make next-trigger-time
// 1、trigger
JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.CRON, -1, null, null, null);
logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId() );
// 2、从当前时间算起，算出这个定时器下次应该执行时间
refreshNextValidTime(jobInfo, new Date());
// next-trigger-time in 5s, pre-read again
// 如果下次发送时间在当前时间之后5秒内，会进行第二次触发，放到另一个线程中执行触发逻辑
if (jobInfo.getTriggerStatus()==1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {
// 1、make ring second
// 这里算出的结果保持在 0 - 59之间
int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
// 2、将当前任务放进一个全局变量map中，让ringThread线程去执行
pushTimeRing(ringSecond, jobInfo.getId());
// 3、从当前时间算起，算出这个定时器下次应该执行时间
refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
}
// 取5秒的直接触发 但是区别于上面是第一次触发
} else {
// 2.3、trigger-pre-read：time-ring trigger && make next-trigger-time
// 1、make ring second
int ringSecond = (int)((jobInfo.getTriggerNextTime()/1000)%60);
// 2、push time ring
pushTimeRing(ringSecond, jobInfo.getId());
// 3、从当前时间算起，算出这个定时器下次应该执行时间
refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
}
}
// 3、更新执行器，主要是更新 上次触发时间，下次触发时间，和当前执行器的状态
for (XxlJobInfo jobInfo: scheduleList) {
XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleUpdate(jobInfo);
}
} else {
preReadSuc = false;
}
// tx stop
} catch (Exception e) {
if (!scheduleThreadToStop) {
logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread error:{}", e);
}
} finally {
// commit
if (conn != null) {
try {
conn.commit();
} catch (SQLException e) {
if (!scheduleThreadToStop) {
logger.error(e.getMessage(), e);
}
}
try {
conn.setAutoCommit(connAutoCommit);
} catch (SQLException e) {
if (!scheduleThreadToStop) {
logger.error(e.getMessage(), e);
}
}
try {
conn.close();
} catch (SQLException e) {
if (!scheduleThreadToStop) {
logger.error(e.getMessage(), e);
}
}
}
// close PreparedStatement
if (null != preparedStatement) {
try {
preparedStatement.close();
} catch (SQLException e) {
if (!scheduleThreadToStop) {
logger.error(e.getMessage(), e);
}
}
}
}
long cost = System.currentTimeMillis()-start;
// Wait seconds, align second
// 如果上述操作耗时大于一秒直接进入下次循环，如果小于一秒需要再判断
if (cost < 1000) { // scan-overtime, not wait
try {
// pre-read period: success > scan each second; fail > skip this period;
// 首先 System.currentTimeMillis()%1000 这里的取余，最大值是999毫秒，可以理解成极限的一秒
// 如果读到了数据 休眠 （1 - （System.currentTimeMillis()%1000））秒，这里最大休眠1秒，最小接近不休眠直接执行
// 因为读到了数据，不知道接下来还有没有数据，这里为了赶工确保满足条件的定时任务能快速被执行。
// 如果读不到数据，休眠 （5 - （System.currentTimeMillis()%1000））秒，这里最大休眠5秒，最小休眠4秒
// 因为已经读不到数据了，多休息一下，让定时器等到需要被执行的时间点
TimeUnit.MILLISECONDS.sleep((preReadSuc?1000:PRE_READ_MS) - System.currentTimeMillis()%1000);
} catch (InterruptedException e) {
if (!scheduleThreadToStop) {
logger.error(e.getMessage(), e);
}
}
}
}
logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#scheduleThread stop");
}
});
scheduleThread.setDaemon(true);
scheduleThread.setName("xxl-job, admin JobScheduleHelper#scheduleThread");
scheduleThread.start();
// ring thread
ringThread = new Thread(new Runnable() {
@Override
public void run() {
while (!ringThreadToStop) {
// align second
try {
// 上来休眠最大1秒，如果scheduleThread有执行任务，保证会向 ringData（map） 里面写
TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
} catch (InterruptedException e) {
if (!ringThreadToStop) {
logger.error(e.getMessage(), e);
}
}
try {
// second data
// 从 ringData（map）里面拿到，
List<Integer> ringItemData = new ArrayList<>();
// 上面介绍过 这个map的key是在0-59之间，直接取当前的秒数（0-59）
int nowSecond = Calendar.getInstance().get(Calendar.SECOND); // 避免处理耗时太长，跨过刻度，向前校验一个刻度；
for (int i = 0; i < 2; i++) {
// 循环两次从 ringData（map）中拿到两个时间点的执行器ID集合 list，然后赋值给 ringItemData
List<Integer> tmpData = ringData.remove( (nowSecond+60-i)%60 );
if (tmpData != null) {
ringItemData.addAll(tmpData);
}
}
// ring trigger
logger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(ringItemData) );
// 如果 ringItemData 集合不为空，说明有需要执行的执行器Id，就遍历执行里面Id对应的执行器
if (ringItemData.size() > 0) {
// do trigger
for (int jobId: ringItemData) {
// do trigger
JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
}
// clear
// 遍历完成后，将 ringItemData 置空，然后等待线程 下次执行
ringItemData.clear();
}
} catch (Exception e) {
if (!ringThreadToStop) {
logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);
}
}
}
logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop");
}
});
ringThread.setDaemon(true);
ringThread.setName("xxl-job, admin JobScheduleHelper#ringThread");
ringThread.start();
}
三、总结
1.xxl-job-admin的启动是利用springboot的扩展接口InitializingBean来实现的，销毁是利用扩展DisposableBean接口来实现的。
2.xxl-job-admin的核心运行流程由JobRegistryHelper，JobFailMonitorHelper，JobCompleteHelper，JobLogReportHelper，JobScheduleHelper 这几个Helper完成。
3.单台xxl-job最多一秒钟可以完成6000个执行器任务的执行。
4.集群环境下，同一个执行器不出现并发执行问题其实是依赖了数据库的行锁实现的。
综上所述，将xxl-job-admin中的启动，以及如何调度核心部分就已经说完了，其实在执行执行器任务的时候里面还涉及到xxl-job的集群分片处理任务的原理，以及集群路由的原理，还有内置server的设计，以及xxl-job-admin远程触发任务使用的RPC调用原理细节，后面有空再整理吧。
网友评论