xxl-job的原理（1）

xxl-job的架构

系统组成

1. 调度中心：进行任务统一调度，可以新增和管理执行器和任务；
2. 执行器：任务执行依赖的组件，一个执行器可以关联多个任务，添加的执行器可以自动注册到调度中心上；
3. 任务：具体的业务执行逻辑，针对该逻辑可以在调度中心配置定时策略，定时策略支持cron表达式和固定速率；

系统架构

xxl-job的主要流程

执行器注册到调度中心

XxlJobConfig 配置类

每创建一个执行器服务需要新建XxlJobConfig 配置类，其核心是创建XxlJobSpringExecutor配置类。

@Configuration
public class XxlJobConfig {
   
   
    private Logger logger = LoggerFactory.getLogger(XxlJobConfig.class);

    @Value("${xxl.job.admin.addresses}")
    private String adminAddresses;

    @Value("${xxl.job.accessToken}")
    private String accessToken;

    @Value("${xxl.job.executor.appname}")
    private String appname;

    @Value("${xxl.job.executor.address}")
    private String address;

    @Value("${xxl.job.executor.ip}")
    private String ip;

    @Value("${xxl.job.executor.port}")
    private int port;

    @Value("${xxl.job.executor.logpath}")
    private String logPath;

    @Value("${xxl.job.executor.logretentiondays}")
    private int logRetentionDays;


    //my-创建XxlJobSpringExecutor配置类
    @Bean
    public XxlJobSpringExecutor xxlJobExecutor() {
   
   
        logger.info(">>>>>>>>>>> xxl-job config init.");
        XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
        xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
        xxlJobSpringExecutor.setAppname(appname);
        xxlJobSpringExecutor.setAddress(address);
        xxlJobSpringExecutor.setIp(ip);
        xxlJobSpringExecutor.setPort(port);
        xxlJobSpringExecutor.setAccessToken(accessToken);
        xxlJobSpringExecutor.setLogPath(logPath);
        xxlJobSpringExecutor.setLogRetentionDays(logRetentionDays);

        return xxlJobSpringExecutor;
    }

    /**
     * 针对多网卡、容器内部署等情况，可借助 "spring-cloud-commons" 提供的 "InetUtils" 组件灵活定制注册IP；
     *
     *      1、引入依赖：
     *          <dependency>
     *             <groupId>org.springframework.cloud</groupId>
     *             <artifactId>spring-cloud-commons</artifactId>
     *             <version>${version}</version>
     *         </dependency>
     *
     *      2、配置文件，或者容器启动变量
     *          spring.cloud.inetutils.preferred-networks: 'xxx.xxx.xxx.'
     *
     *      3、获取IP
     *          String ip_ = inetUtils.findFirstNonLoopbackHostInfo().getIpAddress();
     */


}

XxlJobSpringExecutor执行器实现类

XxlJobSpringExecutor类在xxl-job-core包中，其继承或实现 XxlJobExecutor ,ApplicationContextAware, SmartInitializingSingleton, DisposableBean类。

这三个接口的作用如下：

1. ApplicationContextAware：
   • 这个接口允许一个Spring bean意识到它所属的ApplicationContext（应用程序上下文）。
   • 实现了ApplicationContextAware接口的bean可以访问和与其所在的应用程序上下文交互。
2. SmartInitializingSingleton：
   • 这个接口用于那些需要在所有单例（singleton）bean创建完成后执行某些初始化逻辑的bean。
   • 该接口提供了afterSingletonsInstantiated()方法，该方法在所有单例bean创建完成后调用。
3. DisposableBean：
   • 这个接口用于那些需要在bean销毁（即容器关闭）时执行清理操作的bean。
   • 当应用程序上下文关闭时，Spring容器会调用DisposableBean接口的destroy()方法来销毁bean。

这些接口是Spring框架提供的扩展点，允许开发者编写自定义逻辑以处理ApplicationContext的管理、单例bean的初始化和销毁等方面的任务。根据具体的需求，您可以选择实现其中一个或多个接口来定制您的bean的行为。

    // start
    @Override
    public void afterSingletonsInstantiated() {

        // init JobHandler Repository
        /*initJobHandlerRepository(applicationContext);*/

        // init JobHandler Repository (for method)
        //my-bean实例化后，初始化JobHandler
        initJobHandlerMethodRepository(applicationContext);

        // refresh GlueFactory
        GlueFactory.refreshInstance(1);

        // super start
        try {
        //my-执行器启动的方法
            super.start();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }


    ---
     //my-初始化JobHandler，将其实例化到spring中，将任务注册到调度中心
    private void initJobHandlerMethodRepository(ApplicationContext applicationContext) {
        if (applicationContext == null) {
            return;
        }
        // init job handler from method
        String[] beanDefinitionNames = applicationContext.getBeanNamesForType(Object.class, false, true);
        for (String beanDefinitionName : beanDefinitionNames) {

            // get bean
            Object bean = null;
            Lazy onBean = applicationContext.findAnnotationOnBean(beanDefinitionName, Lazy.class);
            if (onBean!=null){
                logger.debug("xxl-job annotation scan, skip @Lazy Bean:{}", beanDefinitionName);
                continue;
            }else {
                bean = applicationContext.getBean(beanDefinitionName);
            }

            // filter method
            Map<Method, XxlJob> annotatedMethods = null;   // referred to ：org.springframework.context.event.EventListenerMethodProcessor.processBean
            try {
                annotatedMethods = MethodIntrospector.selectMethods(bean.getClass(),
                        new MethodIntrospector.MetadataLookup<XxlJob>() {
                            @Override
                            public XxlJob inspect(Method method) {
                                return AnnotatedElementUtils.findMergedAnnotation(method, XxlJob.class);
                            }
                        });
            } catch (Throwable ex) {
                logger.error("xxl-job method-jobhandler resolve error for bean[" + beanDefinitionName + "].", ex);
            }
            if (annotatedMethods==null || annotatedMethods.isEmpty()) {
                continue;
            }

            // generate and regist method job handler
            //my-将bean实例和方法注册到本地缓存中
            for (Map.Entry<Method, XxlJob> methodXxlJobEntry : annotatedMethods.entrySet()) {
                Method executeMethod = methodXxlJobEntry.getKey();
                XxlJob xxlJob = methodXxlJobEntry.getValue();
                // regist
                registJobHandler(xxlJob, bean, executeMethod);
            }

        }
    }

super.start();这一句是执行器启动的方法，在XxlJobExecutor接口类有具体试实现，在后面有详细说明。

registJobHandler(name, new MethodJobHandler(bean, executeMethod, initMethod, destroyMethod));这句是将JobHandler注册到本地缓存jobHandlerRepository中，该本地缓存由ConcurrentMap来实现，在执行放到本地缓存中前，还解析除了初始化方法和销毁方法，包装成MethodJobHandler放到缓存中。

// ---------------------- job handler repository ----------------------
    //my-将JobHandler注册到本地缓存中
    private static ConcurrentMap<String, IJobHandler> jobHandlerRepository = new ConcurrentHashMap<String, IJobHandler>();
    public static IJobHandler loadJobHandler(String name){
   
   
        return jobHandlerRepository.get(name);
    }
    public static IJobHandler registJobHandler(String name, IJobHandler jobHandler){
   
   
        logger.info(">>>>>>>>>>> xxl-job register jobhandler success, name:{}, jobHandler:{}", name, jobHandler);
        return jobHandlerRepository.put(name, jobHandler);
    }
    protected void registJobHandler(XxlJob xxlJob, Object bean, Method executeMethod){
   
   
        if (xxlJob == null) {
   
   
            return;
        }

        String name = xxlJob.value();
        //make and simplify the variables since they'll be called several times later
        Class<?> clazz = bean.getClass();
        String methodName = executeMethod.getName();
        if (name.trim().length() == 0) {
   
   
            throw new RuntimeException("xxl-job method-jobhandler name invalid, for[" + clazz + "#" + methodName + "] .");
        }
        if (loadJobHandler(name) != null) {
   
   
            throw new RuntimeException("xxl-job jobhandler[" + name + "] naming conflicts.");
        }

        // execute method
        /*if (!(method.getParameterTypes().length == 1 && method.getParameterTypes()[0].isAssignableFrom(String.class))) {
            throw new RuntimeException("xxl-job method-jobhandler param-classtype invalid, for[" + bean.getClass() + "#" + method.getName() + "] , " +
                    "The correct method format like \" public ReturnT<String> execute(String param) \" .");
        }
        if (!method.getReturnType().isAssignableFrom(ReturnT.class)) {
            throw new RuntimeException("xxl-job method-jobhandler return-classtype invalid, for[" + bean.getClass() + "#" + method.getName() + "] , " +
                    "The correct method format like \" public ReturnT<String> execute(String param) \" .");
        }*/

        executeMethod.setAccessible(true);

        // init and destroy
        Method initMethod = null;
        Method destroyMethod = null;

        if (xxlJob.init().trim().length() > 0) {
   
   
            try {
   
   
                initMethod = clazz.getDeclaredMethod(xxlJob.init());
                initMethod.setAccessible(true);
            } catch (NoSuchMethodException e) {
   
   
                throw new RuntimeException("xxl-job method-jobhandler initMethod invalid, for[" + clazz + "#" + methodName + "] .");
            }
        }
        if (xxlJob.destroy().trim().length() > 0) {
   
   
            try {
   
   
                destroyMethod = clazz.getDeclaredMethod(xxlJob.destroy());
                destroyMethod.setAccessible(true);
            } catch (NoSuchMethodException e) {
   
   
                throw new RuntimeException("xxl-job method-jobhandler destroyMethod invalid, for[" + clazz + "#" + methodName + "] .");
            }
        }

        // registry jobhandler
        registJobHandler(name, new MethodJobHandler(bean, executeMethod, initMethod, destroyMethod));

    }

XxlJobExecutor执行器接口类

XxlJobExecutor是执行器的接口类（不是接口，但规定了执行器方法），其主要的执行逻辑是在 start()方法中：

1. 调用XxlJobFileAppender.initLogPath(logPath);方法初始日志目录；
2. 调用initAdminBizList(adminAddresses, accessToken);初始化远程调用的客户端列表，后续这些客户端实现执行器和调度中心的通信；
3. 调用JobLogFileCleanThread.getInstance().start(logRetentionDays);实现日志文件的异步清除，通过创建一个异步线程来实现；
4. 调用TriggerCallbackThread.getInstance().start();创建回调线程；
5. initEmbedServer(address, ip, port, appname, accessToken);初始化内嵌服务器。

//my-执行器的父类，主要定义了启动和停止等一些方法
public class XxlJobExecutor  {
   
   
    private static final Logger logger = LoggerFactory.getLogger(XxlJobExecutor.class);

      //...省略

    // ---------------------- start + stop ----------------------
    //my-执行器启动方法，主要的方法
    public void start() throws Exception {
   
   

        // init logpath
        XxlJobFileAppender.initLogPath(logPath);

        // init invoker, admin-client
        //my-初始化远程调用客户端
        initAdminBizList(adminAddresses, accessToken);


        // init JobLogFileCleanThread
        JobLogFileCleanThread.getInstance().start(logRetentionDays);

        // init TriggerCallbackThread
        TriggerCallbackThread.getInstance().start();

        // init executor-server
        initEmbedServer(address, ip, port, appname, accessToken);
    }    


          //...省略

      // ---------------------- admin-client (rpc invoker) ----------------------
    private static List<AdminBiz> adminBizList;

    //my-远程调用客户端初始化逻辑，将其放入到List<AdminBiz>
    private void initAdminBizList(String adminAddresses, String accessToken) throws Exception {
   
   
        if (adminAddresses!=null && adminAddresses.trim().length()>0) {
   
   
            for (String address: adminAddresses.trim().split(",")) {
   
   
                if (address!=null && address.trim().length()>0) {
   
   

                    AdminBiz adminBiz = new AdminBizClient(address.trim(), accessToken);

                    if (adminBizList == null) {
   
   
                        adminBizList = new ArrayList<AdminBiz>();
                    }
                    adminBizList.add(adminBiz);
                }
            }
        }
    }

  //...省略

  // ---------------------- executor-server (rpc provider) ----------------------
    //my-启动一个内嵌服务器，负责执行器和和调度中心交互
    private EmbedServer embedServer = null;

    private void initEmbedServer(String address, String ip, int port, String appname, String accessToken) throws Exception {
   
   

        // fill ip port
        port = port>0?port: NetUtil.findAvailablePort(9999);
        ip = (ip!=null&&ip.trim().length()>0)?ip: IpUtil.getIp();

        // generate address
        if (address==null || address.trim().length()==0) {
   
   
            String ip_port_address = IpUtil.getIpPort(ip, port);   // registry-address：default use address to registry , otherwise use ip:port if address is null
            address = "http://{ip_port}/".replace("{ip_port}", ip_port_address);
        }

        // accessToken
        if (accessToken==null || accessToken.trim().length()==0) {
   
   
            logger.warn(">>>>>>>>>>> xxl-job accessToken is empty. To ensure system security, please set the accessToken.");
        }

        // start
        embedServer = new EmbedServer();
        embedServer.start(address, port, appname, accessToken);
    }

重点来说明initAdminBizList(String adminAddresses, String accessToken)方法，其会为每一个执行器创建一个AdminBizClient，将其放到类变量List中，AdminBizClient负责和调度中心进行通信。

ExecutorRegistryThread定时注册类

下面来讲内嵌服务器启动的逻辑和定时上报注册信息的逻辑。

public class EmbedServer {
   
   
    private static final Logger logger = LoggerFactory.getLogger(EmbedServer.class);

    private ExecutorBiz executorBiz;
    private Thread thread;

    public void start(final String address, final int port, final String appname, final String accessToken) {
   
   
        executorBiz = new ExecutorBizImpl();
        thread = new Thread(new Runnable() {
   
   
            @Override
            public void run() {
   
   
                // param
                EventLoopGroup bossGroup = new NioEventLoopGroup();
                EventLoopGroup workerGroup = new NioEventLoopGroup();
                ThreadPoolExecutor bizThreadPool = new ThreadPoolExecutor(
                        0,
                        200,
                        60L,
                        TimeUnit.SECONDS,
                        new LinkedBlockingQueue<Runnable>(2000),
                        new ThreadFactory() {
   
   
                            @Override
                            public Thread newThread(Runnable r) {
   
   
                                return new Thread(r, "xxl-job, EmbedServer bizThreadPool-" + r.hashCode());
                            }
                        },
                        new RejectedExecutionHandler() {
   
   
                            @Override
                            public void rejectedExecution(Runnable r, ThreadPoolExecutor executor) {
   
   
                                throw new RuntimeException("xxl-job, EmbedServer bizThreadPool is EXHAUSTED!");
                            }
                        });
                try {
   
   
                    // start server
                    ServerBootstrap bootstrap = new ServerBootstrap();
                    bootstrap.group(bossGroup, workerGroup)
                            .channel(NioServerSocketChannel.class)
                            .childHandler(new ChannelInitializer<SocketChannel>() {
   
   
                                @Override
                                public void initChannel(SocketChannel channel) throws Exception {
   
   
                                    channel.pipeline()
                                            .addLast(new IdleStateHandler(0, 0, 30 * 3, TimeUnit.SECONDS))  // beat 3N, close if idle
                                            .addLast(new HttpServerCodec())
                                            .addLast(new HttpObjectAggregator(5 * 1024 * 1024))  // merge request & reponse to FULL
                                            .addLast(new EmbedHttpServerHandler(executorBiz, accessToken, bizThreadPool));
                                }
                            })
                            .childOption(ChannelOption.SO_KEEPALIVE, true);

                    // bind
                    ChannelFuture future = bootstrap.bind(port).sync();

                    logger.info(">>>>>>>>>>> xxl-job remoting server start success, nettype = {}, port = {}", EmbedServer.class, port);

                    // start registry
                    //my-开始向注册中心报送注册信息
                    startRegistry(appname, address);

                    // wait util stop
                    future.channel().closeFuture().sync();

                } catch (InterruptedException e) {
   
   
                    logger.info(">>>>>>>>>>> xxl-job remoting server stop.");
                } catch (Exception e) {
   
   
                    logger.error(">>>>>>>>>>> xxl-job remoting server error.", e);
                } finally {
   
   
                    // stop
                    try {
   
   
                        workerGroup.shutdownGracefully();
                        bossGroup.shutdownGracefully();
                    } catch (Exception e) {
   
   
                        logger.error(e.getMessage(), e);
                    }
                }
            }
        });
        thread.setDaemon(true);    // daemon, service jvm, user thread leave >>> daemon leave >>> jvm leave
        thread.start();
    }
}

ExecutorRegistryThread是一个单例实现，其主要方法是start(final String appname, final String address)，在该方法中会创建一个异步线程，会在固定周期定时向调度中心注册执行器信息。

public class ExecutorRegistryThread {
   
   
    private static Logger logger = LoggerFactory.getLogger(ExecutorRegistryThread.class);

    private static ExecutorRegistryThread instance = new ExecutorRegistryThread();
    public static ExecutorRegistryThread getInstance(){
   
   
        return instance;
    }

    private Thread registryThread;
    private volatile boolean toStop = false;
    public void start(final String appname, final String address){
   
   

        // valid
        if (appname==null || appname.trim().length()==0) {
   
   
            logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, appname is null.");
            return;
        }
        if (XxlJobExecutor.getAdminBizList() == null) {
   
   
            logger.warn(">>>>>>>>>>> xxl-job, executor registry config fail, adminAddresses is null.");
            return;
        }

        registryThread = new Thread(new Runnable() {
   
   
            @Override
            public void run() {
   
   

                // registry
                //my-固定周期定时向调度中心注册执行器信息
                while (!toStop) {
   
   
                    try {
   
   
                        RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appname, address);
                        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[]{
   
   registryParam, registryResult});
                                    break;
                                } else {
   
   
                                    logger.info(">>>>>>>>>>> xxl-job registry fail, registryParam:{}, registryResult:{}", new Object[]{
   
   registryParam, registryResult});
                                }
                            } catch (Exception e) {
   
   
                                logger.info(">>>>>>>>>>> xxl-job registry error, registryParam:{}", registryParam, e);
                            }

                        }
                    } catch (Exception e) {
   
   
                        if (!toStop) {
   
   
                            logger.error(e.getMessage(), e);
                        }

                    }

                    try {
   
   
                        if (!toStop) {
   
   
                            TimeUnit.SECONDS.sleep(RegistryConfig.BEAT_TIMEOUT);
                        }
                    } catch (InterruptedException e) {
   
   
                        if (!toStop) {
   
   
                            logger.warn(">>>>>>>>>>> xxl-job, executor registry thread interrupted, error msg:{}", e.getMessage());
                        }
                    }
                }

                // registry remove
                try {
   
   
                    RegistryParam registryParam = new RegistryParam(RegistryConfig.RegistType.EXECUTOR.name(), appname, address);
                    for (AdminBiz adminBiz: XxlJobExecutor.getAdminBizList()) {
   
   
                        try {
   
   
                            ReturnT<String> registryResult = adminBiz.registryRemove(registryParam);
                            if (registryResult!=null && ReturnT.SUCCESS_CODE == registryResult.getCode()) {
   
   
                                registryResult = ReturnT.SUCCESS;
                                logger.info(">>>>>>>>>>> xxl-job registry-remove success, registryParam:{}, registryResult:{}", new Object[]{
   
   registryParam, registryResult});
                                break;
                            } else {
   
   
                                logger.info(">>>>>>>>>>> xxl-job registry-remove fail, registryParam:{}, registryResult:{}", new Object[]{
   
   registryParam, registryResult});
                            }
                        } catch (Exception e) {
   
   
                            if (!toStop) {
   
   
                                logger.info(">>>>>>>>>>> xxl-job registry-remove error, registryParam:{}", registryParam, e);
                            }

                        }

                    }
                } catch (Exception e) {
   
   
                    if (!toStop) {
   
   
                        logger.error(e.getMessage(), e);
                    }
                }
                logger.info(">>>>>>>>>>> xxl-job, executor registry thread destroy.");

            }
        });
        registryThread.setDaemon(true);
        registryThread.setName("xxl-job, executor ExecutorRegistryThread");
        registryThread.start();
    }

步骤总结

1. XxlJobConfig 配置类中实例化XxlJobSpringExecutor执行器类，该类是XxlJobExecutor类的实现类，在XxlJobSpringExecutor类中还实现了ApplicationContextAware, SmartInitializingSingleton, DisposableBean的接口，能够在实例化后将带有@Xxljob注解的方法当多JobHandler，将其解析后放到本地缓存jobHandlerRepository中。
2. XxlJobSpringExecutor会调用 super.start()方法，开始将执行器进行初始化，初始化过程中最关键的是将初始化远程调用的客户端，将这些客户端方法放到列表List，后续这些客户端实现执行器和调度中心的通信，实现注册、下线、回调等功能。
3. 在初始化后期，会创建一个嵌入式服务器EmbedServer，然后将其启动，该服务器的启动后会创建一个定时任务线程ExecutorRegistryThread，在该线程中会定时向调度中心上报列表List中的执行器信息，如果服务器下线，也会将关联的执行器下线。

执行器注册到调度中心类间关系：

执行器注册到调度中心时序图：