欢迎您访问程序员文章站本站旨在为大家提供分享程序员计算机编程知识!
您现在的位置是: 首页

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

程序员文章站 2022-03-09 08:33:18
...

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();

}