Quartz 源码分析

矫情吗；* 2022-11-25 11:36 96阅读 0赞

`Quartz` 是运用最广的任务调度框架，它最核心的组成部分是 `Scheduler`、`Trigger`、`JobDetail`。关于如何使用 Quartz 可以参考之前的 Blog – [史上最全的企业级定时任务框架 Quartz 介绍][Quartz]。在 `Scheduler` 初始化的时候会启动线程`QuartzSchedulerThread`，当`Scheduler` 启动的时候，也就是调用 `Scheduler#start` 方法，就会从 `JobStore` 里拿到最近需要触发的`Trigger`，以线程等待的方式等到 `trigger` 触发时间点，之后就是执行 `trigger` 所关联的 `JobDetail` 。此篇文章基于 Quartz-2.3.2，Scheduler初始化、start和 trigger执行的时序图如下所示：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTA3MzM_size_16_color_FFFFFF_t_70_pic_center]

## 1、QuartzSchedulerThread 启动 ##

在 `QuartzSchedulerThread` 初始化的时候会设置 `paused` 属性为 true。

> org.quartz.core.QuartzSchedulerThread\#QuartzSchedulerThread()

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

所以 `QuartzSchedulerThread` 线程在运行的时候会一直阻塞，直到 `Scheduler#start`方法执行。它会调用 `schedThread.togglePause(false);` 设置 `paused`(暂停) 为 false。并且调用 `sigLock.notifyAll()` 释放锁。

> org.quartz.core.QuartzSchedulerThread\#run

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

这个时候 QuartzSchedulerThread 就不会被阻塞了，下面就是任务的运行。

## 2、获取需要运行的 Trigger ##

下面就是QuartzSchedulerResources 获取到 JobStore ，然后从 JobStore 得到需要执行的 OperableTrigger 列表。

> org.quartz.core.QuartzSchedulerThread\#run

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

下面我们以内存保存任务信息 `RAMJobStore` 为例讲解如何获取需要执行的 `Trigger` 信息。

> org.quartz.simpl.RAMJobStore\#acquireNextTriggers

public List<OperableTrigger> acquireNextTriggers(long noLaterThan, int maxCount, long timeWindow) { 
            synchronized (lock) { 
                List<OperableTrigger> result = new ArrayList<OperableTrigger>();
                Set<JobKey> acquiredJobKeysForNoConcurrentExec = new HashSet<JobKey>();
                Set<TriggerWrapper> excludedTriggers = new HashSet<TriggerWrapper>();
                long batchEnd = noLaterThan;
                
                // return empty list if store has no triggers.
                if (timeTriggers.size() == 0)
                    return result;
                
                while (true) { 
                    TriggerWrapper tw;
    
                    try { 
                        tw = timeTriggers.first();
                        if (tw == null)
                            break;
                        timeTriggers.remove(tw);
                    } catch (java.util.NoSuchElementException nsee) { 
                        break;
                    }
    
                    if (tw.trigger.getNextFireTime() == null) { 
                        continue;
                    }
    
                    if (applyMisfire(tw)) { 
                        if (tw.trigger.getNextFireTime() != null) { 
                            timeTriggers.add(tw);
                        }
                        continue;
                    }
    
                    if (tw.getTrigger().getNextFireTime().getTime() > batchEnd) { 
                        timeTriggers.add(tw);
                        break;
                    }
                    
                    // If trigger's job is set as @DisallowConcurrentExecution, and it has already been added to result, then
                    // put it back into the timeTriggers set and continue to search for next trigger.
                    JobKey jobKey = tw.trigger.getJobKey();
                    JobDetail job = jobsByKey.get(tw.trigger.getJobKey()).jobDetail;
                    if (job.isConcurrentExectionDisallowed()) { 
                        if (acquiredJobKeysForNoConcurrentExec.contains(jobKey)) { 
                            excludedTriggers.add(tw);
                            continue; // go to next trigger in store.
                        } else { 
                            acquiredJobKeysForNoConcurrentExec.add(jobKey);
                        }
                    }
    
                    tw.state = TriggerWrapper.STATE_ACQUIRED;
                    tw.trigger.setFireInstanceId(getFiredTriggerRecordId());
                    OperableTrigger trig = (OperableTrigger) tw.trigger.clone();
                    if (result.isEmpty()) { 
                        batchEnd = Math.max(tw.trigger.getNextFireTime().getTime(), System.currentTimeMillis()) + timeWindow;
                    }
                    result.add(trig);
                    if (result.size() == maxCount)
                        break;
                }
    
                // If we did excluded triggers to prevent ACQUIRE state due to DisallowConcurrentExecution, we need to add them back to store.
                if (excludedTriggers.size() > 0)
                    timeTriggers.addAll(excludedTriggers);
                return result;
            }
        }

*  需要执行的 Triggers 都被保存到 TreeSet 里面，所以它是可排序的，在这里面拿出需要执行的第一个 Trigger。(具体的排序信息我们后面分析)
 *  如果获取到的 TriggerWrapper 为空，直接跳出循环并且返回空 Triggers 列表
 *  移除当前 TriggerWrapper，如果获取到的 TriggerWrapper 下一次执行时间为空不进行后续操作，继续拿出需要执行的第一个 TriggerWrapper。
 *  `applyMisfire`，判断当前 TriggerWrapper 是否需要被执行，并且更新下一次需要执行的时候，如果不需要执行把 TriggerWrapper 从新放入 TreeSet 里面再次排序。
 *  如果 TriggerWrapper 的下一次执行时间大于传入时间把获取到的定时任务从新放入到 TreeSet 里面再次排序。
 *  这时 TriggerWrapper 的执行时间就小于传入的时间，就需要执行这个 Trigger，把这个 Trigger 放入到 result 里面，如果 result 的 size 大于 maxCount 就返回结果集(maxCount 的默认值为 1， 可以通过参数 `org.quartz.scheduler.batchTriggerAcquisitionMaxCount` 设置)。

**下面我们来分析一下 TriggerWrapper 的排序规则。**

> org.quartz.simpl.RAMJobStore\#timeTriggers

protected TreeSet<TriggerWrapper> timeTriggers = new TreeSet<TriggerWrapper>(new TriggerWrapperComparator());

可以看到它的排序规则是基于类： `TriggerWrapperComparator`。

> org.quartz.simpl.TriggerWrapperComparator

class TriggerWrapperComparator implements Comparator<TriggerWrapper>, java.io.Serializable { 
      
        private static final long serialVersionUID = 8809557142191514261L;
    
        TriggerTimeComparator ttc = new TriggerTimeComparator();
        
        public int compare(TriggerWrapper trig1, TriggerWrapper trig2) { 
            return ttc.compare(trig1.trigger, trig2.trigger);
        }
    
        @Override
        public boolean equals(Object obj) { 
            return (obj instanceof TriggerWrapperComparator);
        }
    
        @Override
        public int hashCode() { 
            return super.hashCode();
        }
    }

接着我们看 `TriggerTimeComparator`，根据类名我们就可以看出它是根据 Trigger 的执行时间来排序的。

> org.quartz.Trigger.TriggerTimeComparator

class TriggerTimeComparator implements Comparator<Trigger>, Serializable { 
      
        private static final long serialVersionUID = -3904243490805975570L;
        
        // This static method exists for comparator in TC clustered quartz
        public static int compare(Date nextFireTime1, int priority1, TriggerKey key1, Date nextFireTime2, int priority2, TriggerKey key2) { 
            if (nextFireTime1 != null || nextFireTime2 != null) { 
                if (nextFireTime1 == null) { 
                    return 1;
                }
    
                if (nextFireTime2 == null) { 
                    return -1;
                }
    
                if(nextFireTime1.before(nextFireTime2)) { 
                    return -1;
                }
    
                if(nextFireTime1.after(nextFireTime2)) { 
                    return 1;
                }
            }
    
            int comp = priority2 - priority1;
            if (comp != 0) { 
                return comp;
            }
    
            return key1.compareTo(key2);
        }
    
    
        public int compare(Trigger t1, Trigger t2) { 
            return compare(t1.getNextFireTime(), t1.getPriority(), t1.getKey(), t2.getNextFireTime(), t2.getPriority(), t2.getKey());
        }
    }

*  首先根据 Trigger 的下一次的执行时间进行排序，如果时间谁早就先执行
 *  然后根据 Trigger 的执行优先级，默认是 5.
 *  最后根据 Trigger 的 group 以及 name 进行排序

## 3、JobDetail 的执行 ##

把之前获取到的 `OperableTrigger` 转化成 `TriggerFiredBundle`，通过 `TriggerFiredBundle` 创建需要运行的 `JobRunShell`。

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

*  获取 `QuartzSchedulerResources` 中的 `JobRunShellFactory` 的实例 `JTAAnnotationAwareJobRunShellFactory` 用于创建 Job 运行的 `JobRunShell`对象。
 *  `JobRunShellFactory#createJobRunShell` 调用 `new JobRunShell(scheduler, bundle)` 创建 `JobRunShell`。
 *  `JobRunShell#initialize` 初始化创建需要运行的 `Job`实例，并且创建 `Job` 运行上下文`(JobExecutionContext)`.
 *  `qsRsrcs.getThreadPool().runInThread(shell)` 通过线程池运行定时任务。

下面我们来看一下 `JobRunShell` 是如何运行的。

> org.quartz.core.JobRunShell\#run

public void run() { 
            qs.addInternalSchedulerListener(this);
    
            try { 
                OperableTrigger trigger = (OperableTrigger) jec.getTrigger();
                JobDetail jobDetail = jec.getJobDetail();
    
                do { 
    
                    JobExecutionException jobExEx = null;
                    Job job = jec.getJobInstance();
    
                    try { 
                        begin();
                    } catch (SchedulerException se) { 
                        qs.notifySchedulerListenersError("Error executing Job ("
                                + jec.getJobDetail().getKey()
                                + ": couldn't begin execution.", se);
                        break;
                    }
    
                    // notify job & trigger listeners...
                    try { 
                        if (!notifyListenersBeginning(jec)) { 
                            break;
                        }
                    } catch(VetoedException ve) { 
                        try { 
                            CompletedExecutionInstruction instCode = trigger.executionComplete(jec, null);
                            qs.notifyJobStoreJobVetoed(trigger, jobDetail, instCode);
                            
                            // QTZ-205
                            // Even if trigger got vetoed, we still needs to check to see if it's the trigger's finalized run or not.
                            if (jec.getTrigger().getNextFireTime() == null) { 
                                qs.notifySchedulerListenersFinalized(jec.getTrigger());
                            }
    
                            complete(true);
                        } catch (SchedulerException se) { 
                            qs.notifySchedulerListenersError("Error during veto of Job ("
                                    + jec.getJobDetail().getKey()
                                    + ": couldn't finalize execution.", se);
                        }
                        break;
                    }
    
                    long startTime = System.currentTimeMillis();
                    long endTime = startTime;
    
                    // execute the job
                    try { 
                        log.debug("Calling execute on job " + jobDetail.getKey());
                        job.execute(jec);
                        endTime = System.currentTimeMillis();
                    } catch (JobExecutionException jee) { 
                        endTime = System.currentTimeMillis();
                        jobExEx = jee;
                        getLog().info("Job " + jobDetail.getKey() +
                                " threw a JobExecutionException: ", jobExEx);
                    } catch (Throwable e) { 
                        endTime = System.currentTimeMillis();
                        getLog().error("Job " + jobDetail.getKey() +
                                " threw an unhandled Exception: ", e);
                        SchedulerException se = new SchedulerException(
                                "Job threw an unhandled exception.", e);
                        qs.notifySchedulerListenersError("Job ("
                                + jec.getJobDetail().getKey()
                                + " threw an exception.", se);
                        jobExEx = new JobExecutionException(se, false);
                    }
    
                    jec.setJobRunTime(endTime - startTime);
    
                    // notify all job listeners
                    if (!notifyJobListenersComplete(jec, jobExEx)) { 
                        break;
                    }
    
                    CompletedExecutionInstruction instCode = CompletedExecutionInstruction.NOOP;
    
                    // update the trigger
                    try { 
                        instCode = trigger.executionComplete(jec, jobExEx);
                    } catch (Exception e) { 
                        // If this happens, there's a bug in the trigger...
                        SchedulerException se = new SchedulerException(
                                "Trigger threw an unhandled exception.", e);
                        qs.notifySchedulerListenersError(
                                "Please report this error to the Quartz developers.",
                                se);
                    }
    
                    // notify all trigger listeners
                    if (!notifyTriggerListenersComplete(jec, instCode)) { 
                        break;
                    }
    
                    // update job/trigger or re-execute job
                    if (instCode == CompletedExecutionInstruction.RE_EXECUTE_JOB) { 
                        jec.incrementRefireCount();
                        try { 
                            complete(false);
                        } catch (SchedulerException se) { 
                            qs.notifySchedulerListenersError("Error executing Job ("
                                    + jec.getJobDetail().getKey()
                                    + ": couldn't finalize execution.", se);
                        }
                        continue;
                    }
    
                    try { 
                        complete(true);
                    } catch (SchedulerException se) { 
                        qs.notifySchedulerListenersError("Error executing Job ("
                                + jec.getJobDetail().getKey()
                                + ": couldn't finalize execution.", se);
                        continue;
                    }
    
                    qs.notifyJobStoreJobComplete(trigger, jobDetail, instCode);
                    break;
                } while (true);
    
            } finally { 
                qs.removeInternalSchedulerListener(this);
            }
        }

*  给当前运行的 Job 添加 SchedulerListener.
 *  调用扩展方法 `JobRunShell#begin`
 *  调用 Job 和 Trigger 的监听器的开始任务逻辑
 *  调用 Job\#execute(JobExecutionContext),执行定时业务逻辑
 *  调用 Trigger 的监听器的完成任务逻辑
 *  调用扩展方法 `JobRunShell#complete`

从上面的逻辑也印证了我们上面所说的 Quartz 的核心模块就是 `Scheduler`、`Trigger`、`JobDetail`，上面的 3 个章节都是围绕着这 3 个组件展开来的。

**参考文章：**

*  [Quartz源码分析（一）------ 以线程等待的方式实现按时间调度][Quartz_------]

[Quartz]: https://carlzone.blog.csdn.net/article/details/107748768
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTI0MTA3MzM_size_16_color_FFFFFF_t_70_pic_center]: /images/20221124/38b372a8439443358f462aa243f7c5af.png
[Quartz_------]: https://blog.csdn.net/cutesource/article/details/4965520?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522159619751719724848309812%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fall.%2522%257D&request_id=159619751719724848309812&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~all~first_rank_ecpm_v3~pc_rank_v2-14-4965520.first_rank_ecpm_v3_pc_rank_v2&utm_term=quartz+%E6%BA%90%E7%A0%81%E5%89%96%E6%9E%90&spm=1018.2118.3001.4187