摘要:本文讲解如何实现一个任务调度系统的核心逻辑。
本文分享自华为云社区《实现一个任务调度系统,看这篇文章就够了》,作者:勇哥java实战分享 。
Quartz是一款Java开源任务调度框架,也是很多Java工程师接触任务调度的起点。
下图显示了任务调度的整体流程:
Quartz的核心是三个组件。
上图代码中Quartz 的JobStore是 RAMJobStore,Trigger 和 Job 存储在内存中。
执行任务调度的核心类是 QuartzSchedulerThread 。

接下来再聊聊 Quartz 的集群部署方案。
Quartz的集群部署方案,需要针对不同的数据库类型(MySQL , ORACLE) 在数据库实例上创建Quartz表,JobStore是: JobStoreSupport 。
这种方案是分布式的,没有负责集中管理的节点,而是利用数据库行级锁的方式来实现集群环境下的并发控制。
scheduler实例在集群模式下首先获取{0}LOCKS表中的行锁,Mysql 获取行锁的语句:
{0}会替换为配置文件默认配置的QRTZ_。sched_name为应用集群的实例名,lock_name就是行级锁名。Quartz主要有两个行级锁触发器访问锁 (TRIGGER_ACCESS) 和 状态访问锁(STATE_ACCESS)。
这个架构解决了任务的分布式调度问题,同一个任务只能有一个节点运行,其他节点将不执行任务,当碰到大量短任务时,各个节点频繁的竞争数据库锁,节点越多性能就会越差。
Quartz的集群模式可以水平扩展,也可以分布式调度,但需要业务方在数据库中添加对应的表,有一定的强侵入性。
有不少研发同学为了避免这种侵入性,也探索出分布式锁模式。
业务场景:电商项目,用户下单后一段时间没有付款,系统就会在超时后关闭该订单。
通常我们会做一个定时任务每两分钟来检查前半小时的订单,将没有付款的订单列表查询出来,然后对订单中的商品进行库存的恢复,然后将该订单设置为无效。
我们使用Spring Schedule的方式做一个定时任务。
@Scheduled(cron = "0 */2 * * * ? ")
public void doTask() {
log.info("定时任务启动");
//执行关闭订单的操作
orderService.closeExpireUnpayOrders();
log.info("定时任务结束");
}
在单服务器运行正常,考虑到高可用,业务量激增,架构会演进成集群模式,在同一时刻有多个服务执行一个定时任务,有可能会导致业务紊乱。
解决方案是在任务执行的时候,使用Redis 分布式锁来解决这类问题。
@Scheduled(cron = "0 */2 * * * ? ")
public void doTask() {
log.info("定时任务启动");
String lockName = "closeExpireUnpayOrdersLock";
RedisLock redisLock = redisClient.getLock(lockName);
//尝试加锁,最多等待3秒,上锁以后5分钟自动解锁
boolean locked = redisLock.tryLock(3, 300, TimeUnit.SECONDS);
if(!locked){
log.info("没有获得分布式锁:{}" , lockName);
return;
}
try{
//执行关闭订单的操作
orderService.closeExpireUnpayOrders();
} finally {
redisLock.unlock();
}
log.info("定时任务结束");
}
Redis的读写性能极好,分布式锁也比Quartz数据库行级锁更轻量级。当然Redis锁也可以替换成Zookeeper锁,也是同样的机制。
在小型项目中,使用:定时任务框架(Quartz/Spring Schedule)和 分布式锁(redis/zookeeper)有不错的效果。
但是呢?我们可以发现这种组合有两个问题:
ElasticJob-Lite 定位为轻量级无中心化解决方案,使用 jar 的形式提供分布式任务的协调服务。
应用内部定义任务类,实现SimpleJob接口,编写自己任务的实际业务流程即可。
public class MyElasticJob implements SimpleJob {
@Override
public void execute(ShardingContext context) {
switch (context.getShardingItem()) {
case 0:
// do something by sharding item 0
break;
case 1:
// do something by sharding item 1
break;
case 2:
// do something by sharding item 2
break;
// case n: ...
}
}
}
举例:应用A有五个任务需要执行,分别是A,B,C,D,E。任务E需要分成四个子任务,应用部署在两台机器上。
应用A在启动后, 5个任务通过 Zookeeper 协调后被分配到两台机器上,通过Quartz Scheduler 分开执行不同的任务。
ElasticJob 从本质上来讲 ,底层任务调度还是通过 Quartz ,相比Redis分布式锁 或者 Quartz 分布式部署 ,它的优势在于可以依赖 Zookeeper 这个大杀器 ,将任务通过负载均衡算法分配给应用内的 Quartz Scheduler容器。
从使用者的角度来讲,是非常简单易用的。但从架构来看,调度器和执行器依然在同一个应用方JVM内,而且容器在启动后,依然需要做负载均衡。应用假如频繁的重启,不断的去选主,对分片做负载均衡,这些都是相对比较重的操作。
另外,ElasticJob 的控制台是比较粗糙的,通过读取注册中心数据展现作业状态,更新注册中心数据修改全局任务配置。
中心化的原理是:把调度和任务执行,隔离成两个部分:调度中心和执行器。调度中心模块只需要负责任务调度属性,触发调度命令。执行器接收调度命令,去执行具体的业务逻辑,而且两者都可以进行分布式扩容。
先谈谈我在艺龙促销团队接触的第一种中心化架构。
调度中心依赖Quartz集群模式,当任务调度时候,发送消息到RabbitMQ 。业务应用收到任务消息后,消费任务信息。
这种模型充分利用了MQ解耦的特性,调度中心发送任务,应用方作为执行器的角色,接收任务并执行。
但这种设计强依赖消息队列,可扩展性和功能,系统负载都和消息队列有极大的关联。这种架构设计需要架构师对消息队列非常熟悉。
XXL-JOB 是一个分布式任务调度平台,其核心设计目标是开发迅速、学习简单、轻量级、易扩展。现已开放源代码并接入多家公司线上产品线,开箱即用。
我们重点剖析下架构图 :
▍ 网络通讯 server-worker 模型
调度中心和执行器 两个模块之间通讯是 server-worker 模式。调度中心本身就是一个SpringBoot 工程,启动会监听8080端口。
执行器启动后,会启动内置服务( EmbedServer )监听9994端口。这样双方都可以给对方发送命令。
那调度中心如何知道执行器的地址信息呢 ?上图中,执行器会定时发送注册命令 ,这样调度中心就可以获取在线的执行器列表。
通过执行器列表,就可以根据任务配置的路由策略选择节点执行任务。常见的路由策略有如下三种:
▍ 调度器
调度器是任务调度系统里面非常核心的组件。XXL-JOB 的早期版本是依赖Quartz。
但在v2.1.0版本中完全去掉了Quartz的依赖,原来需要创建的 Quartz表也替换成了自研的表。
核心的调度类是:JobTriggerPoolHelper 。调用start方法后,会启动两个线程:scheduleThread 和 ringThread 。
首先 scheduleThread 会定时从数据库加载需要调度的任务,这里从本质上还是基于数据库行锁保证同时只有一个调度中心节点触发任务调度。
Connection 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();
# 触发任务调度 (伪代码)
for (XxlJobInfo jobInfo: scheduleList) {
// 省略代码
}
# 事务提交
conn.commit();
调度线程会根据任务的「下次触发时间」,采取不同的动作:
已过期的任务需要立刻执行的,直接放入线程池中触发执行 ,五秒内需要执行的任务放到 ringData 对象里。
ringThread 启动后,定时从 ringData 对象里获取需要执行的任务列表 ,放入到线程池中触发执行。

2018年,我有一段自研任务调度系统的经历。
背景是:兼容技术团队自研的RPC框架,技术团队不需要修改代码,RPC注解方法可以托管在任务调度系统中,直接当做一个任务来执行。
自研过程中,研读了XXL-JOB 源码,同时从阿里云分布式任务调度 SchedulerX 吸取了很多营养。
我们模仿了SchedulerX的模块,架构设计如下图:
我选择了 RocketMQ 源码的通讯模块 remoting 作为自研调度系统的通讯框架。基于如下两点:
我将 RocketMQ remoting 模块去掉名字服务代码,做了一定程度的定制。
在RocketMQ的remoting里,服务端采用 Processor 模式。
调度中心需要注册两个处理器:回调结果处理器CallBackProcessor和心跳处理器HeartBeatProcessor 。执行器需要注册触发任务处理器TriggerTaskProcessor 。
public void registerProcessor(
int requestCode,
NettyRequestProcessor processor,
ExecutorService executor);
处理器的接口:
public interface NettyRequestProcessor {
RemotingCommand processRequest(
ChannelHandlerContext ctx,
RemotingCommand request) throws Exception;
boolean rejectRequest();
}
对于通讯框架来讲,我并不需要关注通讯细节,只需要实现处理器接口即可。
以触发任务处理器TriggerTaskProcessor举例:
搞定网络通讯后,调度器如何设计 ?最终我还是选择了Quartz 集群模式。主要是基于以下几点原因:
自研版的调度服务花费一个半月上线了。系统运行非常稳定,研发团队接入也很顺畅。 调度量也不大 ,四个月总共接近4000万到5000万之间的调度量。
坦率的讲,自研版的瓶颈,我的脑海里经常能看到。 数据量大,我可以搞定分库分表,但 Quartz 集群基于行级锁的模式 ,注定上限不会太高。
为了解除心中的困惑,我写一个轮子DEMO看看可否work:
这个Demo版本在开发环境可以运行,但有很多细节需要优化,仅仅是个玩具,并没有机会运行到生产环境。
最近读阿里云的一篇文章《如何通过任务调度实现百万规则报警》,SchedulerX2.0 高可用架构见下图:
文章提到:
每个应用都会做三备份,通过 zk 抢锁,一主两备,如果某台 Server 挂了,会进行 failover,由其他 Server 接管调度任务。
这次自研任务调度系统从架构来讲,并不复杂,实现了XXL-JOB的核心功能,也兼容了技术团队的RPC框架,但并没有实现工作流以及mapreduce分片。
SchedulerX 在升级到2.0之后基于全新的Akka 架构,这种架构号称实现高性能工作流引擎,实现进程间通信,减少网络通讯代码。
在我调研的开源任务调度系统中,PowerJob也是基于Akka 架构,同时也实现了工作流和MapReduce执行模式。
我对PowerJob非常感兴趣,也会在学习实践后输出相关文章,敬请期待。
首先我们将任务调度开源产品和商业产品 SchedulerX 放在一起,生成一张对照表:
Quartz 和 ElasticJob从本质上还是属于框架的层面。
中心化产品从架构上来讲更加清晰,调度层面更灵活,可以支持更复杂的调度(mapreduce动态分片,工作流)。
XXL-JOB 从产品层面已经做到极简,开箱即用,调度模式可以满足大部分研发团队的需求。简单易用 + 能打,所以非常受大家欢迎。
其实每个技术团队的技术储备不尽相同,面对的场景也不一样,所以技术选型并不能一概而论。
不管是使用哪种技术,在编写任务业务代码时,还是需要注意两点:
我试图在一个项目中使用rake,如果我把所有东西都放到Rakefile中,它会很大并且很难读取/找到东西,所以我试着将每个命名空间放在lib/rake中它自己的文件中,我添加了这个到我的rake文件的顶部:Dir['#{File.dirname(__FILE__)}/lib/rake/*.rake'].map{|f|requiref}它加载文件没问题,但没有任务。我现在只有一个.rake文件作为测试,名为“servers.rake”,它看起来像这样:namespace:serverdotask:testdoputs"test"endend所以当我运行rakeserver:testid时
如何使用RSpec::Core::RakeTask初始化RSpecRake任务?require'rspec/core/rake_task'RSpec::Core::RakeTask.newdo|t|#whatdoIputinhere?endInitialize函数记录在http://rubydoc.info/github/rspec/rspec-core/RSpec/Core/RakeTask#initialize-instance_method没有很好的记录;它只是说:-(RakeTask)initialize(*args,&task_block)AnewinstanceofRake
电脑0x0000001A蓝屏错误怎么U盘重装系统教学分享。有用户电脑开机之后遇到了系统蓝屏的情况。系统蓝屏问题很多时候都是系统bug,只有通过重装系统来进行解决。那么蓝屏问题如何通过U盘重装新系统来解决呢?来看看以下的详细操作方法教学吧。 准备工作: 1、U盘一个(尽量使用8G以上的U盘)。 2、一台正常联网可使用的电脑。 3、ghost或ISO系统镜像文件(Win10系统下载_Win10专业版_windows10正式版下载-系统之家)。 4、在本页面下载U盘启动盘制作工具:系统之家U盘启动工具。 U盘启动盘制作步骤: 注意:制作期间,U盘会被格式化,因此U盘中的重要文件请注
在应用开发中,有时候我们需要获取系统的设备信息,用于数据上报和行为分析。那在鸿蒙系统中,我们应该怎么去获取设备的系统信息呢,比如说获取手机的系统版本号、手机的制造商、手机型号等数据。1、获取方式这里分为两种情况,一种是设备信息的获取,一种是系统信息的获取。1.1、获取设备信息获取设备信息,鸿蒙的SDK包为我们提供了DeviceInfo类,通过该类的一些静态方法,可以获取设备信息,DeviceInfo类的包路径为:ohos.system.DeviceInfo.具体的方法如下:ModifierandTypeMethodDescriptionstatic StringgetAbiList()Obt
需求:要创建虚拟机,就需要给他提供一个虚拟的磁盘,我们就在/opt目录下创建一个10G大小的raw格式的虚拟磁盘CentOS-7-x86_64.raw命令格式:qemu-imgcreate-f磁盘格式磁盘名称磁盘大小qemu-imgcreate-f磁盘格式-o?1.创建磁盘qemu-imgcreate-fraw/opt/CentOS-7-x86_64.raw10G执行效果#ls/opt/CentOS-7-x86_64.raw2.安装虚拟机使用virt-install命令,基于我们提供的系统镜像和虚拟磁盘来创建一个虚拟机,另外在创建虚拟机之前,提前打开vnc客户端,在创建虚拟机的时候,通过vnc
我写了一个非常简单的rake任务来尝试找到这个问题的根源。namespace:foodotaskbar::environmentdoputs'RUNNING'endend当在控制台中执行rakefoo:bar时,输出为:RUNNINGRUNNING当我执行任何rake任务时会发生这种情况。有没有人遇到过这样的事情?编辑上面的rake任务就是写在那个.rake文件中的所有内容。这是当前正在使用的Rakefile。requireFile.expand_path('../config/application',__FILE__)OurApp::Application.load_tasks这里
因为我现在正在做一些时间测量,我想知道是否可以在不使用Benchmark类或命令行实用程序time的情况下测量用户时间或系统时间。使用Time类只显示挂钟时间,而不显示系统和用户时间,但是我正在寻找具有相同灵active的解决方案,例如time=TimeUtility.now#somecodeuser,system,real=TimeUtility.now-time原因是我有点不喜欢Benchmark,因为它不能只返回数字(编辑:我错了-它可以。请参阅下面的答案。)。当然,我可以解析输出,但感觉不对。*NIX系统的time实用程序也应该可以解决我的问题,但我想知道是否已经在Ruby中实
在Ruby中,以毫秒为单位获取自纪元(1970)以来的当前系统时间的正确方法是什么?我试过了Time.now.to_i,好像不是我想要的结果。我需要结果显示毫秒并且使用long类型,而不是float或double。 最佳答案 (Time.now.to_f*1000).to_iTime.now.to_f显示包含十进制数字的时间。要获得毫秒数,只需将时间乘以1000。 关于ruby-以毫秒为单位获取当前系统时间,我们在StackOverflow上找到一个类似的问题:
我以前没有使用过cron,所以我不能确定我这样做是对的。我想要自动化的任务似乎没有运行。我在终端中执行了这些步骤:sudogeminstall每当切换到应用程序目录无论何时。(这创建了文件schedule.rb)我将此代码添加到schedule.rb:every10.minutesdorunner"User.vote",environment=>"development"endevery:hourdorunner"Digest.rss",:environment=>"development"end我将此代码添加到deploy.rb:after"deploy:symlink","depl
在railscasts项目中你可以看到这段代码:before(:each)dologin_asFactory(:user,:admin=>true)end函数对应的定义是:Factory.define:userdo|f|f.sequence(:github_username){|n|"foo#{n}"}end我不明白admin参数是如何传递给函数的,而函数中没有关于admin参数的字样。谢谢 最佳答案 Factory.define不是一个函数定义,它是一个方法,它接受一个符号或字符串(在本例中是用户)和一个定义你正在制作的工厂的bl