springboot整合quartz定时任务

2018-08-23 本文已影响507人东方舵手

1. quartz的基本实现原理

** Quartz 核心元素 **

Quartz任务调度的核心元素为：

Scheduler——任务调度器

Trigger——触发器

Job——任务

其中trigger和job是任务调度的元数据，scheduler是实际执行调度的控制器。

Trigger是用于定义调度时间的元素，即按照什么时间规则去执行任务。Quartz中主要提供了四种类型的trigger：SimpleTrigger，CronTirgger，DateIntervalTrigger，和NthIncludedDayTrigger。这四种trigger可以满足企业应用中的绝大部分需求。

Job用于表示被调度的任务。主要有两种类型的job：无状态的（stateless）和有状态的（stateful）。对于同一个trigger来说，有状态的job不能被并行执行，只有上一次触发的任务被执行完之后，才能触发下一次执行。Job主要有两种属性：volatility和durability，其中volatility表示任务是否被持久化到数据库存储，而durability表示在没有trigger关联的时候任务是否被保留。两者都是在值为true的时候任务被持久化或保留。一个job可以被多个trigger关联，但是一个trigger只能关联一个job。

Scheduler由scheduler工厂创建：DirectSchedulerFactory或者StdSchedulerFactory。第二种工厂StdSchedulerFactory使用较多，因为DirectSchedulerFactory使用起来不够方便，需要作许多详细的手工编码设置。Scheduler主要有三种：RemoteMBeanScheduler，RemoteScheduler和StdScheduler。

Quartz 线程

在Quartz中，有两类线程，Scheduler调度线程和任务执行线程，其中任务执行线程通常使用一个线程池维护一组线程。

Scheduler调度线程主要有两个：执行常规调度的线程，和执行misfiredtrigger的线程。

常规调度线程轮询存储的所有trigger，如果有需要触发的trigger，即到达了下一次触发的时间，则从任务执行线程池获取一个空闲线程，执行与该trigger关联的任务。

Misfire线程是扫描所有的trigger，查看是否有misfiredtrigger，如果有的话根据misfire的策略分别处理(fire now OR wait for the next fire)

Quartz Job数据存储

Quartz中的trigger和job需要存储下来才能被使用。Quartz中有两种存储方式：RAMJobStore,JobStoreSupport，其中RAMJobStore是将trigger和job存储在内存中，而JobStoreSupport是基于jdbc将trigger和job存储到数据库中。RAMJobStore的存取速度非常快，但是由于其在系统被停止后所有的数据都会丢失，所以在集群应用中，必须使用JobStoreSupport。

Quartz 集群架构

一个Quartz集群中的每个节点是一个独立的Quartz应用，它又管理着其他的节点。这就意味着你必须对每个节点分别启动或停止。Quartz集群中，独立的Quartz节点并不与另一其的节点或是管理节点通信，而是通过相同的数据库表来感知到另一Quartz应用的，如图2.1所示。

** Quartz集群相关数据库表**

因为Quartz集群依赖于数据库，所以必须首先创建Quartz数据库表，Quartz发布包中包括了所有被支持的数据库平台的SQL脚本。这些SQL脚本存放于<quartz_home>/docs/dbTables 目录下。

1、QRTZ_JOB_DETAILS：存储的是job的详细信息，包括：[DESCRIPTION]描述，[IS_DURABLE]是否持久化，[JOB_DATA]持久化对象等基本信息。

2、QRTZ_TRIGGERS：触发器信息,包含：job的名，组外键，[DESCRIPTION]触发器的描述等基本信息，还有[START_TIME]开始执行时间，[END_TIME]结束执行时间，[PREV_FIRE_TIME]上次执行时间，[NEXT_FIRE_TIME]下次执行时间，[TRIGGER_TYPE]触发器类型：simple和cron，[TRIGGER_STATE]执行状态：WAITING，PAUSED，ACQUIRED分别为：等待，暂停，运行中。

3、QRTZ_CRON_TRIGGERS：保存cron表达式。

4、QRTZ_SCHEDULER_STATE：存储集群中note实例信息，quartz会定时读取该表的信息判断集群中每个实例的当前状态，INSTANCE_NAME:之前配置文件中org.quartz.scheduler.instanceId配置的名字，就会写入该字段，如果设置为AUTO,quartz会根据物理机名和当前时间产生一个名字。 [LAST_CHECKIN_TIME]上次检查时间，[CHECKIN_INTERVAL]检查间隔时间。

5、QRTZ_PAUSED_TRIGGER_GRPS：暂停的任务组信息。

6、QRTZ_LOCKS，悲观锁发生的记录信息。

7、QRTZ_FIRED_TRIGGERS，正在运行的触发器信息。

8、QRTZ_SIMPLE_TRIGGERS，简单的触发器详细信息。

9、QRTZ_BLOB_TRIGGERS,触发器存为二进制大对象类型(用于Quartz用户自己触发数据库定制自己的触发器，然而JobStore不明白怎么存放实例的时候)。

10、QRTZ_CALENDARS,以 Blob 类型存储 Quartz 的 Calendar 信息

数据库相关表介绍：http://www.cnblogs.com/zhenyuyaodidiao/p/4755649.html

参数配置介绍：http://blog.csdn.net/zixiao217/article/details/53091812

2. pom.xml引入依赖

<!--集成quartz-->
<dependency>
    <groupId>org.quartz-scheduler</groupId>
    <artifactId>quartz</artifactId>
    <version>${quartz.version}</version>
    <!-- quartz默认使用c3p0连接池，如果项目使用的不是则需要排除依赖包 -->
    <exclusions>
        <exclusion>
            <artifactId>c3p0</artifactId>
            <groupId>c3p0</groupId>
        </exclusion>
    </exclusions>
</dependency>
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-context-support</artifactId>
</dependency>

3. 创建配置文件类

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;

import javax.sql.DataSource;
import java.util.Properties;

/**
 * 定时任务配置
 *
 */
@Configuration
public class QuartzConfig {

    @Bean
    public SchedulerFactoryBean scheduler(@Qualifier("druidDataSource") DataSource dataSource) {

        //quartz参数
        Properties prop = new Properties();
        //配置实例
        //prop.put("org.quartz.scheduler.instanceName", "MyScheduler");//实例名称
        prop.put("org.quartz.scheduler.instanceId", "AUTO");
        //线程池配置
        prop.put("org.quartz.threadPool.threadCount", "5");
        //JobStore配置
        prop.put("org.quartz.jobStore.class", "org.quartz.impl.jdbcjobstore.JobStoreTX");
        prop.put("org.quartz.jobStore.tablePrefix", "QRTZ_");

        SchedulerFactoryBean factory = new SchedulerFactoryBean();
        factory.setDataSource(dataSource);
        factory.setQuartzProperties(prop);
        factory.setSchedulerName("MyScheduler");//数据库中存储的名字
        //QuartzScheduler 延时启动，应用启动5秒后 QuartzScheduler 再启动
        factory.setStartupDelay(5);

        //factory.setApplicationContextSchedulerContextKey("applicationContextKey");
        //可选，QuartzScheduler 启动时更新己存在的Job，这样就不用每次修改targetObject后删除qrtz_job_details表对应记录了
        factory.setOverwriteExistingJobs(true);
        //设置自动启动，默认为true
        factory.setAutoStartup(true);

        return factory;
    }
}

4. 创建自定义Job任务类，继承QuartzJobBean

import com.alibaba.fastjson.JSON;
import com.qfedu.rongzaiboot.entity.ScheduleJob;
import com.qfedu.rongzaiboot.entity.ScheduleJobLog;
import com.qfedu.rongzaiboot.service.ScheduleJobLogService;
import com.qfedu.rongzaiboot.utils.SpringContextUtils;
import org.apache.commons.lang.StringUtils;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.util.ReflectionUtils;

import java.lang.reflect.Method;
import java.util.Date;

public class QuartzJob extends QuartzJobBean {
    private Logger logger = LoggerFactory.getLogger(getClass());

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        System.out.println("执行quartz任务。。。。。");

        String json = context.getMergedJobDataMap().getString("JOB_PARAM_KEY");
        //将获取的对象序列化的json 转化为实体类对象
        ScheduleJob scheduleJob = JSON.parseObject(json, ScheduleJob.class);

        Long jobId = scheduleJob.getJobId();
        String beanName = scheduleJob.getBeanName();
        String methodName = scheduleJob.getMethodName();
        String params = scheduleJob.getParams();

        //quartz没有被spring管理 所以通过其它方式获取service
        ScheduleJobLogService scheduleJobLogService = (ScheduleJobLogService) SpringContextUtils.getBean("scheduleJobLogServiceImpl");
        //保存任务记录日志
        ScheduleJobLog scheduleJobLog = new ScheduleJobLog();
        scheduleJobLog.setJobId(jobId);
        scheduleJobLog.setBeanName(beanName);
        scheduleJobLog.setMethodName(methodName);
        scheduleJobLog.setParams(params);
        scheduleJobLog.setCreateTime(new Date());

        long startTime = System.currentTimeMillis();

        try {
            Object targetClass = SpringContextUtils.getBean(beanName);
            Method method = null;
            //通过反射获取方法
            if (StringUtils.isNotBlank(params)) {
                method = targetClass.getClass().getDeclaredMethod(methodName, String.class);
            } else {
                method = targetClass.getClass().getDeclaredMethod(methodName);
            }

            ReflectionUtils.makeAccessible(method);//使方法具有public权限
            //根据反射执行方法
            if (StringUtils.isNotBlank(params)) {
                method.invoke(targetClass, params);
            } else {
                method.invoke(targetClass);
            }

            long endTime = System.currentTimeMillis() - startTime;

            scheduleJobLog.setStatus((byte) 0);//保存日志里的操作状态 0:成功
            scheduleJobLog.setTimes(endTime);//耗时多长时间

            logger.info("任务执行成功，任务ID：" + jobId + "，总耗时：" + endTime + "毫秒");

        } catch (Exception e) {
            long endTime = System.currentTimeMillis() - startTime;
            scheduleJobLog.setError(StringUtils.substring(e.toString(),2000));//错误消息
            scheduleJobLog.setStatus((byte)1);//失败
            scheduleJobLog.setTimes(endTime);//耗时

            e.printStackTrace();
            logger.error("任务执行失败，任务ID："+jobId);
        } finally {
            //最后调用service保存定时任务日志记录
            scheduleJobLogService.save(scheduleJobLog);
        }

    }

}

5. 创建Scheduler工具类,quartz的操作核心，包括操作quartz在数据库中的表

import com.alibaba.fastjson.JSON;
import com.qfedu.rongzaiboot.entity.ScheduleJob;
import com.qfedu.rongzaiboot.quartz.QuartzJob;
import org.quartz.*;

public class SchedulerUtils {

    /**
     * 创建任务
     */
    public static void createJob(Scheduler scheduler, ScheduleJob scheduleJob) {

        try {
            Long jobId = scheduleJob.getJobId();
            //创建Job对象
            JobDetail job = JobBuilder.newJob(QuartzJob.class).withIdentity("JOB_" + jobId).build();
            //获取cron表达式 并创建对象
            CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(scheduleJob.getCronExpression())
                    .withMisfireHandlingInstructionDoNothing();
            //创建触发器
            CronTrigger trigger = TriggerBuilder.newTrigger()
                    .withIdentity("TRIGGET_" + jobId)
                    .withSchedule(cronScheduleBuilder) //将cron表达式配置到触发器
                    .build();

            //将对象josn序列化存储到Job的getJobDataMap()方法中，为后续根据获取属性执行对应的类的任务
            job.getJobDataMap().put("JOB_PARAM_KEY", JSON.toJSONString(scheduleJob));
            //存数据
            scheduler.scheduleJob(job, trigger);
            scheduler.pauseJob(JobKey.jobKey("JOB_" + jobId));//使任务处于等待状态,创建后不会执行
        } catch (SchedulerException e) {
            throw new RRException("创建任务失败", e);
        }
    }

    /**
     * 更新任务
     */
    public static void updateJob(Scheduler scheduler, ScheduleJob scheduleJob) {
        //获取新的cron表达式
        CronScheduleBuilder cronScheduleBuilder = CronScheduleBuilder.cronSchedule(scheduleJob.getCronExpression())
                .withMisfireHandlingInstructionDoNothing();

        Long jobId = scheduleJob.getJobId();

        try {
            //拿到原有的trigger
            TriggerKey triggerKey = TriggerKey.triggerKey("TRIGGER_" + jobId);
            CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
            //为原有的trigger赋予新的cron表达式
            trigger = trigger.getTriggerBuilder().withIdentity(triggerKey)
                    .withSchedule(cronScheduleBuilder).build();
            //执行原有的trigger更新
            scheduler.rescheduleJob(triggerKey, trigger);
        } catch (SchedulerException e) {
            e.printStackTrace();
            throw new RRException("更新定时任务失败", e);
        }
    }

    /**
     * 删除任务
     */
    public static void deleteJob(Scheduler scheduler, Long jobId) {
        try {
            scheduler.deleteJob(JobKey.jobKey("JOB_" + jobId));
        } catch (SchedulerException e) {
            e.printStackTrace();
            throw new RRException("删除定时任务失败", e);
        }
    }

    /**
     * 恢复任务
     */
    public static void resumeJob(Scheduler scheduler, Long jobId) {
        try {
            scheduler.resumeJob(JobKey.jobKey("JOB_" + jobId));
        } catch (SchedulerException e) {
            e.printStackTrace();
            throw new RRException("恢复定时任务失败", e);
        }
    }

    /**
     * 立即执行定时任务
     */
    public static void run(Scheduler scheduler, Long jobId) {
        try {
            //只执行一次并且不会改变任务的状态
            scheduler.triggerJob(JobKey.jobKey("JOB_" + jobId));
        } catch (SchedulerException e) {
            e.printStackTrace();
            throw new RRException("立即执行定时任务失败", e);
        }
    }

    /**
     * 暂停任务
     *
     * @param scheduler
     * @param jobId
     */
    public static void pauseJob(Scheduler scheduler, Long jobId) {
        try {
            scheduler.pauseJob(JobKey.jobKey("JOB_" + jobId));
        } catch (SchedulerException e) {
            e.printStackTrace();
            throw new RRException("暂停定时任务失败", e);
        }
    }
}

6. 保存任务的自定义类实体类

public class ScheduleJob implements Serializable {
    private static final long serialVersionUID = 1L;

    private Long jobId;

    private String beanName; //执行的类名

    private String methodName; //方法名

    private String params; //参数

    private String cronExpression; //cron表达式

    private Byte status; //任务状态 0，运行 1，暂停

    private String remark; //备注

    private Date createTime; //创建时间
}

7. 创建quartz任务

controller

    /**
     * 保存定时任务
     */
    @MyLog("保存定时任务")
    @RequestMapping("/save")
    @RequiresPermissions("schedule:job:save")
    public R save(@RequestBody ScheduleJob scheduleJob){
        scheduleJobService.save(scheduleJob);
        return R.ok();
    }

service,impl

    /**
     * 保存定时任务
     */
    void save(ScheduleJob scheduleJob);
//------------------
@Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void save(ScheduleJob scheduleJob) {
        //保存实体类的信息
        scheduleJob.setCreateTime(new Date());
        scheduleJob.setStatus(Constant.ScheduleStatus.PAUSE.getValue());
        scheduleJobMapper.insertSelective(scheduleJob);

        //创建定时任务 并保存到对应的quatrz表中
        SchedulerUtils.createJob(scheduler, scheduleJob);
    }

dao,mapper

修改任务

controller

   /**
     * 定时任务信息
     */
    @GetMapping("/info/{jobId}")
    @RequiresPermissions(value={"schedule:job:info"})
    public R info(@PathVariable("jobId") Long jobId){
        ScheduleJob scheduleJob = scheduleJobService.queryObject(jobId);
        return R.ok().put("scheduleJob", scheduleJob);
    }
-------------------------------------------
    @MyLog("修改定时任务")
    @PostMapping("/update")
    @RequiresPermissions(value={"schedule:job:update"})
    public R update(@RequestBody ScheduleJob scheduleJob){

        scheduleJobService.update(scheduleJob);
        return R.ok();
    }

service,impl

    /**
     * 查询
     */
    ScheduleJob queryObject(Long jobId);

    /**
     * 更新定时任务
     */
    void update(ScheduleJob scheduleJob);
--------------------------------------------------------
    @Override
    public ScheduleJob queryObject(Long menuId) {
        return scheduleJobMapper.selectByPrimaryKey(menuId);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void update(ScheduleJob scheduleJob) {

        SchedulerUtils.updateJob(scheduler, scheduleJob);

        scheduleJobMapper.updateByPrimaryKeySelective(scheduleJob);
    }

dao,mapper

  <select id="selectByPrimaryKey" resultMap="BaseResultMap" parameterType="java.lang.Long" >
    select
    <include refid="Base_Column_List" />
    from schedule_job
    where job_id = #{jobId,jdbcType=BIGINT}
  </select>

  <update id="updateByPrimaryKeySelective" parameterType="com.qfedu.rongzaiboot.entity.ScheduleJob" >
    update schedule_job
    <set >
      <if test="beanName != null" >
        bean_name = #{beanName,jdbcType=VARCHAR},
      </if>
      <if test="methodName != null" >
        method_name = #{methodName,jdbcType=VARCHAR},
      </if>
      <if test="params != null" >
        params = #{params,jdbcType=VARCHAR},
      </if>
      <if test="cronExpression != null" >
        cron_expression = #{cronExpression,jdbcType=VARCHAR},
      </if>
      <if test="status != null" >
        status = #{status,jdbcType=TINYINT},
      </if>
      <if test="remark != null" >
        remark = #{remark,jdbcType=VARCHAR},
      </if>
      <if test="createTime != null" >
        create_time = #{createTime,jdbcType=TIMESTAMP},
      </if>
    </set>
    where job_id = #{jobId,jdbcType=BIGINT}
  </update>

删除任务

controller

    /**
     * 删除定时任务
     */
    @MyLog("删除定时任务")
    @PostMapping("/del")
    @RequiresPermissions("schedule:job:delete")
    public R delete(@RequestBody Long[] jobIds){
        scheduleJobService.deleteBatch(jobIds);

        return R.ok();
    }

service,impl

    /**
     * 批量删除
     */
    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void deleteBatch(Long[] jobIds) {

        for(Long jobId : jobIds){
            SchedulerUtils.deleteJob(scheduler, jobId);
        }

        //删除数据
        scheduleJobMapper.deleteBatch(jobIds);
    }

dao,mapper

    <!-- 批量删除任务记录 -->
    <delete id="deleteBatch">
        delete from schedule_job where job_id in
        <foreach collection="array" item="jobId" open="(" separator="," close=")">
            #{jobId}
        </foreach>
    </delete>

暂停任务

controller

    /**
     * 暂停定时任务
     */
    @MyLog("暂停定时任务")
    @PostMapping("/pause")
    @RequiresPermissions("schedule:job:pause")
    public R pause(@RequestBody Long[] jobIds){
        scheduleJobService.pause(jobIds);
        return R.ok();
    }

service,impl

    /**
     * 批量暂停任务
     */
    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void pause(Long[] jobIds) {

        for(Long jobId : jobIds){
            SchedulerUtils.pauseJob(scheduler, jobId);
        }

        Map<String, Object> map = new HashMap<>();
        map.put("list", jobIds);
        map.put("status", Constant.ScheduleStatus.PAUSE.getValue());
        scheduleJobMapper.updateBatch(map);
    }

dao,mapper

    <!-- 批量更新状态 -->
    <update id="updateBatch">
        update schedule_job set status = #{status} where job_id in
        <foreach collection="list" item="jobId" open="(" separator="," close=")">
            #{jobId}
        </foreach>
    </update>

恢复任务

controller

    /**
     * 恢复定时任务
     */
    @MyLog("恢复定时任务")
    @PostMapping("/resume")
    @RequiresPermissions("schedule:job:resume")
    public R resume(@RequestBody Long[] jobIds){
        scheduleJobService.resume(jobIds);
        return R.ok();
    }

service,impl

    /**
     * 恢复定时任务
     */
    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void resume(Long[] jobIds) {

        for(Long jobId : jobIds){
            SchedulerUtils.resumeJob(scheduler, jobId);
        }

        Map<String, Object> map = new HashMap<>();
        map.put("list", jobIds);
        map.put("status", Constant.ScheduleStatus.NORMAL.getValue());
        scheduleJobMapper.updateBatch(map);
    }

dao,mapper

    <!-- 批量更新状态 -->
    <update id="updateBatch">
        update schedule_job set status = #{status} where job_id in
        <foreach collection="list" item="jobId" open="(" separator="," close=")">
            #{jobId}
        </foreach>
    </update>

立即执行任务（项目里被设置只会执行一次）

controller

    /**
     * 立即执行定时任务
     */
    @MyLog("立即执行定时任务")
    @PostMapping("/run")
    @RequiresPermissions("schedule:job:run")
    public R run(@RequestBody Long[] jobIds){
        scheduleJobService.run(jobIds);
        return R.ok();
    }

service,impl

    /**
     * 立即执行定时任务
     */
    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void run(Long[] jobIds) {
        for(Long jobId : jobIds){
            SchedulerUtils.run(scheduler, jobId);
        }
    }

dao,mapper

8. 创建quartz的相关的数据库中的表

在官网下载quartz
解压进入到目录 docs > daTables 目录里面有数据库表sql脚本选择自己对应的数据库脚本，直接执行就可以生成quartz相关的表

springboot整合quartz定时任务

1. quartz的基本实现原理

2. pom.xml引入依赖

3. 创建配置文件类

4. 创建自定义Job任务类，继承QuartzJobBean

5. 创建Scheduler工具类,quartz的操作核心，包括操作quartz在数据库中的表

6. 保存任务的自定义类实体类

7. 创建quartz任务

修改任务

删除任务

暂停任务

恢复任务

立即执行任务（项目里被设置只会执行一次）

8. 创建quartz的相关的数据库中的表

猜你喜欢

热点阅读