quartz 定时器核心源代码阅读

淩亂°似流年 2021-08-27 17:15 343阅读 0赞

![20190827153602686.png][]

![20190827153714607.png][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlc2hpeXVhbjE0MDYxNDY4NTQ_size_16_color_FFFFFF_t_70][]

this.schedThread 的类型是 org.quartz.core.QuartzSchedulerThread extends java.lang.Thread,下面是截取的类QuartzSchedulerThread 的 run 方法的一部分:

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlc2hpeXVhbjE0MDYxNDY4NTQ_size_16_color_FFFFFF_t_70 1][]

可见一直死循环,锁等待在 this.sigLock 处,直到 this.togglePause( false ) ( 将 this.paused置为false,并通过 this.sigLock.notifyAll( ) 唤醒此线程  ) 结束这个死循环

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlc2hpeXVhbjE0MDYxNDY4NTQ_size_16_color_FFFFFF_t_70 2][]

以下是 QuartzSchedulerThread.run( ) 方法中的一行代码,用于查询未来一段时间( 即 this.idleWaitTime,默认是 30 秒,可通过 "org.quartz.scheduler.idleWaitTime" 配置,单位是毫秒 )需要触发的任务:

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

进入 JobStoreSupport 的 acquireNextTriggers 这个方法( 上面将 now + this.idleWaitTime 作为 noLaterThan 传递给此方法, noLaterThan 是不晚于的意思,即是表达未来 this.idleWaitTime 时间的意思 ),有如下一行代码:

return JobStoreSupport.this.acquireNextTrigger(conn, noLaterThan, maxCount, timeWindow);  
进入这个方法,发现有如下一行代码:

List<TriggerKey> keys = this.getDelegate().selectTriggerToAcquire(conn, noLaterThan + timeWindow, this.getMisfireTime(), maxCount);
    this.getMisfireTime(  ) 代码如下:
    
    protected long getMisfireTime() {
        long misfireTime = System.currentTimeMillis();
        if (this.getMisfireThreshold() > 0L) {
            misfireTime -= this.getMisfireThreshold();
        }
     
        return misfireTime > 0L ? misfireTime : 0L;
    }

进入 StdJDBCDelegate 的 selectTriggerToAcquire 方法,有如下代码:

ps = conn.prepareStatement(this.rtp("SELECT TRIGGER_NAME, TRIGGER_GROUP, NEXT_FIRE_TIME, PRIORITY FROM {0}TRIGGERS WHERE SCHED_NAME = {1} AND TRIGGER_STATE = ? AND NEXT_FIRE_TIME <= ? AND (MISFIRE_INSTR = -1 OR (MISFIRE_INSTR != -1 AND NEXT_FIRE_TIME >= ?)) ORDER BY NEXT_FIRE_TIME ASC, PRIORITY DESC"));
    if (maxCount < 1) {
        maxCount = 1;
    }
     
    ps.setMaxRows(maxCount);
    ps.setFetchSize(maxCount);
    ps.setString(1, "WAITING");
    ps.setBigDecimal(2, new BigDecimal(String.valueOf(noLaterThan)));
    ps.setBigDecimal(3, new BigDecimal(String.valueOf(noEarlierThan)));
    rs = ps.executeQuery();

sql 格式化之后:

SELECT 
            TRIGGER_NAME, 
            TRIGGER_GROUP, 
            NEXT_FIRE_TIME, 
            PRIORITY 
    FROM {0}
            TRIGGERS 
    WHERE 
            SCHED_NAME = {1} 
    AND 
            TRIGGER_STATE = "WAITING"
    AND 
            NEXT_FIRE_TIME <= new BigDecimal(String.valueOf(noLaterThan))
    AND 
            (  
                    MISFIRE_INSTR = -1 
                OR 
                    (
                            MISFIRE_INSTR != -1 
                        AND 
                            NEXT_FIRE_TIME >= new BigDecimal(String.valueOf(noEarlierThan))
                    )
            ) 
    ORDER BY 
            NEXT_FIRE_TIME ASC, 
            PRIORITY DESC

即每次会查询 介于 noEarlierThan( 过去的 ) 和 noLaterThan ( 未来的 ) 之间的待触发任务

quartz  调度线程 QuartzSchedulerThread 核心 loop 代码分析:

// 由于代码过于庞大,"干扰"阅读, 所以去掉了一些不影响阅读主流程的代码( 如错误日志输出等 )
    @Override
    public void run() {
        // 预获取 triggers 失败次数计数器
        int acquiresFailed = 0;
        while ( !halted.get( ) ) {
            // 检测是否应该暂停
            synchronized ( sigLock ) {
                while ( paused && !halted.get( ) ) {
                    // 等待,直到 togglePause( false ) 被调用 或者 超时
                    sigLock.wait( 1000L );
                    // 暂停后重置 失败次数计数器,这样当恢复执行后就不用再等了
                    acquiresFailed = 0;
                }
                // 为 true  ,表示停止了,退出 while  循环 调度主线程终止
                if ( halted.get( ) ) {
                    break;
                }
            }
            // 如果从 job store 中预获取 triggers  失败后睡一小会( 防止由于数据库挂掉或者什么其他原因导致一直获取失败,如果不睡一小会儿,可能会导致 CPU 爆满 )
            if ( acquiresFailed > 1) {
                long delay = computeDelayForRepeatedErrors( qsRsrcs.getJobStore( ), acquiresFailed );
                Thread.sleep( delay );
            }
            // 获取可用的工作线程数 ( availThreadCount 总会大于0 ,因为 qsRsrcs.getThreadPool().blockForAvailableThreads() 是阻塞的,知道 > 0 才返回 )
            int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
            List<OperableTrigger> triggers;
            long now = System.currentTimeMillis();
            clearSignaledSchedulingChange();
            try {
                // 预获取 min( availThreadCount, this.qsRsrcs.getMaxBatchSize() ) 个介于 [ now + this.idleWaitTime + this.qsRsrcs.getBatchTimeWindow(),now - misfireThreshold ] 时间区间的上的触发器
                triggers = qsRsrcs.getJobStore( ).acquireNextTriggers( now + idleWaitTime, Math.min( availThreadCount, qsRsrcs.getMaxBatchSize( ) ), qsRsrcs.getBatchTimeWindow( ) );
                acquiresFailed = 0;
            } catch ( Exception jpe ) {
                if ( acquiresFailed < Integer.MAX_VALUE ){
                    // 失败次数计数器累加
                    acquiresFailed++;
                }
                // 本次预获取 triggers  失败,继续进行下一次预获取 triggers  while  循环
                continue;
            }
            if ( triggers != null && !triggers.isEmpty( ) ) {
                now = System.currentTimeMillis();
                // 预获取的 triggers  中最早需要执行的 trigger  的触发时间
                long triggerTime = triggers.get( 0 ).getNextFireTime( ).getTime( );
                // 距离本次触发的时间
                long timeUntilTrigger = triggerTime - now;
                // 循环一直等到本轮最先需要触发的的 trigger 的触发的时间非常临近了
                while( timeUntilTrigger > 2 ) {
                    synchronized ( sigLock ) {
                        if ( halted.get( ) ) {
                            break;
                        }
                        if ( !isCandidateNewTimeEarlierWithinReason( triggerTime, false ) ) {
                            // 可能已经在 'synchronize' 上阻塞了一段时间,所以要重新计算 now
                            now = System.currentTimeMillis();
                            timeUntilTrigger = triggerTime - now;
                            if( timeUntilTrigger >= 1 ){
                                sigLock.wait( timeUntilTrigger );
                            }
                        }
                    }
                    // 检查 triggers  是否应该释放( 释放其实就是删除表 FIRED_TRIGGERS 中的记录 )
                    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'
                // 开始执行这些 triggers
                List<TriggerFiredResult> bndles = new ArrayList<TriggerFiredResult>( );
                boolean goAhead = true;
                synchronized( sigLock ) {
                    goAhead = !halted.get();
                }
                if( goAhead ) {
                    try {
                        // 主要是遍历 triggers ,生成绑定对象的集合( 例如如果死数据库存储的话,则从数据库中查询出每个trigger  对应的 job detail,将 trigger 和 job  封装成绑定对象 )
                        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);
                        // 如果查询封装 trigger 和 job 的绑定对象时发生了异常,则依次释放本轮预获取的 trigger ,然后 contine 继续下一次while  循环( 即本次没有任何 trigger  触发 )
                        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 ) {
                        // 可能根据 本轮 预获取的 triggers 查询生成的 "job 和 trigger  绑定对象"集合中存在失败的,则通过 continue 忽略失败的绑定对象
                        getLog().error("RuntimeException while firing trigger " + triggers.get(i), exception);
                        qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
                        continue;
                    }
                    // 由于可能发生 trigger 被暂停, 阻塞, 其他类似的阻止本次 trigger 触发的状况, 或者干脆调度器被干死, 因此此 "job trigger 绑定对象" 可能为空, 需要通过 continue 忽略此 trigger
                    if ( bndle == null ) {
                        qsRsrcs.getJobStore( ).releaseAcquiredTrigger( triggers.get( i ) );
                        continue;
                    }
                    JobRunShell shell = null;
                    try {
                        // 根据 "job trigger 绑定对象"生成 可以直接交给工作线程处理的 job shell
                        shell = qsRsrcs.getJobRunShellFactory( ).createJobRunShell( bndle );
                        shell.initialize( qs );
                    } catch ( SchedulerException se ) {
                        // 生成 job shell  失败,将该 trigger 的状态置为 error
                        qsRsrcs.getJobStore( ).triggeredJobComplete( triggers.get( i ), bndle.getJobDetail( ), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR );
                        continue;
                    }
                    // 将生成的 job shell 交给工作线程池处理
                    if ( qsRsrcs.getThreadPool( ).runInThread( shell ) == false ) {
                        // 此状况应该不会发生,除非调度器正在关闭,或者交给线程池的 job 任务体业务代码本身存在 bug, 或者一些其他 文档不推荐的非法操作,此时将数据库中此任务的执行状态置为 error
                        qsRsrcs.getJobStore( ).triggeredJobComplete( triggers.get( i ), bndle.getJobDetail( ), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR );
                    }
                }
                // 继续下一次 while 循环 预获取 trigger
                continue;
            }
            // 本轮没有预获取到任何 trigger ,等待一个随机的一小会儿时间后 继续下一次 while 循环 预获取 trigger
            long now = System.currentTimeMillis();
            long waitTime = now + getRandomizedIdleWaitTime();
            long timeUntilContinue = waitTime - now;
            synchronized( sigLock ) {
                if( !halted.get( ) ) {
                    if ( !isScheduleChanged( ) ) {
                        sigLock.wait( timeUntilContinue );
                    }
                }
            }
        }
    }

————————————————  
版权声明：本文为CSDN博主「heshiyuan1406146854」的原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接及本声明。  
原文链接：https://blog.csdn.net/heshiyuan1406146854/article/details/100101276

[20190827153602686.png]: /images/20210518/8449103953bb4b548c06352873d4c039.png
[20190827153714607.png]: /images/20210518/d39f1e25aa5a4316b272cfc983f82aad.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlc2hpeXVhbjE0MDYxNDY4NTQ_size_16_color_FFFFFF_t_70]: /images/20210518/78a681b4fb3b4d37ad2548dc1eea15ef.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlc2hpeXVhbjE0MDYxNDY4NTQ_size_16_color_FFFFFF_t_70 1]: /images/20210518/7e9c0572d34f4a32bc8ec5dbd7cd3163.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2hlc2hpeXVhbjE0MDYxNDY4NTQ_size_16_color_FFFFFF_t_70 2]: /images/20210518/4fdba0b67c5d47a8a9a2a9b2b86f1e24.png