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

news/2024/11/8 14:40:11/

你好,今天我想和你分享一下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() {@Overridepublic void run() {// registrywhile (!toStop) {// do registryadminBiz.registry(registryParam);TimeUnit.SECONDS.sleep(JobConstants.HEARTBEAT_INTERVAL);// 30s}// registry removeadminBiz.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调度框架,这里我打算向你介绍一下时间轮的实现方案,核心源码如下:

 @Componentpublic class JobScheduleHandler {​private Thread scheduler;private Thread ringConsumer;private final Map<Integer, List<Integer>> ring;@PostConstructpublic 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 {@Overridepublic 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 {@Overridepublic 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】


http://www.ppmy.cn/news/2064.html

相关文章

C++中的野指针

一篇非常全面的介绍指针的博客: C/C 中的指针[非常全面] 一&#xff1a;维基百科的定义 在计算机编程领域中&#xff0c;迷途指针&#xff0c;或称悬空指针、野指针&#xff0c;指的是不指向任何合法的对象的指针。 当所指向的对象被释放或者收回&#xff0c;但是对该指针没有…

内存可见性问题

目录 1.什么是内存可见性问题 2.内存可见性问题是怎么发生的 3.解决方法&#xff1a;volatile 4.volatile使用的注意事项 5.内存可见性问题的延伸 缓存&#xff08;cache&#xff09; 1.什么是内存可见性问题 首先来看一段代码 class Counter{public int flag 0; } pu…

免费分享20套微信小程序源码 源码免费下载【强烈推荐】

淘源码&#xff1a;国内知名的源码免费下载平台 微信小程序源码包括&#xff1a;商城系统、点餐外卖、垃圾分类、预约洗车、物业管理、校园跑腿、驾考学习、会议预约、图书管理、智能停车、在线答题等小程序源码。 源码分享&#xff0c;文末获取源码&#xff01; 1、JAVA微信…

并查集介绍

文章目录&#xff1a;并查集原理并查集实现并查集的类结构并查集的合并统计集合数量并查集原理 在一些应用问题中&#xff0c;需要将 n 个不同的元素划分成一些不相交的集合。开始时&#xff0c;每个元素自成一个单元素集合&#xff0c;然后按照一定的规律将归于同一组元素的集…

ASP.NET Core 3.1系列(18)——EFCore中执行原生SQL语句

1、前言 前一篇博客介绍了EFCore中常见的一些查询操作&#xff0c;使用Linq或Lambda结合实体类的操作相当方便。但在某些特殊情况下&#xff0c;我们仍旧需要使用原生SQL来获取数据。好在EFCore中提供了完整的方法支持原生SQL&#xff0c;下面开始介绍。 2、构建测试数据库 …

【Opencv实战】高手勿入,Python使用Opencv+Canny实现边缘检测以及轮廓检测(详细步骤+源码分享)

前言 有温度 有深度 有广度 就等你来关注哦~ 所有文章完整的素材源码都在&#x1f447;&#x1f447; 粉丝白嫖源码福利&#xff0c;请移步至CSDN社区或文末公众hao即可免费。 在这次的案例实战中&#xff0c;我们将使用Python 3和OpenCV。我们将使用OpenCV&#xff0c;因为它是…

C# 11 中的新增功能

我们很高兴地宣布 C# 11 已经发布&#xff01;与往常一样&#xff0c;C# 开辟了一些全新的领域&#xff0c;同时推进了过去版本中一直在运行的几个主题。我们的文档页面上的 C# 11 的新增功能下有许多功能和详细信息&#xff0c;这些内容都得到了很好的介绍。 随着每个版本的发…

数据结构 树练习题

目录 判断 选择 判断 1.一棵有124个结点的完全二叉树&#xff0c;其 叶结点个数是确定的。 【答案】正确 【解析】完全二叉树 若设二叉树的深度为h 除第 h 层外 其它各层 1&#xff5e;(h-1) 的结点数都达到最大个数(即1~(h-1)层为一个满二叉树) 第 h 层所有的结点都连续集…