Quartz使用(6) - Quartz项目实战

本片博文将阐述项目工作中使用Quartz的情况，包含项目背景、项目框架、Quartz集群部署等方面，重点讲述如何在实际项目中使用Quartz。

1. 背景

因项目需求，需要定时调用数据下载接口，并将数据存储至诸如mongo、redis、elasticsearch等数据库或缓存中。具体涉及到的需求如下：

a. 调用接口的任务均从mongo数据库读取;

b. 任务的个数随着业务量的增加而增加;

c. 每个调用任务的定时执行时间可能不同，且定时执行时间在mongo中可配置;

d. 任务的执行需要动态更新，如检测到某一任务的定时时间发生变化，则任务的执行也需要实时修改

e. mongo、redis、elasticsearch等数据库中所存储的字段也由mongo进行配置;

f. 任务执行需要实时性较高、可靠性较强、可扩展性较高等

综上需求，调研了一番，发现任务调度框架Quartz可满足项目需求。

2. 框架

基于项目的需求，结合任务调度框架Quartz，大体的流程框架如下图所示：

1) 首先构建从mongo加载任务

2) 将任务的配置信息初始化至Quartz

3) 通过Quartz的Job任务实现定时调用下载接口任务

4) 将下载的数据依据配置，存储至数据库中

5) 定时检测任务通过定时扫描mongo数据库，查看相关任务信息的配置是否发生变化，如果发生变化，则进行动态更新

6) 为了实现高可用性、可扩展性，可以直接使用Quartz原生的集群特性。

3. 核心代码

核心代码将会涵盖上述流程图中的相关环节，为了项目的保密性，相关信息也会隐藏。

3.1 任务主流程　　

import com.quartz.conf.Configuration;
import com.quartz.conf.OcpConfHelper;
import com.quartz.module.TaskInfo;
import org.apache.log4j.PropertyConfigurator;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.impl.StdSchedulerFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

public class SchedulerRunner {
    static Logger logger = LoggerFactory.getLogger(SchedulerRunner.class);

    public static void main(String[] args) {

        // 加载日志配置文件
        PropertyConfigurator.configure("./conf/log4j.properties");

        // 加载quartz配置文件
        System.setProperty("org.quartz.properties", "./conf/quartz.properties");

        // 执行任务解析与调度
        run();
    }

    public static void run(){
        // 获取配置信息表
        List<TaskInfo> taskInfos = GenerateTaskInfo.generateTaskInfoFromMysql();
        if(taskInfos.size() == 0){
            logger.info("there is no tasks from mongoInfo");
            return;
        }

        // 过滤下线任务
        taskInfos = GenerateTaskInfo.filterTask(taskInfos);
        if(taskInfos.size() == 0){
            logger.info("all tasks if offline, no need to run");
            return;
        }

        Scheduler scheduler = null;
        try {
            scheduler = StdSchedulerFactory.getDefaultScheduler();
        } catch (SchedulerException e) {
            e.printStackTrace();
        }

        if(scheduler == null){
            logger.error("create scheduler failed");
            return;
        }

        if(isSchedulerClear()){
            clearSchedulerJob(scheduler);
        }

        // 加入任务调度
        for(TaskInfo task : taskInfos){
            SchedulerFactory.addJob2Scheduler(task, scheduler);
        }

        // 加入动态更新任务
        SchedulerFactory.addDynamicUpdateJob2Scheduler(scheduler);

        // 开启任务
        try {
            scheduler.start();
        } catch (SchedulerException e) {
            logger.error("start scheduler error!");
        }
    }

    public static void clearSchedulerJob(Scheduler scheduler){
        try {
            scheduler.clear();
        } catch (SchedulerException e) {
            logger.error("clear  scheduler error!");
        }
    }

    /**
     * 基于配置文件中的信息，加载调度器开始运行时的清洗标识
     * @return
     */
    private static boolean isSchedulerClear(){
        Configuration conf = OcpConfHelper.getInstance().getOcpConf();
        return conf.getBooleanValue("cleanSchedulerFlag", "true");
    }
}