揭秘XXLJOB：任务调度和执行的全面指南

本文主要向大家介绍一下xxljob在调度任务时执行了哪些操作，这也是xxljob最核心的功能

表结构

xxljob是如何触发任务的，首先我们先了解一下xxljob的表结构

xxl_job_info 记录的是各个具体job的信息是xxljob中最重要的表这张表记录的job 的调度类型，调度时机，路由策略，阻塞策略等信息

CREATE TABLE `xxl_job_info` (
  `id` int(11) NOT NULL AUTO_INCREMENT,
  `job_group` int(11) NOT NULL COMMENT '执行器主键ID',
  `job_desc` varchar(255) NOT NULL,
  `add_time` datetime DEFAULT NULL,
  `update_time` datetime DEFAULT NULL,
  `author` varchar(64) DEFAULT NULL COMMENT '作者',
  `alarm_email` varchar(255) DEFAULT NULL COMMENT '报警邮件',
  `schedule_type` varchar(50) NOT NULL DEFAULT 'NONE' COMMENT '调度类型',
  `schedule_conf` varchar(128) DEFAULT NULL COMMENT '调度配置，值含义取决于调度类型',
  `misfire_strategy` varchar(50) NOT NULL DEFAULT 'DO_NOTHING' COMMENT '调度过期策略',
  `executor_route_strategy` varchar(50) DEFAULT NULL COMMENT '执行器路由策略',
  `executor_handler` varchar(255) DEFAULT NULL COMMENT '执行器任务handler',
  `executor_param` varchar(512) DEFAULT NULL COMMENT '执行器任务参数',
  `executor_block_strategy` varchar(50) DEFAULT NULL COMMENT '阻塞处理策略',
  `executor_timeout` int(11) NOT NULL DEFAULT '0' COMMENT '任务执行超时时间，单位秒',
  `executor_fail_retry_count` int(11) NOT NULL DEFAULT '0' COMMENT '失败重试次数',
  `glue_type` varchar(50) NOT NULL COMMENT 'GLUE类型',
  `glue_source` mediumtext COMMENT 'GLUE源代码',
  `glue_remark` varchar(128) DEFAULT NULL COMMENT 'GLUE备注',
  `glue_updatetime` datetime DEFAULT NULL COMMENT 'GLUE更新时间',
  `child_jobid` varchar(255) DEFAULT NULL COMMENT '子任务ID，多个逗号分隔',
  `trigger_status` tinyint(4) NOT NULL DEFAULT '0' COMMENT '调度状态：0-停止，1-运行',
  `trigger_last_time` bigint(13) NOT NULL DEFAULT '0' COMMENT '上次调度时间',
  `trigger_next_time` bigint(13) NOT NULL DEFAULT '0' COMMENT '下次调度时间',
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

xxl_job_log 这张表记录了job执行时的一些参数包括了执行次数执行参数 jobid，调度的时机和结果 xxl_job_log_report 记录了调度中心的统计数据 xxl_job_logglue记录了glue类型job的日志

xxl_job_registry 记录了在调度中心注册的执行器信息 xxl_job_group 记录了调度中心的分组情况 xxl_job_user 记录了用户信息 xxl_job_lock记录了锁信息其实我们通过表的设计就以了解一些xxljob的设计思想，下面我们就来看一下一个任务是如何被调度的

架构简述

执行器注册到调度中心
页面新建/修改job ->根据执行策略计算下次执行时间
执行线程轮询表 5s 执行job -> 阻塞策略
记录执行信息 ->更新下次执行

大家可以看到上图是一个简要的job执行的流程图在了解代码前和大家简要的介绍一下XXLJOB的通信结构，调度中心和执行器各维护一个netty的服务，双方使用HTTP通信，执行器的ip和端口在启动时会通过http调用通知调度中心，各位在启动服务时是可以看到XXLJOB对应的日志的，在看XXLJOB的代码时大家会看到一些以client或impl结尾的实现，对应的就是调度中心和执行器的netty的实现。

下面笔者带大家深入了解一下具体的代码在执行器端需配置一些参数，从而时执行器能够找到调度中心

包括了调度中心的地址，执行器的ip,端口（可以不填）,日志的过期时间（非必填）

代码详解

启动/初始化阶段在服务启动时会启动netty的服务

XxlJobSpringExecutor#afterSingletonsInstantiated()
->super.start()
->initEmbedServer();
->embedServer.start();
->startRegistry();
->ExecutorRegistryThread.getInstance().start()

最终启动一个名为registryThread的线程，在这个线程中调用 adminBiz.registry()方法，调用调度中心的接口实现注册此时调度中心与执行器（客户端）就建立起了连接此外在spring的bean初始化完成后在XxlJobAdminConfig这个类中还利用afterPropertiesSet实现了初始化了一些线程这些线程的作用分别为处理国际化对应的语言，快慢线程池的初始化，服务注册监控，任务执行监控，任务完成监控，日志报表处理，任务执行，限于篇幅，这些线程的作用我们下期再详细的解释，到现在为止，XXLJOB的启动就已经完成了。

页面处理

在调度中心的界面我们可以新建一个调度任务，其中有几个参数需要注意，jobhandler,运行模式，阻塞策略，cron。由于目前大部分是在springboot中运行的，所以大部分是依赖Bean执行的，jobHandler则是代码中定义的job名称。运行模式决定了调度中心选择执行器的逻辑

阻塞策略决定了连续执行时的策略

cron则决定了调度的时间

执行

下面我们深入了解一下一个job是如果被调度中心执行的。执行的逻辑在上文中提到的JobScheduleHelper中实现。首先会先获取锁，此处是利用mysql的for update实现的，然后会获取5秒内的任务列表

逐个遍历拿到的任务，如果任务已经超过了执行时间+预读时间（5s），根据过期策略决定是丢弃还是立即执行一次，并且更新下次执行时间如果当前时间大于任务下次执行时间，则将任务放入执行线程池，等待执行，更新下次执行时间，如果下次执行时间在5s内，则会将任务放入时间轮中，等待其他线程处理，再此更新此任务执行时间如果均不是上述两种情况，则会将任务放入时间轮中，更新执行时间，等待其他线程处理。


for (XxlJobInfo jobInfo: scheduleList) {
    // time-ring jump
    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());
        // 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());
    } 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、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()));
        }
    } 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、fresh next
        refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
    }
}