Quartz+Spring Java配置

技术

记录一下在Spring中整合Quartz的配置，当时遇到的问题有以下两个个方面：

  1. 能自动配置Job，新建Job无需手动修改配置。
  2. Job中的依赖需要Spring来装配

Job标记注解

用于识别是否是Quartz的Job实现类

1
2
3
4
5
6
7
8
9
10
11
12
import java.lang.annotation.*;

/**
 * Quartz Job标记注解
 * @author fun90 12/12/16
 * @since 2.1.0
 */
@Target({ ElementType.TYPE })
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface JobLabel {
}

Job配置信息注解

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
import java.lang.annotation.*;

/**
 * Quartz Job配置信息注解
 * @author fun90 12/12/16
 * @since 2.1.0
 */
@Target({ ElementType.METHOD, ElementType.PARAMETER })
@Retention(RetentionPolicy.RUNTIME)
@Documented
public @interface JobAnnotation {
    String name();
    String group() default "DEFAULT_GROUP";
    String cronExp();
    String timeZone() default "Asia/Shanghai";
    String description();
}

自动配置Job

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
import org.quartz.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.util.ReflectionUtils;
import org.springframework.util.ReflectionUtils.MethodCallback;

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

/**
 * Quartz Job自动
 * @author fun90 21/10/16
 * @since 1.0.0
 */
public class QuartzJobSchedulingListener implements ApplicationListener<ContextRefreshedEvent> {

    private Logger logger = LoggerFactory.getLogger(QuartzJobSchedulingListener.class);
    private boolean initialed;
    @Autowired
    private Scheduler scheduler;

    @Override
    public synchronized void onApplicationEvent(ContextRefreshedEvent event) {
        if(!initialed) {
            try {
                logger.info("Quartz清理全部任务...");
                scheduler.clear();
            } catch (SchedulerException e) {
                logger.error(e.getMessage(), e);
            }
            initialed = true;
            ApplicationContext applicationContext = event.getApplicationContext();
            Map<JobDetail, CronTrigger> jobDetailCronTriggerMap = this.loadJobAndCronTriggers(applicationContext);
            logger.info("loadJobAndCronTriggers size:" + jobDetailCronTriggerMap.size());
            Set<Map.Entry<JobDetail, CronTrigger>> entries = jobDetailCronTriggerMap.entrySet();
            for (Map.Entry<JobDetail, CronTrigger> entry : entries) {
                try {
                    scheduler.scheduleJob(entry.getKey(), entry.getValue());
                } catch (SchedulerException e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }
    }

    private Map<JobDetail, CronTrigger> loadJobAndCronTriggers(ApplicationContext applicationContext) {
        // 获取标记为Quartz Job实现的类
        final Map<String, Object> quartzJobBeans = applicationContext.getBeansWithAnnotation(JobLabel.class);
        Set<String> beanNames = quartzJobBeans.keySet();
        final Map<JobDetail, CronTrigger> jobDetailCronTriggerMap = new HashMap<>();
        for (final String beanName : beanNames) {
            final Object bean = quartzJobBeans.get(beanName);
            ReflectionUtils.doWithMethods(bean.getClass(), new MethodCallback() {
                        @Override
                        public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
                            if (method.isAnnotationPresent(JobConfig.class)) {
                                JobConfig jobConfig = method.getAnnotation(JobConfig.class);
                                logger.debug("Scheduling a job for " + bean.getClass() + " and method " + method.getName());
                                Class jobClass = bean.getClass();
                                String className = jobClass.getName();
                                String jobName = className + "." + jobConfig.name();
                                String jobGroup = jobConfig.group();
                                String cronExpression = jobConfig.cronExp();
                                //构建job信息
                                JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(jobName, jobGroup).storeDurably(true).build();
                                //表达式调度构建器
                                CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cronExpression)
                                        .inTimeZone(TimeZone.getTimeZone(jobConfig.timeZone()));
                                GregorianCalendar calendar = new GregorianCalendar(TimeZone.getTimeZone(jobConfig.timeZone()));
                                //按新的cronExpression表达式构建一个新的trigger
                                CronTrigger trigger = TriggerBuilder.newTrigger().withIdentity(jobName, jobGroup)
                                        .withSchedule(scheduleBuilder).startAt(calendar.getTime()).build();
                                jobDetailCronTriggerMap.put(jobDetail, trigger);
                            }
                        }
                    }
            );
        }
        return jobDetailCronTriggerMap;
    }
}

Job自动装配

在Job实现类中直接使用Spring注入是没有效果的，需要重新定义一个Job工厂类，重写createJobInstance方法。至于为什么，请查看org.springframework.scheduling.quartz.AdaptableJobFactory的源码就清楚了。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
import org.quartz.spi.TriggerFiredBundle;
import org.springframework.context.ApplicationContext;
import org.springframework.scheduling.quartz.AdaptableJobFactory;

/**
 * Quartz Job自动装配
 * @author xionglingcong 21/10/16
 * @since 1.0.0
 */
public class AutowiringSpringBeanJobFactory extends AdaptableJobFactory {

    private ApplicationContext applicationContext;

    public AutowiringSpringBeanJobFactory(ApplicationContext applicationContext){
        this.applicationContext = applicationContext;
    }

    @Override
    protected Object createJobInstance(final TriggerFiredBundle bundle) throws Exception {
        // 第一种：new一个Job对象
        //final Object job = super.createJobInstance(bundle);
        //applicationContext.getAutowireCapableBeanFactory().autowireBean(job); // 自动装配Job
        // 第二种：从Spring容器中获取Job对象
        final Object job = applicationContext.getBean(bundle.getJobDetail().getJobClass());
        return job;
    }
}

Spring配置

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.io.ClassPathResource;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.scheduling.quartz.AdaptableJobFactory;
import org.springframework.scheduling.quartz.SchedulerFactoryBean;

import javax.sql.DataSource;

/**
 * Quartz配置
 *
 * @author fun90 21/10/16
 * @since 1.0.0
 */
@Configuration
public class QuartzConfig {
    private Logger logger = LoggerFactory.getLogger(QuartzConfig.class);
    @Autowired
    private ApplicationContext applicationContext;
    @Autowired
    private DataSource dataSource;
    @Autowired
    private DataSourceTransactionManager transactionManager;

    @Bean
    public AdaptableJobFactory springBeanJobFactory() {
        AutowiringSpringBeanJobFactory jobFactory = new AutowiringSpringBeanJobFactory(applicationContext);
        logger.debug("Configuring Job factory");
        return jobFactory;
    }

    @Bean
    public QuartzJobSchedulingListener quartJobSchedulingListener(){
        return new QuartzJobSchedulingListener();
    }

    @Bean
    public SchedulerFactoryBean scheduler(/*List<CronTrigger> triggerFactoryBeans*/) {
        SchedulerFactoryBean schedulerFactory = new SchedulerFactoryBean();
        schedulerFactory.setConfigLocation(new ClassPathResource("quartz.properties"));
        schedulerFactory.setDataSource(dataSource);
        schedulerFactory.setTransactionManager(transactionManager);
        schedulerFactory.setApplicationContext(applicationContext);
        schedulerFactory.setSchedulerName("AneNetBillScheduler");
        // 必须的，QuartzScheduler 延时启动，应用启动完后 QuartzScheduler 再启动
        schedulerFactory.setStartupDelay(10);
        schedulerFactory.setApplicationContextSchedulerContextKey("applicationContext");
        // 可选，QuartzScheduler 启动时更新己存在的Job
        schedulerFactory.setOverwriteExistingJobs(true);
        schedulerFactory.setAutoStartup(true);
        schedulerFactory.setJobFactory(springBeanJobFactory());
        return schedulerFactory;
    }
}

使用

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
import org.quartz.DisallowConcurrentExecution;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

@Component
@DisallowConcurrentExecution
@JobLabel
public class SendMessageJob implements Job {
    @Autowired
    private MessageService messageService;

    @Override
    @JobConfig(name = "send", cronExp = "0 */10 * * * ?", description = "")
    public void execute(JobExecutionContext context) throws JobExecutionException {
        messageService.send("消息内容");
    }
}