xxl-job

系统说明

安装

安装部署参考文档：分布式任务调度平台xxl-job

功能

定时调度、服务解耦、灵活控制跑批时间（停止、开启、重新设定时间、手动触发）

XXL-JOB是一个轻量级分布式任务调度平台，其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线，开箱即用

概念

执行器列表：一个执行器是一个项目

任务：一个任务是一个项目中的 JobHandler

一个xxl-job服务可以有多个执行器（项目），一个项目下可以有多个任务（JobHandler），他们是如何关联的？

页面操作：

1. 在管理平台可以新增执行器（项目）
2. 在任务列表可以指定执行器（项目）下新增多个任务（JobHandler）

代码操作：

1. 项目配置中增加 xxl.job.executor.appname = "执行器名称"
2. 在实现类中增加 @JobHandler(value="xxl-job-demo") 注解，并继承 IJobHandler

架构图

抛出疑问

1. 调度中心启动过程？
2. 执行器启动过程？
3. 执行器如何注册到调度中心？
4. 调度中心怎么调用执行器？
5. 集群调度时如何控制一个任务在该时刻不会重复执行
6. 集群部署应该注意什么？

系统分析

执行器依赖jar包

com.xuxueli:xxl-job-core:2.1.0

com.xuxueli:xxl-registry-client:1.0.2

com.xuxueli:xxl-rpc-core:1.4.1

调度中心启动过程

// 1. 加载 XxlJobAdminConfig，adminConfig = this
XxlJobAdminConfig.java

// 启动过程代码
@Component
public class XxlJobScheduler implements InitializingBean, DisposableBean {
    private static final Logger logger = LoggerFactory.getLogger(XxlJobScheduler.class);

    @Override
    public void afterPropertiesSet() throws Exception {
        // init i18n
        initI18n();

        // admin registry monitor run
        // 2. 启动注册监控器（将注册到register表中的IP加载到group表）/ 30执行一次
        JobRegistryMonitorHelper.getInstance().start();

        // admin monitor run
        // 3. 启动失败日志监控器（失败重试，失败邮件发送）
        JobFailMonitorHelper.getInstance().start();

        // admin-server
        // 4. 初始化RPC服务
        initRpcProvider();

        // start-schedule
        // 5. 启动定时任务调度器（执行任务，缓存任务）
        JobScheduleHelper.getInstance().start();

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

执行器启动过程

@Override
public void start() throws Exception {

    // init JobHandler Repository
    // 将执行 JobHandler 注册到缓存中 jobHandlerRepository（ConcurrentMap）
    initJobHandlerRepository(applicationContext);

    // refresh GlueFactory
    // 刷新GLUE
    GlueFactory.refreshInstance(1);

    // super start
    // 核心启动项
    super.start();
}

public void start() throws Exception {
    // 初始化日志路径 
    // private static String logBasePath = "/data/applogs/xxl-job/jobhandler";
    XxlJobFileAppender.initLogPath(this.logPath);
    // 初始化注册中心列表 （把注册地址放到 List）
    this.initAdminBizList(this.adminAddresses, this.accessToken);
    // 启动日志文件清理线程 （一天清理一次）
    // 每天清理一次过期日志，配置参数必须大于3才有效
    JobLogFileCleanThread.getInstance().start((long)this.logRetentionDays);
    // 开启触发器回调线程
    TriggerCallbackThread.getInstance().start();
    // 指定端口
    this.port = this.port > 0 ? this.port : NetUtil.findAvailablePort(9999);
    // 指定IP
    this.ip = this.ip != null && this.ip.trim().length() > 0 ? this.ip : IpUtil.getIp();
    // 初始化RPC 将执行器注册到调度中心 30秒一次
    this.initRpcProvider(this.ip, this.port, this.appName, this.accessToken);
}

执行器注册到调度中心

执行器

// 注册执行器入口
XxlJobExecutor.java->initRpcProvider()->xxlRpcProviderFactory.start();

// 开启注册
XxlRpcProviderFactory.java->start();

// 执行注册
ExecutorRegistryThread.java->start();
// RPC 注册代码
for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
    try {
        ReturnT<String> registryResult = adminBiz.registry(registryParam);
        if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
            registryResult = ReturnT.SUCCESS;
            logger.debug(">>>>>>>>>>> xxl-job registry success, registryParam:{}, registryResult:{}", new Object[]{r