使用java.util.concurrent.ScheduledExecutorService结合Spring实现quartz的cron配置

java的几种定时任务的比较：几种任务调度java实现方案

 

我最近看了spring的定时任务源码，发现可以使用ScheduledExecutor来实现quartz的cron表达式的配置。并且我加入了可以通过页面修改和配置cron表达式来达到更灵活的配置。

 

加入的功能如下：

1、开启和关闭任务

2、修改cron表达式并自动重新发布任务

3、基于注解的配置

4、任务持久化到数据库

 

Spring容器启动时扫描所有被注解的方法，并发布需要启动的任务(代码48行)：

@Component
public class TaskConfig implements BeanPostProcessor, ApplicationContextAware, ApplicationListener<ContextRefreshedEvent> {

    private ApplicationContext applicationContext;

    private Set<JobDetail> jobDetails = new HashSet<JobDetail>();

    @Override
    public Object postProcessBeforeInitialization(Object bean, String beanName) throws BeansException {
        return bean;
    }

    @Override
    public Object postProcessAfterInitialization(final Object bean, String beanName) throws BeansException {
        if (!bean.getClass().getCanonicalName().startsWith("com.xxx.task")) {
            return bean;
        }
        final Class<?> targetClass = AopUtils.getTargetClass(bean);
        ReflectionUtils.doWithMethods(targetClass, new ReflectionUtils.MethodCallback() {
            public void doWith(Method method) throws IllegalArgumentException, IllegalAccessException {
                Job annotation = AnnotationUtils.getAnnotation(method, Job.class);
                if (annotation != null) {
                    JobDetail job = new JobDetail();
                    job.setCronExp(annotation.cron());
                    job.setJobClass(bean.getClass());
                    job.setJobName(annotation.name());
                    job.setValid(true);
                    job.setMethodName(method.getName());
                    jobDetails.add(job);
                }
            }
        });
        return bean;
    }

    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        if (event.getApplicationContext() != this.applicationContext) {
            return;
        }

        for (JobDetail job : jobDetails) {
            JobService jobService = applicationContext.getBean(JobService.class);
            JobDetail jobDetail = jobService.findJobByName(job.getJobName());
            if (jobDetail == null) {
                jobDetail = job;
            }
            jobService.schedule(jobDetail);
        }
    }

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.applicationContext = applicationContext;
    }
}

 

再来jobService的：

@Service
public class JobService {
    private TaskScheduler taskScheduler = new ConcurrentTaskScheduler(Executors.newScheduledThreadPool(1));
    private final Map<String, ScheduledFuture<?>> scheduledFutures = new HashMap<String, ScheduledFuture<?>>();
    /**
     * 保存jobName与task的映射关系，等
     */
    private Map<String, Runnable> runnableMap = new ConcurrentHashMap<String, Runnable>();
    private TriggerContext triggerContext = new SimpleTriggerContext();
    /**
     * 根据jobDetail计划一次新的任务
     * 任务执行完毕自动计算下次任务执行时间并保存
     * @param job
     */
    @Transactional
    public void schedule(final JobDetail job) {
        CronTrigger trigger = new CronTrigger(job.getCronExp());
        final CronSequenceGenerator sequenceGenerator = new CronSequenceGenerator(job.getCronExp(), TimeZone.getDefault());

        job.setNextExecTime(sequenceGenerator.next(new Date()));
        commonDao.saveOrUpdate(job);

        ErrorHandler errorHandler = new JobErrorHandler(job);
        Runnable runnable = new Runnable() {
            @Override
            public void run() {
                // 每一次执行任务都需要获取最新的任务状态
                JobDetail currentJob = findJobByName(job.getJobName());
                if (currentJob != null && currentJob.isValid()) {
                    Object o = SpringContextUtil.getBean(currentJob.getJobClass());
                    ReflectionUtils.invokeMethod(ReflectionUtils.findMethod(currentJob.getJobClass(),
                            currentJob.getMethodName()), o);
                    Date d = new Date();
                    currentJob.setLastExecTime(d);
                    currentJob.setNextExecTime(sequenceGenerator.next(new Date()));
                    commonDao.saveOrUpdate(currentJob);
                } else {
                    cancel(job.getJobName());
                }
            }
        };
        Runnable delegatingErrorHandlingRunnable = TaskUtils.decorateTaskWithErrorHandler(runnable, errorHandler, false);
        this.scheduledFutures.put(job.getJobName(), taskScheduler.schedule(delegatingErrorHandlingRunnable, trigger));
    }

    /**
     * 停止任务
     * @param jobName
     */
    @Transactional
    public void stopJob(String jobName) {
        String update = "update cms_sys_jobdetails set VALID='N' where JOB_NAME='" + jobName + "'";
        commonDao.getCurrentSession().createSQLQuery(update).executeUpdate();
        this.cancel(jobName);
    }

    /**
     * 结束任务
     * @param jobName
     */
    @Transactional
    public void startJob(String jobName) {
        JobDetail job = this.findJobByName(jobName);
        job.setValid(true);
        this.schedule(job);
    }

    /**
     * 重置cron表达式
     * @param jobName
     * @param cron
     */
    @Transactional
    public void resetCron(String jobName, String cron) {

        new CronSequenceGenerator(cron, TimeZone.getDefault());

        JobDetail job = this.findJobByName(jobName);
        job.setCronExp(cron);
        commonDao.saveOrUpdate(job);
        if (job.isValid()) {
            this.cancel(jobName);
            this.schedule(job);
        }
    }

    /**
     * 取消未执行的任务
     * @param jobName
     */
    private void cancel(String jobName) {
        ScheduledFuture r = scheduledFutures.get(jobName);
        if (r != null) {
            r.cancel(true); //参数true表示如果任务正在执行，则强行关闭，否则等待执行完毕关闭
        }
    }

    /**
     * 任务处理失败时的ErrorHandler
     */
    private class JobErrorHandler implements ErrorHandler {
        private JobDetail job;

        public JobErrorHandler(JobDetail job) {
            this.job = job;
        }

        private final Log logger = LogFactory.getLog(JobErrorHandler.class);

        @Override
        public void handleError(Throwable t) {
            logger.error("Unexpected error occurred in scheduled task===>" + job.getJobName(), t);
        }
    }

}

 

 

JobDetai是需要持久化到数据库的实体：

@Entity
@Table(name = "CMS_SYS_JOBDETAILS")
public class JobDetail implements Serializable {
    private static final long serialVersionUID = 6899388713399265016L;

    @Id
    @GeneratedValue(strategy = GenerationType.SEQUENCE)
    private Long id;

    @Column(name="JOB_NAME", length = 255)
    private String jobName;

    @Column(name="CRON_EXP", length = 255)
    private String cronExp;

    @Column(name = "LAST_EXEC_TIME")
    @Temporal(TemporalType.TIMESTAMP)
    private Date lastExecTime;

    @Column(name = "NEXT_EXEC_TIME")
    @Temporal(TemporalType.TIMESTAMP)
    private Date nextExecTime;

    @Type(type = "yes_no")
    @Column(name = "VALID")
    private boolean valid;

    @Column(name = "JOB_CLASS")
    private Class jobClass;

    @Column(name = "METHOD_NAME")
    private String methodName;

    public String getMethodName() {
        return methodName;
    }

    public void setMethodName(String methodName) {
        this.methodName = methodName;
    }

    public Class getJobClass() {
        return jobClass;
    }

    public void setJobClass(Class jobClass) {
        this.jobClass = jobClass;
    }

    public Long getId() {
        return id;
    }

    public void setId(Long id) {
        this.id = id;
    }

    public String getJobName() {
        return jobName;
    }

    public void setJobName(String jobName) {
        this.jobName = jobName;
    }

    public String getCronExp() {
        return cronExp;
    }

    public void setCronExp(String cronExp) {
        this.cronExp = cronExp;
    }

    public Date getLastExecTime() {
        return lastExecTime;
    }

    public void setLastExecTime(Date lastExecTime) {
        this.lastExecTime = lastExecTime;
    }

    public Date getNextExecTime() {
        return nextExecTime;
    }

    public boolean isValid() {
        return valid;
    }

    public void setValid(boolean valid) {
        this.valid = valid;
    }

    public void setNextExecTime(Date nextExecTime) {
        this.nextExecTime = nextExecTime;
    }

}

 

 

自定义注解Job，可以加在所有需要定时任务的方法上

/**
 * 用来注解所有需要做定时任务的方法，name和cron是必填项，分别代表任务名称和任务的cron
 * 加了这个注解的方法在被加载到spring上下文时会被{@link TaskConfig}拦截并将所有的配置
 * 信息加载到数据库里面，接着初始化所有的定时任务。定时任务所有的对外暴露的API都被放到了
 * {@link com.xxx.service.JobService}里面了。
 *
 * @see com.xxx.service.JobService
 * @see TaskConfig
 * @author: bigtian
 * Date: 12-6-20
 * Time: 下午7:26
 */
@Target({ ElementType.TYPE, ElementType.METHOD })
@Retention(RetentionPolicy.RUNTIME)
public @interface Job {
    String name();

    String cron();

}