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