quartz原理java_Quartz原理解密

Quartz原理解密

Author: Dorae

Date:2018年7月17日15:55:02

转载请注明出处

一、quartz概述

quartz是一个用java实现的开源任务调度框架&＃xff0c;可以用来创建简单或者复杂的任务调度&＃xff0c;并且可以提供许多企业级的功能&＃xff0c;比如JTA以及集群等&＃xff0c;是当今比较流行的JAVA任务调度框架。

1. 可以用来做什么

Quartz是一个任务调度框架&＃xff0c;当遇到以下问题时&＃xff1a;

想在每月25号&＃xff0c;自动还款&＃xff1b;

想在每年4月1日给当年自己暗恋的女神发一封匿名贺卡&＃xff1b;

想每隔1小时&＃xff0c;备份一下自己的各种资料。

那么总结起来就是&＃xff0c;在一个有规律的时间点做一些事情&＃xff0c;并且这个规律可以非常复杂&＃xff0c;复杂到了需要一个框架来帮助我们。Quartz的出现就是为了解决这个问题&＃xff0c;定义一个触发条件&＃xff0c;那么其负责到了特定的时间点&＃xff0c;触发相应的job干活。

2. 特点

强大的调度功能&＃xff0c;例如丰富多样的调度方法&＃xff0c;可以满足各种常规和特殊需求&＃xff1b;

灵活的应用方式&＃xff0c;比如支持任务调度和任务的多种组合&＃xff0c;支持数据的多种存储(DB&＃xff0c;RAM等&＃xff1b;

支持分布式集群&＃xff0c;在被Terracotta收购之后&＃xff0c;在原来基础上进行了进一步的改造。

二、quartz基本原理

1. 核心元素

Quartz核心要素有Scheduler、Trigger、Job、JobDetail&＃xff0c;其中trigger和job、jobDetail为元数据&＃xff0c;而Scheduler为实际进行调度的控制器。

Trigger

Trigger用于定义调度任务的时间规则&＃xff0c;在Quartz中主要有四种类型的Trigger&＃xff1a;SimpleTrigger、CronTrigger、DataIntervalTrigger和NthIncludedTrigger。

Job&Jodetail

Quartz将任务分为Job、JobDetail两部分&＃xff0c;其中Job用来定义任务的执行逻辑&＃xff0c;而JobDetail用来描述Job的定义(例如Job接口的实现类以及其他相关的静态信息)。对Quartz而言&＃xff0c;主要有两种类型的Job&＃xff0c;StateLessJob、StateFulJob

Scheduler

实际执行调度逻辑的控制器&＃xff0c;Quartz提供了DirectSchedulerFactory和StdSchedulerFactory等工厂类&＃xff0c;用于支持Scheduler相关对象的产生。

2. 核心元素间关系

图 1-1

3. 主要线程

在Quartz中&＃xff0c;有两类线程&＃xff0c;也即执行线程和调度线程&＃xff0c;其中执行任务的线程通常用一个线程池维护。线程间关系如图1-2所示。

图 1-2

图 1-2

在quartz中&＃xff0c;Scheduler调度线程主要有两个&＃xff1a;regular Scheduler Thread(执行常规调度)和Misfire Scheduler Thread(执行错失的任务)。其中Regular Thread 轮询Trigger&＃xff0c;如果有将要触发的Trigger&＃xff0c;则从任务线程池中获取一个空闲线程&＃xff0c;然后执行与改Trigger关联的job&＃xff1b;Misfire Thraed则是扫描所有的trigger&＃xff0c;查看是否有错失的&＃xff0c;如果有的话&＃xff0c;根据一定的策略进行处理。

4. 数据存储

Quartz中的trigger和job需要存储下来才能被使用。Quartz中有两种存储方式&＃xff1a;RAMJobStore,JobStoreSupport&＃xff0c;其中RAMJobStore是将trigger和job存储在内存中&＃xff0c;而JobStoreSupport是基于jdbc将trigger和job存储到数据库中。RAMJobStore的存取速度非常快&＃xff0c;但是由于其在系统被停止后所有的数据都会丢失&＃xff0c;所以在集群应用中&＃xff0c;必须使用JobStoreSupport。其中表结构如表1-1所示。

表 1-1

Table name

Description

QRTZ_CALENDARS

存储Quartz的Calendar信息

QRTZ_CRON_TRIGGERS

存储CronTrigger&＃xff0c;包括Cron表达式和时区信息

QRTZ_FIRED_TRIGGERS

存储与已触发的Trigger相关的状态信息&＃xff0c;以及相联Job的执行信息

QRTZ_PAUSED_TRIGGER_GRPS

存储已暂停的Trigger组的信息

QRTZ_SCHEDULER_STATE

存储少量的有关Scheduler的状态信息&＃xff0c;和别的Scheduler实例

QRTZ_LOCKS

存储程序的悲观锁的信息

QRTZ_JOB_DETAILS

存储每一个已配置的Job的详细信息

QRTZ_SIMPLE_TRIGGERS

存储简单的Trigger&＃xff0c;包括重复次数、间隔、以及已触的次数

QRTZ_BLOG_TRIGGERS

Trigger作为Blob类型存储

QRTZ_TRIGGERS

存储已配置的Trigger的信息

QRTZ_SIMPROP_TRIGGERS

三、quartz集群原理

一个Quartz集群中的每个节点是一个独立的Quartz应用&＃xff0c;它又管理着其他的节点。这就意味着你必须对每个节点分别启动或停止。Quartz集群中&＃xff0c;独立的Quartz节点并不与另一其的节点或是管理节点通信&＃xff0c;而是通过相同的数据库表来感知到另一Quartz应用的&＃xff0c;如图1-3所示。

图 1-3

四、quartz主要流程

1. 启动流程

若quartz是配置在spring中&＃xff0c;当服务器启动时&＃xff0c;就会装载相关的bean。SchedulerFactoryBean实现了InitializingBean接口&＃xff0c;因此在初始化bean的时候&＃xff0c;会执行afterPropertiesSet方法&＃xff0c;该方法将会调用SchedulerFactory(DirectSchedulerFactory 或者 StdSchedulerFactory&＃xff0c;通常用StdSchedulerFactory)创建Scheduler。SchedulerFactory在创建quartzScheduler的过程中&＃xff0c;将会读取配置参数&＃xff0c;初始化各个组件&＃xff0c;关键组件如下&＃xff1a;

ThreadPool:一般是使用SimpleThreadPool,SimpleThreadPool创建了一定数量的WorkerThread实例来使得Job能够在线程中进行处理。WorkerThread是定义在SimpleThreadPool类中的内部类&＃xff0c;它实质上就是一个线程。在SimpleThreadPool中有三个list&＃xff1a;workers-存放池中所有的线程引用&＃xff0c;availWorkers-存放所有空闲的线程&＃xff0c;busyWorkers-存放所有工作中的线程&＃xff1b;

线程池的配置参数如下所示&＃xff1a;

org.quartz.threadPool.class&＃61;org.quartz.simpl.SimpleThreadPool

org.quartz.threadPool.threadCount&＃61;3

org.quartz.threadPool.threadPriority&＃61;5

JobStore:分为存储在内存的RAMJobStore和存储在数据库的JobStoreSupport(包括JobStoreTX和JobStoreCMT两种实现&＃xff0c;JobStoreCMT是依赖于容器来进行事务的管理&＃xff0c;而JobStoreTX是自己管理事务)&＃xff0c;若要使用集群要使用JobStoreSupport的方式&＃xff1b;

QuartzSchedulerThread:用来进行任务调度的线程&＃xff0c;在初始化的时候paused&＃61;true,halted&＃61;false,虽然线程开始运行了&＃xff0c;但是paused&＃61;true&＃xff0c;线程会一直等待&＃xff0c;直到start方法将paused置为false&＃xff1b;

另外&＃xff0c;SchedulerFactoryBean还实现了SmartLifeCycle接口&＃xff0c;因此初始化完成后&＃xff0c;会执行start()方法&＃xff0c;该方法将主要会执行以下的几个动作&＃xff1a;

创建ClusterManager线程并启动线程:该线程用来进行集群故障检测和处理&＃xff0c;将在下文详细讨论&＃xff1b;

创建MisfireHandler线程并启动线程:该线程用来进行misfire任务的处理&＃xff0c;将在下文详细讨论&＃xff1b;

置QuartzSchedulerThread的paused&＃61;false&＃xff0c;调度线程才真正开始调度&＃xff1b;

Quartz的整个启动流程如图1-4所示。

图 1-4

2. QuartzSchedulerThread线程

QuartzSchedulerThread线程是实际执行任务调度的线程&＃xff0c;其中主要代码如下。

while (!halted.get()) {

int availThreadCount &＃61; qsRsrcs.getThreadPool().blockForAvailableThreads();

triggers &＃61; qsRsrcs.getJobStore().acquireNextTriggers(now &＃43; idleWaitTime,

Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());

long triggerTime &＃61; triggers.get(0).getNextFireTime().getTime();

long timeUntilTrigger &＃61; triggerTime - now;

while (timeUntilTrigger > 2) {

now &＃61; System.currentTimeMillis();

timeUntilTrigger &＃61; triggerTime - now;

}

List bndle &＃61; qsRsrcs.getJobStore().triggersFired(triggers);

for (int i &＃61; 0; i

JobRunShell shell &＃61; qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle);

shell.initialize(qs);

qsRsrcs.getThreadPool().runInThread(shell);

}

}

先获取线程池中的可用线程数量(若没有可用的会阻塞&＃xff0c;直到有可用的)&＃xff1b;

获取30m内要执行的trigger(即acquireNextTriggers)&＃xff1a;

获取trigger的锁&＃xff0c;通过select …for update方式实现&＃xff1b;获取30m内(可配置)要执行的triggers(需要保证集群节点的时间一致)&＃xff0c;若&＃64;ConcurrentExectionDisallowed且列表存在该条trigger则跳过&＃xff0c;否则更新trigger状态为ACQUIRED(刚开始为WAITING)&＃xff1b;插入firedTrigger表&＃xff0c;状态为ACQUIRED;(注意&＃xff1a;在RAMJobStore中&＃xff0c;有个timeTriggers&＃xff0c;排序方式是按触发时间nextFireTime排的&＃xff1b;JobStoreSupport从数据库取出triggers时是按照nextFireTime排序);

等待直到获取的trigger中最先执行的trigger在2ms内&＃xff1b;

triggersFired&＃xff1a;

更新firedTrigger的status&＃61;EXECUTING;

更新trigger下一次触发的时间&＃xff1b;

更新trigger的状态&＃xff1a;无状态的trigger->WAITING&＃xff0c;有状态的trigger->BLOCKED&＃xff0c;若nextFireTime&＃61;&＃61;null ->COMPLETE&＃xff1b;

commit connection,释放锁&＃xff1b;

针对每个要执行的trigger&＃xff0c;创建JobRunShell&＃xff0c;并放入线程池执行&＃xff1a;

execute:执行job

获取TRIGGER_ACCESS锁

若是有状态的job&＃xff1a;更新trigger状态&＃xff1a;BLOCKED->WAITING,PAUSED_BLOCKED->BLOCKED

若&＃64;PersistJobDataAfterExecution&＃xff0c;则updateJobData

删除firedTrigger

commit connection&＃xff0c;释放锁

调度线程的执行流程如图1-5所示。

图 1-5

调度过程中Trigger状态变化如图1-6所示。

图 1-6

3. MisfireHandler线程

下面这些原因可能造成 misfired job:

系统因为某些原因被重启。在系统关闭到重新启动之间的一段时间里&＃xff0c;可能有些任务会被 misfire&＃xff1b;

Trigger 被暂停(suspend)的一段时间里&＃xff0c;有些任务可能会被 misfire&＃xff1b;

线程池中所有线程都被占用&＃xff0c;导致任务无法被触发执行&＃xff0c;造成 misfire&＃xff1b;

有状态任务在下次触发时间到达时&＃xff0c;上次执行还没有结束&＃xff1b;为了处理 misfired job&＃xff0c;Quartz 中为 trigger 定义了处理策略&＃xff0c;主要有下面两种&＃xff1a;

MISFIRE_INSTRUCTION_FIRE_ONCE_NOW&＃xff1a;针对 misfired job 马上执行一次&＃xff1b;

MISFIRE_INSTRUCTION_DO_NOTHING&＃xff1a;忽略 misfired job&＃xff0c;等待下次触发&＃xff1b;默认是MISFIRE_INSTRUCTION_SMART_POLICY&＃xff0c;该策略在CronTrigger中&＃61;MISFIRE_INSTRUCTION_FIRE_ONCE_NOW线程默认1分钟执行一次&＃xff1b;在一个事务中&＃xff0c;默认一次最多recovery 20个&＃xff1b;

执行流程&＃xff1a;

若配置(默认为true&＃xff0c;可配置)成获取锁前先检查是否有需要recovery的trigger&＃xff0c;先获取misfireCount&＃xff1b;

获取TRIGGER_ACCESS锁&＃xff1b;

hasMisfiredTriggersInState&＃xff1a;获取misfired的trigger&＃xff0c;默认一个事务里只能最大20个misfired trigger(可配置)&＃xff0c;misfired判断依据&＃xff1a;status&＃61;waiting,next_fire_time

notifyTriggerListenersMisfired

updateAfterMisfire:获取misfire策略(默认是MISFIRE_INSTRUCTION_SMART_POLICY&＃xff0c;该策略在CronTrigger中&＃61;MISFIRE_INSTRUCTION_FIRE_ONCE_NOW)&＃xff0c;根据策略更新nextFireTime&＃xff1b;

将nextFireTime等更新到trigger表&＃xff1b;

commit connection&＃xff0c;释放锁8.如果还有更多的misfired&＃xff0c;sleep短暂时间(为了集群负载均衡)&＃xff0c;否则sleep misfirethreshold时间&＃xff0c;后继续轮询&＃xff1b;

misfireHandler线程执行流程如图1-7所示&＃xff1a;

图 1-7

4. ClusterManager集群管理线程

初始化&＃xff1a;

failedInstance&＃61;failed&＃43;self&＃43;firedTrigger表中的schedulerName在scheduler_state表中找不到的(孤儿)

线程执行&＃xff1a;

每个服务器会定时(org.quartz.jobStore.clusterCheckinInterval这个时间)更新SCHEDULER_STATE表的LAST_CHECKIN_TIME&＃xff0c;若这个字段远远超出了该更新的时间&＃xff0c;则认为该服务器实例挂了&＃xff1b;

注意&＃xff1a;每个服务器实例有唯一的id&＃xff0c;若配置为AUTO&＃xff0c;则为hostname&＃43;current_time

线程执行的具体流程&＃xff1a;

检查是否有超时的实例failedInstances;

更新该服务器实例的LAST_CHECKIN_TIME&＃xff1b;

若有超时的实例&＃xff1a;

获取STATE_ACCESS锁&＃xff1b;

获取超时的实例failedInstances;

获取TRIGGER_ACCESS锁&＃xff1b;

clusterRecover:

针对每个failedInstances&＃xff0c;通过instanceId获取每个实例的firedTriggers;

针对每个firedTrigger&＃xff1a;

更新trigger状态&＃xff1a;

BLOCKED->WAITING

PAUSED_BLOCKED->PAUSED

ACQUIRED->WAITING

若firedTrigger不是ACQUIRED状态(在执行状态),且jobRequestRecovery&＃61;true:

创建一个SimpleTrigger&＃xff0c;存储到trigger表&＃xff0c;status&＃61;waiting,MISFIRE_INSTR&＃61;MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY.

删除firedTrigger

clusterManager线程执行时序图如图1-8所示&＃xff1a;

图 1-8

五、注意问题

时间同步问题

Quartz实际并不关心你是在相同还是不同的机器上运行节点。当集群放置在不同的机器上时&＃xff0c;称之为水平集群。节点跑在同一台机器上时&＃xff0c;称之为垂直集群。对于垂直集群&＃xff0c;存在着单点故障的问题。这对高可用性的应用来说是无法接受的&＃xff0c;因为一旦机器崩溃了&＃xff0c;所有的节点也就被终止了。对于水平集群&＃xff0c;存在着时间同步问题。

节点用时间戳来通知其他实例它自己的最后检入时间。假如节点的时钟被设置为将来的时间&＃xff0c;那么运行中的Scheduler将再也意识不到那个结点已经宕掉了。另一方面&＃xff0c;如果某个节点的时钟被设置为过去的时间&＃xff0c;也许另一节点就会认定那个节点已宕掉并试图接过它的Job重运行。最简单的同步计算机时钟的方式是使用某一个Internet时间服务器(Internet Time Server ITS)。

节点争抢Job问题

因为Quartz使用了一个随机的负载均衡算法&＃xff0c;Job以随机的方式由不同的实例执行。Quartz官网上提到当前&＃xff0c;还不存在一个方法来指派(钉住) 一个 Job 到集群中特定的节点。

从集群获取Job列表问题

当前&＃xff0c;如果不直接进到数据库查询的话&＃xff0c;还没有一个简单的方式来得到集群中所有正在执行的Job列表。请求一个Scheduler实例&＃xff0c;将只能得到在那个实例上正运行Job的列表。Quartz官网建议可以通过写一些访问数据库JDBC代码来从相应的表中获取全部的Job信息。

六、参考文献