1.scheduler
1. Scheduler就是Quartz的大脑,所有任务都是由它来设施。
Scheduler包含一个两个重要组件: JobStore和ThreadPool。
JobStore是会来存储运行时信息的,包括Job、JobDetail、Trigger以及业务锁等。它有多种实现RAMJob(内存实现),JobStoreTX(JDBC,事务由Quartz管理),JobStoreCMT(JDBC,使用容器事务),ClusteredJobStore(集群实现)。
ThreadPool就是线程池,Quartz有自己的线程池实现。所有任务的都会由线程池执行。
2.SchdulerFactory,顾名思义就是来用创建Schduler了,有两个实现:DirectSchedulerFactory和 StdSchdulerFactory。前者可以用来在代码里定制你自己的Schduler参数。后者是直接读取classpath下的quartz.properties(不存在就都使用默认值)配置来实例化Schduler。通常来讲,我们使用StdSchdulerFactory也就足够了。
org.quartz.scheduler.instanceName = DefaultQuartzScheduler 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.class = org.quartz.simpl.RAMJobStore
2.Trigger
2.1 simpleTrigger
SimpleTrigger可以满足的调度需求是:在具体的时间点执行一次,或者在具体的时间点执行,并且以指定的间隔重复执行若干次。比如,你有一个trigger,你可以设置它在2018年12月9日的上午11:23:54准时触发,或者在这个时间点触发,并且每隔2秒触发一次,一共重复5次。
public static void main(String[] args) throws SchedulerException { // 创建Scheduler SchedulerFactory sf = new StdSchedulerFactory(); Scheduler scheduler = sf.getScheduler(); // 需求:我要在5秒之后执行任务,时间间隔为2秒,最多执行100次 long currentTime = System.currentTimeMillis(); long delayTime = currentTime + 5 * 1000l; // 5秒之后执行任务 // 设置Trigger(不再使用静态方法) Trigger trigger = TriggerBuilder.newTrigger() // 使用TriggerBuilder创建Trigger .withIdentity("trigger1", "group1") .startAt(new Date(delayTime)) .withSchedule(SimpleScheduleBuilder .simpleSchedule() // 使用SimpleScheduleBuilder创建simpleSchedule .withIntervalInSeconds(2) // 时间间隔为2秒 .withRepeatCount(99)) // 最多执行100次,此处需要注意,不包括第一次执行的 .build(); // 设置JobDetail(不再使用静态方法) JobDetail jobDetail = JobBuilder.newJob((MyJobDetail.class)) // 使用JobBuilder创建JobDetail .withIdentity("jobDetail1", "group1") .usingJobData("user", "AlanShelby") .build(); // 设置scheduler scheduler.scheduleJob(jobDetail, trigger); // 启动、停止Scheduler scheduler.start(); try { Thread.sleep(500000); } catch (InterruptedException e) { e.printStackTrace(); } scheduler.shutdown(); }
2.2 cronTrigger
1、适用于更为复杂的需求,它类似于Linux系统中的crontab,但要比crontab更强大。它基本上覆盖了之前章节讲到的三个类型的功能(并不是全部功能),相对于前三个类型,cronTrigger也更难理解。
2、它适合的任务类似于:每天0:00,9:00,18:00各执行一次。
3、它的属性只有一个Cron表达式,下面有对cron表达式详细的讲解。
Trigger trigger = TriggerBuilder.newTrigger() .withIdentity("trigger1", "group1") .withSchedule(CronScheduleBuilder.cronSchedule("*/5 * * * * ?")) // 每5秒钟执行一次 .build();
3.job
Job是有可能并发执行的,比如一个任务要执行10秒中,而调度算法是每秒中触发1次,那么就有可能多个任务被并发执行。
有时候我们并不想任务并发执行,比如这个任务要去”获得数据库中所有未发送邮件的名单”,如果是并发执行,就需要一个数据库锁去避免一个数据被多次处理。这个时候一个@DisallowConcurrentExecution解决这个问题。
@DisallowConcurrentExecution public class DoNothingJob implements Job { public void execute(JobExecutionContext context) throws JobExecutionException { System.out.println("do nothing"); } }
4.调度核心类QuartzSchedulerThread
/** * <p> * The main processing loop of the <code>QuartzSchedulerThread</code>. * </p> */ @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; }
4.1 blockForAvailableThreads
就是qsRsrcs.getThreadPool().blockForAvailableThreads(),如果线程池满了的话,则会阻塞,因而会影响调度的准确性。 int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads(); 获取可用的线程数量。通常在第一次时候这个数量等于配置中配置的参数: org.quartz.threadPool.threadCount = 20
4.2 maxBatchSize
triggers = qsRsrcs.getJobStore().acquireNextTriggers( now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());
这个参数的意思是批量查询的数量,但并不是你配置多少它每次就能查询多少,这算一个优化的配置项,因为在jdbc store的时候,减少对数据库的轮询次数算是一个比较大的优化;Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow())可以看到这里会取配置的批量的数和可用线程的最小数,所以批量数可以配置成和线程数大小一致:
org.quartz.threadPool.threadCount= 20 org.quartz.scheduler.batchTriggerAcquisitionMaxCount= 20