Quartz定时任务执行原理

深碍√TFBOYSˉ_ 2022-01-26 00:59 688阅读 0赞

## 1、原理分析 ##

在java中常见的定时调度方案有：ScheduledExecutorService和quartz两种方案。其本质上都是通过native的wait方法来实现的.

quartz定时调度是通过Object.wait方式（native方法）实现的，其本质是通过操作系统的时钟来实现的。Quartz主要的执行类和执行方法。  
其主要流程如下：

public class QuartzSchedulerThread extends Thread{
       public void run() {
            boolean lastAcquireFailed = false;
    
            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) {
                            }
                        }
    
                        if (halted.get()) {
                            break;
                        }
                    }
    
                    int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
                    if(availThreadCount > 0) { // will always be true, due to semantics of blockForAvailableThreads...
    
                        List<OperableTrigger> triggers = null;
    
                        long now = System.currentTimeMillis();
    
                        clearSignaledSchedulingChange();
                        try {
                            triggers = qsRsrcs.getJobStore().acquireNextTriggers(
                                    now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());
                            lastAcquireFailed = false;
                            if (log.isDebugEnabled()) 
                                log.debug("batch acquisition of " + (triggers == null ? 0 : triggers.size()) + " triggers");
                        } catch (JobPersistenceException jpe) {
                            if(!lastAcquireFailed) {
                                qs.notifySchedulerListenersError(
                                    "An error occurred while scanning for the next triggers to fire.",
                                    jpe);
                            }
                            lastAcquireFailed = true;
                            continue;
                        } catch (RuntimeException e) {
                            if(!lastAcquireFailed) {
                                getLog().error("quartzSchedulerThreadLoop: RuntimeException "
                                        +e.getMessage(), e);
                            }
                            lastAcquireFailed = true;
                            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;
        }
    
    }

*  获取最近需要执行的任务列表（30s内的将要执行的任务）。然后根据执行时间进行排序，然后计算出需要wait()的时间。当调度时间来临，欢迎主线程，将任务交给一个线程池进行执行。

triggers = qsRsrcs.getJobStore().acquireNextTriggers(
                                    now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());

*  30秒内没有需要执行的任务，则等待一个随机时间。getRandomizedIdleWaitTime产生一个30秒内随机等待时间。

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) {
                        }
                    }

## 2.quartz-db定时调度执行过程分析 ##

### 2.1 执行流程图 ###

在QuartzSchedulerThread主线程中，首先会从QRTZ\_TRIGGERS表中取出最近30秒内将要执行的任务，然后等待executeTime-now时间，然后在等待唤醒时交给线程池处理。当任务执行完成时，会通过事件机制，更新QRTZ\_TRIGGERS中的nextFireTime。在每次获取QRTZ\_TRIGGERS最近30秒的Trigger时，都会先对QRTZ\_LOCKS表中的Trigger行进行加锁，从而保证了一个任务只会在分布式环境中的一台机器上执行。  
![这里写图片描述][wKioOVGZjcayYIDlAACPrqg-Xsw581.jpg]

### 2.2 结合DB分析执行过程 ###

*  QRTZ\_TRIGGERS：用于存储接下来需要进行调度的triiger.

SELECT TRIGGER\_NAME, TRIGGER\_GROUP, NEXT\_FIRE\_TIME, PRIORITY FROM QRTZ\_TRIGGERS WHERE SCHED\_NAME = ‘quartzScheduler’ AND TRIGGER\_STATE = ‘WAITING’ AND NEXT\_FIRE\_TIME <= 1506322290680 AND (MISFIRE\_INSTR = -1 OR (MISFIRE\_INSTR != -1 AND NEXT\_FIRE\_TIME >= 1506322375502)) ORDER BY NEXT\_FIRE\_TIME ASC, PRIORITY DESC

例如因为机器重启被中断的trigger  
(该trigger是在机器启动时从QRTZ\_FIRED\_TRIGGER扫描到的记录，都会加入到QRTZ\_TRIGGERS等待进一步调用，其NEXT\_FIRE\_TIME为空，表示优先级最高)

mysql> select TRIGGER_NAME,TRIGGER_GROUP,from_unixtime(NEXT_FIRE_TIME/1000) from QRTZ_TRIGGERS;
    +----------------------------------------------------------------+-----------------+------------------------------------+
    | TRIGGER_NAME                                                   | TRIGGER_GROUP   | from_unixtime(NEXT_FIRE_TIME/1000) |
    +----------------------------------------------------------------+-----------------+------------------------------------+
    | recover_caowenyideMacBook-Pro.local1506316013536_1506322246054 | RECOVERING_JOBS | 2017-09-25 14:48:30.0070           |
    | sampleJobCronTrigger                                           | DEFAULT         | 2017-09-25 14:56:45.8750           |
    +----------------------------------------------------------------+-----------------+------------------------------------+

QRTZ\_FIRED\_TRIGGERS: 记录正在进行调度的TRIIGER  
INSERT INTO QRTZ\_FIRED\_TRIGGERS (SCHED\_NAME, ENTRY\_ID, TRIGGER\_NAME, TRIGGER\_GROUP, INSTANCE\_NAME, FIRED\_TIME, SCHED\_TIME, STATE, JOB\_NAME, JOB\_GROUP, IS\_NONCONCURRENT, REQUESTS\_RECOVERY, PRIORITY) VALUES(‘quartzScheduler’, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)

mysql> select TRIGGER_NAME,TRIGGER_GROUP,TRIGGER_STATE from QRTZ_TRIGGERS;
    +----------------------------------------------------------------+-----------------+---------------+
    | TRIGGER_NAME                                                   | TRIGGER_GROUP   | TRIGGER_STATE |
    +----------------------------------------------------------------+-----------------+---------------+
    | recover_caowenyideMacBook-Pro.local1506316013536_1506322246054 | RECOVERING_JOBS | ACQUIRED      |
    | sampleJobCronTrigger                                           | DEFAULT         | WAITING       |
    +----------------------------------------------------------------+-----------------+---------------+
    
    mysql> select TRIGGER_NAME,TRIGGER_GROUP from QRTZ_FIRED_TRIGGERS;
    +----------------------------------------------------------------+-----------------+
    | TRIGGER_NAME                                                   | TRIGGER_GROUP   |
    +----------------------------------------------------------------+-----------------+
    | recover_caowenyideMacBook-Pro.local1506316013536_1506322246054 | RECOVERING_JOBS |
    +----------------------------------------------------------------+-----------------+

SELECT \* FROM QRTZ\_LOCKS WHERE SCHED\_NAME = ‘quartzScheduler’ AND LOCK\_NAME = ? FOR UPDATE  
INSERT INTO QRTZ\_LOCKS(SCHED\_NAME, LOCK\_NAME) VALUES (‘quartzScheduler’, ?)  
//获取锁失败直接抛异常，保证了只会有一个任务会获得锁

*  执行任务  
    job = sched.getJobFactory().newJob(firedTriggerBundle, scheduler);
 *  任务执行完成（事件机制）  
    DELETE FROM QRTZ\_TRIGGERS WHERE SCHED\_NAME = ‘quartzScheduler’ AND TRIGGER\_NAME = ? AND TRIGGER\_GROUP = ?  
    DELETE FROM QRTZ\_FIRED\_TRIGGERS WHERE SCHED\_NAME = ‘quartzScheduler’ AND ENTRY\_ID = ?
 *  并且更新下次要调度的trigger时间  
    UPDATE QRTZ\_TRIGGERS SET JOB\_NAME = ?, JOB\_GROUP = ?, DESCRIPTION = ?, NEXT\_FIRE\_TIME = ?, PREV\_FIRE\_TIME = ?, TRIGGER\_STATE = ?, TRIGGER\_TYPE = ?, START\_TIME = ?, END\_TIME = ?, CALENDAR\_NAME = ?, MISFIRE\_INSTR = ?, PRIORITY = ?, JOB\_DATA = ? WHERE SCHED\_NAME = ‘quartzScheduler’ AND TRIGGER\_NAME = ? AND TRIGGER\_GROUP = ?

mysql> select TRIGGER_NAME,from_unixtime(NEXT_FIRE_TIME/1000),from_unixtime(PREV_FIRE_TIME/1000),from_unixtime(START_TIME/1000),TRIGGER_STATE from QRTZ_TRIGGERS;
    +----------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    | TRIGGER_NAME         | from_unixtime(NEXT_FIRE_TIME/1000) | from_unixtime(PREV_FIRE_TIME/1000) | from_unixtime(START_TIME/1000) | TRIGGER_STATE |
    +----------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    | sampleJobCronTrigger | 2017-09-25 15:45:00.0000           | 2017-09-25 15:40:00.0000           | 2017-09-25 11:55:02.0000       | WAITING       |
    +----------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    1 row in set (0.00 sec)

PS：QRTZ\_TRIGGERS每一个任务对应一条记录

mysql> select TRIGGER_NAME,from_unixtime(NEXT_FIRE_TIME/1000),from_unixtime(PREV_FIRE_TIME/1000),from_unixtime(START_TIME/1000),TRIGGER_STATE from QRTZ_TRIGGERS;
    +------------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    | TRIGGER_NAME           | from_unixtime(NEXT_FIRE_TIME/1000) | from_unixtime(PREV_FIRE_TIME/1000) | from_unixtime(START_TIME/1000) | TRIGGER_STATE |
    +------------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    | sampleJobCronTrigger   | 2017-09-25 16:05:00.0000           | 2017-09-25 16:00:00.0000           | 2017-09-25 11:55:02.0000       | WAITING       |
    | sampleJobV2CronTrigger | 2017-09-25 16:05:00.0000           | 2017-09-25 16:00:00.0000           | 2017-09-25 15:57:17.0000       | WAITING       |
    +------------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    2 rows in set (0.00 sec)

案例1：线上有个任务有问题，立即kill掉jvm，但是机器重启后任务还是会执行，怎么解决？  
答案：在机器重启前删除掉QRTZ\_FIRED\_TRIGGERS中对应的记录

数据库截图如下：

mysql> select * from QRTZ_FIRED_TRIGGERS;;
    +-----------------+-------------------------------------------------------+----------------------+---------------+------------------------------------------+---------------+---------------+----------+-----------+-----------------+-----------+------------------+-------------------+
    | SCHED_NAME      | ENTRY_ID                                              | TRIGGER_NAME         | TRIGGER_GROUP | INSTANCE_NAME                            | FIRED_TIME    | SCHED_TIME    | PRIORITY | STATE     | JOB_NAME        | JOB_GROUP | IS_NONCONCURRENT | REQUESTS_RECOVERY |
    +-----------------+-------------------------------------------------------+----------------------+---------------+------------------------------------------+---------------+---------------+----------+-----------+-----------------+-----------+------------------+-------------------+
    | quartzScheduler | caowenyideMacBook-Pro.local15063261797261506326179669 | sampleJobCronTrigger | DEFAULT       | caowenyideMacBook-Pro.local1506326179726 | 1506327900007 | 1506327900000 |        5 | EXECUTING | sampleJobDetail | DEFAULT   | 1                | 1                 |
    +-----------------+-------------------------------------------------------+----------------------+---------------+------------------------------------------+---------------+---------------+----------+-----------+-----------------+-----------+------------------+-------------------+
    1 row in set (0.00 sec)
    
    ERROR:
    No query specified
    
    mysql> select TRIGGER_NAME,from_unixtime(NEXT_FIRE_TIME/1000),from_unixtime(PREV_FIRE_TIME/1000),from_unixtime(START_TIME/1000),TRIGGER_STATE from QRTZ_TRIGGERS;
    +------------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    | TRIGGER_NAME           | from_unixtime(NEXT_FIRE_TIME/1000) | from_unixtime(PREV_FIRE_TIME/1000) | from_unixtime(START_TIME/1000) | TRIGGER_STATE |
    +------------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    | sampleJobCronTrigger   | 2017-09-25 16:30:00.0000           | 2017-09-25 16:25:00.0000           | 2017-09-25 11:55:02.0000       | BLOCKED       |
    | sampleJobV2CronTrigger | 2017-09-25 16:30:00.0000           | 2017-09-25 16:25:00.0000           | 2017-09-25 16:18:00.0000       | WAITING       |
    +------------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    2 rows in set (0.00 sec)
    
    mysql> select TRIGGER_NAME,from_unixtime(NEXT_FIRE_TIME/1000),from_unixtime(PREV_FIRE_TIME/1000),from_unixtime(START_TIME/1000),TRIGGER_STATE from QRTZ_TRIGGERS;
    +----------------------------------------------------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    | TRIGGER_NAME                                                   | from_unixtime(NEXT_FIRE_TIME/1000) | from_unixtime(PREV_FIRE_TIME/1000) | from_unixtime(START_TIME/1000) | TRIGGER_STATE |
    +----------------------------------------------------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    | recover_caowenyideMacBook-Pro.local1506326179726_1506328016310 | NULL                               | 2017-09-25 16:25:00.0000           | 2017-09-25 16:25:00.0000       | COMPLETE      |
    | sampleJobCronTrigger                                           | 2017-09-25 16:30:00.0000           | 2017-09-25 16:25:00.0000           | 2017-09-25 11:55:02.0000       | BLOCKED       |
    | sampleJobV2CronTrigger                                         | 2017-09-25 16:30:00.0000           | 2017-09-25 16:25:00.0000           | 2017-09-25 16:18:00.0000       | WAITING       |
    +----------------------------------------------------------------+------------------------------------+------------------------------------+--------------------------------+---------------+
    3 rows in set (0.00 sec)

[wKioOVGZjcayYIDlAACPrqg-Xsw581.jpg]: https://timgsa.baidu.com/timg?image&quality=80&size=b9999_10000&sec=1506426861995&di=706e5e58cb9db06268bd241d811cb8fe&imgtype=0&src=http://s3.51cto.com/wyfs01/M00/09/EE/wKioOVGZjcayYIDlAACPrqg-Xsw581.jpg