Quartz定时任务执行原理分析

时间:2021-01-05 07:56:32

1.定时任务执行原理

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

1.1 ScheduledExecutorService定时执行原理

ScheduledExecutorService的schedule方法,其根据delay周期性的执行任务。

public ScheduledFuture<?> schedule(Runnable command,
                                       long delay, TimeUnit unit);

其核心执行类和方法:其主要流程就是根据initTime和period时间计算出第一次执行的时间差,然后调用ReentrantLock.newCondition().awaitNanos(long nanosTimeout)方法,到指定的时间进行唤醒,分配线程进行执行。对于后续的周期性执行的await时间为period.
从代码下面可以发现,在每次执行完成后会调用setNextRunTime来设置下次需要调度的时间,然后根据当前时间和下次执行时间的差值进行await(),当下次触发时间到达则唤醒线程进行执行。

public class ScheduledThreadPoolExecutor extends ThreadPoolExecutor implements ScheduledExecutorService {
        //获取下一个将要任务,并根据计算的delay时间进行等待,知道等待时间到达则唤醒任务,并进行执行。
        public RunnableScheduledFuture<?> take() throws InterruptedException {
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
                for (;;) {
                    RunnableScheduledFuture<?> first = queue[0];
                    if (first == null)
                        available.await();
                    else {
                        long delay = first.getDelay(NANOSECONDS);
                        if (delay <= 0)
                            return finishPoll(first);
                        first = null; // don't retain ref while waiting
                        if (leader != null)
                            available.await();
                        else {
                            Thread thisThread = Thread.currentThread();
                            leader = thisThread;
                            try {
                                available.awaitNanos(delay);
                            } finally {
                                if (leader == thisThread)
                                    leader = null;
                            }
                        }
                    }
                }
            } finally {
                if (leader == null && queue[0] != null)
                    available.signal();
                lock.unlock();
            }
        }

      private void setNextRunTime() {
            long p = period;
            if (p > 0)
                time += p;
            else
                time = triggerTime(-p);
        }

1.2 Quartz定时调度的执行原理

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行进行加锁,从而保证了一个任务只会在分布式环境中的一台机器上执行。
Quartz定时任务执行原理分析

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’, ?)
//获取锁失败直接抛异常,保证了只会有一个任务会获得锁

mysql> select * from QRTZ_LOCKS; +-----------------+----------------+
| SCHED_NAME | LOCK_NAME | +-----------------+----------------+
| quartzScheduler | STATE_ACCESS | | quartzScheduler | TRIGGER_ACCESS |
+-----------------+----------------+
  • 执行任务
    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)