Quartz應用與集群原理分析

ymc4 9年前發布 | 34K 次閱讀 Quartz 作業調度框架

為什么選擇Quartz

Quartz是Java領域最著名的開源任務調度工具。Quartz提供了極為廣泛的特性如持久化任務，集群和分布式任務等，其特點如下：

完全由Java寫成，方便集成(Spring)
伸縮性
負載均衡
高可用性

任務調度其他方案

1.Crontab+SQL

每天晚上運行定時任務，通過SQL腳本+crontab方式執行，例如，

0 2 * * * /xxx/mtcrm/shell/mtcrm_daily_stat.sql    //每天凌晨2:00執行統計
30 7 * * * /xxx/mtcrm/shell/mtcrm_data_fix.sql     //每天早上7:30執行數據修復

該方案存在以下問題：

直接訪問數據庫，各系統業務接口沒有重用。完成復雜業務需求時，會引入過多中間表。業務邏輯計算完全依賴SQL，增大數據庫壓力。任務失敗無法自動恢復。

2.Python+SQL

采用python腳本(多數據源)+SQL方式執行，例如，

def connectCRM():
  return MySQLdb.Connection("host1", "uname", "xxx", "crm", 3306, charset="utf8")

 def connectTemp():
  return MySQLdb.Connection("host1", "uname", "xxx", "temp", 3306, charset="utf8")

該方案存在問題：

直接訪問數據，需要理解各系統的數據結構，無法滿足動態任務問題，各系統業務接口沒有重用。無負載均衡。任務失敗無法恢復。在JAVA語言開發中出現異構，且很難統一到自動部署系統中。

3.Spring+JDK Timer

該方案使用spring+JDK Timer方式，調用接口完成定時任務，在分布式部署環境下，防止多個節點同時運行任務，需要寫死host，控制在一臺服務器上執行task。

<bean id="accountStatusTaskScanner"  class="xxx.crm.service.impl.AccountStatusTaskScanner" />
    <task:scheduler id="taskScheduler" pool-size="5" />
    <task:scheduled-tasks scheduler="taskScheduler">
    <task:scheduled ref="accountStatusTaskScanner" method="execute" cron="0 0 1 * * ?" />
</task:scheduled-tasks>

該方案較方案1，2有很大改進，但仍存在以下問題：

步驟復雜、分散，任務量增大的情況下，很難擴展使用寫死服務器Host的方式執行task，存在單點風險，負載均衡手動完成。應用重啟，任務無法自動恢復。定時任務走過了很多彎路：定時任務多種實現方式，使配置和代碼分散在多處，難以維護和監控；任務執行過程沒有保證，沒有錯誤恢復；任務執行異常沒有反饋 (郵件)；沒有集群支持、負載均衡。CRM系統需要分布式的任務調度框架，統一解決問題，Java可以使用的任務調度框架有 Quartz，Jcrontab，cron4j，擇優選擇Quartz。

Quartz集群部署實踐

Quartz與Spring結合使用，Spring通過提供org.springframework.scheduling.quartz下的封裝類對Quartz支持。

Quartz集群部署：

Quartz應用與集群原理分析

Quartz集群中的每個節點是一個獨立的Quartz應用，它又管理著其他的節點。該集群需要分別對每個節點分別啟動或停止，不像應用服務器的集群，獨立的Quartz節點并不與另一個節點或是管理節點通信。Quartz應用是通過數據庫表來感知到另一應用。只有使用持久的JobStore才能完成Quqrtz集群。

基于Spring的集群配置：

<!-- 調度工廠 -->
<bean id="quartzScheduler"
    class="org.springframework.scheduling.quartz.SchedulerFactoryBean">
    <property name="dataSource" ref="dataSource" />
    <property name="quartzProperties">
        <props>
            <prop key="org.quartz.scheduler.instanceName">CRMscheduler</prop>
            <prop key="org.quartz.scheduler.instanceId">AUTO</prop>
            <!-- 線程池配置 -->
            <prop key="org.quartz.threadPool.class">org.quartz.simpl.SimpleThreadPool</prop>
            <prop key="org.quartz.threadPool.threadCount">20</prop>
            <prop key="org.quartz.threadPool.threadPriority">5</prop>
            <!-- JobStore 配置 -->
            <prop key="org.quartz.jobStore.class">org.quartz.impl.jdbcjobstore.JobStoreTX</prop>

            <!-- 集群配置 -->
            <prop key="org.quartz.jobStore.isClustered">true</prop>
            <prop key="org.quartz.jobStore.clusterCheckinInterval">15000</prop>
            <prop key="org.quartz.jobStore.maxMisfiresToHandleAtATime">1</prop>

            <prop key="org.quartz.jobStore.misfireThreshold">120000</prop>

            <prop key="org.quartz.jobStore.tablePrefix">QRTZ_</prop>
        </props>

    </property>

    <property name="schedulerName" value="CRMscheduler" />

    <!--必須的，QuartzScheduler 延時啟動，應用啟動完后 QuartzScheduler 再啟動 -->
    <property name="startupDelay" value="30" />

    <property name="applicationContextSchedulerContextKey" value="applicationContextKey" />

    <!--可選，QuartzScheduler 啟動時更新己存在的Job，這樣就不用每次修改targetObject后刪除qrtz_job_details表對應記錄了 -->
    <property name="overwriteExistingJobs" value="true" />

    <!-- 設置自動啟動 -->
    <property name="autoStartup" value="true" />

    <!-- 注冊觸發器 -->
    <property name="triggers">
        <list>
            <ref bean="userSyncScannerTrigger" />
                     ......
        </list>
    </property>

    <!-- 注冊jobDetail -->
    <property name="jobDetails">
        <list>
        </list>
    </property>

    <property name="schedulerListeners">
        <list>
            <ref bean="quartzExceptionSchedulerListener" />
        </list>
    </property>
</bean>

org.quartz.jobStore.class屬性為JobStoreTX，將任務持久化到數據中。因為集群中節點依賴于數據庫來傳播Scheduler實例的狀態，你只能在使用JDBC JobStore時應用Quartz集群。

org.quartz.jobStore.isClustered屬性為true，通知Scheduler實例要它參與到一個集群當中。

org.quartz.jobStore.clusterCheckinInterval屬性定義了Scheduler實例檢入到數據庫中的頻率(單位：毫秒)。Scheduler檢查是否其他的實例到了它們應當檢入的時候未檢入；這能指出一個失敗的Scheduler實例，且當前 Scheduler會以此來接管任何執行失敗并可恢復的Job。通過檢入操作，Scheduler 也會更新自身的狀態記錄。clusterChedkinInterval越小，Scheduler節點檢查失敗的Scheduler實例就越頻繁。默認值是 15000 (即15 秒)。

一個簡單實例

https://github.com/hunng/spring-quartz

Quartz監控

Triggers監控：

Quartz應用與集群原理分析

JobDetails監控：

Quartz應用與集群原理分析

Quartz集群原理分析

Quartz集群數據庫表

Quartz的集群部署方案在架構上是分布式的，沒有負責集中管理的節點，而是利用數據庫鎖的方式來實現集群環境下進行并發控制。BTW，分布式部署時需要保證各個節點的系統時間一致。

Quartz數據庫核心表如下：

Table Name	Description
QRTZ_CALENDARS	存儲Quartz的Calendar信息
QRTZCRONTRIGGERS	存儲CronTrigger，包括Cron表達式和時區信息
QRTZFIREDTRIGGERS	存儲與已觸發的Trigger相關的狀態信息，以及相聯Job的執行信息
QRTZPAUSEDTRIGGER_GRPS	存儲已暫停的Trigger組的信息
QRTZSCHEDULERSTATE	存儲少量的有關Scheduler的狀態信息，和別的Scheduler實例
QRTZ_LOCKS	存儲程序的悲觀鎖的信息
QRTZJOBDETAILS	存儲每一個已配置的Job的詳細信息
QRTZJOBLISTENERS	存儲有關已配置的JobListener的信息
QRTZSIMPLETRIGGERS	存儲簡單的Trigger，包括重復次數、間隔、以及已觸的次數
QRTZBLOGTRIGGERS	Trigger作為Blob類型存儲
QRTZTRIGGERLISTENERS	存儲已配置的TriggerListener的信息
QRTZ_TRIGGERS	存儲已配置的Trigger的信息

其中，QRTZ_LOCKS就是Quartz集群實現同步機制的行鎖表，其表結構如下：

--QRTZ_LOCKS表結構
CREATE TABLE `QRTZ_LOCKS` (
  `LOCK_NAME` varchar(40) NOT NULL,
   PRIMARY KEY (`LOCK_NAME`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8;

--QRTZ_LOCKS記錄
+-----------------+ 
| LOCK_NAME       |
+-----------------+ 
| CALENDAR_ACCESS |
| JOB_ACCESS      |
| MISFIRE_ACCESS  |
| STATE_ACCESS    |
| TRIGGER_ACCESS  |
+-----------------+

可以看出QRTZ_LOCKS中有5條記錄，代表5把鎖，分別用于實現多個Quartz Node對Job、Trigger、Calendar訪問的同步控制。

Quartz線程模型

Quartz 調度線程流程圖：

Quartz應用與集群原理分析

在Quartz中有兩類線程：Scheduler調度線程和任務執行線程。任務執行線程：Quartz不會在主線程 (QuartzSchedulerThread)中處理用戶的Job。Quartz把線程管理的職責委托給ThreadPool，一般的設置使用 SimpleThreadPool。SimpleThreadPool創建了一定數量的WorkerThread實例來使得Job能夠在線程中進行處理。 WorkerThread是定義在SimpleThreadPool類中的內部類，它實質上就是一個線程。

<!-- 線程池配置 -->
<prop key="org.quartz.threadPool.class">org.quartz.simpl.SimpleThreadPool</prop>
<prop key="org.quartz.threadPool.threadCount">20</prop>
<prop key="org.quartz.threadPool.threadPriority">5</prop>

QuartzSchedulerThread調度主線程：QuartzScheduler被創建時創建一個QuartzSchedulerThread實例。

集群源碼分析

Quartz究竟是如何保證集群情況下trgger處理的信息同步？

下面跟著源碼一步一步分析，QuartzSchedulerThread包含有決定何時下一個Job將被觸發的處理循環，主要邏輯在其run()方法中：

public void run() {
   boolean lastAcquireFailed = false;
   while (!halted.get()) {
     ......

     int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
     if(availThreadCount > 0) { 

     ......

     //調度器在trigger隊列中尋找30秒內一定數目的trigger(需要保證集群節點的系統時間一致)
     triggers = qsRsrcs.getJobStore().acquireNextTriggers(
                            now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());

     ......

     //觸發trigger
     List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);

     ......

     //釋放trigger
     for (int i = 0; i < triggers.size(); i++) {
         qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
     }

   }                
}

由此可知，QuartzScheduler調度線程不斷獲取trigger，觸發trigger，釋放trigger。下面分析trigger的獲取過程，qsRsrcs.getJobStore()返回對象是JobStore，集群環境配置如下：

<!-- JobStore 配置 -->
<prop key="org.quartz.jobStore.class">org.quartz.impl.jdbcjobstore.JobStoreTX</prop>

JobStoreTX繼承自JobStoreSupport，而JobStoreSupport的acquireNextTriggers、 triggersFired、releaseAcquiredTrigger方法負責具體trigger相關操作，都必須獲得 TRIGGER_ACCESS鎖。核心邏輯在executeInNonManagedTXLock方法中：

protected <T> T executeInNonManagedTXLock(
        String lockName, 
        TransactionCallback<T> txCallback, final TransactionValidator<T> txValidator) throws JobPersistenceException {
    boolean transOwner = false;
    Connection conn = null;
    try {
        if (lockName != null) {
            if (getLockHandler().requiresConnection()) {
                conn = getNonManagedTXConnection();
            }

            //獲取鎖
            transOwner = getLockHandler().obtainLock(conn, lockName);
        }

        if (conn == null) {
            conn = getNonManagedTXConnection();
        }

        final T result = txCallback.execute(conn);
        try {
            commitConnection(conn);
        } catch (JobPersistenceException e) {
            rollbackConnection(conn);
            if (txValidator == null || !retryExecuteInNonManagedTXLock(lockName, new TransactionCallback<Boolean>() {
                @Override
                public Boolean execute(Connection conn) throws JobPersistenceException {
                    return txValidator.validate(conn, result);
                }
            })) {
                throw e;
            }
        }

        Long sigTime = clearAndGetSignalSchedulingChangeOnTxCompletion();
        if(sigTime != null && sigTime >= 0) {
            signalSchedulingChangeImmediately(sigTime);
        }

        return result;
    } catch (JobPersistenceException e) {
        rollbackConnection(conn);
        throw e;
    } catch (RuntimeException e) {
        rollbackConnection(conn);
        throw new JobPersistenceException("Unexpected runtime exception: "
                + e.getMessage(), e);
    } finally {
        try {
            releaseLock(lockName, transOwner);      //釋放鎖
        } finally {
            cleanupConnection(conn);
        }
    }
}

由上代碼可知Quartz集群基于數據庫鎖的同步操作流程如下圖所示：

Quartz應用與集群原理分析

一個調度器實例在執行涉及到分布式問題的數據庫操作前，首先要獲取QUARTZ_LOCKS表中對應的行級鎖，獲取鎖后即可執行其他表中的數據庫操作，隨著操作事務的提交，行級鎖被釋放，供其他調度實例獲取。集群中的每一個調度器實例都遵循這樣一種嚴格的操作規程。

getLockHandler()方法返回的對象類型是Semaphore，獲取鎖和釋放鎖的具體邏輯由該對象維護

public interface Semaphore {

     boolean obtainLock(Connection conn, String lockName) throws LockException;

     void releaseLock(String lockName) throws LockException;

     boolean requiresConnection();
}

該接口的實現類完成具體操作鎖的邏輯，在JobStoreSupport的初始化方法中注入的Semaphore具體類型是StdRowLockSemaphore

setLockHandler(new StdRowLockSemaphore(getTablePrefix(), getInstanceName(), getSelectWithLockSQL()));

StdRowLockSemaphore的源碼如下所示：

public class StdRowLockSemaphore extends DBSemaphore {
//鎖定SQL語句
public static final String SELECT_FOR_LOCK = "SELECT * FROM "
        + TABLE_PREFIX_SUBST + TABLE_LOCKS + " WHERE " + COL_LOCK_NAME
        + " = ? FOR UPDATE";

public static final String INSERT_LOCK = "INSERT INTO " + TABLE_PREFIX_SUBST 
        + TABLE_LOCKS + "(" + COL_SCHEDULER_NAME + ", " 
        + COL_LOCK_NAME + ") VALUES (" + SCHED_NAME_SUBST + ", ?)"; 

//指定鎖定SQL
protected void executeSQL(Connection conn, String lockName, String expandedSQL) throws LockException {
    PreparedStatement ps = null;
    ResultSet rs = null;
    try {
        ps = conn.prepareStatement(expandedSQL);
        ps.setString(1, lockName);
        ......
        rs = ps.executeQuery();
        if (!rs.next()) {
            throw new SQLException(Util.rtp(
                "No row exists in table " + TABLE_PREFIX_SUBST +
                TABLE_LOCKS + " for lock named: " + lockName, getTablePrefix()));
        }
    } catch (SQLException sqle) {

    } finally {
      ...... //release resources
    }
  }
}

//獲取QRTZ_LOCKS行級鎖
public boolean obtainLock(Connection conn, String lockName) throws LockException {
    lockName = lockName.intern();

    if (!isLockOwner(conn, lockName)) {
        executeSQL(conn, lockName, expandedSQL);

        getThreadLocks().add(lockName);
    }
    return true;
}

//釋放QRTZ_LOCKS行級鎖
public void releaseLock(Connection conn, String lockName) {
    lockName = lockName.intern();

    if (isLockOwner(conn, lockName)) {
        getThreadLocks().remove(lockName);
    }
    ......
}

小結

至此，總結一下Quartz集群同步機制：每當要進行與某種業務相關的數據庫操作時，先去QRTZLOCKS表中查詢操作相關的業務對象所需要的鎖，在select語句之后加for update來實現。例如，TRIGGERACCESS表示對任務觸發器相關的信息進行修改、刪除操作時所需要獲得的鎖。這時，執行查詢這個表數據的SQL形如：

select * from QRTZ_LOCKS t where t.lock_name='TRIGGER_ACCESS' for update

當一個線程使用上述的SQL對表中的數據執行查詢操作時，若查詢結果中包含相關的行，數據庫就對該行進行ROW LOCK；若此時，另外一個線程使用相同的SQL對表的數據進行查詢，由于查詢出的數據行已經被數據庫鎖住了，此時這個線程就只能等待，直到擁有該行鎖的線程完成了相關的業務操作，執行了commit動作后，數據庫才會釋放了相關行的鎖，這個線程才能繼續執行。

通過這樣的機制，在集群環境下，結合悲觀鎖的機制就可以防止一個線程對數據庫數據的操作的結果被另外一個線程所覆蓋，從而可以避免一些難以覺察的錯誤發生。當然，達到這種效果的前提是需要把Connection設置為手動提交，即autoCommit為false。

參考資料

本文由用戶 ymc4 自行上傳分享，僅供網友學習交流。所有權歸原作者，若您的權利被侵害，請聯系管理員。

轉載本站原創文章，請注明出處，并保留原始鏈接、圖片水印。

本站是一個以用戶分享為主的開源技術平臺，歡迎各類分享！

本文地址：http://www.baiduhome.net/lib/view/open1423663712389.html

Quartz 作業調度框架

Quartz應用與集群原理分析

為什么選擇Quartz

任務調度其他方案

Quartz集群部署實踐

Quartz集群部署：

一個簡單實例

Quartz監控

Quartz集群原理分析

Quartz集群數據庫表

Quartz線程模型

集群源碼分析

小結

參考資料

相關經驗

相關資訊

相關文檔

目錄