一起来学SpringBoot(十一)优雅的整合Quartz

Quartz是一个功能丰富的开源作业调度库,几乎可以集成在任何Java应用程序中 - 从最小的独立应用程序到最大的电子商务系统。Quartz可用于创建简单或复杂的计划,以执行数十,数百甚至数万个作业; 将任务定义为标准Java组件的作业,这些组件可以执行几乎任何可以编程的程序。Quartz Scheduler包含许多企业级功能,例如支持JTA事务和集群。springboot在2.x后加入了quartz的支持,可以说真的是这个start还是做的不错的!!

spring定时器

首先呢说一下spring自带的定时器

package com.maoxs.conf;

import org.springframework.beans.factory.annotation.Configurable;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;

@Component
@Configurable
@EnableScheduling
public class ScheduledTasks {

@Scheduled(fixedRate = 1000 * 30)
public void reportCurrentTime() {
System.out.println("任务执行 " + dateFormat().format(new Date()));
}

//每1分钟执行一次
@Scheduled(cron = "0 */1 * * * * ")
public void reportCurrentByCron() {
System.out.println("任务执行" + dateFormat().format(new Date()));
}

private SimpleDateFormat dateFormat() {
return new SimpleDateFormat("HH:mm:ss");
}

}

其中呢  @EnableScheduling:标注启动定时任务 @Scheduled(fixedRate = 1000 * 30) 定义某个定时任务。当然也可以使用cron表达式。但是呢需要进行任务的持久和复杂的管控,肯定就不行了,下面上quartz。


quartz

首先呢要使用首先加入依赖

<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-quartz</artifactId>
</dependency>

然后在yml中加入

spring:
quartz:
properties:
org:
quartz:
scheduler:
instanceName: MyScheduler
instanceId: AUTO #如果使用集群,instanceId必须唯一,设置成AUTO
threadPool:
class: org.quartz.simpl.SimpleThreadPool
threadCount: 10
threadPriority: 5
threadsInheritContextClassLoaderOfInitializingThread: true
job-store-type: memory #jdbc 为持久化,memory 为内存

这里呢我封装了一个操作的类

package com.maoxs.conf;

import org.quartz.*;
import org.quartz.impl.matchers.GroupMatcher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import java.util.*;

/**
* quartz管理类
*
* @author fulin
*/
@Service
public class QuartzManager {
@Autowired
private Scheduler scheduler;

/**
* 增加一个job
*
* @param jobClass 任务实现类
* @param jobName 任务名称
* @param jobGroupName 任务组名
* @param jobCron cron表达式(如:0/5 * * * * ? )
*/
public void addJob(Class<? extends Job> jobClass, String jobName, String jobGroupName, String jobCron) {
try {
JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(jobName, jobGroupName).build();
Trigger trigger = TriggerBuilder.newTrigger().withIdentity(jobName, jobGroupName)
.startAt(DateBuilder.futureDate(1, DateBuilder.IntervalUnit.SECOND))
.withSchedule(CronScheduleBuilder.cronSchedule(jobCron)).startNow().build();
scheduler.scheduleJob(jobDetail, trigger);
if (!scheduler.isShutdown()) {
scheduler.start();
}
} catch (SchedulerException e) {
e.printStackTrace();
}
}


/**
* 创建or更新任务,存在则更新不存在创建
*
* @param jobClass 任务类
* @param jobName 任务名称
* @param jobGroupName 任务组名称
* @param jobCron cron表达式
*/
public void addOrUpdateJob(Class<? extends Job> jobClass, String jobName, String jobGroupName, String jobCron) {
try {
TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroupName);
CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
if (trigger == null) {
addJob(jobClass, jobName, jobGroupName, jobCron);
} else {
if (trigger.getCronExpression().equals(jobCron)) {
return;
}
updateJob(jobName, jobGroupName, jobCron);
}
} catch (SchedulerException e) {
e.printStackTrace();
}
}

/**
* @param jobClass
* @param jobName
* @param jobGroupName
* @param jobTime
*/
public void addJob(Class<? extends Job> jobClass, String jobName, String jobGroupName, int jobTime) {
addJob(jobClass, jobName, jobGroupName, jobTime, -1);
}

public void addJob(Class<? extends Job> jobClass, String jobName, String jobGroupName, int jobTime, int jobTimes) {
try {
JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(jobName, jobGroupName)// 任务名称和组构成任务key
.build();
// 使用simpleTrigger规则
Trigger trigger = null;
if (jobTimes < 0) {
trigger = TriggerBuilder.newTrigger().withIdentity(jobName, jobGroupName)
.withSchedule(SimpleScheduleBuilder.repeatSecondlyForever(1).withIntervalInSeconds(jobTime))
.startNow().build();
} else {
trigger = TriggerBuilder
.newTrigger().withIdentity(jobName, jobGroupName).withSchedule(SimpleScheduleBuilder
.repeatSecondlyForever(1).withIntervalInSeconds(jobTime).withRepeatCount(jobTimes))
.startNow().build();
}
scheduler.scheduleJob(jobDetail, trigger);
if (!scheduler.isShutdown()) {
scheduler.start();
}
} catch (SchedulerException e) {
e.printStackTrace();
}
}

public void updateJob(String jobName, String jobGroupName, String jobTime) {
try {
TriggerKey triggerKey = TriggerKey.triggerKey(jobName, jobGroupName);
CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
trigger = trigger.getTriggerBuilder().withIdentity(triggerKey)
.withSchedule(CronScheduleBuilder.cronSchedule(jobTime)).build();
// 重启触发器
scheduler.rescheduleJob(triggerKey, trigger);
} catch (SchedulerException e) {
e.printStackTrace();
}
}

/**
* 删除任务一个job
*
* @param jobName 任务名称
* @param jobGroupName 任务组名
*/
public void deleteJob(String jobName, String jobGroupName) {
try {
scheduler.deleteJob(new JobKey(jobName, jobGroupName));
} catch (Exception e) {
e.printStackTrace();
}
}

/**
* 暂停一个job
*
* @param jobName
* @param jobGroupName
*/
public void pauseJob(String jobName, String jobGroupName) {
try {
JobKey jobKey = JobKey.jobKey(jobName, jobGroupName);
scheduler.pauseJob(jobKey);
} catch (SchedulerException e) {
e.printStackTrace();
}
}

/**
* 恢复一个job
*
* @param jobName
* @param jobGroupName
*/
public void resumeJob(String jobName, String jobGroupName) {
try {
JobKey jobKey = JobKey.jobKey(jobName, jobGroupName);
scheduler.resumeJob(jobKey);
} catch (SchedulerException e) {
e.printStackTrace();
}
}


/**
* 立即执行一个job
*
* @param jobName
* @param jobGroupName
*/
public void runAJobNow(String jobName, String jobGroupName) {
try {
JobKey jobKey = JobKey.jobKey(jobName, jobGroupName);
scheduler.triggerJob(jobKey);
} catch (SchedulerException e) {
e.printStackTrace();
}
}


/**
* 获取所有计划中的任务列表
*
* @return
*/
public List<Map<String, Object>> queryAllJob() {
List<Map<String, Object>> jobList = null;
try {
GroupMatcher<JobKey> matcher = GroupMatcher.anyJobGroup();
Set<JobKey> jobKeys = scheduler.getJobKeys(matcher);
jobList = new ArrayList<Map<String, Object>>();
for (JobKey jobKey : jobKeys) {
List<? extends Trigger> triggers = scheduler.getTriggersOfJob(jobKey);
for (Trigger trigger : triggers) {
Map<String, Object> map = new HashMap<>();
map.put("jobName", jobKey.getName());
map.put("jobGroupName", jobKey.getGroup());
map.put("description", "触发器:" + trigger.getKey());
Trigger.TriggerState triggerState = scheduler.getTriggerState(trigger.getKey());
map.put("jobStatus", triggerState.name());
if (trigger instanceof CronTrigger) {
CronTrigger cronTrigger = (CronTrigger) trigger;
String cronExpression = cronTrigger.getCronExpression();
map.put("jobTime", cronExpression);
}
jobList.add(map);
}
}
} catch (SchedulerException e) {
e.printStackTrace();
}
return jobList;
}


/**
* 获取所有正在运行的job
*
* @return
*/
public List<Map<String, Object>> queryRunJon() {
List<Map<String, Object>> jobList = null;
try {
List<JobExecutionContext> executingJobs = scheduler.getCurrentlyExecutingJobs();
jobList = new ArrayList<Map<String, Object>>(executingJobs.size());
for (JobExecutionContext executingJob : executingJobs) {
Map<String, Object> map = new HashMap<String, Object>();
JobDetail jobDetail = executingJob.getJobDetail();
JobKey jobKey = jobDetail.getKey();
Trigger trigger = executingJob.getTrigger();
map.put("jobName", jobKey.getName());
map.put("jobGroupName", jobKey.getGroup());
map.put("description", "触发器:" + trigger.getKey());
Trigger.TriggerState triggerState = scheduler.getTriggerState(trigger.getKey());
map.put("jobStatus", triggerState.name());
if (trigger instanceof CronTrigger) {
CronTrigger cronTrigger = (CronTrigger) trigger;
String cronExpression = cronTrigger.getCronExpression();
map.put("jobTime", cronExpression);
}
jobList.add(map);
}
} catch (SchedulerException e) {
e.printStackTrace();
}
return jobList;
}

}

然后呢新建一个任务类,任务类一定要继承QuartzJobBean,其实跟之前一样,QuartzJobBean也是实现的job类。

package com.maoxs.job;

import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.scheduling.quartz.QuartzJobBean;

public class OrderTimeoutJob extends QuartzJobBean {
@Override
protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
System.out.println("武器强化失败");
}
}

创建一个测试类玩玩

package com.maoxs;

import com.maoxs.conf.QuartzManager;
import com.maoxs.dao.QuartzMapper;
import com.maoxs.job.OrderTimeoutJob;
import com.maoxs.pojo.QuartzEntity;
import com.maoxs.service.JobService;
import lombok.extern.slf4j.Slf4j;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.quartz.*;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.test.context.junit4.SpringRunner;

import java.util.Collections;
import java.util.List;
import java.util.Map;

@RunWith(SpringRunner.class)
@SpringBootTest
@Slf4j
public class SpringbootQuartzApplicationTests {
@Autowired
private QuartzManager quartzManager;
@Test
public void mcreateTask() throws InterruptedException {
quartzManager.addJob(OrderTimeoutJob.class, "maoxs", "q1", "0/1 * * * * ?");
log.info("添加任务成功");
Thread.sleep(10000);
quartzManager.pauseJob("maoxs", "q1");
log.info("暂停任务成功");
Thread.sleep(2000);
log.info("=============查询任务================");
List<Map<String, Object>> maps = quartzManager.queryAllJob();
System.out.println(Collections.unmodifiableCollection(maps));
while (true) ;
}
}

然后运行看看效果 嗯,效果还是很完美的

肯定很多小伙伴关系怎么持久化任务,首先呢肯定是要先创建quartz的表点击下载

然后呢需要改下yml

spring:
quartz:
properties:
org:
quartz:
scheduler:
instanceName: MyScheduler
instanceId: AUTO #如果使用集群,instanceId必须唯一,设置成AUTO
dataSource: #配置数据源
myDS:
driver: com.mysql.jdbc.Driver
URL: jdbc:mysql://192.168.8.100:3306/quartz_database?characterEncoding=utf-8&useSSL=false
user: root
password: 123456
maxConnections: 5
threadPool:
class: org.quartz.simpl.SimpleThreadPool
threadCount: 10
threadPriority: 5
threadsInheritContextClassLoaderOfInitializingThread: true
job-store-type: jdbc #jdbc 为持久化,memory 为内存
jdbc: #存储方式使用JobStoreTX(持久化到数据库中)
initialize-schema: never
comment-prefix: QRTZ_
schema: org.quartz.impl.jdbcjobstore.StdJDBCDelegate

然后呢我这里使用的是mysql 记得加入数据库驱动

<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
</dependency>

然后呢我们重新运行那个测试

ok没问题,再来看下数据库 也是有的

这里缺少一个持久化集群的示例,等等我会补上。

本博文是基于springboot2.x 如果有什么不对的请在下方留言。