百度360必应搜狗淘宝本站头条
当前位置:网站首页 > 技术文章 > 正文

Quartz触发Trigger及状态转换(quartz 多个trigger)

ccwgpt 2024-09-18 12:26 34 浏览 0 评论

前言

分布式定时任务框架Quartz的定时任务依赖触发器Trigger来触发执行,那么Trigger如何被触发,在整个触发流程中,Trigger的状态如何变化,本文将对这部分内容进行详细分析。

Quartz框架的基本概念和基本使用可以参考翻翻Quartz框架的旧账

本文基于Quartz2.3.2版本展开分析。

正文

先回顾一下QuartzSchedulerThread的作用,其是由QuartzScheduler持有的 调度线程,在QuartzScheduler创建出来并被调用start() 方法后,QuartzSchedulerThread就会开始运行,会不断的去判断哪些Trigger到点需要触发了,需要触发的Trigger就会被从ThreadPool中分配一个线程,然后执行Trigger关联的JobDetail

具体的整套逻辑,全部在QuartzSchedulerThreadrun() 方法中,下面一起来看一下。

下面方法比较长,分支也比较多,所以重点看有注释的部分,再结合后面的补充说明进行理解

Bash
@Override
public void run() {
    int acquiresFailed = 0;

    while (!halted.get()) {
        try {
            synchronized (sigLock) {
                while (paused && !halted.get()) {
                    try {
                        sigLock.wait(1000L);
                    } catch (InterruptedException ignore) {

                    }
                    acquiresFailed = 0;
                }

                if (halted.get()) {
                    break;
                }
            }

            if (acquiresFailed > 1) {
                try {
                    long delay = computeDelayForRepeatedErrors(qsRsrcs.getJobStore(), acquiresFailed);
                    Thread.sleep(delay);
                } catch (Exception ignore) {

                }
            }

            // 从ThreadPool中获取当前可用线程数量
            // 若没有可用线程则阻塞直到有可用线程
            int availThreadCount = qsRsrcs.getThreadPool().blockForAvailableThreads();
            if(availThreadCount > 0) {

                List<OperableTrigger> triggers;

                long now = System.currentTimeMillis();

                clearSignaledSchedulingChange();
                try {
                    // 获取下一次触发时间在30分钟内的Trigger
                    // 这里的步骤可以分解如下
                    // 1. 从qrtz_locks表获取TRIGGER_ACCESS锁
                    // 2. 从qrtz_triggers表获取触发时间在30分钟内且状态是WAITING的Trigger
                    // 3. 遍历每一个步骤2拿到的Trigger
                    // 4. 从qrtz_job_details表查询出Trigger对应的JobDetail
                    // 5. 如果JobDetail不允许并发执行则判断一下当前JobDetail是否已经由另外一个Tragger执行
                    //    若已经由另外一个Trigger执行则当前Trigger本次不执行
                    // 6. 将确定要执行的Trigger在qrtz_triggers表中的状态设置为ACQUIRED
                    // 7. 将确定要执行的Trigger插入qrtz_fired_triggers表且状态为ACQUIRED
                    // 8. 继续遍历步骤2拿到的Trigger直至全部遍历完
                    // 9. 释放TRIGGER_ACCESS锁
                    // 10. 返回所有符合条件的Trigger
                    triggers = qsRsrcs.getJobStore().acquireNextTriggers(
                            now + idleWaitTime, Math.min(availThreadCount, qsRsrcs.getMaxBatchSize()), qsRsrcs.getBatchTimeWindow());
                    acquiresFailed = 0;
                    if (log.isDebugEnabled())
                        log.debug("batch acquisition of " + (triggers == null ? 0 : triggers.size()) + " triggers");
                } catch (JobPersistenceException jpe) {
                    if (acquiresFailed == 0) {
                        qs.notifySchedulerListenersError(
                            "An error occurred while scanning for the next triggers to fire.",
                            jpe);
                    }
                    if (acquiresFailed < Integer.MAX_VALUE)
                        acquiresFailed++;
                    continue;
                } catch (RuntimeException e) {
                    if (acquiresFailed == 0) {
                        getLog().error("quartzSchedulerThreadLoop: RuntimeException "
                                +e.getMessage(), e);
                    }
                    if (acquiresFailed < Integer.MAX_VALUE)
                        acquiresFailed++;
                    continue;
                }

                if (triggers != null && !triggers.isEmpty()) {

                    now = System.currentTimeMillis();
                    long triggerTime = triggers.get(0).getNextFireTime().getTime();
                    long timeUntilTrigger = triggerTime - now;
                    // 所有Trigger中最先会触发的Trigger的触发时间如果距离当前大于2ms则等待
                    // 直到最先会触发的Trigger的触发时间距离当前小于2ms
                    while(timeUntilTrigger > 2) {
                        synchronized (sigLock) {
                            if (halted.get()) {
                                break;
                            }
                            if (!isCandidateNewTimeEarlierWithinReason(triggerTime, false)) {
                                try {
                                    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;
                    }

                    if(triggers.isEmpty())
                        continue;

                    List<TriggerFiredResult> bndles = new ArrayList<TriggerFiredResult>();

                    boolean goAhead = true;
                    synchronized(sigLock) {
                        goAhead = !halted.get();
                    }
                    if(goAhead) {
                        try {
                            // 将Trigger进行fire
                            // 但是这里并不会执行对应的任务逻辑
                            // 对应的步骤可以分解如下
                            // 1. 从qrtz_locks表获取TRIGGER_ACCESS锁
                            // 2. 遍历每一个需要执行的Trigger
                            // 3. 将qrtz_fired_triggers表中需要执行的Trigger的状态设置为EXECUTING
                            // 4. 将Trigger对象的下一次执行时间nextFireTime更新
                            // 5. 如果Trigger执行的JobDetail不允许并发执行
                            //    将Trigger对象的状态更新为BLOCKED
                            //    否则将Trigger对象的状态更新为WAITING
                            // 6. 如果Trigger执行的JobDetail不允许并发执行
                            //    将JobDetail关联的其它Trigger在qrtz_triggers表里的状态更新
                            //    如果是WAITING或ACQUIRED则更新为BLOCKED
                            // 7. 如果Trigger是最后一次执行则将Trigger对象的状态更新为COMPLETE
                            // 8. 将Trigger对象更新回qrtz_triggers表
                            // 9. 基于这个Trigger创建一个TriggerFiredResult并添加到集合
                            // 10. 继续遍历下一个需要执行的Trigger直至遍历完毕
                            // 11. 释放TRIGGER_ACCESS锁
                            // 12. 此时得到了所有执行的Trigger对应的TriggerFiredResult的集合
                            // 上面步骤执行完后所有fire的Trigger在qrtz_fired_triggers表中的状态是EXECUTING
                            // 在qrtz_triggers表中的状态可能是WAITING,BLOCKED或COMPLETE
                            // 但是此时Trigger对应的JobDetail实际是还没有被执行的
                            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);
                            for (int i = 0; i < triggers.size(); i++) {
                                qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
                            }
                            continue;
                        }

                    }

                    // 遍历每一个执行的Trigger对应的TriggerFiredResult
                    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;
                        }

                        if (bndle == null) {
                            qsRsrcs.getJobStore().releaseAcquiredTrigger(triggers.get(i));
                            continue;
                        }

                        JobRunShell shell = null;
                        try {
                            // 基于TriggerFiredResult创建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;
                        }

                        // 在ThreadPool中分配一个线程来执行JobRunShell
                        // 随后就会在JobRunShell的run()方法中执行JobDetail
                        // 执行完毕后会再执行Trigger的完成逻辑
                        // 对应的步骤可以拆分如下
                        // 1. 从qrtz_locks表获取TRIGGER_ACCESS锁
                        // 2. 如果Trigger后续不再执行则在qrtz_triggers表里删除这个Trigger
                        // 3. 如果Trigger执行的任务是不允许并发执行则将所有关联的Trigger状态做如下更新
                        //    将Trigger状态由BLOCKED恢复成WAITING
                        // 4. 如果任务类由@PersistJobDataAfterExecution注解修饰则
                        //    将qrtz_job_details表里的JobDetail的JOB_DATA字段更新
                        // 5. 删除Trigger在qrtz_fired_triggers表中对应的记录
                        if (qsRsrcs.getThreadPool().runInThread(shell) == false) {
                            getLog().error("ThreadPool.runInThread() return false!");
                            qsRsrcs.getJobStore().triggeredJobComplete(triggers.get(i), bndle.getJobDetail(), CompletedExecutionInstruction.SET_ALL_JOB_TRIGGERS_ERROR);
                        }

                    }

                    continue;
                }
            } else {
                continue;
            }

            long now = System.currentTimeMillis();
            long waitTime = now + getRandomizedIdleWaitTime();
            long timeUntilContinue = waitTime - now;
            synchronized(sigLock) {
                try {
                    if(!halted.get()) {
                    if (!isScheduleChanged()) {
                        sigLock.wait(timeUntilContinue);
                    }
                    }
                } catch (InterruptedException ignore) {
                }
            }

        } catch(RuntimeException re) {
            getLog().error("Runtime error occurred in main trigger firing loop.", re);
        }
    }
    qs = null;
    qsRsrcs = null;
}

下面分小节进行讲解

1.拿到即将触发的所有Trigger

这里的即将触发,就是触发时间在30分钟内且状态是WAITINGTrigger

针对每一个即将触发的Trigger,其在qrtz_triggers表里的状态此时会被置为ACQUIRED,针对这个Trigger同时也会插入一条记录到qrtz_fired_triggers表中,状态也是ACQUIRED,表示这个Trigger已经在fire处理了。

假如我们有一个trigger-1,对应任务允许并发执行,还有一个trigger-2,对应任务不允许并发执行(@DisallowConcurrentExecution注解修饰),并且这两个Trigger的触发时间均在30分钟内。

那么此时在qrtz_triggers表中,它们的状态是这样的。

TRIGGER_NAME

...

TRIGGER_STATE

trigger-1

...

ACQUIRED

trigger-2

...

ACQUIRED

qrtz_fired_triggers表中,它们的状态是这样的。

TRIGGER_NAME

...

STATE

trigger-1

...

ACQUIRED

trigger-2

...

ACQUIRED

2.等待最先触发的Trigger的触发时间在2ms内

如果最先触发的Trigger的触发时间距离当前大于2ms,则进行等待,直到小于等于2ms

3.将Trigger进行fire

fire一个Trigger其实就是将这个Triggerqrtz_fired_triggers表中记录的状态设置为EXECUTING,后面会为这个Trigger分配线程来执行任务,注意此时Trigger对应的任务实际上是还没有执行的

Triggerfire之后,这个Triggerqrtz_triggers里面的状态及下一次fire的时间会被更新,这里需要关注一下 状态 的更新。

如果Trigger对应的任务没有被@DisallowConcurrentExecution注解修饰,那么这个Trigger状态更新为WAITING;如果Trigger对应的任务被@DisallowConcurrentExecution注解修饰,那么这个Trigger状态会更新为BLOCKED,并且还会将这个被@DisallowConcurrentExecution注解修饰的任务所有关联的Trigger的状态更新为BLOCKED

注意到一个Trigger会被fire,首先就是需要满足触发时间在30分钟内且状态是WAITING,所以如果一个被@DisallowConcurrentExecution注解修饰的任务正在被执行,那么这个任务关联的所有Trigger的状态都应该被置为BLOCKED,以防止这些Trigger再次被fire

回到第1小节中的例子,此时在qrtz_triggers表中,trigger-1trigger-2它们的状态是这样的。

TRIGGER_NAME

...

TRIGGER_STATE

trigger-1

...

WAITING

trigger-2

...

BLOCKED

qrtz_fired_triggers表中,它们的状态是这样的。

TRIGGER_NAME

...

STATE

trigger-1

...

EXECUTING

trigger-2

...

EXECUTING

4.为fire的Trigger分配线程并执行任务

fireTrigger会在qrtz_fired_triggers表中插入一条记录,随后就会被分配一个线程来执行这个Trigger关联的JobDetail

执行JobDetail没什么好说的,就是调用到这个任务的execute() 方法,我们这里需要关注的是任务执行完毕后的对于Triggercomplete逻辑。

首先会判断当前这个Trigger是不是不会再执行了,如果不会再执行了,那么就会在qrtz_triggers表里删除这个Trigger

然后就是如果这个Trigger执行的任务是不允许并发执行的,那么此时这个任务关联的所有Trigger的状态肯定都是BLOCKED,所以还需要将这些Trigger的状态由BLOCKED还原为WAITING

最后就是删除fireTriggerqrtz_fired_triggers表中的记录。

回到第1小节中的例子,此时在qrtz_triggers表中,trigger-1trigger-2它们的状态是这样的。

TRIGGER_NAME

...

TRIGGER_STATE

trigger-1

...

WAITING

trigger-2

...

WAITING

qrtz_fired_triggers表中,它们都没有记录了。

5.暂停一个Trigger

在上面的所有讨论中,都没有提及Trigger暂停状态,也就是PAUSED状态,因为这个状态相对独立,没必要和上面的其余状态转换混在一起讨论。

我们可以通过如下手段将Trigger的状态置为PAUSED

  1. 通过Triggergroupname找到Trigger,然后将其状态置为PAUSED。这种可以理解为暂停一个Trigger
  2. 通过JobDetail找到所有关联的Trigger,然后将这些Trigger的状态置为PAUSED。这种可以理解为暂停一个JobDetail

Triggerqrtz_triggers表中的状态是PAUSED之后,就不再满足触发时间在30分钟内且状态是WAITING,从而Trigger就不会被fire,对应的任务也不会被执行。

总结

阅读完本文后,应该能够回答下面的问题。

1. Trigger的触发流程是怎么样的

首先触发时间在30分钟内且状态是WAITINGTrigger会被获取出来;

其次最先触发的Trigger的触发时间在2s内时就会开始fire这些Trigger

fire一个Trigger就是将这个Trigger插入一条数据到qrtz_fired_triggers表,然后会为这个Trigger对应的任务分配一个线程来执行,执行完毕后删除Triggerqrtz_fired_triggers表里的记录。

2. Trigger的状态是怎么变化的

Trigger不触发时状态是WAITING,表示等待着被触发并且允许被触发;

Trigger触发时间在30分钟内时会被获取出来等待被fire,此时Trigger状态是ACQUIRED,表示已经被获取;

Triggerfire后,如果Trigger关联的任务允许并发执行,此时Trigger状态还原为WAITING,表示等待着下一次触发,如果Trigger关联的任务不允许并发执行,此时这个任务关联的所有Trigger的状态会被设置为BLOCKED,表示这些Trigger都阻塞住了;

Trigger对应的任务被执行完毕后,如果Trigger后续不会再触发了,则删除Trigger,如果执行的任务是不允许并发执行的,则需要将这个任务关联的所有Trigger的状态从BLOCKED还原为WAITING

3. Trigger如何暂停

我们可以通过暂停JobDetail来暂停其关联的所有Trigger,也可以单独暂停某一个Trigger

所谓暂停Trigger,其实就是将这个Trigger的状态设置为PAUSED,一旦设置为PAUSED,这些Trigger就不满足触发时间在30分钟内且状态是WAITING,从而就不会被触发了。

4. Quartz如何保证同时只有一个实例执行定时任务

Quartz基于数据库实现了一套分布式锁,可以理解为抢占到锁的实例才有资格来触发Trigger从而执行定时任务。

相关推荐

定时任务工具,《此刻我要...》软件体验

之前果核给大家介绍过一款小众但实用的软件——小说规则下载器,可以把网页里的小说章节按照规则下载到本地,非常适合喜欢阅读小说的朋友。有意思的是,软件作者当时看到果核写的体验内容后,给反推荐到他的帖子里去...

前端定时任务的神库:Node-cron,让你的项目更高效!

在前端开发中,定时任务是一个常见的需求。无论是定时刷新数据、轮询接口,还是发送提醒,都需要一个可靠且灵活的定时任务解决方案。今天,我要向大家介绍一个强大的工具——Node-cron,它不仅能解决定时任...

Shutter Pro!一款多功能定时执行任务工具

这是一款可以在电脑上定时执行多种任务的小工具,使用它可以根据时间,电量等来设定一些定时任务,像定时打开程序、打开文件,定时关机重启,以及定时弹窗提醒等都可以轻松做到。这是个即开即用的小工具,无需安装,...

深度解析 Redis 缓存击穿及解决方案

在当今互联网大厂的后端开发体系中,Redis缓存占据着极为关键的地位。其凭借高性能、丰富的数据类型以及原子性操作等显著优势,助力众多高并发系统从容应对海量用户的访问冲击,已然成为后端开发从业者不可或...

从零搭建体育比分网站完整步骤(比较好的体育比分软件)

搭建一个体育比分网站是一个涉及前端、后端、数据源、部署和维护的完整项目。以下是从零开始搭建的详细流程:一、明确项目需求1.功能需求:实时比分展示(如足球、篮球、网球等)支持多个联赛和赛事历史数据查询比...

告别复杂命令行:GoCron 图形界面让定时任务触手可及

如果你是运维人员或者经常接触一些定时任务的配置,那么你一定希望有一款图形界面来帮助你方便的轻松配置定时任务,而GoCron就是这样一款软件,让你的配置可视化。什么是GoCron从名字你就可以大概猜到,...

Java任务管理框架核心技术解析与分布式高并发实战指南

在当今数字化时代,Java任务管理框架在众多应用场景中发挥着关键作用。随着业务规模的不断扩大,面对分布式高并发的复杂环境,掌握其核心技术并进行实战显得尤为重要。Java任务管理框架的核心技术涵盖多个方...

链表和结构体实现:MCU软件定时器(链表在单片机中的应用)

在一般的嵌入式产品设计中,介于成本、功耗等,所选型的MCU基本都是资源受限的,而里面的定时器的数量更是有限。在我们软件设计中往往有多种定时需求,例如脉冲输出、按键检测、LCD切屏延时等等,我们不可能...

SpringBoot定时任务(springboot定时任务每小时执行一次)

前言在我们开发中,经常碰到在某个时间点去执行某些操作,而我们不能人为的干预执行,这个时候就需要我们使用定时任务去完成该任务,下面我们来介绍下载springBoot中定时任务实现的方式。定时任务实现方式...

定时任务新玩法!systemd timer 完整实战详解

原文链接:「链接」Hello,大家好啊!今天给大家带来一篇使用systemdtimer实现定时任务调度的详细实战文章。相比传统的crontab,systemdtimer更加现代化、结构清晰...

Celery与Django:打造高效DevOps的定时任务与异步处理神器

本文详细介绍了Celery这一强大的异步任务队列系统,以及如何在Django框架中应用它来实现定时任务和异步处理,从而提高运维开发(DevOps)的效率和应用性能。下面我们先认识一下Cele...

订单超时自动取消的7种方案,我用这种!

前言在电商、外卖、票务等系统中,订单超时未支付自动取消是一个常见的需求。这个功能乍一看很简单,甚至很多初学者会觉得:"不就是加个定时器么?"但真到了实际工作中,细节的复杂程度往往会超...

裸机下多任务框架设计与实现(gd32裸机配置lwip 网络ping不通)

在嵌入式系统中,特别是在没有操作系统支持的裸机环境下,实现多任务执行是一个常见的挑战。本文将详细介绍一种基于定时器的多任务框架设计,通过全局时钟和状态机机制,实现任务的非阻塞调度,确保任务执行中不会出...

亿级高性能通知系统构建,小白也能拿来即用

作者介绍赵培龙,采货侠JAVA开发工程师分享概要一、服务划分二、系统设计1、首次消息发送2、重试消息发送三、稳定性的保障1、流量突增2、问题服务的资源隔离3、第三方服务的保护4、中间件的容错5、完善...

运维实战:深度拆解Systemd定时任务原理,90%的人不知道的玩法

运维实战:深度拆解Systemd定时任务原理,90%的人不知道的高效玩法一、Systemd定时任务的核心原理Systemd定时任务是Linux系统中替代传统cron的现代化解决方案,通过...

取消回复欢迎 发表评论: