Quartz——任务调度框架

落日映苍穹つ 2024-04-18 19:52 38阅读 0赞

Quartz是一个完全由Java编写的开源任务调度的框架，通过触发器设置作业定时运行规则，控制作业的运行时间。其中Quartz集群通过故障切换和负载平衡的功能，能给调度器带来高可用性和伸缩性。主要用来执行定时任务，如：定时发送信息、定时生成报表等等。

# 1.Quartz核心概念 #

![11627468\_1438829844Zbz8.png][11627468_1438829844Zbz8.png]

<table> 
 <tbody> 
 <tr> 
 <th>概念</th> 
 <th>含义</th> 
 </tr> 
 <tr> 
 <td>Job</td> 
 <td>任务，即被调度的任务。是一个接口，只有一个方法void&nbsp;execute(JobExecutionContext&nbsp;context)，开发者实现该接口定义运行任务，JobExecutionContext类提供了调度上下文的各种信息。Job运行时的信息保存在JobDataMap实例中</td> 
 </tr> 
 <tr> 
 <td>JobDetail</td> 
 <td>Quartz在每次执行Job时，都重新创建一个Job实例，所以它不直接接受一个Job的实例，相反它接收一个Job实现类，以便运行时通过newInstance()的反射机制实例化Job。因此需要通过一个类来描述Job的实现类及其它相关的静态信息，如Job名字、描述、关联监听器等信息，JobDetail承担了这一角色。</td> 
 </tr> 
 <tr> 
 <td>Trigger</td> 
 <td>是一个类，描述触发Job执行的时间触发规则。主要有SimpleTrigger和CronTrigger这两个子类。当仅需触发一次或者以固定时间间隔周期执行，SimpleTrigger是最适合的选择；而CronTrigger则可以通过Cron表达式定义出各种复杂时间规则的调度方案：如每早晨9:00执行，周一、周三、周五下午5:00执行等</td> 
 </tr> 
 <tr> 
 <td>Calendar</td> 
 <td>org.quartz.Calendar和java.util.Calendar不同，它是一些日历特定时间点的集合（可以简单地将org.quartz.Calendar看作java.util.Calendar的集合——java.util.Calendar代表一个日历时间点，无特殊说明后面的Calendar即指org.quartz.Calendar）。一个Trigger可以和多个Calendar关联，以便排除或包含某些时间点。假设，我们安排每周星期一早上10:00执行任务，但是如果碰到法定的节日，任务则不执行，这时就需要在Trigger触发机制的基础上使用Calendar进行定点排除。</td> 
 </tr> 
 <tr> 
 <td>Scheduler</td> 
 <td>代表一个Quartz的独立运行容器，Trigger和JobDetail可以注册到Scheduler中，两者在Scheduler中拥有各自的组及名称，组及名称是Scheduler查找定位容器中某一对象的依据，Trigger的组及名称必须唯一，JobDetail的组和名称也必须唯一（但可以和Trigger的组和名称相同，因为它们是不同类型的）。Scheduler定义了多个接口方法，允许外部通过组及名称访问和控制容器中Trigger和JobDetail。</td> 
 </tr> 
 <tr> 
 <td>ThreadPool</td> 
 <td>Scheduler使用一个线程池作为任务运行的基础设施，任务通过共享线程池中的线程提高运行效率。</td> 
 </tr> 
 <tr> 
 <td>misfire</td> 
 <td>错过执行的任务，指本来应该被执行但实际没有被执行的任务调度。</td> 
 </tr> 
 </tbody> 
</table>

Scheduler可以将Trigger绑定到某一JobDetail中，这样当Trigger触发时，对应的Job就被执行。一个Job可以对应多个Trigger（如：每天早上发送邮件，节假日除外，就可以定义两个Trigger），但一个Trigger只能对应一个Job。可以通过SchedulerFactory创建一个Scheduler实例。Scheduler拥有一个SchedulerContext，它类似于ServletContext，保存着Scheduler上下文信息，Job和Trigger都可以访问SchedulerContext内的信息。SchedulerContext内部通过一个Map，以键值对的方式维护这些上下文数据，SchedulerContext为保存和获取数据提供了多个put()和getXxx()的方法。可以通过Scheduler\# getContext()获取对应的SchedulerContext实例；

# 2.Quartz设计分析 #

## 2.1 quartz.properties ##

Quartz的jar包中有一个叫做quartz.properties的配置文件，它允许你修改框架运行时环境。缺省是使用Quartz.jar里面的quartz.properties文件。当然，你应该创建一个quartz.properties文件的副本并且把它放入你工程的classes目录中以便类加载器找到它。一旦将Quartz.jar文件和第三方库加到自己的工程里面并且quartz.properties文件在工程的classes目录中，就可以创建作业了。

Quartz默认的配置文件quartz.properties如下：

# Default Properties file for use by StdSchedulerFactory 
         # to create a Quartz Scheduler Instance, if a different 
         # properties file is not explicitly specified. 
         # 
         
         org .quartz .scheduler .instanceName: DefaultQuartzScheduler 
         org .quartz .scheduler .rmi .export: false 
         org .quartz .scheduler .rmi .proxy: false 
         org .quartz .scheduler .wrapJobExecutionInUserTransaction: false 
         
         org .quartz .threadPool .class: org .quartz .simpl .SimpleThreadPool 
         org .quartz .threadPool .threadCount: 10 
         org .quartz .threadPool .threadPriority: 5 
         org .quartz .threadPool .threadsInheritContextClassLoaderOfInitializingThread: true 
         
         org .quartz .jobStore .misfireThreshold: 60000 
         
         org .quartz .jobStore .class: org .quartz .simpl .RAMJobStore

## 2.2 Quartz Demo ##

### 2.2.1SimpleTrigger创建简单的任务 ###

package com.test.quartzdemo.helloworld; 
         
         import org.quartz.*; 
         import org.quartz.impl.StdSchedulerFactory; 
         
         /** 
         * @Author zhouguanya 
         * @Date 2018/6/10 
         * @Description 
         */ 
         public class SimpleTriggerDemo { 
         public static void main(String[] args) throws InterruptedException { 
         
         //通过schedulerFactory获取一个调度器 
         SchedulerFactory schedulerfactory = new StdSchedulerFactory(); 
         Scheduler scheduler = null; 
         try { 
         // 通过schedulerFactory获取一个调度器 
         scheduler = schedulerfactory.getScheduler(); 
         
         // 创建jobDetail实例，绑定Job实现类 
         // 指明job的名称，所在组的名称，以及绑定job类 
         JobDetail job = JobBuilder.newJob(HelloQuartz.class).withIdentity( "JobName", "JobGroupName").build(); 
         
         // 定义调度触发规则 
         
         // SimpleTrigger 
         Trigger trigger = TriggerBuilder.newTrigger().withIdentity( "SimpleTrigger", "SimpleTriggerGroup") 
         .withSchedule(SimpleScheduleBuilder.repeatSecondlyForever( 1).withRepeatCount( 5)) 
         //默认当前时间启动 
         .startNow().build(); 
         
         // cron表达式 每1秒执行一次 (http://cron.qqe2.com/) 
         // Trigger trigger=TriggerBuilder.newTrigger().withIdentity("CronTrigger1", "CronTriggerGroup") 
         // .withSchedule(CronScheduleBuilder.cronSchedule("*/1 * * * * ?")) 
         // .startNow().build(); 
         
         //把作业和触发器注册到任务调度中 
         scheduler.scheduleJob(job, trigger); 
         
         // 启动调度 
         scheduler.start(); 
         
         Thread.sleep( 10000); 
         
         // 停止调度 
         scheduler.shutdown(); 
         
         } catch (SchedulerException e) { 
         e.printStackTrace(); 
         } 
         
         } 
         }

### 2.2.1CronTrigger创建基于Crontab的任务 ###

package com.test.quartzdemo.helloworld; 
         
         import org.quartz.*; 
         import org.quartz.impl.StdSchedulerFactory; 
         
         /** 
         * @Author zhouguanya 
         * @Date 2018/6/10 
         * @Description 
         */ 
         public class CronTriggerDemo { 
         public static void main(String[] args) throws InterruptedException { 
         
         //通过schedulerFactory获取一个调度器 
         SchedulerFactory schedulerfactory = new StdSchedulerFactory(); 
         Scheduler scheduler = null; 
         try { 
         // 通过schedulerFactory获取一个调度器 
         scheduler = schedulerfactory.getScheduler(); 
         
         // 创建jobDetail实例，绑定Job实现类 
         // 指明job的名称，所在组的名称，以及绑定job类 
         JobDetail job = JobBuilder.newJob(HelloQuartz.class).withIdentity( "JobName", "JobGroupName").build(); 
         
         // 定义调度触发规则 
         // cron表达式 每1秒执行一次 (http://cron.qqe2.com/) 
         Trigger trigger=TriggerBuilder.newTrigger().withIdentity( "CronTrigger", "CronTriggerGroup") 
         .withSchedule(CronScheduleBuilder.cronSchedule( "*/1 * * * * ?")) 
         .startNow().build(); 
         
         //把作业和触发器注册到任务调度中 
         scheduler.scheduleJob(job, trigger); 
         
         // 启动调度 
         scheduler.start(); 
         
         Thread.sleep( 10000); 
         
         // 停止调度 
         scheduler.shutdown(); 
         
         } catch (SchedulerException e) { 
         e.printStackTrace(); 
         } 
         
         } 
         }

## 2.3 作业/任务 ##

用Quartz的属于来说，任务是一个执行任务的简单Java类。任务可以是任何Java代码。只需你实现org.quartz.Job接口并且在出现严重错误情况下抛出JobExecutionException异常即可。Job接口包含唯一的一个方法execute()，作业从这里开始执行。一旦实现了Job接口和execute()方法，当Quartz确定该是作业运行的时候，它将调用你的作业。execute()方法内就完全是你要做的事情。

### 2.3.1作业管理和存储 ###

作业一旦被调度，调度器需要记住并且跟踪作业和它们的执行次数。如果你的作业是30分钟后或每30秒调用，这不是很有用。事实上，作业执行需要非常准确和即时调用在被调度作业上的execute()方法。Quartz通过一个称之为作业存储（JobStore）的概念来做作业存储和管理。

Quartz提供两种基本作业存储类型。

*  第一种类型叫做RAMJobStore，它利用通常的内存来持久化调度程序信息。这种作业存储类型最容易配置、构造和运行。对许多应用来说，这种作业存储已经足够了。
 *  第二种类型的作业存储称为JDBC作业存储。Quartz支持两种JDBC作业存储，都需要JDBC驱动程序和后台数据库来持久化调度程序信息。
    
    这两种类型的不同在于你是否想要控制数据库事务或这释放控制给应用服务器例如BEA's WebLogic或Jboss。（这类似于J2EE领域中，Bean管理的事务和和容器管理事务之间的区别）这两种JDBC作业存储是：
    
    （1）JobStoreTX：当你想要控制事务或工作在非应用服务器环境中的时候使用
    
    （2）JobStoreCMT：当你工作在应用服务器环境中和想要容器控制事务时使用

## 2.4作业和触发器 ##

### 2.4.1 触发器 ###

Quartz将调度和作业分离。Quartz中的触发器用来告诉调度程序作业什么时候触发。框架提供了一些触发器类型，但两个最常用的是SimpleTrigger和CronTrigger。SimpleTrigger为需要简单调度而设计。

典型地，如果你需要在给定的时间和重复次数或者两次调度之间等待的固定的时间后再次调度个作业，那么SimpleTrigger适合你。另一方面，如果你有许多复杂的作业调度（基于日期的调度），那么或许需要CronTrigger。

CronTrigger是基于Calendar-like调度的。当你需要在除星期六和星期天外的每天上午10点半执行作业时，那么应该使用CronTrigger。正如它的名字所暗示的那样，CronTrigger是基于Unix crontab表达式的。

作为一个例子，下面的Quartz克隆表达式将在星期一到星期五的每天上午10点15分执行一个作业。

0 15 10 ? \* MON-FRI

将在2018年到2028年的每个月的最后一个星期五上午10点15分执行作业。你不可能用SimpleTrigger来做这些事情。你可以用两者之中的任何一个，但哪个跟合适则取决于你的调度需要。

0 15 10 ? \* 6L 2018-2028

cron表达式在线验证：[http://cron.qqe2.com/][http_cron.qqe2.com]

### 2.4.2 有状态作业和无状态作业 ###

无状态的作业。这意味着在两次作业执行之间，不会去维护作业执行时JobDataMap的状态改变。如果你需要能增、删，改JobDataMap的值，而且能让作业在下次执行时能看到这个状态改变，则需要用Quartz有状态作业。Quartz有状态作业实现了org.quartz.StatefulJob接口。

无状态和有状态作业的关键不同是有状态作业在每次执行时只有一个实例。大多数情况下，有状态的作业不回带来大的问题。然而，如果你有一个需要频繁执行的作业或者需要很长时间才能完成的作业，那么有状态作业可能给你带来伸缩性问题。

## 2.5错过触发(misfire) ##

trigger还有一个重要的属性misfire；如果scheduler关闭了，或者Quartz线程池中没有可用的线程来执行job，此时持久性的trigger就会错过(miss)其触发时间，即错过触发(misfire)。不同类型的trigger，有不同的misfire机制。它们默认都使用“智能机制(smart policy)”，即根据trigger的类型和配置动态调整行为。当scheduler启动的时候，查询所有错过触发(misfire)的持久性trigger。然后根据它们各自的misfire机制更新trigger的信息。当你在项目中使用Quartz时，你应该对各种类型的trigger的misfire机制都比较熟悉，这些misfire机制在JavaDoc中有说明。关于misfire机制的细节，会在讲到具体的trigger时作介绍。

# 3.执行时序图 #

![image.png][]

# 4.QuartzSchedulerThread解析 #

QuartzScheduler在哪里调用初始化的？

QuartzScheduler的构造函数中会初始化一个**private QuartzSchedulerThread schedThread**对象，就是在QuartzScheduler的构造函数中，调用了的构造函数QuartzSchedulerThread，实例化QuartzSchedulerThread后，会将这个线程提交到线程池去执行。**ThreadExecutor schedThreadExecutor = resources.getThreadExecutor()**; **schedThreadExecutor.execute(this.schedThread);**

public QuartzScheduler(QuartzSchedulerResources resources, long idleWaitTime, @Deprecated long dbRetryInterval) 
         throws SchedulerException { 
         this.resources = resources; 
         if (resources.getJobStore() instanceof JobListener) { 
         addInternalJobListener((JobListener)resources.getJobStore()); 
         } 
         
         this.schedThread = new QuartzSchedulerThread( this, resources); 
         ThreadExecutor schedThreadExecutor = resources.getThreadExecutor(); 
         schedThreadExecutor.execute( this.schedThread); 
         if (idleWaitTime > 0) { 
         this.schedThread.setIdleWaitTime(idleWaitTime); 
         } 
         
         jobMgr = new ExecutingJobsManager(); 
         addInternalJobListener(jobMgr); 
         errLogger = new ErrorLogger(); 
         addInternalSchedulerListener(errLogger); 
         
         signaler = new SchedulerSignalerImpl( this, this.schedThread); 
         
         getLog().info( "Quartz Scheduler v." + getVersion() + " created."); 
         }

QuartzSchedulerThread是负责执行在QuartzScheduler中注册的触发器的这样一个线程。他负责寻找满足触发条件的触发器，然后将与这些触发器相关的任务放到线程池中执行。

QuartzScheduleThread类的构造函数：

QuartzSchedulerThread(QuartzScheduler qs, QuartzSchedulerResources qsRsrcs) { 
         this(qs, qsRsrcs, qsRsrcs.getMakeSchedulerThreadDaemon(), Thread.NORM_PRIORITY); 
         }

QuartzSchedulerThread(QuartzScheduler qs, QuartzSchedulerResources qsRsrcs, boolean setDaemon, int threadPrio) { 
         super(qs.getSchedulerThreadGroup(), qsRsrcs.getThreadName()); 
         this.qs = qs; 
         this.qsRsrcs = qsRsrcs; 
         this.setDaemon(setDaemon); 
         if(qsRsrcs.isThreadsInheritInitializersClassLoadContext()) { 
         log.info( "QuartzSchedulerThread Inheriting ContextClassLoader of thread: " + Thread.currentThread().getName()); 
         this.setContextClassLoader(Thread.currentThread().getContextClassLoader()); 
         } 
         
         this.setPriority(threadPrio); 
         
         // start the underlying thread, but put this object into the 'paused' 
         // state 
         // so processing doesn't start yet... 
         paused = true; 
         halted = new AtomicBoolean( false); 
         }

重点看下一下这段代码：

// start the underlying thread, but put this object into the 'paused' 
         // state 
         // so processing doesn't start yet... 
         paused = true; 
         halted = new AtomicBoolean( false);

这段代码上面的注释的大概意思是：启动线程，但是将paused置'paused'，字面意思上来说，就是设置为暂停的状态。halted初始值置为false。halted是暂停下一次线程执行的一个信号。下面看下QuartzSchedulerThread的run()方法，看下QuartzSchedulerThread线程到底做了什么！

@ Override 
 public void run( ) { 
 int acquiresFailed = 0; 
 
 while (!halted. get()) { 
 try { 
 // check if we're supposed to pause... 
 synchronized (sigLock) { 
 while (paused && !halted. get()) { 
 try { 
 // wait until togglePause(false) is called... 
 sigLock.wait( 1000L); 
 } catch (InterruptedException ignore) { 
 } 
 
 // reset failure counter when paused, so that we don't 
 // wait again after unpausing 
 acquiresFailed = 0; 
 } 
 
 if (halted. get()) { 
 break; 
 } 
 } 
 
 // wait a bit, if reading from job store is consistently 
 // failing (e.g. DB is down or restarting).. 
 if (acquiresFailed > 1) { 
 try { 
 long delay = computeDelayForRepeatedErrors(qsRsrcs.getJobStore(), acquiresFailed); 
 Thread.sleep(delay); 
 } catch (Exception ignore) { 
 } 
 } 
 
 int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads(); 
 if(availThreadCount > 0) { // will always be true, due to semantics of blockForAvailableThreads... 
 
 List<OperableTrigger> triggers; 
 
 long now = System.currentTimeMillis(); 
 
 clearSignaledSchedulingChange(); 
 try { 
 triggers = qsRsrcs.getJobStore().acquireNextTriggers( 
 now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow()); 
 acquiresFailed = 0; 
 if (log.isDebugEnabled()) 
 log.debug( "batch acquisition of " + (triggers == null ? 0 : triggers.size()) + " triggers"); 
 } catch (JobPersistenceException jpe) { 
 if (acquiresFailed == 0) { 
 qs.notifySchedulerListenersError( 
 "An error occurred while scanning for the next triggers to fire.", 
 jpe); 
 } 
 if (acquiresFailed < Integer.MAX_VALUE) 
 acquiresFailed++; 
 continue; 
 } catch (RuntimeException e) { 
 if (acquiresFailed == 0) { 
 getLog().error( "quartzSchedulerThreadLoop: RuntimeException " 
 +e.getMessage(), e); 
 } 
 if (acquiresFailed < Integer.MAX_VALUE) 
 acquiresFailed++; 
 continue; 
 } 
 
 if (triggers != null && !triggers.isEmpty()) { 
 
 now = System.currentTimeMillis(); 
 long triggerTime = triggers. get( 0).getNextFireTime().getTime(); 
 long timeUntilTrigger = triggerTime - now; 
 while(timeUntilTrigger > 2) { 
 synchronized (sigLock) { 
 if (halted. get()) { 
 break; 
 } 
 if (!isCandidateNewTimeEarlierWithinReason(triggerTime, false)) { 
 try { 
 // we could have blocked a long while 
 // on 'synchronize', so we must recompute 
 now = System.currentTimeMillis(); 
 timeUntilTrigger = triggerTime - now; 
 if(timeUntilTrigger >= 1) 
 sigLock.wait(timeUntilTrigger); 
 } catch (InterruptedException ignore) { 
 } 
 } 
 } 
 if(releaseIfScheduleChangedSignificantly(triggers, triggerTime)) { 
 break; 
 } 
 now = System.currentTimeMillis(); 
 timeUntilTrigger = triggerTime - now; 
 } 
 
 // this happens if releaseIfScheduleChangedSignificantly decided to release triggers 
 if(triggers.isEmpty()) 
 continue; 
 
 // set triggers to 'executing' 
 List<TriggerFiredResult> bndles = new ArrayList<TriggerFiredResult>(); 
 
 boolean goAhead = true; 
 synchronized(sigLock) { 
 goAhead = !halted. get(); 
 } 
 if(goAhead) { 
 try { 
 List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers); 
 if(res != null) 
 bndles = res; 
 } catch (SchedulerException se) { 
 qs.notifySchedulerListenersError( 
 "An error occurred while firing triggers '" 
 + triggers + "'", se); 
 //QTZ-179 : a problem occurred interacting with the triggers from the db 
 //we release them and loop again 
 for ( int i = 0; i < triggers.size(); i++) { 
 qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers. get(i)); 
 } 
 continue; 
 } 
 
 } 
 
 for ( int i = 0; i < bndles.size(); i++) { 
 TriggerFiredResult result = bndles. get(i); 
 TriggerFiredBundle bndle = result.getTriggerFiredBundle(); 
 Exception exception = result.getException(); 
 
 if (exception instanceof RuntimeException) { 
 getLog().error( "RuntimeException while firing trigger " + triggers. get(i), exception); 
 qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers. get(i)); 
 continue; 
 } 
 
 // it's possible to get 'null' if the triggers was paused, 
 // blocked, or other similar occurrences that prevent it being 
 // fired at this time... or if the scheduler was shutdown (halted) 
 if (bndle == null) { 
 qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers. get(i)); 
 continue; 
 } 
 
 JobRunShell shell = null; 
 try { 
 shell = qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle); 
 shell.initialize(qs); 
 } catch (SchedulerException se) { 
 qsRsrcs.getJobStore().triggeredJobComplete(triggers. get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR); 
 continue; 
 } 
 
 if (qsRsrcs.getThreadPool().runInThread(shell) == false) { 
 // this case should never happen, as it is indicative of the 
 // scheduler being shutdown or a bug in the thread pool or 
 // a thread pool being used concurrently - which the docs 
 // say not to do... 
 getLog().error( "ThreadPool.runInThread() return false!"); 
 qsRsrcs.getJobStore().triggeredJobComplete(triggers. get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR); 
 } 
 
 } 
 
 continue; // while (!halted) 
 } 
 } else { // if(availThreadCount > 0) 
 // should never happen, if threadPool.blockForAvailableThreads() follows contract 
 continue; // while (!halted) 
 } 
 
 long now = System.currentTimeMillis(); 
 long waitTime = now + getRandomizedIdleWaitTime(); 
 long timeUntilContinue = waitTime - now; 
 synchronized(sigLock) { 
 try { 
 if(!halted. get()) { 
 // QTZ-336 A job might have been completed in the mean time and we might have 
 // missed the scheduled changed signal by not waiting for the notify() yet 
 // Check that before waiting for too long in case this very job needs to be 
 // scheduled very soon 
 if (!isScheduleChanged()) { 
 sigLock.wait(timeUntilContinue); 
 } 
 } 
 } catch (InterruptedException ignore) { 
 } 
 } 
 
 } catch(RuntimeException re) { 
 getLog().error( "Runtime error occurred in main trigger firing loop.", re); 
 } 
 } // while (!halted) 
 
 // drop references to scheduler stuff to aid garbage collection... 
 qs = null; 
 qsRsrcs = null; 
 }

看下线程QuartzSchedulerThread被刚刚初始化后，会做什么操作？run()方式里面有个循环体，halted=false（线程没有暂停的时候），线程的run()方法会一直while循环执行操作。

看到while循环中第一个同步块的代码：

// check if we're supposed to pause... 
         synchronized (sigLock) { 
         while (paused && !halted. get()) { 
         try { 
         // wait until togglePause(false) is called... 
         sigLock.wait( 1000L); 
         } catch (InterruptedException ignore) { 
         } 
         
         // reset failure counter when paused, so that we don't 
         // wait again after unpausing 
         acquiresFailed = 0; 
         } 
         
         if (halted. get()) { 
         break; 
         } 
         }

可以看出，线程刚被初始化的情况下，paused=false，halted=false，所以会直接等待1s。也就是说QuartzSchedulerThread刚被初始化提交到线程池后，线程等于是没有执行逻辑的，就是无限的等待。这也就是对应了时序图中的“执行线程并未执行，等待执行指令”这个状态。

// wait a bit, if reading from job store is consistently 
         // failing (e.g. DB is down or restarting).. 
         if (acquiresFailed > 1) { 
         try { 
         long delay = computeDelayForRepeatedErrors(qsRsrcs.getJobStore(), acquiresFailed); 
         Thread.sleep(delay); 
         } catch (Exception ignore) { 
         } 
         }

这上面一段代码，从注释上就可以看出其功能：如果持续的从jobstore取数据失败，就会计算一个延迟时间，让线程休眠一段时间（休眠的时间长短 = delay）。

int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads(); 
         if(availThreadCount > 0) { // will always be true, due to semantics of blockForAvailableThreads... 
         //todo 正常的逻辑继续往下走 
         ...... 
         } else { 
         continue; 
         }

上面这段代码是查询当前的线程池中，可用的线程数量。如果有可用的线程，则继续向下执行，否则直接continue，进入了一轮新的循环中。可以看一下SimpleThreadPool中blockForAvailableThreads()方法的实现如下：

public int blockForAvailableThreads() { 
 synchronized(nextRunnableLock) { 
 
 while((availWorkers.size() < 1 || handoffPending) && !isShutdown) { 
 try { 
 nextRunnableLock.wait( 500); 
 } catch (InterruptedException ignore) { 
 } 
 } 
 
 return availWorkers.size(); 
 } 
 }

在blockForAvailableThreads方法实现中，可以发现，while循环的条件，当没有可用的线程时，会等待500ms，然后在查询一下有没有可用的线程。所以在QuartzSchedulerThread的run()的源码中，有这样一句注释：// will always be true, due to semantics of blockForAvailableThreads... 这就是说从blockForAvailableThreads()的语义来看，这个if(availThreadCount > 0)条件总是为true。下面就if(availThreadCount > 0)这个条件内了，即线程池中有可用的空闲的线程。

triggers = qsRsrcs.getJobStore().acquireNextTriggers( 
         now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());

这段代码的作用：QuartzSchedulerThread线程在trigger队列中寻找未来30秒内要执行一定数目的触发器。JobStore接口的实现有很多，这里拿RAMJobStore举例，RAMJobStore中acquireNextTriggers()的实现，其目的是：获取要触发的下一个触发器的句柄，并将其标记为调用调度程序的“保留”。

List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);

上面这段代码是让触发器去执行。

如果执行触发器发生异常了，怎么办？

catch (SchedulerException se) { 
 qs.notifySchedulerListenersError( 
 "An error occurred while firing triggers '" 
 + triggers + "'", se); 
 //QTZ-179 : a problem occurred interacting with the triggers from the db 
 //we release them and loop again 
 for ( int i = 0; i < triggers.size(); i++) { 
 qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers. get(i)); 
 } 
 continue; 
 }

从上面代码看出，在让触发器执行的过程中如果出现SchedulerException异常，那么就释放trigger。

正常流程继续往下走。

List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);做了以下的处理：

1.处理每个trigger

2.检验每个trigger的状态，只有可运行状态才是可执行的。

3.拿到JobDetail中的Job

4.修改trigger状态为执行中

5.修改下一次trigger触发时间

接下来解释执行满足触发条件的这些触发器关联的任务了。

JobRunShell shell = null; 
         try { 
         shell = qsRsrcs.getJobRunShellFactory().createJobRunShell(bndle); 
         shell.initialize(qs); 
         } catch (SchedulerException se) { 
         qsRsrcs.getJobStore().triggeredJobComplete(triggers. get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR); 
         continue; 
         }

可以看到创建了一个JobRunShell对象，JobRunShell可以看成是一个执行Job的一个安全的环境，执行所有作业要做的工作，捕获所有的一次，在Job执行完成时，更新trigger状态。initialize()是初始化JobRunShell，initialize()中有一行代码，会通过反射创建Job实例化对象。

job = sched.getJobFactory().newJob(firedTriggerBundle, scheduler);

看下newJob()方法的实现

public Job newJob(TriggerFiredBundle bundle, Scheduler Scheduler) throws SchedulerException { 
 
 JobDetail jobDetail = bundle.getJobDetail(); 
 Class<? extends Job> jobClass = jobDetail.getJobClass(); 
 try { 
 if( log.isDebugEnabled()) { 
 log.debug( 
 "Producing instance of Job '" + jobDetail.getKey() + 
 "', class=" + jobClass.getName()); 
 } 
 
 return jobClass.newInstance(); 
 } catch (Exception e) { 
 SchedulerException se = new SchedulerException( 
 "Problem instantiating class '" 
 + jobDetail.getJobClass().getName() + "'", e); 
 throw se; 
 } 
 }

initialize()执行结束后，就可以在JobRunShell中的JobExecutionContextImpl jec对象中得到Job实例化的对象(JobExecutionContextImpl是包含了执行环境需要的变量。由JobRunShell在初使化时创建)。（与Spring明显不同的是，这里每个Job都是一个实例化对象，而非单利。如果要想在Quartz中实现单利对象，需要使用StatefulJob接口。本例默认描述的Job接口的相关实现）

准备好了Job执行环境后，就可以将Job提交到线程池中去执行了。

qsRsrcs.getThreadPool().runInThread(shell)//这是把shell真正放到线程池中的一个线程上去运行。

至此，一个完成的Quartz执行流程算是结束了。

流程图：https://www.processon.com/view/link/5b33730be4b063f71f4b7f40

[11627468_1438829844Zbz8.png]: https://123.sankuai.com/km/api/file/57639854/58096866
[http_cron.qqe2.com]: http://cron.qqe2.com/
[image.png]: https://123.sankuai.com/km/api/file/57639854/58301041