xxl-job源码解读：调度器schedule

本文基于xxl-job的2.3.1版本

基本说明

基本原理概述

调用器主要的用于判断定时任务的执行时间，按时调用触发器(trigger)，再由触发器去获取任务信息，执行预先定义好的程序。

调度器的基本原理为启用一个线程，循环去查询所有任务状态，获取一个时间段内，已经到执行时间或者即将执行的任务，通知触发器去执行任务。

任务信息存储

定时任务的信息存储大致可分为两类：

应用内存：应用启动时将任务信息加载到内存里，调度器通过读取这些信息进行调度，这种功能一般比较单一，且不支持集群（即集群情况下任务会出现每个节点都执行一次的情况），例如spring scheduled-task
数据库存储：将任务信息存储于数据库或某中间件中，调度器通过锁的方式进行集群调度，能实现的功能更为丰富，xxl-job归属于此类

调度器源码解读

xxl-job调度器代码主要在 com.xxl.job.admin.core.thread.JobScheduleHelper 中

代码逻辑流程图

调度器执行主流程

调度器线程启动加载完毕之后，在收到调度器终止命令之前（一般为应用关闭），会一直循环执行：

通过xxl_job_lock表，获取数据库锁，防止集群下调度器任务调度并发冲突

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();

读取xxl_job_info 信息，获取未来5秒内执行的任务信息

// 1、pre read 预读任务执行时间，获取可执行任务集合
long nowTime = System.currentTimeMillis();
List<XxlJobInfo> scheduleList = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleJobQuery(nowTime + PRE_READ_MS, preReadCount);

    <select id="scheduleJobQuery" parameterType="java.util.HashMap" resultMap="XxlJobInfo">SELECT <include refid="Base_Column_List" />FROM xxl_job_info AS tWHERE t.trigger_status = 1and t.trigger_next_time <![CDATA[ <= ]]> #{maxNextTime}ORDER BY id ASCLIMIT #{pagesize}</select>

根据任务的触发时间与当前时间的对比，判断进入三种处理方式：

过期处理策略：跳过执行（默认跳过，可配置为执行一次），根据当前时间，重新计算下次触发时间并更新

// 过期处理策略: 过期超5s, 本次忽略, 当前时间开始计算下次触发时间
logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = {}", jobInfo.getId());// 1、misfire match
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、fresh next
refreshNextValidTime(jobInfo, new Date());

交给触发器立即执行，并更新下次执行时间。如果下次执行时间在5秒以内，计算更新下下次执行时间，并将任务加入预执行集合ringData中

// 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、fresh next
refreshNextValidTime(jobInfo, new Date());// next-trigger-time in 5s, pre-read again
if (jobInfo.getTriggerStatus() == 1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {// 1、make ring second
int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);// 2、push time ring
pushTimeRing(ringSecond, jobInfo.getId());// 3、fresh next
refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));

加入预执行集合ringData，并更新下次执行时间

    /*** 预提取执行的任务集合 触发秒->任务ID集合*/private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();

// 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、fresh next
refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));

    private void pushTimeRing(int ringSecond, int jobId) {// push async ringList<Integer> ringItemData = ringData.computeIfAbsent(ringSecond, k -> new ArrayList<>());ringItemData.add(jobId);logger.debug(">>>>>>>>>>> xxl-job, schedule push time-ring : {} = {}" , ringSecond, Collections.singletonList(ringItemData));}

根据当次循环耗时，选择睡眠时长，同时进行秒数对齐，保证下次循环开始时间为整秒(毫秒级)

while (!scheduleThreadToStop) {// Scan Joblong start = System.currentTimeMillis();// 此处省略流程代码long cost = System.currentTimeMillis() - start;// Wait seconds, align secondif (cost < 1000) {  // scan-overtime, not waittry {// pre-read period: success > scan each second; fail > skip this period;TimeUnit.MILLISECONDS.sleep((preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);} catch (InterruptedException e) {if (!scheduleThreadToStop) {logger.error(e.getMessage(), e);Thread.currentThread().interrupt();}}}}

预执行线程

用于处理被调度器抓取，但未到执行时间的任务，循环根据触发的秒数进行判断，对到达执行秒数的任务，调用触发器进行任务执行

        // ring threadringThread = new Thread(() -> {while (!ringThreadToStop) {// align secondtry {TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);} catch (InterruptedException e) {if (!ringThreadToStop) {logger.error(e.getMessage(), e);Thread.currentThread().interrupt();}}try {// second dataList<Integer> ringItemData = new ArrayList<>();int nowSecond = Calendar.getInstance().get(Calendar.SECOND);   // 避免处理耗时太长，跨过刻度，向前校验一个刻度；for (int i = 0; i < 2; i++) {List<Integer> tmpData = ringData.remove((nowSecond + 60 - i) % 60);if (tmpData != null) {ringItemData.addAll(tmpData);}}// ring triggerlogger.debug(">>>>>>>>>>> xxl-job, time-ring beat : " + nowSecond + " = " + Arrays.asList(ringItemData));if (!ringItemData.isEmpty()) {// do triggerfor (int jobId : ringItemData) {// do triggerJobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);}// clearringItemData.clear();}} catch (Exception e) {if (!ringThreadToStop) {logger.error(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread error:{}", e);}}}logger.info(">>>>>>>>>>> xxl-job, JobScheduleHelper#ringThread stop");});