一、Quartz 的幾個概念類

這幾個概念類，是我們呼叫Quartz任務排程的基礎。瞭解清楚之後，我們再來看一下如何去啟動和關閉一個Quartz排程程式。

• 1、org.quartz.Job
  它是一個抽象介面，表示一個工作，也就是我們要執行的具體內容，他只定義了一個幾口方法：
  void execute(JobExecutionContext context)
  作用等同Spring的：
  org.springframework.scheduling.quartz.QuartzJobBean
• 2、org.quartz.JobDetail
  JobDetail表示一個具體的可執行的排程程式，Job是這個可執行程排程程式所要執行的內容，它包含了這個任務排程的方案和策略。
  他的實現類：
  org.quartz.impl.JobDetailImpl
  
  作用等同Spring：
  org.springframework.scheduling.quartz.MethodInvokingJobDetailFactoryBean
• 3、org.quartz.Trigger
  它是一個抽象介面，表示一個排程引數的配置，通過配置它，來告訴排程容器什麼時候去呼叫JobDetail。
  他的兩個實現類:
  org.quartz.impl.triggers.SimpleTriggerImpl
  org.quartz.impl.triggers.CronTriggerImpl
  等同於Spring的:
  org.springframework.scheduling.quartz.SimpleTriggerBean
  
  org.springframework.scheduling.quartz.CronTriggerBean
  前者只支援按照一定頻度呼叫任務，如每隔30分鐘執行一次。
  後者既支援按照一定頻度呼叫任務，又支援定時任務。
• 4、org.quartz.Scheduler
  代表一個排程容器，一個排程容器中可以註冊多個JobDetail和Trigger。當Trigger與JobDetail組合，就可以被Scheduler容器排程了。它的方法有start()、shutdown()等方法，負責管理整個排程作業。
  等同Spring的： org.springframework.scheduling.quartz.SchedulerFactoryBean

二、Quartz 入門示例

import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.SchedulerFactory;
import org.quartz.SimpleScheduleBuilder;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.impl.StdSchedulerFactory;

public class Quartz {

    public static void main(String[] args) {
        // 1、建立JobDetial物件 , 並且設定選項
        JobDetail jobDetail= JobBuilder.newJob(MyJob.class).withIdentity("testJob_1","group_1").build();

        // 2、通過 TriggerBuilder 建立Trigger物件
        TriggerBuilder<Trigger> triggerBuilder = TriggerBuilder.newTrigger();
        triggerBuilder.withIdentity("trigger_1", "group_1");
        triggerBuilder.startNow();
        // 設定重複次數和間隔時間
        triggerBuilder.withSchedule(SimpleScheduleBuilder.simpleSchedule()
                .withIntervalInMilliseconds(1) //時間間隔
                .withRepeatCount(5) // 重複次數
         );
        // 設定停止時間
        //triggerBuilder.endAt(new Date(System.currentTimeMillis() + 3));
        // 建立Trigger物件
        Trigger trigger = triggerBuilder.build();

        // 3、建立Scheduler物件，並配置JobDetail和Trigger物件
        SchedulerFactory sf = new StdSchedulerFactory();
        try {
            Scheduler scheduler = sf.getScheduler();
            scheduler.scheduleJob(jobDetail, trigger);
            // 4、並執行啟動、關閉等操作
            scheduler.start();

          //關閉排程器
          //scheduler.shutdown(true); 
        } catch (SchedulerException e) {
            e.printStackTrace(); 
        }
    }
}

import java.util.Date;

import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

public class MyJob implements Job{

    public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
        System.out.println(new Date() + ": doing something...");
    }
}

執行結果：

Wed Aug 03 19:12:53 CST 2016: doing something…
Wed Aug 03 19:12:53 CST 2016: doing something…
Wed Aug 03 19:12:53 CST 2016: doing something…
Wed Aug 03 19:12:53 CST 2016: doing something…
Wed Aug 03 19:12:53 CST 2016: doing something…
Wed Aug 03 19:12:53 CST 2016: doing something…

當把結束時間改為：

// 設定停止時間
triggerBuilder.endAt(new Date(System.currentTimeMillis() + 3));

執行結果：

Wed Aug 03 19:17:12 CST 2016: doing something…
Wed Aug 03 19:17:12 CST 2016: doing something…
Wed Aug 03 19:17:12 CST 2016: doing something…

結果分析：
可以看出，設定重複執行5次，沒設定停止時間的時候，執行了6次，當停止時間設定為new Date(System.currentTimeMillis() + 3時， 果只執行了3次。 也就是說當到了停止時間，不管執行的次數是否到達最大次數，都不在執行了。

當添加了關閉排程器：

//關閉排程器
scheduler.shutdown(true); 

執行結果：

Wed Aug 03 19:17:12 CST 2016: doing something…

三、Quartz 簡單總結

• 1、scheduler.shutdown(true); // true表示等到本次執行結束
• 2、為了配置強大時間排程策略，可以研究專門的CronTrigger。
• 3、整合Spring時，Quartz容器關閉方式有兩種方法，一種是關閉- Spring容器，一種是獲取到SchedulerFactoryBean例項，然後呼叫一個shutdown就搞定了。
• 4、Quartz的JobDetail、Trigger都可以在執行時重新設定，並且在下次呼叫時候起作用。這就為動態作業的實現提供了依據。你可以將排程時間策略存放到資料庫，然後通過資料庫資料來設定Trigger，這樣就能產生動態的排程。
• 5、JobDetail不儲存具體的例項，但它允許你定義一個例項，JobDetail 又指向JobDataMap。 JobDetail持有Job的詳細資訊，如它所屬的組，名稱等資訊
• 6、JobDataMap儲存著任務例項的物件，並保持著他們狀態資訊，它是Ｍap介面的實現，即你可以往裡面put和get一些你想儲存和獲取的資訊．
• 7、scheduler容器包含多個JobDetail和Trigger。scheduler是個容器，容器中有一個執行緒池，用來並行排程執行每個作業，這樣可以提高容器效率。

四、Quartz 執行過程

quartz執行時由QuartzSchedulerThread類作為主體，迴圈執行排程流程。JobStore作為中間層，按照quartz的併發策略執行資料庫操作，完成主要的排程邏輯。JobRunShellFactory負責例項化JobDetail物件，將其放入執行緒池執行。LockHandler負責獲取LOCKS表中的資料庫鎖。
整個quartz對任務排程的時序大致如下:

0.排程器執行緒run()

1.獲取待觸發trigger
1.1讀取JobDetail資訊
1.2讀取trigger表中觸發器資訊並標記為”已獲取”

2.觸發trigger
2.1確認trigger的狀態
2.2讀取trigger的JobDetail資訊
2.3讀取trigger的Calendar資訊
2.4更新trigger資訊

3例項化並執行Job
3.1從執行緒池獲取執行緒執行JobRunShell的run方法

我們看到初始化一個排程器需要用工廠類獲取例項:

SchedulerFactory sf = new StdSchedulerFactory();
Scheduler sch = sf.getScheduler(); 
// 然後啟動:
sch.start();

下面跟進StdSchedulerFactory的getScheduler()方法:

public Scheduler getScheduler() throws SchedulerException {
        if (cfg == null) {
            initialize();
        }
        SchedulerRepository schedRep = SchedulerRepository.getInstance();
        //從"排程器倉庫"中根據properties的SchedulerName配置獲取一個排程器例項
        Scheduler sched = schedRep.lookup(getSchedulerName());
        if (sched != null) {
            if (sched.isShutdown()) {
                schedRep.remove(getSchedulerName());
            } else {
                return sched;
            }
        }
        //初始化排程器
        sched = instantiate();
        return sched;
    }

我們再看一下 sched = instantiate(); 排程器的初始化方法
- 讀取配置資源,
- 生成QuartzScheduler物件,
- 建立該物件的執行執行緒,並啟動執行緒;
- 初始化JobStore,QuartzScheduler,DBConnectionManager等重要元件,
- 至此,排程器的初始化工作已完成,初始化工作中quratz讀取了資料庫中存放的對應當前排程器的鎖資訊,對應CRM中的表QRTZ2_LOCKS,中的STATE_ACCESS,TRIGGER_ACCESS兩個LOCK_NAME.

public void initialize(ClassLoadHelper loadHelper,
            SchedulerSignaler signaler) throws SchedulerConfigException {
        if (dsName == null) {
            throw new SchedulerConfigException("DataSource name not set.");
        }
        classLoadHelper = loadHelper;
        if(isThreadsInheritInitializersClassLoadContext()) {
            log.info("JDBCJobStore threads will inherit ContextClassLoader of thread: " + Thread.currentThread().getName());
            initializersLoader = Thread.currentThread().getContextClassLoader();
        }

        this.schedSignaler = signaler;
        // If the user hasn't specified an explicit lock handler, then
        // choose one based on CMT/Clustered/UseDBLocks.
        if (getLockHandler() == null) {

            // If the user hasn't specified an explicit lock handler,
            // then we *must* use DB locks with clustering
            if (isClustered()) {
                setUseDBLocks(true);
            }

            if (getUseDBLocks()) {
                if(getDriverDelegateClass() != null && getDriverDelegateClass().equals(MSSQLDelegate.class.getName())) {
                    if(getSelectWithLockSQL() == null) {
                        //讀取資料庫LOCKS表中對應當前排程器的鎖資訊
                        String msSqlDflt = "SELECT * FROM {0}LOCKS WITH (UPDLOCK,ROWLOCK) WHERE " + COL_SCHEDULER_NAME + " = {1} AND LOCK_NAME = ?";
                        getLog().info("Detected usage of MSSQLDelegate class - defaulting 'selectWithLockSQL' to '" + msSqlDflt + "'.");
                        setSelectWithLockSQL(msSqlDflt);
                    }
                }
                getLog().info("Using db table-based data access locking (synchronization).");
                setLockHandler(new StdRowLockSemaphore(getTablePrefix(), getInstanceName(), getSelectWithLockSQL()));
            } else {
                getLog().info(
                    "Using thread monitor-based data access locking (synchronization).");
                setLockHandler(new SimpleSemaphore());
            }
        }
    }

當呼叫sch.start();方法時, scheduler做了如下工作:
1.通知listener開始啟動
2.啟動排程器執行緒
3.啟動plugin
4.通知listener啟動完成

public void start() throws SchedulerException {
        if (shuttingDown|| closed) {
            throw new SchedulerException(
                    "The Scheduler cannot be restarted after shutdown() has been called.");
        }
        // QTZ-212 : calling new schedulerStarting() method on the listeners
        // right after entering start()
        //通知該排程器的listener啟動開始
        notifySchedulerListenersStarting();
        if (initialStart == null) {
            initialStart = new Date();
            //啟動排程器的執行緒
            this.resources.getJobStore().schedulerStarted();            
            //啟動plugins
            startPlugins();
        } else {
            resources.getJobStore().schedulerResumed();
        }
        schedThread.togglePause(false);
        getLog().info(
                "Scheduler " + resources.getUniqueIdentifier() + " started.");
        //通知該排程器的listener啟動完成
        notifySchedulerListenersStarted();
    }

排程過程

排程器啟動後,排程器的執行緒就處於執行狀態了,開始執行quartz的主要工作–排程任務. 前面已介紹過,任務的排程過程大致分為三步:
1.獲取待觸發trigger
2.觸發trigger
3.例項化並執行Job
而排程過程的三個步驟就承載在QuartzSchedulerThread這個排程器執行緒類的run()方法中。程式碼大家可以看一下原始碼。我在這裡就略過性的支出run()方法中對應上面三個步驟的原始碼。

觸發器的獲取，run()方法中獲取待觸發trigger
呼叫JobStoreSupport.acquireNextTriggers方法:

 //排程器在trigger佇列中尋找30秒內一定數目的trigger準備執行排程,
                        //引數1:nolaterthan = now+3000ms,引數2 最大獲取數量,大小取執行緒池執行緒剩餘量與定義值得較小者
                        //引數3 時間視窗 預設為0,程式會在nolaterthan後加上視窗大小來選擇trigger
                        triggers = qsRsrcs.getJobStore().acquireNextTriggers(
                                now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());

引數的意義如下:
引數1:nolaterthan = now+3000ms,即未來30s內將會被觸發.
引數2 最大獲取數量,大小取執行緒池執行緒剩餘量與定義值得較小者.
引數3 時間視窗 預設為0,程式會在nolaterthan後加上視窗大小來選擇

觸發trigger: QuartzSchedulerThread.run()
呼叫JobStoreSupport.triggersFired方法:

List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);

該方法做了以下工作:
1.獲取trigger當前狀態
2.通過trigger中的JobKey讀取trigger包含的Job資訊
3.將trigger更新至觸發狀態
4.結合calendar的資訊觸發trigger,涉及多次狀態更新
5.更新資料庫中trigger的資訊,包括更改狀態至STATE_COMPLETE,及計算下一次觸發時間.
6.返回trigger觸發結果的資料傳輸類TriggerFiredBundle

Job執行過程:QuartzSchedulerThread.run()

qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
shell.initialize(qs);

為每個Job生成一個可執行的RunShell,並放入執行緒池執行.
在最後排程執行緒生成了一個隨機的等待時間,進入短暫的等待,這使得其他節點的排程器都有機會獲取資料庫資源.如此就實現了quratz的負載平衡.
這樣一次完整的排程過程就結束了.排程器執行緒進入下一次迴圈.