Spring+Quartz 实现任务动态管理监控

Spring+Quartz实现任务动态管理监控

环境：Quartz2.2.3+Spring4.0

配置文件很简单，只需要一个配置就可以了,注入一个FactoryBean.

<!-- 任务工厂，注入需注入scheduler -->
<bean id="schedulerFactoryBean"class="org.springframework.scheduling.quartz.SchedulerFactoryBean" />
<!-- 注册SpringContextUtil用于获取ApplicationContext -->
<bean id="springContextUtil" class="com.hmj.service.quartz.SpringContextUtil"></bean>

Job的实体类(省略GETTER)

private Integer id;
/**
* 任务名称
*/
private String job_name;
/**
* 任务内容
*/
private String job_content;
/**
*  表达式
*/
private String cron_expression;
/**
*  是否并发
*/
private String concurrent;
/**
* 任务分组
*/
private String job_group;
/**
* 调用类
*/
private String target_object;
/**
* 调用类的方法
*/
private String target_method;
/**
* 子任务
*/
private String childJobs;
/**
*  任务状态
*/
private Byte job_status;
/**
* 任务运行状态
*/
private Byte running_status;
/**
* 任务开始时间
*/
@DateTimeFormat(pattern="yyyy-MM-dd")
private Date start_time;
/**
* 任务结束时间
*/
@DateTimeFormat(pattern="yyyy-MM-dd")
private Date end_time;

无实现Job的加上这个注解即可，可以动态添加Job，不用写死！

@DisallowConcurrentExecution//无实现

重点的Quartz管理的Service,作为Spring的Service，这里也可以设置并发，默认是不并发，如果需要并发，就加一个字段，然后添加的时候设置并发为true就行了。

@Service
public class QuartzServiceImpl implements QuartzService {
//这里注入Scheduler,不能注入schedulerFactoryBean
@Resource
private Scheduler scheduler ;

@Override
public Scheduler getScheduler() {
return scheduler;
}

@Resource
private WyFeeJobDao wyFeeJobDao;

/**
* 开启所有任务
*/
@Override
public void startJobs() {
try {
scheduler.start();
} catch (SchedulerException e) {
throw new RuntimeException(e);
}
}

@Override
public boolean addJob(WyFeeJobPO job) throws Exception {
boolean flag = false;
CronTrigger cronTrigger = checkTrigger(job);
// 不存在该任务的触发器
if (cronTrigger == null) {
// 新建一个基于Spring的管理Job类
MethodInvokingJobDetailFactoryBean methodInvJobDetailFB = new MethodInvokingJobDetailFactoryBean();
methodInvJobDetailFB.setName(job.getJob_name());
methodInvJobDetailFB.setGroup(job.getJob_group());
//设置任务类(这里的任务类是交给Spring管理的Bean，如果不是Spring的Bean,需要类的全路径)
methodInvJobDetailFB.setTargetObject( SpringContextUtil.getApplicationContext().getBean(job.getTarget_object()));
//设置任务方法
methodInvJobDetailFB.setTargetMethod(job.getTarget_method());
// 将管理Job类提交到计划管理类
methodInvJobDetailFB.afterPropertiesSet();

JobDetail jobDetail = methodInvJobDetailFB.getObject();
jobDetail.getJobDataMap().put(job.getJob_name(), job);

CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(job.getCron_expression());
// 按新的cronExpression表达式构建一个新的trigger
CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(job.getJob_name(), job.getJob_group()).withSchedule(scheduleBuilder).build();
if(scheduler.isShutdown()) {
startJobs();
}
scheduler.scheduleJob(jobDetail, trigger);// 注入到管理类

flag = true;
}else {
flag = false;
}

return flag;
}

//检查是否有相同的触发器
private CronTrigger checkTrigger(WyFeeJobPO job) throws SchedulerException {
// 获得触发器key
TriggerKey triggerKey = new TriggerKey(job.getJob_name(), job.getJob_group());
// 获得触发器Cron
CronTrigger cronTrigger = (CronTrigger) scheduler.getTrigger(triggerKey);
return cronTrigger;
}

@Override
public boolean updatejob(WyFeeJobPO job) throws Exception {
boolean flag = false;
// 获得触发器key
TriggerKey triggerKey = new TriggerKey(job.getJob_name(), job.getJob_group());
// 获得触发器Cron
CronTrigger cronTrigger = (CronTrigger) scheduler.getTrigger(triggerKey);

if(cronTrigger == null) {
return flag;
}else {

//表达式调度构建器
CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(job
.getCron_expression());

//按新的cronExpression表达式重新构建trigger
cronTrigger = cronTrigger.getTriggerBuilder()
.withIdentity(triggerKey)
.withSchedule(scheduleBuilder)
.build();
if(scheduler.isShutdown()) {
startJobs();
}

//按新的trigger重新设置job执行
scheduler.rescheduleJob(triggerKey, cronTrigger);
flag = true;
WyFeeJobPO jobPO = wyFeeJobDao.getJobById(job.getId());
// Trigger已存在，那么更新相应的定时设置
if(job.getJob_status() == JobStatusEnum.CLOSE.getStatus() || jobPO.getRunning_status() == JobRunningStatusEnum.STOP.getStatus()) {
//暂停任务
pauseJob(job);
flag=true;
}

}
return flag;
}

@Override
public boolean deletJob(WyFeeJobPO job) throws Exception {
boolean flag = false;
CronTrigger checkTrigger = checkTrigger(job);
if(checkTrigger == null) {
return flag;
}else {
//删除任务
JobKey jobKey = JobKey.jobKey(job.getJob_name(), job.getJob_group());
scheduler.deleteJob(jobKey);
flag = true;
}
return flag;
}

@Override
public boolean startJob(WyFeeJobPO job) throws Exception {
boolean flag = false;
CronTrigger checkTrigger = checkTrigger(job);
if(checkTrigger == null) {
return flag;
}else {
//开始任务
JobKey jobKey = JobKey.jobKey(job.getJob_name(),job.getJob_group());
scheduler.resumeJob(jobKey);
flag = true;
}
return flag;
}

@Override
public boolean pauseJob(WyFeeJobPO job) throws Exception {
boolean flag = false;
CronTrigger checkTrigger = checkTrigger(job);
if(checkTrigger == null) {
return flag;
}else {
//暂停任务
JobKey jobKey = JobKey.jobKey(job.getJob_name(),job.getJob_group());
scheduler.pauseJob(jobKey);
flag = true;
}
return flag;
}

当服务器启动时，初始化任务。需要写一个监听器。

@Service
public class StartedListener implements ApplicationListener<ContextRefreshedEvent>{

private Logger logger =  Logger.getLogger(QuartzServiceImpl.class);

@Resource
private WyFeeJobDao wyFeeJobDao;

@Resource
private QuartzService quartzService;

@Override
public void onApplicationEvent(ContextRefreshedEvent event) {
if(event.getApplicationContext().getParent() == null)//root application context 没有parent，他就是老大.
{
//添加监听器
try {
quartzService.getScheduler().getListenerManager().addJobListener(new WyFeeJobListenner());
} catch (Exception e1) {
e1.printStackTrace();
}
//需要执行的逻辑代码，当spring容器初始化完成后就会执行该方法。
WyFeeJobQuery wyFeeJobQuery = new WyFeeJobQuery();
wyFeeJobQuery.setJob_status(JobStatusEnum.START.getStatus());

List<WyFeeJobPO> wyFeeJobPOs = wyFeeJobDao.queryByWyFeeJobQuery(wyFeeJobQuery);
for (WyFeeJobPO wyFeeJobPO : wyFeeJobPOs) {
try {
quartzService.addJob(wyFeeJobPO);
if (wyFeeJobPO.getRunning_status() == JobRunningStatusEnum.STOP.getStatus()) {
quartzService.pauseJob(wyFeeJobPO);
}
logger.info(wyFeeJobPO.getJob_name()+"任务开启！");
} catch (Exception e) {
e.printStackTrace();
logger.warning(wyFeeJobPO.getJob_name()+"任务开启失败");
}
}
}
}

}

Job监听器

public class WyFeeJobListenner implements JobListener{

private Logger logger =  Logger.getLogger(WyFeeJobListenner.class);

public static final String LISTENER_NAME = "JobListener";

@Resource
private JobLogService jobLogService;
//这里必须有个不为null的name
@Override
public String getName() {
return LISTENER_NAME;
}

/**
* 任务执行之前
*/
@Override
public void jobToBeExecuted(JobExecutionContext arg0) {

}

/**
* 但是如果当TriggerListener中的vetoJobExecution方法返回true时,那么执行这个方法.
* 需要注意的是 如果方法(2)执行 那么(1),(3)这个俩个方法不会执行,因为任务被终止了嘛.
*/
@Override
public void jobExecutionVetoed(JobExecutionContext arg0) {

}
/**
* 任务执行完成后执行,jobException如果它不为空则说明任务在执行过程中出现了异常
*/
@SuppressWarnings("null")
@Override
public void jobWasExecuted(JobExecutionContext context, JobExecutionException exception) {
JobKey key = context.getJobDetail().getKey();
if(exception==null) {
//TODO 插入记录表,任务执行完成后的操作

}else {
//TODO 记录错误信息，插入记录表，报错的操作

}
}
}

SpringCotext工具类

public class SpringContextUtil implements ApplicationContextAware {
// Spring应用上下文环境
private static ApplicationContext applicationContext;
/**
* 实现ApplicationContextAware接口的回调方法，设置上下文环境
*
* @param applicationContext
*/
public void setApplicationContext(ApplicationContext applicationContext) {
SpringContextUtil.applicationContext = applicationContext;
}
/**
* @return ApplicationContext
*/
public static ApplicationContext getApplicationContext() {
return applicationContext;
}
/**
* 获取对象
*
* @param name
* @return Object
* @throws BeansException
*/
public static Object getBean(String name) throws BeansException {
return applicationContext.getBean(name);
}
}

Quartz工具类，写了一个转CronExpression的方法。

public class QuartzUtils {

/**
* 将cronExpression转换成中文
* @param cronExp
* @return
*/
public static String translateToChinese(String cronExp)
{
if (cronExp == null || cronExp.length() < 1)
{
return "cron表达式为空";
}
CronExpression exp = null;
// 初始化cron表达式解析器
try
{
exp = new CronExpression(cronExp);
}
catch (ParseException e)
{
return "corn表达式不正确";
}
String[] tmpCorns = cronExp.split(" ");
StringBuffer sBuffer = new StringBuffer();
if(tmpCorns.length == 6)
{
//解析月
if(!tmpCorns[4].equals("*"))
{
sBuffer.append(tmpCorns[4]).append("月");
}
else
{
sBuffer.append("每月");
}
//解析周
if(!tmpCorns[5].equals("*") && !tmpCorns[5].equals("?"))
{
char[] tmpArray =  tmpCorns[5].toCharArray();
for(char tmp:tmpArray)
{
switch (tmp)
{
case '1':
sBuffer.append("星期天");
break;
case '2':
sBuffer.append("星期一");
break;
case '3':
sBuffer.append("星期二");
break;
case '4':
sBuffer.append("星期三");
break;
case '5':
sBuffer.append("星期四");
break;
case '6':
sBuffer.append("星期五");
break;
case '7':
sBuffer.append("星期六");
break;
case '-':
sBuffer.append("至");
break;
default:
sBuffer.append(tmp);
break;
}
}
}

//解析日
if(!tmpCorns[3].equals("?"))
{
if(!tmpCorns[3].equals("*"))
{
sBuffer.append(tmpCorns[3]).append("日");
}
else
{
sBuffer.append("每日");
}
}

//解析时
if(!tmpCorns[2].equals("*"))
{
sBuffer.append(tmpCorns[2]).append("时");
}
else
{
sBuffer.append("每时");
}

//解析分
if(!tmpCorns[1].equals("*"))
{
sBuffer.append(tmpCorns[1]).append("分");
}
else
{
sBuffer.append("每分");
}

//解析秒
if(!tmpCorns[0].equals("*"))
{
sBuffer.append(tmpCorns[0]).append("秒");
}
else
{
sBuffer.append("每秒");
}
}

return sBuffer.toString();

}
}