XXL-JOB- admin 源码

发表于

xxl-job 的 admin 服务是 xxl-job 的调度中心，负责管理和调度注册的 job，关于 xxl-job 的使用，可以阅读 “参考阅读” 中的《XXL-JOB分布式调度框架全面详解》，这里主要是介绍 admin 中的源码。

admin 服务除了管理页面上的一些接口外，还有一些核心功能，比如：

1、根据 job 的配置，自动调度 job；

2、接收 executor 实例的请求，实现注册和下线；

3、监视失败的 job，进行重试；

4、结束一些异常的 job；

5、清理和统计日志；

这些功能都是在 admin 服务启动后，在后台自动运行的，下面将详细介绍 admin 服务这些功能的实现。

XxlJobAdminConfig

admin 的配置类

XxlJobAdminConfig 是 admin 服务的配置类，在 admin 服务启动时，它除了配置 admin 服务的一些参数外，还会启动 admin 服务的所有后台线程。

属性

该类的属性主要分为5类：

1、配置文件中的参数，比如 accessToken；

2、DAO 层各个数据表的 mapper；

3、Spring 容器中的一些 Bean，比如 JobAlarmer、DataSource 等；

4、私有变量 XxlJobScheduler 对象；

1
java复制代码private XxlJobScheduler xxlJobScheduler;

5、私有静态变量 adminConfig，指向实例自身。

1
2
3
4
5
java复制代码private static XxlJobAdminConfig adminConfig = null;

public static XxlJobAdminConfig getAdminConfig() {
    return adminConfig;
}

方法

该类有两个重要方法，分别实现自接口 InitializingBean、DisposableBean，作用如下：

  • afterPropertiesSet 方法，在 Spring 容器中 Bean 初始化完成之后，在该方法中进行初始化；
  • destroy 方法，在容器销毁 Bean 时，会执行销毁操作；

这两个方法分别调用了 XxlJobScheduler 对象的 init、destroy 方法，源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
java复制代码@Override
public void afterPropertiesSet() throws Exception {
    adminConfig = this;

    xxlJobScheduler = new XxlJobScheduler();
    xxlJobScheduler.init();
}

@Override
public void destroy() throws Exception {
    xxlJobScheduler.destroy();
}

XxlJobAdminConfig 作为 admin 服务的配置类，作用就是在 Spring 容器启动时，调用 XxlJobScheduler 的初始化方法，来初始化和启动 admin 服务的功能。

XxlJobScheduler

XxlJobScheduler 的作用就是调用各个辅助类（xxxHelper）来启动和结束不同的线程和功能，初始化方法 init 的代码如下：

如果把 XxlJobScheduler 看做是一个启动器，那么 init 方法就是启动按钮，XxlJobAdminConfig 的作用就是按下这个按钮。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
java复制代码public void init() throws Exception {
  // 0、初始化国际化消息，不是很重要忽略
  initI18n();

  // 1、初始化调度器的线程池
  JobTriggerPoolHelper.toStart();

  // 2、启动注册监视器线程
  JobRegistryHelper.getInstance().start();

  // 3、启动 失败 job 监视器线程，查询失败日志进行重试
  JobFailMonitorHelper.getInstance().start();

  // 4、启动 丢失 job 监视器线程，一些 job 发出调度指令后，一直没有响应，状态一直是“运行中”
  JobCompleteHelper.getInstance().start();

  // 5、启动日志统计和清理线程
  JobLogReportHelper.getInstance().start();

  // 6、启动调度线程，定时调度 job
  JobScheduleHelper.getInstance().start();

  logger.info(">>>>>>>>> init xxl-job admin success.");
}

下面我们主要介绍 init 中各个类及其作用，最后再简单一下介绍 destroy 的作用。

JobTriggerPoolHelper

Trigger 线程池的辅助类：管理 Trigger 线程池、添加 trigger 线程到线程池

当 admin 服务向 executor 实例发出一个调度请求来执行 job 时，会调用 XxlJobTrigger.trigger() 方法把要传输的参数（比如 job_id、jobHandler、job_log_id、阻塞策略等，包装成 TriggerParam 对象）传给 ExecutorBiz 对象来执行一次调度。

xxl-job 对调度过程做了两个优化：

  • 每次发出调度请求时，会新建一个线程，异步执行 XxlJobTrigger 的方法；
  • 在新建线程时，会根据执行 XxlJobTrigger 方法的耗时，选择不同的线程池；

属性 和 start

初始化线程池

JobTriggerPoolHelper 在 toStart 方法中初始化了它的两个线程池属性，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
java复制代码/**
 * 快速、慢速线程池，分别执行调度任务不一样的任务，实现隔离，避免相互阻塞
 */
private ThreadPoolExecutor fastTriggerPool = null;
private ThreadPoolExecutor slowTriggerPool = null;

public void start() {
    fastTriggerPool = new ThreadPoolExecutor(
        10,
        // 至少200
        XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax(),60L,TimeUnit.SECONDS,new LinkedBlockingQueue<>(1000),      
        r -> new Thread(r, "xxl-job, admin JobTriggerPoolHelper-fastTriggerPool-" + r.hashCode()));

    slowTriggerPool = new ThreadPoolExecutor(
        10,
        // 至少100
        XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax(),60L,TimeUnit.SECONDS,new LinkedBlockingQueue<>(2000),
        r -> new Thread(r, "xxl-job, admin JobTriggerPoolHelper-slowTriggerPool-" + r.hashCode()));
}

每次有调度请求时，就会在这两个线程池中创建线程，创建线程的逻辑在 addTrigger 方法中。

addTrigger

新建线程，调用 XxlJobTrigger.trigger() 方法

不同 job 存在执行时长的差异，为了避免不同耗时 job 之间相互阻塞，xxl-job 根据 job 的响应时间，对 job 进行了区分，主要体现在：

  • 如果 job 耗时短，就在 fastTriggerPool 线程池中创建线程；
  • 如果 job 耗时长且调用频繁，就在 slowTriggerPool 线程池中创建线程；

如果快 job 与调用频繁的慢 job 在同一个线程池中创建线程，慢 job 会占用大量的线程，导致快 job 线程不能及时运行，降低了线程池和线程的利用率。xxl-job 通过快慢隔离，避免了这个问题。

问题：如果快慢 job 使用同一个线程池时，慢 job 占用了线程，导致快 job 线程不能及时运行，正常情况下，我们的反应是增加线程池的线程数，这样做能否解决问题？

不能，因为慢 job 还是会占用大量线程，抢占了快 job 的线程资源；增加线程池中的线程数不但没有提升利用率，还会导致大量线程看空闲，利用率反而降低了。最好的方法还是用两个线程池把两者隔离，可以合理地使用各自线程池的资源。

为了记录慢 job 的超时次数，代码中使用一个 map（变量 jobTimeoutCountMap ）来记录一分钟内 job 超时次数，key 值是 job_id，value 是超时次数。在调用 XxlJobTrigger.trigger() 方法之前，会先判断 map 中，该 job_id 的超时次数是否大于 10，如果大于10，就是使用 slowTriggerPool，代码如下：

1
2
3
4
5
6
7
8
9
java复制代码// 属性变量
private volatile ConcurrentMap<Integer, AtomicInteger> jobTimeoutCountMap = new ConcurrentHashMap<>();

// 选择线程池，如果在一分钟内调度超过10次，使用 slowTriggerPool
ThreadPoolExecutor triggerPool_ = fastTriggerPool;
AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId);
if (jobTimeoutCount != null && jobTimeoutCount.get() > 10) {
  triggerPool_ = slowTriggerPool;
}

调用 XxlJobTrigger.trigger() 方法后，根据两个值来更新 jobTimeoutCountMap 的值：

  • 当前时间与上次调用是否在一分钟以内，如果不在一分钟以内，就清空 map；
  • 本次 XxlJobTrigger.trigger() 的调用是否超过 500 毫秒，如果超过 500 毫秒，就在 map 中增加 job_id 的超时次数；

和上面的代码相结合，一个 job 在一分钟内有10次调用超过 500 毫秒，就认为该 job 是一个 频繁调度且耗时的 job。

代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
java复制代码// 属性变量，初始值等于 JobTriggerPoolHelper 对象构造时的分钟数
// 每次调用 XxlJobTrigger.trigger() 方法时，值等于上一次调用的分钟数
private volatile long minTim = System.currentTimeMillis() / 60000;

// 当前时间的分钟数，如果和前一次调用不在同一分钟内，就清空 jobTimeoutCountMap
long minTim_now = System.currentTimeMillis() / 60000;
if (minTim != minTim_now) {
    minTim = minTim_now;
    jobTimeoutCountMap.clear();
}

// 开始调用 XxlJobTrigger.trigger() 的时间
long start = System.currentTimeMillis();

// ... 调用 XxlJobTrigger.trigger() 方法
XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);

// 如果用时超过 500 毫秒，就增加一次它的慢调用次数
long cost = System.currentTimeMillis() - start;
if (cost > 500) {
    AtomicInteger timeoutCount = jobTimeoutCountMap.putIfAbsent(jobId, new AtomicInteger(1));
    if (timeoutCount != null) {
        timeoutCount.incrementAndGet();
    }
}

XxlJobTrigger.trigger() 方法在下面的 XxlJobTrigger 类中有详细介绍，这里只需要知道它会对一个 job 发起一次执行请求。

在该类中，属性变量 minTim 和 jobTimeoutCountMap 都使用 volatile 来修饰，保证了并发调用 addTrigger 时数据的一致性和可见性。

问题：为什么要每分钟清空一次 map 中的数据？

admin 服务发起 job 调度请求时，是在静态方法 public static void trigger() 中调用静态变量 private static JobTriggerPoolHelper helper 的 addTrigger 方法来发起请求的。minTim 和 jobTimeoutCountMap 虽然不是 static 修饰的，但可以看做是全局唯一的（因为持有它们的对象是全局唯一的），因此这两个参数维护的是 admin 服务全局的调度时间和超时次数，为了避免记录的数据量过大，需要每分钟清空一次数据的操作。

JobRegistryHelper

executor 注册和下线的辅助类

admin 服务提供了接口给 executor 来注册和下线，另外，当 executor 长时间（90秒）没有发心跳时，要把 executor 自动下线。前一个功能通过暴露一个接口来接收请求，后一个功能需要开启一个线程，定时更新过期 executor 的状态。

xxl-job 为了提升 admin 服务的性能，在前一个功能的接口接收到 executor 的请求时，不是同步执行，而是在线程池中开启一个线程，异步执行 executor 的注册和下线请求。

JobRegistryHelper 类就负责管理这个线程池和定时线程的。

注册和下线

线程池的定义和初始化代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
java复制代码// 注册或移除 executor 的线程池
private ThreadPoolExecutor registryOrRemoveThreadPool = null;

// 注册或移除线程池
registryOrRemoveThreadPool = new ThreadPoolExecutor(
    2,    10,    30L,    TimeUnit.SECONDS,
    new LinkedBlockingQueue<>(2000),
    r -> new Thread(r, "xxl-job, admin JobRegistryMonitorHelper-registryOrRemoveThreadPool-" + r.hashCode()),
    (r, executor) -> {
        r.run();
        logger.warn(">>>>>>>>>>> xxl-job, registry or remove too fast, match thread pool rejected handler(run now).");
    });

线程池的核心线程数是 2，最大线程数是10，允许一个 2000 的队列，如果 executor 实例很多，会导致注册延迟的。当然，一般不会把2000个 executor 注册到同一个 admin 服务。

executor 实例在发起注册和下线请求时，会调用 AdminBizImpl 类的对应方法，该类的方法如下：


可以看到，AdminBizImpl 类的两个方法都是调用了 JobRegistryHelper 方法来实现，其中 JobRegistryHelper.registry 方法代码如下（registryRemove 代码与之相似）：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
java复制代码public ReturnT<String> registry(RegistryParam registryParam) {
	// 校验参数
    if (!StringUtils.hasText(registryParam.getRegistryGroup())
        || !StringUtils.hasText(registryParam.getRegistryKey())
        || !StringUtils.hasText(registryParam.getRegistryValue())) {
        return new ReturnT<>(ReturnT.FAIL_CODE, "Illegal Argument.");
    }
	// 在线程池中创建线程
    registryOrRemoveThreadPool.execute(() -> {
        // update
        int ret = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao()
            .registryUpdate(registryParam.getRegistryGroup(), registryParam.getRegistryKey(),
                            registryParam.getRegistryValue(), new Date());
        // update 失败，insert
        if (ret < 1) {
            XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao()
                .registrySave(registryParam.getRegistryGroup(), registryParam.getRegistryKey(),
                              registryParam.getRegistryValue(), new Date());
            // 刷新，空方法
            freshGroupRegistryInfo(registryParam);
        }
    });
    return ReturnT.SUCCESS;
}

这两个方法是通过在线程池 registryOrRemoveThreadPool 中创建线程来异步执行请求，然后把数据更新或新建到数据表 xxl_job_registry 中。

更新和管理 Job_group

当 executor 注册到 admin 服务后（数据入库到 xxl_job_registry 表），是不会在页面上显示的，需要要用户手动添加 job_group 数据（添加到 xxl_job_group 表），admin 服务会自动把用户添加的 job_group 数据与 xxl_job_registry 数据关联。这就需要 admin 定时从 xxl_job_group 表读取数据，关联 xxl_job_registry 表和 xxl_job_group 表的数据。

这个功能是与 “executor 自动下线” 功能在同一个线程中实现，该线程的主要逻辑是：

  • 从 xxl_job_group 表查询出 “自动设置 address” 的 group 列表，如果 group 列表不为空，才继续向下执行；
  • 从 xxl_job_registry 表删除不再存活（90秒内都没有更新）的记录，避免无效记录影响后续操作；
  • 从 xxl_job_registry 表取出存活的记录，根据 appName 设置 xxl_job_group 记录的 address_list 值，多个 address 使用逗号拼接；
  • sleep 30 秒，这个线程每 30 秒执行一次。

相关代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
java复制代码// 注册监视器线程
private Thread registryMonitorThread;

// 停止标志位
private volatile boolean toStop = false;


// 自动注册的 job group
List<XxlJobGroup> groupList = XxlJobAdminConfig.getAdminConfig().getXxlJobGroupDao().findByAddressType(0);

// 删除已经下线（90 秒内没有心跳）的注册 admin/executor
List<Integer> ids = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findDead(RegistryConfig.DEAD_TIMEOUT, new Date());
if (ids != null && ids.size() > 0) {
    XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().removeDead(ids);
}

// 刷线还在线（90秒内有心跳）的 admin/executor 的地址 map<appName,List<address>>
List<XxlJobRegistry> list = XxlJobAdminConfig.getAdminConfig().getXxlJobRegistryDao().findAll(RegistryConfig.DEAD_TIMEOUT, new Date());
HashMap<String, List<String>> appAddressMap = new HashMap<>(list.size());
// 略...


// 每 30 秒执行一次
TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);

从这里可以看出，如果是对外接口（接收请求等）的功能，使用线程池和异步线程来实现；如果是一些自动任务，则是通过一个线程来定时执行。

JobFailMonitorHelper

Job 执行失败的监视线程辅助类

如果一个 Job 调度后，没有响应返回，需要定时重试。作为一种“自动执行”的任务，很显然可以像前面 JobRegistryHelper 一样，使用一个线程定时重试。

在这个类中，定义了一个监视线程，以每10 秒一次的频率运行，对失败的 job 进行重试。如果 job 剩余的重试次数大于0，就会 job 进行重试，并把发送告警信息。线程的定义如下：

1
2
3
4
java复制代码/**
 * 监视器线程
 */
private Thread monitorThread;

这里需要关注的问题是：当 admin 服务是集群部署时（共用一个数据库），怎么避免一个 job 被多个实例多次重试？需要有一个“分布式锁”。

加锁

在这个线程中，它利用 “数据库执行 UPDATE 语句时会加上互斥锁” 的特性，使用了 “基于数据库的分布式锁”，代码如下所示：

1
2
3
4
5
6
7
8
9
10
java复制代码// UPDATE 语句给该条记录加互斥锁，如果能加上，说明没有其他线程在修改该记录，也说明该记录还没被修改过
// 设置新值 -1，表示该记录已经被加锁了
int lockRet = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao()
  .updateAlarmStatus(failLogId, 0, -1);
if (lockRet < 1) {
  continue;
}

// 解锁
xlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateAlarmStatus(failLogId,-1, newAlarmStatus);

在这个语句中，会把 jobLog 的状态设置为 -1，这是一个无效状态值，当其他线程通过有效状态值来搜索失败记录时，会略过该记录，这样该记录就不会被其他线程重试，达到的分布式锁的功能（这个锁是一个行锁）。或者说，-1状态类似于 java 中的对象头的锁标志位，表明该记录已经被加锁了，其他线程会“忽略”该记录。

问题：这里的加锁解锁代码有什么问题？

在 try 代码块中加锁和解锁，如果加锁后重试时抛出异常，会导致该记录永远无法解锁。所以，应该在 finnally 块中执行解锁操作，或者使用 redis 给锁加一个过期时间来实现分布式锁。

重试

从失败的日志中取出 jobId，查询出对应的 jobInfo 数据，如果日志中的剩余重试次数大于 0，就执行重试。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
java复制代码// 取出失败的日志和对应的 job
XxlJobLog log = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().load(failLogId);
XxlJobInfo info = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(log.getJobId());

// 1、重试失败的 job，更新日志的 trigger_msg 字段值
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);
    // 更新 log
    XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateTriggerInfo(log);
}

// 2、如果 job 不为空，说明存在失败的任务，发送告警消息
// 告警状态：0-默认、-1=锁定状态、1-无需告警、2-告警成功、3-告警失败
int newAlarmStatus = 0;
if (info != null) {
    boolean alarmResult = XxlJobAdminConfig.getAdminConfig().getJobAlarmer().alarm(info, log);
    newAlarmStatus = alarmResult ? 2 : 3;
} else {
    newAlarmStatus = 1;
}

// 3、更新 jobLog 的 alarm_status 值
XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().updateAlarmStatus(failLogId, -1, newAlarmStatus);

调度任务使用的就是前面介绍的 JobTriggerPoolHelper.trigger 方法，最后更新 jobLog 的 alarm_status 值，有两个作用：

  • 释放分布式锁；
  • 日志记录的 alarm_status 被设置为大于 0 的值，不会再被作为失败日志查询出来（findFailJobLogIds 方法的查询条件之一是 alarm_status == 0），避免了在下一次线程执行时再被重试。

JobCompleteHelper

Job 完成线程的辅助类

这个类与 JobRegistryHelper 类似，都有一个线程池、一个线程，通过前面 JobRegistryHelper 的学习，可以大胆猜测：

  • 线程池用来创建线程，处理接收到的请求；
  • 线程用来执行执行一些定“定时任务”。

实际上，该类中线程池和线程的作用就是用来 “完成” 一个 job。

接收回调

当 executor 接收到 admin 的调度请求后，会异步执行 job，并立刻返回一个回调。

admin 接受到回调后，和前面的 “注册、下线” 一样，在线程池中创建线程来处理回调，主要是更新 job 和日志。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
java复制代码/**
 * 接收回调请求的线程池
 */
private ThreadPoolExecutor callbackThreadPool = null;

// 初始化线程池
callbackThreadPool = new ThreadPoolExecutor(
  2, 20, 30L, TimeUnit.SECONDS,
  new LinkedBlockingQueue<>(3000),
  r -> new Thread(r, "xxl-job, admin JobLosedMonitorHelper-callbackThreadPool-" + r.hashCode()),
  (r, executor) -> {
    r.run();
    logger.warn(">>>>>>>>>>> xxl-job, callback too fast, match threadpool rejected handler(run now).");
  });

当有回调请求时，public callback 方法（该方法被 AdminBizImpl 调用）会在线程池中创建一个线程，遍历回调请求的参数列表，依次处理回调参数，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
java复制代码// 在线程池中创建线程处理回调参数
public ReturnT<String> callback(List<HandleCallbackParam> callbackParamList) {

    callbackThreadPool.execute(new Runnable() {
        @Override
        public void run() {
            for (HandleCallbackParam handleCallbackParam : callbackParamList) {
                ReturnT<String> callbackResult = callback(handleCallbackParam);
                // ...
            }
        }
    });
    return ReturnT.SUCCESS;
}

private ReturnT<String> callback(HandleCallbackParam handleCallbackParam) {
    // valid log item
    XxlJobLog log = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().load(handleCallbackParam.getLogId());
    // 更新 log 数据，略...
    
    // 更新和完成 job
    XxlJobCompleter.updateHandleInfoAndFinish(log);

    return ReturnT.SUCCESS;
}

callBack 的调用顺序：JobApiController -> AdminBizImpl -> public allback -> private callback.

从代码可以看出，最后调用 XxlJobCompleter.updateHandleInfoAndFinish 方法完成回调逻辑。

更新 Job

如果一个 job 较长时间前被调度，但是一直处于 “运行中” 且它所属的 executor 已经超过 90 秒没有心跳了，那么可以认为该 job 已经丢失了，需要把该 job 结束掉。这个就是线程 monitorThread 的主要功能。

monitorThread 会以 60秒 一次的频率，从 xxl_job_log 表中找出 10分钟前调度、仍处于”运行中“状态、executor 已经下线 的 job，然后调用 XxlJobCompleter.updateHandleInfoAndFinish 来更新 handler 的信息和结束 job，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
java复制代码/**
 * 监视 丢失job 的线程
 */
private Thread monitorThread;

// 任务结果丢失处理：调度记录停留在 "运行中" 状态超过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());
        // 设置 handler_code
        jobLog.setHandleCode(ReturnT.FAIL_CODE);
        jobLog.setHandleMsg(I18nUtil.getString("joblog_lost_fail"));
        // 更新执行信息和结束 job
        XxlJobCompleter.updateHandleInfoAndFinish(jobLog);
    }
}

从代码可以看出，上面的两个功能最后都调用了 XxlJobCompleter.updateHandleInfoAndFinish 方法，关于该方法的介绍，可以看后面 XxlJobCompleter 部分的介绍，这里不详细展开。

JobLogReportHelper

Job 日志统计辅助类

如果去看 XxlJobTrigger.triger 方法，会发现每次调度 job 时，都会先新增一个 jobLog 记录，这也是为什么 JobFailMonitorHelper 中的线程在重试时，先查询 jobLog 的原因。

JobLog 作为 job 的调度记录，还可以用来统计一段时间内 job 的调度次数、成功数等；另外，会清理超出有效期（配置的参数 logretentiondays）的日志，避免日志数据过大。很显然，这又是一个 ”自动任务“，可以使用一个线程定时完成。

该类持有一个线程变量，线程以 每分钟一次的频率，执行两个操作：

  • 统计一段时间的 job 数据，主要统计指标有：总的调度次数、处于调度运行中的次数、调度失败的次数、调度成功的次数；
  • 清理过期的日志数。

在线程 run 方法的前半部分，线程会统计 3 天内，每天的调度次数、运行次数、成功运行数、失败次数；然后更新或新增 xxl_job_log_report 表的数据。

清理日志

在线程 run 方法的后半部分，线程按天对日志进行清理，如果当前时间与上次清理的时间相隔超过一天，就会清理日志记录，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
java复制代码// 根据上次执行时间、配置的过期参数，来决定是否执行清理
// 上次清理时间与当前超过1天才清理
long lastCleanLogTime = 0;


if (XxlJobAdminConfig.getAdminConfig().getLogretentiondays() > 0
    && System.currentTimeMillis() - lastCleanLogTime > 24 * 60 * 60 * 1000) {
  // 清理的开始时间... 略

  // 开始清理日志
  List<Long> logIds = null;
  do {
    logIds = XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().findClearLogIds(
      0, 0, clearBeforeTime, 0, 1000);
    if (logIds != null && logIds.size() > 0) {
      XxlJobAdminConfig.getAdminConfig().getXxlJobLogDao().clearLog(logIds);
    }
  } while (logIds != null && logIds.size() > 0);

  // 更新执行清理操作的时间
  lastCleanLogTime = System.currentTimeMillis();
}

问题：为什么要用 lastCleanLogTime 记录上一次的清理时间？每次执行时，不能直接清理一天前创建的数据吗？

如果不使用参数 lastCleanLogTime 来记录上次清理的时间，只是清理一天前创建的数据记录。那么该线程每分钟执行一次时，都会删除前天当前时刻的数据，导致前一年的数不完整。

使用参数 lastCleanLogTime 来记录上次清理的时间，并且与当前时间相差超过一天时才清理，能保证前一天的日志是完整的。

问题：为什么每次只删除 1000 条日志？

不明白为什么清理日志时，不是一次性删除全部的过期日志，而是每次删除 1000条。按理说，这些旧的日志数据应该已经不在 buffer pool 中了，trigger_time 字段又是普通索引，那么 DELETE 操作会先更新到 change buffer 中，之后再合并。现在先查询再删除，相当于多了一次 IO 且没有使用到 change buffer。

JobScheduleHelper

Job 调度辅助类

admin 服务是用来管理和调度 job 的，用户也可以在它的管理后台新建一个 job，配置 CRON 和 JobHandler，然后 admin 服务就会按照配置的参数来调度 job。很显然，这种“自动化工作”也是由线程定时执行的。

1、如果使用线程调度 Job，存在的第一个问题是：如果某个 Job 在调度时比较耗时，就可能阻塞后续的 Job，导致后续 job 的执行有延迟，怎么解决这个问题？

在前面 JobTriggerPoolHelper 我们已经知道，admin 在调度 job 时是 ”使用线程池、线程“ 异步执行调度任务，避免了主线程的阻塞。

2、使用线程定时调度 job，存在的第二个问题是：怎么保证 job 在指定的时间执行，而不会出现大量延迟？

admin 使用 ”预读“ 的方式，提前读取在未来一段时间内要执行的 job，提前取到内存中，并使用 “时间轮算法” 按时间分组 job，把未来要执行的 job 下一个时间段执行。

3、还隐藏第三个问题：admin 服务是可以多实例部署的，在这种情况下该怎么避免一个 job 被多个实例重复调度？

admin 把一张数据表作为 “分布式锁” 来保证只有一个 admin 实例能执行 job 调度，又通过随机 sleep 线程一段时间，来降低线程之间的竞争。

下面我们就通过代码来了解 xxl-job 是怎么解决上述问题的。

调度线程

在该类中，定义了一个调度线程，用来调度要执行的 job 和已经过期一段时间的 job，定义代码如下：

1
2
3
4
5
6
7
8
java复制代码/**
 * 预读的毫秒数
 */
public static final long PRE_READ_MS = 5000;
/**
 * 预读和调度过期任务的线程
 */
private Thread scheduleThread;

预读

下面代码中的 pushTimeRing，是把 job 添加到一个 map 对象 ringData 中，然后让另一个线程从该 map 对象中取出，再次调度

该线程会预读出 “下次执行时间 <= now + 5000 毫秒内” 的部分 job，根据它们下一次执行时间划分成三段，执行三种不同的逻辑。

1、下次执行时间在 (−∞,now−5000)(-∞, now - 5000)(−∞,now−5000) 范围内

说明过期时间已经大于 5000 毫秒，这时如果过期策略要求调度，就调度一次。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
java复制代码if (nowTime > jobInfo.getTriggerNextTime() + PRE_READ_MS) {
  logger.warn(">>>>>>>>>>> xxl-job, schedule misfire, jobId = " + jobInfo.getId());

  MisfireStrategyEnum misfireStrategyEnum = MisfireStrategyEnum.match(jobInfo.getMisfireStrategy(), MisfireStrategyEnum.DO_NOTHING);
  if (MisfireStrategyEnum.FIRE_ONCE_NOW == misfireStrategyEnum) {
    // 调度一次
    JobTriggerPoolHelper.trigger(jobInfo.getId(), TriggerTypeEnum.MISFIRE, -1, null, null, null);
    logger.debug(">>>>>>>>>>> xxl-job, schedule push trigger : jobId = " + jobInfo.getId());
  }
  // 更新下一次执行时间
  refreshNextValidTime(jobInfo, new Date());
}

2、下次执行时间在 [now−5000,now)[now - 5000, now)[now−5000,now) 范围内

说明过期时间小于5000毫秒，只能算是延迟不能算是过期，直接调度一次，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
java复制代码if (nowTime > jobInfo.getTriggerNextTime()) {
    // 1、 调度一次
    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());

    // 3、 如果当前 job 处于 ”可以被调度“ 的状态，且下一次执行时间在 5000 毫秒内，就记录下 job Id，等待后面轮询
    if (jobInfo.getTriggerStatus() == 1 && nowTime + PRE_READ_MS > jobInfo.getTriggerNextTime()) {
        // 下次调度的时刻：秒
        int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);
        // 保存进 ringData 中
        pushTimeRing(ringSecond, jobInfo.getId());
        // 刷新下一次的调度时间
        refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));
    }
}

如果 job 的下一次执行时间在 5000 毫秒以内，为了省下下次预读的 IO 耗时，这里会记录下 job id，等待后面的调度。

3、下次执行时间在 [now,now+5000)[now, now + 5000)[now,now+5000) 范围内

说明还没到执行时间，先记录下 job id，等待后面的调度，代码如下：

1
2
3
java复制代码int ringSecond = (int) ((jobInfo.getTriggerNextTime() / 1000) % 60);
pushTimeRing(ringSecond, jobInfo.getId());
refreshNextValidTime(jobInfo, new Date(jobInfo.getTriggerNextTime()));

上面的3个步骤结束后，会更新 jobInfo 的 trigger_last_time、trigger_next_time、trigger_status 字段：

1
2
3
4
java复制代码// 更新 job 数据
for (XxlJobInfo jobInfo : scheduleList) {
  XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().scheduleUpdate(jobInfo);
}

可以看到，通过预读，一方面会把过期一小段时间的 job 执行一遍，另一方面会把未来一小段时间内要执行的 job 取出，保存进一个 map 对象 ringData 中，等待另一个线程调度。这样就避免了某些 job 到了时间还没执行。

分布式锁

因为 admin 是可以多实例部署的，所以在调度 job 时，需要考虑怎么避免 job 被多次调度。

xxl-job 在前面 JobFailMonitorHelper 中遍历失败的 job 时，会对每个 job 设置一个无效的状态作为 ”分布式行锁“，如果设置失败就跳过。而在这里，如果还使用该方法，有可能出现，一个 job 被设置为无效状态后，线程就崩溃了，导致该 job 永远无法被调度。因此，要尽量避免对 job 状态的修改。

在这里，admin 服务使用一张表 xxl_job_lock 作为分布式锁，每个 admin 实例都要先尝试获取该表的锁，获取成功才能继续执行；同时，为了降低不同实例之间的竞争，会在线程开始执勤随机 sleep 一段时间。

如何获取分布式锁？

在线程中会开启一个事务，设置为手动提交，然后对表 xxl_job_lock 执行 FOR UPDATE 查询。如果该线程执行语句成功，其他实例的线程就会排队等待该表的锁，实现了分布式锁功能。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
java复制代码// 获取数据库链接，通过 SELECT FOR UPDATE 来尝试获取 X锁
// 在事务提交前一直持有该锁，其他实例的线程想获取该锁就会失败，并且会排队等待，直到第一个事务提交释放或锁超时
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();


// 提交事务，释放 X锁
if (conn != null) {
  try {
    conn.commit();
  } catch (SQLException e) {
    // ...
  }
  try {
    conn.setAutoCommit(connAutoCommit);
  } catch (SQLException e) {
    // ...
  }
  try {
    conn.close();
  } catch (SQLException e) {
    // ...
  }
}

怎么降低锁的竞争？

为了降低锁竞争，在线程开始前会先 sleep 4000～5000 毫秒的随机值（不能大于 5000 毫秒，5000 毫秒是预读的时间范围）；在线程结束当前循环时，会根据耗时和是否有预读数据，选择不同的 sleep 策略：

  • 耗时超过1000 毫秒，不sleep，直接开始下一次循环；
  • 耗时小于1000 毫秒，根据是否有预读数据，sleep 一个大小不同的随机时长：
    • 有预读数据，sleep 时间短一些，在 0～1000 毫秒范围内；
    • 没有预读数据，sleep 时间长一些，在 0～4000 毫秒范围内；

代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
java复制代码try {
  // 随机 sleep 4000~5000 毫秒，通过这种方式，降低多实例部署时对锁的竞争
  // 这里也看出来，最多部署 5000 台实例 ==..==
  TimeUnit.MILLISECONDS.sleep(5000 - System.currentTimeMillis() % 1000);
} catch (InterruptedException e) {
  if (!scheduleThreadToStop) {
    logger.error(e.getMessage(), e);
  }
}


// 耗时超过 1000 毫秒，就不 sleep
// 不超过 1000 ms，就 sleep 一个随机时长
long cost = System.currentTimeMillis() - start;
if (cost < 1000) {
  try {
    // 没有预读数据，就 sleep 时间长一点；有预读数据，就 sleep 时间短一些
    TimeUnit.MILLISECONDS.sleep((preReadSuc ? 1000 : PRE_READ_MS) - System.currentTimeMillis() % 1000);
  } catch (InterruptedException e) {
    if (!scheduleThreadToStop) {
      logger.error(e.getMessage(), e);
    }
  }
}

ringThread：时间轮

在前面的线程中，对即将要开始的 job，不是立刻调度，而是按照执行的时刻（秒），把 job id 保存进一个 map 中，然后由 ringThread 线程按时刻进行调度，这只典型的“时间轮算法”。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
java复制代码/**
 * 调度线程2
 */
private Thread ringThread;
/**
 * 按时刻（秒）调度 job
 */
private volatile static Map<Integer, List<Integer>> ringData = new ConcurrentHashMap<>();

// 调度过程

List<Integer> ringItemData = new ArrayList<>();
// 每次取出 2 个时刻的 job 来调度
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);
  }
}
// 遍历 job Id，执行调度
if (ringItemData.size() > 0) {
  for (int jobId : ringItemData) {
    JobTriggerPoolHelper.trigger(jobId, TriggerTypeEnum.CRON, -1, null, null, null);
  }
  ringItemData.clear();
}

每次轮询调度时，只取出当前时刻（秒）、前一秒内的 job，不会去调度与现在相隔太久的 job。

在执行轮询调度前，有一个时间在 0～1000 毫秒范围内的 sleep。如果没有这个 sleep，该线程会一直执行，而 ringData 中当前时刻（秒）的数据可能已经为空，会导致大量无效的操作；增加了这个 sleep 之后，可以避免这种无效的操作。之所以 sleep 时间在 1000 毫秒以内，是因为调度时刻最小精确到秒，一秒的 sleep 可以避免 job 的延迟。

1
2
3
4
5
6
7
java复制代码try {
  TimeUnit.MILLISECONDS.sleep(1000 - System.currentTimeMillis() % 1000);
} catch (InterruptedException e) {
  if (!ringThreadToStop) {
    logger.error(e.getMessage(), e);
  }
}

问题：为什么在时间轮调度时，没有加分布式锁？

因为在前面的 scheduleThread 线程中，最后一个操作是把 job 的 next_trigger_time 值更新为大于 now + 5000 毫秒，其他 admin 实例 scheduleThread 线程的查询条件是：next_trigger_time < now + 5000，不会查询出这里调度的 job，所以不需要加分布式锁。

至此，XxlJobScheduler-init 方法的作用我们介绍完毕，下面我们简单介绍一下 XxlJobScheduler-destroy 方法

XxlJobScheduler-destroy

destroy 方法很简单，就是销毁前面初始化的线程池和线程，它销毁的顺序与前面启动的顺序相反。

代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
java复制代码/**
 * 销毁，销毁过程与 init 顺序相反
 */
public void destroy() throws Exception {

  // 1、销毁 调度线程
  JobScheduleHelper.getInstance().toStop();

  // 2、销毁 日志统计和清理线程
  JobLogReportHelper.getInstance().toStop();

  // 3、销毁 丢失 job 监视器线程
  JobCompleteHelper.getInstance().toStop();

  // 4、销毁 失败 job 监视器线程
  JobFailMonitorHelper.getInstance().toStop();

  // admin registry stop
  JobRegistryHelper.getInstance().toStop();

  // admin trigger pool stop
  JobTriggerPoolHelper.toStop();
}

因为各个 toStop 方法都很相似，所以我们只介绍 JobScheduleHelper 的 toStop 方法。

该方法的步骤如下：

1、设置停止标志位为 true；

2、sleep 一段时间，让出 CPU 时间片给线程执行任务；

3、如果线程不是终止状态（线程正在 sleep），中断它；

4、线程执行 join 方法，直到线程结束，执行最后一次。

代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
java复制代码scheduleThreadToStop = true;
// 给线程 1s 的时间去执行任务
try {
  TimeUnit.SECONDS.sleep(1);
} catch (InterruptedException e) {
  logger.error(e.getMessage(), e);
}
// 如果线程不是终止状态，就让它执行完所有任务
if (scheduleThread.getState() != Thread.State.TERMINATED) {
  scheduleThread.interrupt();
  try {
    scheduleThread.join();
  } catch (InterruptedException e) {
    logger.error(e.getMessage(), e);
  }
}

至此，JobScheduleHelper 的主要功能就介绍完了，可以看出， admin 服务在启动时，启动了多个线程池和线程，异步执行任务和异步响应 executor 的请求。

下面，我们介绍前面涉及到的 XxlJobTrigger 和 XxlJobCompleter。

XxlJobTrigger

调度 job 时的封装类

XxlJobTrigger 是调度 job 时的封装类，它主要工作就是接受传入的 jobId、调度参数等，查询对应的 jobGroup、jobInfo，然后调用 ExecutorBiz 对象来执行调度（run 方法）。

注意：这个类本身不会执行 http 请求，http 请求是在 core 包下的工具类 XxlJobRemotingUtil 中执行的。

该类中三个核心方法及其调用关系如下：trigger -> processTrigger -> runExecutor，

trigger

该方法的功能比较简单，就是根据传入的参数查询 jobGroup 和 jobInfo 对象，设置相关的字段值，然后调用 processTrigger 方法。

processTrigger

该方法的主要工作分为以下几步：

1、保存一条调度日志；

2、从 jobInfo、jobGroup 中取出字段值，构造 TriggerParam 对象；

3、根据 jobInfo 的路由策略，从 jobGroup 中取出要调度的 executor 地址；

4、调用 runExecutor 方法执行调度；

5、保存调度参数、设置调度信息、更新日志。

这里不会修改 jobInfo、jobGroup 对象的字段值，只取出字段值来使用，对这两个对象字段的修改，是在前一步 trigger方法中进行的。

runExecutor

该方法会执行调度，并返回调度结果，它的核心代码如下：

1
2
java复制代码ExecutorBiz executorBiz = XxlJobScheduler.getExecutorBiz(address);
runResult = executorBiz.run(triggerParam);

这里使用 XxlJobScheduler 类取出 ExecutorBiz 对象，以 “懒加载” 的方式给每个 address 创建一个 ExecutorBiz 对象，代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
java复制代码private static ConcurrentMap<String, ExecutorBiz> executorBizRepository = new ConcurrentHashMap<String, ExecutorBiz>();

public static ExecutorBiz getExecutorBiz(String address) throws Exception {
  // valid
  if (address == null || address.trim().length() == 0) {
    return null;
  }

  // load-cache
  address = address.trim();
  ExecutorBiz executorBiz = executorBizRepository.get(address);
  if (executorBiz != null) {
    return executorBiz;
  }

  // set-cache
  executorBiz = new ExecutorBizClient(address, XxlJobAdminConfig.getAdminConfig().getAccessToken());

  executorBizRepository.put(address, executorBiz);
  return executorBiz;
}

吐槽一句：这个功能完全可以放在 XxlJobTrigger 类中、或者封装在 ExecutorBiz 内部，不知道为什么要放在 XxlJobScheduler，平白无故多了一层调用。

可以看出，该类中的三个方法其实可以归类为：pre -> execute -> post，在执行前、执行时、执行后做一些前置和收尾工作。

XxlJobCompleter

job 的完成类

该类在前面 JobCompleteHelper 中被使用，最终 job 的完成就是在该类中执行的，该类有两个主要方法：

  • updateHandleInfoAndFinish：公共方法，调用 finishJob 方法和更新日志；
  • finishJob：私有方法，执行子任务和更新日志；

下面主要介绍 finishJob 方法。

finishJob

finishJob 的主要功能是：如果当前任务执行成功了，就调度它的所有子任务，最后把子任务的调度消息添加到当前 job 的日志中。代码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
java复制代码private static void finishJob(XxlJobLog xxlJobLog) {
  // 1、job 执行成功，开始调度子任务handle success, to trigger child job
  StringBuilder triggerChildMsg = null;
  if (XxlJobContext.HANDLE_CODE_SUCCESS == xxlJobLog.getHandleCode()) {
    XxlJobInfo xxlJobInfo = XxlJobAdminConfig.getAdminConfig().getXxlJobInfoDao().loadById(xxlJobLog.getJobId());
    if (xxlJobInfo != null && xxlJobInfo.getChildJobId() != null && xxlJobInfo.getChildJobId().trim().length() > 0) {
	  // 2、遍历子任务ID
      String[] childJobIds = xxlJobInfo.getChildJobId().split(",");
      for (int i = 0; i < childJobIds.length; i++) {
        int childJobId = (childJobIds[i] != null && childJobIds[i].trim().length() > 0 && isNumeric(childJobIds[i])) ? Integer.parseInt(childJobIds[i]) : -1;
        if (childJobId > 0) {
          // 3、调度子任务
          JobTriggerPoolHelper.trigger(childJobId, TriggerTypeEnum.PARENT, -1, null, null, null);
          ReturnT<String> triggerChildResult = ReturnT.SUCCESS;
          // 4、添加日志信息....略 
      }
    }
  }
  // 5、保存子任务的调度消息到日志
  if (triggerChildMsg != null) {
    xxlJobLog.setHandleMsg(xxlJobLog.getHandleMsg() + triggerChildMsg);
  }
}

需要注意的是：

1、这里依赖于 JobTriggerPoolHelper 来调度 job，所以在 JobCompleteHelper 的监视线程开始时，有一个 50 秒的等待，就是等待 JobTriggerPoolHelper 启动完成；

2、在 finishJob 方法中，调度子任务的时候，默认子任务的调度结果是成功，注意，这里是指 “调度” 这个行为是成功的，而不是指子任务执行是成功的。

总结

1、XxlJobAdminConfig 作为 admin 服务的启动入口，要尽可能保持简洁，作用类似于一个仓库，来管理和持有所有的类和对象，并不会去启动具体的线程，它只需要“按下启动器的按钮”就可以了；

2、XxlJobScheduler 是 admin 服务的启动器类，它会调用各个辅助类（xxxHelper）来启动对应的线程；

3、对外的接口，比如调度 job、接收注册或下线等，都是使用线程池 + 线程 的异步方式实现，避免 job 对主线程的阻塞；

4、对“自动任务“类的功能，都是使用线程定时执行；

参考阅读

XXL-JOB分布式调度框架全面详解

时间轮算法

一个开源的时间轮算法介绍

原文链接：mp.weixin.qq.com/s/Qti9UEpUL…

本文转载自: 掘金

开发者博客 – 和开发相关的 这里全都有

0%