这是我参与「日新计划 12 月更文挑战」的第4天，点击查看活动详情

你好，今日我想和你共享一下XXL-JOB的中心完结。假如你是XXL-JOB的用户，那么你肯定思考过它的完结原理；假如你还未接触过这个产品，那么能够经过本文了解一下。

XXL-JOB的架构图（2.0版本）如下：

它是如何工作的呢？从运用方的角度来看，首先履行器要向服务端注册。那么这儿你或许就有疑问了，履行器向服务端注册？怎样注册的？多久注册一次？选用什么通讯协议？

注册完了之后，服务端才干知道有哪些履行器，并触发使命调度。那么服务端是如何记载每个使命的触发时机，并完结精准调度的呢？XXL-JOB选用的是Quartz调度结构，本文我计划用时刻轮计划来替换。

最后，履行器接纳到调度恳求，是怎样履行使命的呢？

带着这些问题，咱们开启XXL-JOB的探索之旅。我先来说说XXL-JOB项目模块，项目模块很简单，有2个：

xxl-job-core：这个模块是给履行器依赖的；
xxl-job-admin：对应架构图中的调度中心；

本文内容较干，请调配源码食用。源码版本是：2.0.2

1、Job服务自动注册

第一个中心技术点，服务注册。

服务注册要从xxl-job-core模块的XxlJobSpringExecutor类说起，这是一个 Spring 的 Bean，它是这么界说的：

@Bean(initMethod = "start", destroyMethod = "destroy")
public XxlJobSpringExecutor xxlJobExecutor() {
  XxlJobSpringExecutor xxlJobSpringExecutor = new XxlJobSpringExecutor();
  xxlJobSpringExecutor.setAdminAddresses(adminAddresses);
  // 其他的一些注册信息
  return xxlJobSpringExecutor;
}

进行代码追寻，终究会是下面的调用链路：

xxl-job-core模块
spring bean: XxlJobSpringExecutor # start()
-> XxlJobExecutor # start() -> initRpcProvider()

xxl-rpc-core.jar
-> XxlRpcProviderFactory # start() 
-> ServiceRegistry # start()
-> ExecutorServiceRegistry # start()
-> ExecutorRegistryThread # start()

ExecutorRegistryThread便是服务注册的中心完结了，start()办法中心代码如下：

public void start(String appName, String address) {
  registryThread = new Thread(new Runnable() {
    @Override
    public void run() {
      // registry
      while (!toStop) {
        // do registry
        adminBiz.registry(registryParam);
        TimeUnit.SECONDS.sleep(JobConstants.HEARTBEAT_INTERVAL);// 30s
       }
      // registry remove
      adminBiz.registryRemove(registryParam);
     }
   });
  registryThread.setDaemon(true);
  registryThread.start();
}

能够看到履行器每 30s 履行注册一次，咱们继续往下看。

2、自动注册通讯技术完结

经过上面ExecutorRegistryThread # start()办法中心代码，能够看到，注册是经过adminBiz.registry(registryParam)代码完结的，调用链路总结如下：

xxl-job-core模块

AdminBiz # registry()
-> AdminBizClient # registry()
-> XxlJobRemotingUtil # postBody()
-> POST api/registry (jdk HttpURLConnection)

终究还是经过 HTTP 协议的 POST 恳求，注册数据格式如下：

{
 "registryGroup": "EXECUTOR",
 "registryKey": "example-job-executor",
 "registryValue": "10.0.0.10:9999"
}

看到这儿，咱们回到文章开头问题部分。

履行器向服务端注册？怎样注册的？多久注册一次？选用什么通讯协议？

答案现已很明显了。

3、使命调度完结

咱们接着来看第二个中心技术点，使命调度。

XXL-JOB选用的是Quartz调度结构，这儿我计划向你介绍一下时刻轮的完结计划，中心源码如下：

@Component
public class JobScheduleHandler {

  private Thread scheduler;
  private Thread ringConsumer;
  private final Map<Integer, List<Integer>> ring;
  
  @PostConstruct
  public void start() {
    scheduler = new Thread(new JobScheduler(), "job-scheduler");
    scheduler.setDaemon(true);
    scheduler.start();

    ringConsumer = new Thread(new RingConsumer(), "job-ring-handler");
    ringConsumer.setDaemon(true);
    ringConsumer.start();
   }
  
  class JobScheduler implements Runnable {
    @Override
    public void run() {
      sleep(5000 - System.currentTimeMillis() % 1000);
      while (!schedulerStop) {
        try {
          lock.lock();
          // pre read to ring
         } catch (Exception e) {
          log.error("JobScheduler error", e);
         } finally {
          lock.unlock();
         }
        sleep(1000);
       }
     }
   }
  
  class RingConsumer implements Runnable {
    @Override
    public void run() {
      sleep(1000 - System.currentTimeMillis() % 1000);
      while (!ringConsumerStop) {
        try {
          int nowSecond = Calendar.getInstance().get(Calendar.SECOND);
          List<Integer> jobIds = ring.remove(nowSecond % 60);
          // 触发使命调度
         } catch (Exception e) {
          log.error("ring consumer error", e);
         }
        sleep(1000 - System.currentTimeMillis() % 1000);
       }
     }
   }
}

上述经过两个线程池来完结，job-scheduler为预读线程，job-ring-handler为时刻轮线程。那么时刻轮是怎样完结使命的精准调度的呢？

时刻轮的完结原理

咱们常见的时钟依据秒针滚动的类型，能够分为嘀嗒式秒针和流动式秒针。

我以嘀嗒式秒针时钟为例，能够把时钟环看作一个数组，秒针 1~60 秒停留的位置作为数组下标，60s 为数组下标 0。假定现在有 3 个待履行的使命，分别如下：

jobid: 101  0秒时刻开端履行，2s/次
jobid: 102  0秒时刻开端履行，3s/次
jobid: 103  3秒时刻开端履行，4s/次

对应 0 秒时刻的数组模型如下图所示：

这儿我把 0 时刻拆成了三个阶段，分别是：

履行前：读取该时刻有哪些使命待履行，拿到使命 id；
履行中：经过使命 id 查询使命的运转策略，履行使命；
履行后：更新使命的下次履行时刻；

然后时刻指针往前推动一个时刻，到了 1 秒时刻。此时刻时刻轮中的使命并未发生变化。

到了第 2 秒时刻，预读线程将 jobid 103 加入时刻轮，并履行该数组下标下的使命：

这样到了第 3 秒时刻，使命的数组下标又会被更新。

那么这种以秒为刻度的时刻轮有没有差错呢？

使命调度的精准度是取决于时刻轮的刻度的。举个比如，咱们把 0 秒时刻的这 1s 拆成 1000ms。

假定使命都是在第 500ms 完结该时刻秒内所有使命的调度的，501ms 有一个新的使命被预读线程加载进来了，那么轮到下次调度，就要等到第 1 秒时刻的第 500ms，差错相差了一个刻度即 1s。假如以 0.5 秒为一个刻度，那么差错就变小了，是 500ms。

所以说，刻度越小，差错越小。不过这也要依据事务的实际情况来决议，毕竟要想削减差错，就要耗费更多的 CPU 资源。

了解完使命调度的完结原理，那调度器与履行器间的服务通讯是如何完结的呢？

4、使命调度通讯技术完结

在xxl-job-admin模块，整理调用链路如下：

xxl-job-admin模块

JobTriggerPoolHelper # trigger()
-> ThreadPoolExecutor # execute() (分快慢线程池)
-> XxlJobTrigger # trigger() -> processTrigger() -> runExecutor()
-> XxlJobDynamicScheduler # getExecutorBiz()  
-> ExecutorBiz # run() (动态代理完结, 这儿调用的 run 会作为参数) [1]
-> XxlRpcReferenceBean. new InvocationHandler() # invoke()

xxl-rpc-core.jar
-> NettyHttpClient # asyncSend()
(POST...恳求参数 XxlRpcRequest 设置 methodName 为[1]处的调用办法即 "run")

终究是经过 HTTP 协议进行通讯的，中心通讯代码如下：

public void send(XxlRpcRequest xxlRpcRequest) throws Exception {
  byte[] requestBytes = serializer.serialize(xxlRpcRequest);
  DefaultFullHttpRequest request = new DefaultFullHttpRequest(HttpVersion.HTTP_1_1, HttpMethod.POST, new URI(address).getRawPath(), Unpooled.wrappedBuffer(requestBytes));
  request.headers().set(HttpHeaderNames.HOST, host);
  request.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
  request.headers().set(HttpHeaderNames.CONTENT_LENGTH, request.content().readableBytes());
  this.channel.writeAndFlush(request).sync();
}

调度器将履行恳求发送到履行器后，接着便是履行器的工作了。

5、履行器接纳使命接口完结

履行器的工作，整理调用链路如下：

xxl-job-core模块
spring bean: XxlJobSpringExecutor # start()
-> XxlJobExecutor # start() -> initRpcProvider()

xxl-rpc-core.jar
-> XxlRpcProviderFactory # start() 
-> Server # start()
-> NettyHttpServer # start()

netty 接口完结
NettyHttpServerHandler # channelRead0() -> process() (线程池履行)
-> XxlRpcProviderFactory # invokeService()
(依据恳求参数 XxlRpcRequest 里的 methodName 反射调用)
-> ExecutorBizImpl # run()

咱们也能够经过 HTTP 恳求查看接口完结：

GET http://localhost:17711/services

成果如下：

<ui>
    <li>com.xxl.job.core.biz.ExecutorBiz: com.xxl.job.core.biz.impl.ExecutorBizImpl@d579177</li>
</ui>

履行器接纳使命，总结来说用的是下面的接口：

POST http://localhost:17711

要注意的是，这儿假如经过 Postman 来调用是调不通的，由于序列化方法和 HTTP 协议是不一样的。

接下来便是履行器接纳到使命逻辑，代码链路如下：

xxl-job-core模块
spring bean: XxlJobSpringExecutor # start()
-> XxlJobExecutor # start() -> initRpcProvider()
-> new ExecutorBizImpl()
-> JobThread # pushTriggerQueue()

spring bean: XxlJobExecutor # registJobThread() 启动 jobThead
-> JobThread # run()

到这儿，咱们就把中心流程整理了一遍。

小结

经过上文的整理，假如想要从 0 搭建一个分布式使命调度系统，想必你已胸中有数了。本文所描述的时刻轮计划，也是敝司根据XXL-JOB的重构计划，后来也应用在了音讯中间件的推迟音讯完结中。

欢迎交流，大众号【杨同学technotes】

XXL-JOB核心源码解读及时间轮原理剖析

1、Job服务自动注册

2、自动注册通讯技术完结

3、使命调度完结

时刻轮的完结原理

4、使命调度通讯技术完结

5、履行器接纳使命接口完结

小结

作者信息

XXL-JOB核心源码解读及时间轮原理剖析

1、Job服务自动注册

2、自动注册通讯技术完结

3、使命调度完结

时刻轮的完结原理

4、使命调度通讯技术完结

5、履行器接纳使命接口完结

小结

相关文章

大型并发应用中如何使用 ForkJoinPool 实现高效计算？

Coze：赋予每个人创造 AI 的平台🌟

DNS（域名系统）是如何工作的？

主流的四种限流策略，我都可以通过redis实现｜周末学习

作者信息