quartz框架(七)-JobStore - 幕友皎敖奔乾

JobStore

在之前的博文中，博主已经写了关于Job的相关内容。本篇博文，博主将介绍JobStore相关的内容。

JobStore是存放Job和Trigger的地方。当我们调用Scheduler对象的scheduleJob时就会将其存入JobStore中，然后供quartzSchedulerThread使用。

为什么需要JobStore？

因为我们需要被Scheduler调用的任务大多数并不是一次性的任务，而是需要被定时触发，或者某个时间点才能被触发的。因此我们需要一个容器来存储Job和Trigger的相关内容。

其次，quartz框架还有考虑到持久化存储的场景，比如说将对应的数据存放到数据库，这时候存放或者读取数据库里面的数据都需要有一个与之对应的容器。

JobStore接口中定义的方法太多，这里博主只列出一下比较重要的方法。接口定义如下所示：

public interface JobStore {
   
   //存储job和tigger
   void storeJobAndTrigger(JobDetail newJob, OperableTrigger newTrigger) 
        throws ObjectAlreadyExistsException, JobPersistenceException;
		
   //获取下一次需要进行触发的触发器
   List<OperableTrigger> acquireNextTriggers(long noLaterThan, int maxCount, long timeWindow)
        throws JobPersistenceException;
	
   //释放获取到的触发器
   void releaseAcquiredTrigger(OperableTrigger trigger);
   
   //触发器被触发
   List<TriggerFiredResult> triggersFired(List<OperableTrigger> triggers) throws JobPersistenceException;
   
   //触发器触发完成
   void triggeredJobComplete(OperableTrigger trigger, JobDetail jobDetail, CompletedExecutionInstruction triggerInstCode);
}

RAMJobStore

在quartz的默认配置中使用的就是RAMJobStore。顾名思义，RAMJobStore是基于内存来存储的Job相关数据，也就是在程序重启之后，对应的数据就会消失，并且它不支持集群，也就是说它不可以把需要调度的任务分配到多台机器上面进行执行。博主这里就以RAMJobStore为例，讲解一下RAMJobStore的相关实现。

RAMJobStore属性

1. jobsByKey，按jobKey进行分组的HashMap集合
2. triggersByKey，按jobKey进行分组的HashMap集合
3. jobsByGroup，按job的分组名进行分组的HashMap集合
4. triggersByGroup，按trigger的分组名进行分组的HashMap集合
5. timeTriggers，具有下次触发时间的trigger集合
6. triggers，trigger集合
7. lock，操作RAMJobStore时需要获取的锁
8. pausedTriggerGroups，被暂停的trigger分组名的HashSet集合
9. pausedJobGroups，被暂停的job分组名的HashSet集合
10. blockedJobs，被锁住的任务key的HashSet集合
11. misfireThreshold，失火阈值
12. signaler，SchedulerSignaler信号器

acquireNextTriggers方法

总体来说，quartzSchedulerThread有设置idlewaitime时间。idlewaitime时间就是在这个空闲时间内如果没有接收到调度器发生变化的信号(sigLock锁的notify)，它就会阻塞对应的时间(sigLock的wait(idlewaitime)方法)。因此quartzSchedulerThread需要提前获取到这个（now + idlewaitime）时间点内的trigger，否则会造成触发器失火的情况，并且它会一次性获取当前可用空闲线程个数的trigger。

acquireNextTriggers方法逻辑：

1. 先获取到lock锁，防止此时的job和trigger发生变化。
2. 如果timeTriggers的集合为空，那么直接返回。
3. while循环获取triggerWrapper，从timetriggers中获取第一个triggerWrapper。因为timeTriggers是有序的triggerWrapper集合（按照触发时间和优先级排序），接着从timeTriggers中移除triggerWrapper。
4. 如果trigger的下一次触发时间为空，则重新获取triggerWrapper。
5. 如果trigger的下一次触发时间大于需要获取的时间点，则跳出循环（因为timeTrigger是有序的，第一个时间都不满足了，就不用再继续循环了），然后返回获取到的trigger集合。
6. 接着判断该trigger是否失火，失火的条件为：当前时间小于这个（now-misfireTime）的时间点 并且失火策略为不能忽略（Trigger.MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY）。
   a. 通知triggerListener该trigger发生失火
   b. 接着调用trigger的updateAfterMisfire方法
   c. 最后进行下一次触发时间的判断，如果下一次触发时间为空，那么就说明trigger已经完成，从timeTriggers中移除自己
7. 如果处理后的trigger的下一次执行时间跟之前的执行时间不一样的话，则需要重新加入timeTrigger计算触发顺序。
8. 判断该trigger对应的job是否禁止并发执行，如果禁止，则需要将该job对应的其它trigger暂时从timeTriggers中移除。
9. 设置triggerWrapper的状态为ACQUIRED，设置本次触发的triggerInstanceId。
10. 如果获取到的trigger数量等于需要获取的数量，则可以跳出循环。
11. 结束循环后需要判断是否有被临时移除的trigger，有的话，需要将triggerWrapper放回timeTriggers。

releaseAcquiredTrigger方法

在JobStore中调用triggerWrapper相关的方法时（如triggersFired方法），如果发生了异常，则会调用该方法，将其重新放入timeTriggers中。

releaseAcquiredTrigger方法逻辑：

1. 先获取到lock锁。
2. 判断triggerWrapper的状态是否为已获取状态。
3. 如果是已获取状态，则将其重新加入到timeTriggers中。

triggersFired方法

1. 先获取到lock锁。
2. 循环进行处理firedTriggers集合，每次从firedTriggers集合获取triggerWrapper。
3. 判断triggerWrapper的状态是否为ACQUIRED状态，不是的话则continue。
4. 调用trigger对应的triggered方法，更新trigger内部的属性。
   a. 计算下一次执行时间
   b. 已执行的次数
   c. 存储上一次执行时间
5. 设置triggerWrapper的状态为WAITING（之前是ACQUIRED状态）
6. 判断对应的job是否允许并发执行
   a. 如果不允许的话，则将该job对应的所有triggerWrapper改为加锁状态(BLOCKED)
   b. 如果允许的话，判断下一次执行时间是否为null，不为null的话，重新加入timeTriggers中。
7. 返回触发器触发执行结果（TriggerFiredResult）。

triggeredJobComplete方法

正常情况下，Job被执行完毕的时候，会通知JobStore执行该方法。

1. 先获取到lock锁
2. 根据isPersistJobDataAfterExecution判断需要是否持久化存储jobDataMap
3. 根据isConcurrentExectionDisallowed判断job是否不允许并发执行
   a. 如果不允许并发执行，需要从blockJobs中移除该job，解除对应trigger的锁定状态。如果解除状态后的trigger状态为WAITING状态，则将trigger放入到timeTriggers结合中。
4. 根据完成策略CompletedExecutionInstruction对triggerWrapper进行处理
   a. 如果CompletedExecutionInstruction默认值是NOOP，什么都不做。
   b. 如果值是DELETE_TRIGGER，则需要删除对应的trigger。
   c. ...

博主微信公众号