Quartz任务调度框架--任务执行流程（二）

谁借莪１个温暖的怀抱￠ 2022-05-08 02:12 634阅读 0赞

上一篇博客[Quartz任务调度框架--简介与示例（一）][Quartz_--]中我们已经简介和示例代码对quartz有了初步的认识，这篇博客我们通过追踪quartz的定时任务执行流程来加深对quartz的了解。

1、运行活动线程

![70][]

（1）Quartz\_Worker-\*线程是quartz的线程池初始化的线程数目，在配置文件中可以进行配置

（2）Timer-0是Timer定时任务，作用不是很大，主要是保活

（3）Quartz\_QuartzSchedulerThread线程，quartz中最重要的线程，此线程用来计算定时任务，将定时任务提交到线程池进行执行等操作。

2、流程解析

在Quartz初始化时需要创建调度器工厂，并根据调度器工厂获取调度器

//1.创建Scheduler的工厂
    SchedulerFactory sf = new StdSchedulerFactory();
    //2.从工厂中获取调度器实例
    Scheduler scheduler = sf.getScheduler();

在初始化调度器Scheduler时会初始化QuartzSchedulerThread线程，在scheduler调用start方法时会启动QuartzSchedulerThread线程。

public QuartzScheduler(QuartzSchedulerResources resources, long idleWaitTime, @Deprecated long dbRetryInterval)
            throws SchedulerException {
            this.resources = resources;
            if (resources.getJobStore() instanceof JobListener) {
                addInternalJobListener((JobListener)resources.getJobStore());
            }
            //初始化QuartzSchedulerThread线程
            this.schedThread = new QuartzSchedulerThread(this, resources);
            ThreadExecutor schedThreadExecutor = resources.getThreadExecutor();
            schedThreadExecutor.execute(this.schedThread);
            if (idleWaitTime > 0) {
                this.schedThread.setIdleWaitTime(idleWaitTime);
            }
    
            jobMgr = new ExecutingJobsManager();
            addInternalJobListener(jobMgr);
            errLogger = new ErrorLogger();
            addInternalSchedulerListener(errLogger);
    
            signaler = new SchedulerSignalerImpl(this, this.schedThread);
            
            if(shouldRunUpdateCheck()) 
                updateTimer = scheduleUpdateCheck();
            else
                updateTimer = null;
            
            getLog().info("Quartz Scheduler v." + getVersion() + " created.");
        }

在线程QuartzSchedulerThread的run方法中做了如下处理操作

（1）线程是永久运行的，当quartz暂停时，线程会阻塞一段时间

（2）在执行定时之前首先会检查线程池中可执行线程数，当可执行线程数小于0时不执行定时任务

（3）从数据库或缓存中获取某一段时间内（30秒）要执行的任务

（4）根据任务处理器获取要执行的任务

（5）将任务创建JobRunShell，并初始化Job接口的实现类

（6）将JobRunShell添加到线程池中去运行

（7）当线程任务是持久化到数据库中时，这中间包含了一些数据库事物及锁相关的处理操作，完成集群及分布式相关的处理操作。

@Override
        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...
                                 //等待直到togglePause（false）被调用...  
                                sigLock.wait(1000L);
                            } catch (InterruptedException ignore) {
                            }
                        }
    
                        if (halted.get()) {
                            break;
                        }
                    }
    
                    //2.1获取可用线程的数量
                    int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
                     //将永远是true，由于blockForAvailableThreads的语义...
                    if(availThreadCount > 0) { // will always be true, due to semantics of blockForAvailableThreads...
    
                        List<OperableTrigger> triggers = null;//定义触发器集合
    
                        long now = System.currentTimeMillis();//获取当前的时间
    
                        clearSignaledSchedulingChange();
                        try {
                            //2.2 从jobStore中获取下次要触发的触发器集合
                            //idleWaitTime == 30L * 1000L; 当调度程序发现没有当前触发器要触发，它应该等待多长时间再检查...
                            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();
                            //若有没有触发的Trigger，下次触发时间 next_fire_time 这个会在启动的时候有个默认的misfire机制，如上一篇中分析的 。setNextFireTime(); 即start（）启动时候的当前时间。
                            long timeUntilTrigger = triggerTime - now;
                            while(timeUntilTrigger > 2) {//这里为什么是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
                             //这种情况发生，如果releaseIfScheduleChangedSignificantly 决定 释放Trigger
                            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 {
                                    //2.3 通知JobStore调度程序现在正在触发其先前已获取（保留）的给定触发器（执行其关联的作业）。
                                    List<TriggerFiredResult> res = qsRsrcs.getJobStore().triggersFired(triggers);
                                    if(res != null)
                                        bndles = res;  //下面的2.3方法返回的数据赋值到bndles
                                } 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;
                                }
    
                            }
            //循环List<TriggerFiredResult> bndles 集合，获取TriggerFiredResult和TriggerFiredBundle等 
                            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)
    //如果触发器被暂停，阻塞或其他类似的事件阻止它在这时被触发，或者如果调度器被关闭（暂停），则可以获得'null'
                                if (bndle == null) {
                                    qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
                                    continue;
                                }
    
                                JobRunShell shell = null;
                                try {
                                    //创建 JobRunShell ，并初始化
                                    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
                        应该永远不会发生，如果threadPool.blockForAvailableThreads（）遵循约定
                        continue; // while (!halted)
                    }
    
                    long now = System.currentTimeMillis();
                    long waitTime = now + getRandomizedIdleWaitTime();
                    long timeUntilContinue = waitTime - now;
                    //idleWaitTime  == 30L * 1000L;  idleWaitVariablness ==  7 * 1000;              
                    //计算getRandomizedIdleWaitTime()的值 : idleWaitTime - random.nextInt(idleWaitVariablness);
    
                    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;
       }

整个quartz对任务调度的时序大致如下:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70][]

引用：

[https://www.jianshu.com/p/5625d595a49a][https_www.jianshu.com_p_5625d595a49a]

[Quartz_--]: https://blog.csdn.net/qq924862077/article/details/82972014
[70]: /images/20220508/0c43fd5a02884a69a43e1d48c59fd428.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxOTI0ODYyMDc3_size_16_color_FFFFFF_t_70]: /images/20220508/b884c9cd94e84447aa9fd72f9b6a477d.png
[https_www.jianshu.com_p_5625d595a49a]: https://www.jianshu.com/p/5625d595a49a