Quartz分布式调度在洞窝智能营销平台中的应用-每日运维

QQ截图20230915165829.png

一、背景

在洞窝智能营销平台中存在很多定时调度场景，我们选择了Quartz调度框架。

我们最初选择Quartz有以下原因：

1、Quartz是一套轻量级的任务调度框架，可以方便的集成到我们的应用中，不用额外部署服务；
2、Quartz基于数据库实现了分布式调度能力；
3、Quartz提供了丰富的调度管理API，可以方便的动态添加、修改、删除、暂停、重启调度任务；
4、SpringBoot提供了spring-boot-starter-quartz，可以快速与SpringBoot项目整合；

我们不仅把Quartz作为调度引擎，并且利用Quartz的内置线程池作为任务的执行引擎。所有任务都通过Quartz的线程池去执行，这样我们统一了任务执行的入口，实现对任务的统一管理。

二、简介

1、基本介绍

Quartz是OpenSymphony开源组织在Job scheduling领域又一个开源项目，它可以与J2EE与J2SE应用程序相结合，也可以单独使用。

Quartz是开源且具有丰富特性的“任务调度库”，能够集成于任何的Java应用，小到独立的应用，大至电子商业系统。Quartz能够创建亦简单亦复杂的调度，以执行上十、上百，甚至上万的任务。任务job被定义为标准的Java组件，能够执行任何你想要实现的功能。Quartz调度框架包含许多企业级的特性，如JTA事务、集群的支持。

简而言之，Quartz就是基于Java实现的任务调度框架，用于执行你想要执行的任何任务。

官方网址：www.quartz-scheduler.org/
官方文档：www.quartz-scheduler.org/documentati…
原码地址：github.com/quartz-sche…

2、Quartz运行环境

Quartz可以运行嵌入在另一个独立式应用程序。
Quartz可以在应用程序服务器（或Servlet容器）内被实例化，并且参与事务。
Quartz可以作为一个独立的程序运行（其自己的Java虚拟机内），可以通过RMI使用。
Quartz可以被实例化，作为独立的项目集群（负载平衡和故障转移功能），用于作业的执行。

3、Quartz核心概念

任务 Job
Job 就是你想要实现的任务类，每一个 Job 必须实现 org.quartz.job 接口，且只需实现接口定义的 execute() 方法。

触发器 Trigger
Trigger 为你执行任务的触发器，比如你想每天定时3点发送一份统计邮件，Trigger 将会设置3点执行该任务。

Trigger 主要包含两种 SimplerTrigger 和 CronTrigger 两种。

调度器 Scheduler
Scheduler 为任务的调度器，它会将任务 Job 及触发器 Trigger 整合起来，负责基于 Trigger 设定的时间来执行 Job。

三、整合步骤

1、引入maven依赖


    org.springframework.boot
    spring-boot-starter-quartz

2、创建Quartz依赖的数据表

在Quartz的jar包里有各种数据库的sql脚本，脚本路径：org/quartz/impl/jdbcjobstore，因为使用的mysql数据库，所以选择tables_mysql_innodb.sql。

tapd_41194427_1694597228_475

3、在yaml配置文件中加入Quartz配置

spring:
  quartz:
    auto-startup: true
    job-store-type: jdbc
    startup-delay: 10000
    jdbc:
      initialize-schema: never
    properties:
      org:
        quartz:
          scheduler:
            instanceName: dmaQuartzScheduler
            instanceId: AUTO
          jobStore:
            driverDelegateClass: org.quartz.impl.jdbcjobstore.StdJDBCDelegate
            tablePrefix: QRTZ_
            isClustered: true
            clusterCheckinInterval: 10000
            useProperties: false
          threadPool:
            class: org.quartz.simpl.SimpleThreadPool
            threadCount: 100
            threadPriority: 5
            threadsInheritContextClassLoaderOfInitializingThread: true

4、实现任务接口

import lombok.extern.slf4j.Slf4j;
import org.quartz.*;
import org.springframework.scheduling.quartz.QuartzJobBean;

import javax.annotation.Resource;

@Slf4j
@DisallowConcurrentExecution
public class QuartzDemoJob extends QuartzJobBean {

    @Resource
    private ExecuteManager executeManager;

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        JobDetail jobDetail = context.getJobDetail();
        JobKey key = jobDetail.getKey();
        String name = key.getName();
        log.info("job start, name: {}", name);

        // 下面是具体的任务逻辑，省略

    }
}

通过以上四步就完成了Quartz的整合，可以在业务代码里使用Quartz来创建定时任务了。

import com.easyhome.dma.data.enums.JobStatus;
import com.easyhome.dma.data.job.QuartzDemoJob;
import com.easyhome.dma.data.mapper.label.BaseLabelMapper;
import org.quartz.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;

@Service
public class QuartzDemoManager {

    @Resource
    private BaseLabelMapper baseLabelMapper;

    @Resource
    private Scheduler scheduler;


    @Transactional(rollbackFor = Exception.class)
    public void createLabel(Label label) throws SchedulerException {

        //添加调度
        JobDetail jobDetail = JobBuilder.newJob(QuartzDemoJob.class)
                .withIdentity(String.valueOf(label.getId()))
                .storeDurably()
                .build();
        Trigger trigger = TriggerBuilder.newTrigger()
                .withIdentity(String.valueOf(label.getId()))
                .withSchedule(SimpleScheduleBuilder.repeatMinutelyForever(30).withMisfireHandlingInstructionIgnoreMisfires())
                .startNow()
                .build();
        scheduler.scheduleJob(jobDetail, trigger);
        //保存标签
        baseLabelMapper.insertSelective(label);
    }
}

四、应用

1、异步执行

我们的系统还支持了任务的“手动计算”功能，如果等待任务执行完再返回给前端，用户需要等待很长时间，体验不太好。因此我们在service层只更新了任务的执行状态，就交给Quartz框架异步执行，返回前端。

public void compute(Long labelId) throws SchedulerException {
    Label label = new Label();
    label.setId(labelId);
    label.setState(JobStatus.EXECUTING.name());
    //更新状态
    baseLabelMapper.updateSelective(label);

    //异步执行
    scheduler.triggerJob(new JobKey(String.valueOf(labelId)));
}

2、并发控制

有两种情况会导致同一任务的不同实例同时执行：

1、用户重复对同一任务执行手动计算。
2、有的任务执行时间比较长，上一次还没有执行完，下一次调度已经开始。

如果同一个任务的不同实例同时调度会造成数据混乱，为了避免这种情况，我们在任务接口上增加@DisallowConcurrentExecution注解，让同一个任务的不同实例串行执行。

3、资源控制

由于计算任务都要占用一定的服务器资源，当大量任务同时执行时会造成服务器负载过高，因此我们通过调整Quartz的threadPool.threadCount参数来限制任务的最大并发数。

4、依赖管理

对于有依赖关系的调度任务，假如底层数据没有更新，上层任务每次计算得到的结果都是一样的，这种情况会造成服务器资源的浪费。我们不希望这种情况发生，因此在任务执行前加了前置逻辑来判断任务是否真的需要执行。

1）如果用户手动更新，可以执行；
2）如果当前任务没有依赖，可以执行；
3）如果当前任务没有执行过，可以执行；
4）如果有依赖任务更新过（依赖任务的最后执行时间晚于当前任务的最后执行时间），可以执行；

通过上述逻辑，简单实现了依赖任务的调度编排。

import com.easyhome.dma.data.model.ScheduleTask;

import java.time.LocalDateTime;
import java.util.List;
import java.util.stream.Collectors;

public abstract class DependExecutor extends IExecutor {
    @Override
    public boolean needExecute(ScheduleTask task, ScheduleContext sc, LocalDateTime startTime) {

        //手动调度直接执行
        if (sc.getTriggerType() == 3) {
            return true;
        }

        List dependTasks = scheduleTaskManager.getDependTasks(task.getId());
        //没有依赖直接执行
        if (dependTasks.size() == 0) {
            return true;
        }
        //没有执行过直接执行
        if (task.getLastExecuteTime() == null) {
            return true;
        }
        
		//依赖任务有更新
        List updatedTasks = dependTasks.stream().filter(item -> item.getLastExecuteTime() != null && item.getLastExecuteTime().isAfter(task.getLastExecuteTime())).collect(Collectors.toList());
        if (updatedTasks.size() != 0) {
            return true;
        } else {
            return false;
        }
    }

}

5、事务控制

在洞窝智能营销平台中，新建一个标签要同步创建一个调度，删除标签也要同步删除调度。既然Quartz和业务用了同一个数据库，那么这两个操作是否在同一个事务中呢？我们用代码试验了一下。

import com.easyhome.dma.data.enums.JobStatus;
import com.easyhome.dma.data.job.QuartzDemoJob;
import com.easyhome.dma.data.mapper.label.BaseLabelMapper;
import org.quartz.*;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.Resource;

@Service
public class QuartzDemoManager {

    @Resource
    private BaseLabelMapper baseLabelMapper;

    @Resource
    private Scheduler scheduler;


    @Transactional(rollbackFor = Exception.class)
    public void createLabel(Label label) throws SchedulerException {

        //添加调度
        JobDetail jobDetail = JobBuilder.newJob(QuartzDemoJob.class)
                .withIdentity(String.valueOf(label.getId()))
                .storeDurably()
                .build();
        Trigger trigger = TriggerBuilder.newTrigger()
                .withIdentity(String.valueOf(label.getId()))
                .withSchedule(SimpleScheduleBuilder.repeatMinutelyForever(30).withMisfireHandlingInstructionIgnoreMisfires())
                .startNow()
                .build();
        scheduler.scheduleJob(jobDetail, trigger);
		
		// 抛异常
		int i = 1 / 0;
		
        //保存标签
        baseLabelMapper.insertSelective(label);
    }
}

试验结果说明确实在一个事务里，可以保证两个操作的原子性。那么为什么呢？带着这个疑问我们阅读了下Quartz的源码。

enter image description here

当datasource不为null时，设置JobStore为LocalDataSourceJobStore。

tapd_41194427_1694745666_705

LocalDatasourceJobStore提供了一个匿名的ConnectionProvider。

tapd_41194427_1694746020_320

当有事务控制时，从当前线程上下文获取connection，而这个connection和spring事务使用的connection是同一个，因此Quartz和业务操作是在同一个事务里。

五、遇到的问题

当我们使用cron表达式设置调度时，会出现还没到定时时间就会执行一次的问题，后来发现是因为默认的misfire策略的问题。因此我又研究了一下Quartz的misfire机制。

如果持久性触发器由于调度程序被关闭或因为 Quartz 的线程池中没有可用于执行作业的线程而“错过”其触发时间，则会发生失火。不同的触发器类型有不同的失火指令可供使用。默认情况下，它们使用“智能策略”指令 - 该指令具有基于触发器类型和配置的动态行为。当计划程序启动时，它会搜索任何已触发错误的持久性触发器，然后根据其单独配置的错误触发指令更新每个触发器。

有以下几种misfire策略：

MISFIRE_INSTRUCTION_IGNORE_MISFIRE_POLICY 其处理方式为misfire了多少次就立即触发多少次，之后的触发按照设定的频率正常触发。

MISFIRE_INSTRUCTION_FIRE_NOW 其处理方式为无论misfire多少次，立即触发一次，并且后续的触发时间的计算都要以当前时间为基准计算，直至结束。例如每隔五分钟触发一次，整点开始的，那么每一次触发的时间应该是05分，10分这样5的倍数的。如果misfire的处理在某小时的12分，那么后续的触发时间就是17分，22分这样以此类推，直至任务结束。

MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_EXISTING_COUNT 其处理方式为misfire了多少次都不管，之后的触发按照正常的设定触发，直至结束，但是该方法保证整个任务的生命周期触发的次数是符合预期的。比如已经触发了5次，misfire了3次，总共计划要触发10次，那么该任务不受misfire影响，必须要再触发5次才会结束。

MISFIRE_INSTRUCTION_RESCHEDULE_NEXT_WITH_REMAINING_COUNT 其处理方式类似于上面的NextWithExistingCount，同样不管misfire多少次，但是把misfire的次数也算到了总的计划触发次数中，之后只需要触发剩余的次数即可。例如一个任务计划触发10次，已经触发了5次，misfire了3次，则只需要再触发2次就可以结束了。

MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_EXISTING_REPEAT_COUNT 其处理方式是无论misfire了多少次都立即触发一次，之后以当前时间为基准计算此后的触发时间，直至触发完计划要触发的次数。例如计划触发10次，每隔5分钟触发一次，9点开始。之后misfire了3次，已经触发了5次，在9.53检测到了misfire，则53分立即触发一次，下一次的触发时间为58分，还需触发5次结束。

MISFIRE_INSTRUCTION_RESCHEDULE_NOW_WITH_REMAINING_REPEAT_COUNT 该策略与上面的策略唯一不同的地方在于只需要再触发检测到misfire后剩余的触发次数，misfire的次数也算到计划触发次数之后，剩余次数等于计划触发次数减去已经触发次数和misfire次数。

MISFIRE_INSTRUCTION_DO_NOTHING 该策略会忽略misfire，什么都不做

实际使用时根据实际需求设置相应的策略即可。

六、现状

目前，洞窝智能营销平台有100多个定时任务，每天调度几千次。Quartz作为平台的调度引擎，保证我们的任务准时无误的运行。同时作为执行引擎保证我们的集群负载维持在一个稳定的水平。同时支持业务的事务，提高了平台的可靠性。

Quartz分布式调度在洞窝智能营销平台中的应用

1、基本介绍

2、Quartz运行环境

3、Quartz核心概念

1、引入maven依赖

2、创建Quartz依赖的数据表

3、在yaml配置文件中加入Quartz配置

4、实现任务接口

1、异步执行

2、并发控制

3、资源控制

4、依赖管理

5、事务控制

相关文章

发布评论取消回复

剑圣无痕

百度搜索：蓝易云 – centos7系统手动安装ceph教程

出行365：用国产数据库为国人出行“护航”

tidb8.1的磁盘选择，关于网络ssd，和本地ssd的选择对性能影响很大，差距60倍。

Quartz分布式调度在洞窝智能营销平台中的应用

1、基本介绍

2、Quartz运行环境

3、Quartz核心概念

1、引入maven依赖

2、创建Quartz依赖的数据表

3、在yaml配置文件中加入Quartz配置

4、实现任务接口

1、异步执行

2、并发控制

3、资源控制

4、依赖管理

5、事务控制

相关文章

发布评论 取消回复

剑圣无痕

百度搜索：蓝易云 – centos7系统手动安装ceph教程

出行365：用国产数据库为国人出行“护航”

tidb8.1的磁盘选择，关于网络ssd，和本地ssd的选择对性能影响很大，差距60倍。

发布评论取消回复