Quartz源码解析 ---- 触发器按时启动原理

89 阅读 0 评论 59 点赞

我是靠谱客的博主开朗万宝路，最近开发中收集的这篇文章主要介绍Quartz源码解析 ---- 触发器按时启动原理，觉得挺不错的，现在分享给大家，希望可以做个参考。

概述

1.不停的轮询(while)，直到收到停止任务的要求
2.算出下次触发的时间，比如是 09:10:00 ，减去当前时间 09:00:00 ，也就是还有10分钟才会触发下次任务。那就wait 10分钟
p.s. 最近刚好在看quartz的源码，还是比较容易看懂，建议你也好好看看

QuartzSchedulerThread.class

 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;
}

从线程开始，我们来看下QuartzSchedulerThread类(负责执行触发的Trigger的工作) :

[java] 
view plain
 copy


 print
?
 @Override  
     public void run() {  
         boolean lastAcquireFailed = false;  
   
         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) {  
                         }  
                     }  
   
                     if (halted.get()) {  
                         break;  
                     }  
                 }  
   
                 int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();  

使用了AtomicBoolean类halted.get()检查触发器是否是暂停状态，早期版本的Quartz用的还是Boolean， AtomicBoolean类只有在值稳定后才会更新，保持操作原子性。从上面代码可以看出，只要是触发器停止状态或者等待状态， Thread就会一直wait。sigLock同步对象用来随时唤醒将被触发的Trigger(使用notifyAll来进行对wait中线程的唤醒，如下源码)

[java] 
view plain
 copy


 print
?
 synchronized(sigLock) {  
            signaled = true;  
            signaledNextFireTime = candidateNewNextFireTime;  
            sigLock.notifyAll();  
        }  

qsRsrcs.getThreadPool().blockForAvailableThreads() // 获取线程池现在可以用线程数

跟spring集成的配置(建议一般使用10~50个线程): 下面一行代码配置了最大线程数

[html] 
view plain
 copy


 print
?
 <prop key="org.quartz.threadPool.threadCount">40</prop>  

接上面run方法代码:

[java] 
view plain
 copy


 print
?
 if(availThreadCount > 0) { // will always be true, due to semantics of blockForAvailableThreads...  
   
                     List<OperableTrigger> triggers = null;  
   
                     long now = System.currentTimeMillis();  
   
                     clearSignaledSchedulingChange();  
                     try {  
                         // 查找30秒内一定数量的Trigger  
                         triggers = qsRsrcs.getJobStore().acquireNextTriggers(  
                                 now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());  
                         lastAcquireFailed = false;  
                         ...  
                     }   
                         ...  
                           
                     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) {  
                                     }  
                                 }  
                             ...  
                         }  
   
                         ...  
   
                         boolean goAhead = true;  
                         synchronized(sigLock) {  
                             goAhead = !halted.get();  
                         }  
                         if(goAhead) {  
                             try {  
                                 List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);  
                                 if(res != null)  
                                     bndles = res;  
                             }   
                             ...  
   
                         }  
   
                         for (int i = 0; i < bndles.size(); i++) {                             
                             ...  
                             
                             if (bndle == null) {  
                                 qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));  
                                 continue;  
                             }  