鉴于对Spring实现的@Scheduled的调度和SchedulerFactoryBean的研究发现,基于Spring的调度封装虽满足了大多需求,但为了简化使用方式使得Job并不容易得到控制,导致开发对Job的控制和运维成本上升;下面是本人基于Quartz和Spring及Annotation开发的单机版调度配置,满足单机调度的大部分需求和管理、运维操作并解放对配置文件的繁琐操作;
功能点 | Spring @Scheduled | 自定义@SchedulerJob |
---|---|---|
可控制 | 否 | 是 |
可运维 | 否 | 是 |
可页面化 | 否 | 是 |
可统一跟踪业务状态 | 否 | 是 |
可统一跟踪调度状态 | 否 | 是 |
支持cron表达式 | 是 | 是 |
支持类似ScheduledExecutorService的定时调度 | 是 | 否 |
@Slf4j(topic = "dynamic-datasource") @Component public class DetectJob { /** * 作业配置 value=作业名,group=作业所属组,init=true为容器创建完毕时立即触发 */ @SchedulerJob(value = "detectDataSource",cron = "${cron.detect.data.source}",group = "dynamic-datasource", descrption="动态数据源切换",init = true) public void detectDataSource(){ log.info(VariableUtils.join(SystemConstants.Symbol.DELIMITER,"dynamic-datasource","detectDataSource")); } } 复制代码
##cron表达式 cron.detect.data.source=1 * * * * ? 复制代码
通过页面可对作业进行统一的监控和管理(触发、暂停、恢复、动态添加、参数下发)及报警等操作;
简要列出以下功能点:
@Target(ElementType.METHOD) @Retention(RetentionPolicy.RUNTIME) public @interface SchedulerJob { /** * 作业名 * @return */ String value(); /** * 表达式 * @return */ String cron(); /** * 是否初始化时立即执行 * @return */ boolean init() default false; /** * 是否人为控制 * @return */ boolean control() default false; /** * 所属组 * @return */ String group() default "default"; /** * 作业描述 * @return */ String descrption() default ""; /** * 作业执行器 * @return */ Class jobClass() default SimpleJob.class; } @Target({ElementType.METHOD,ElementType.ANNOTATION_TYPE}) @Retention(RetentionPolicy.RUNTIME) public @interface SchedulerJobs { /** * 注解集 * @return */ SchedulerJob[] value(); } 复制代码
@Slf4j @Configuration public class SchedulerBean implements InitializingBean, DisposableBean { private Scheduler scheduler; @Value("#{schdulerProperties['quartz.thread.count']}") private String threadCount; @Override public void destroy() throws Exception { scheduler.shutdown(); } @Override public void afterPropertiesSet() throws Exception { createScheduler(); } /** * 创建调度 * @throws SchedulerException */ public void createScheduler() throws SchedulerException { StdSchedulerFactory factory = new StdSchedulerFactory(); factory.initialize(getBaseQuartzProperties()); this.scheduler = factory.getScheduler(); } /** * 作业配置 * @return */ private Properties getBaseQuartzProperties() { Properties result = new Properties(); result.put("org.quartz.threadPool.class", org.quartz.simpl.SimpleThreadPool.class.getName()); result.put("org.quartz.threadPool.threadCount", threadCount); result.put("org.quartz.scheduler.threadName", "baiyunpeng-scheduler"); result.put("org.quartz.scheduler.instanceName", "baiyunpeng-scheduler"); result.put("org.quartz.jobStore.misfireThreshold", "1"); return result; } /** * 创建作业 * @param jobParam * @throws SchedulerException */ public void createJob(JobParam jobParam) throws SchedulerException { SchedulerJob schedulerJob = jobParam.getSchedulerJob(); JobDetail jobDetail = JobBuilder.newJob(schedulerJob.jobClass()) .withIdentity(jobParam.getJobKey()) .withDescription(jobParam.getJobKey().getName()) .build(); addJobDataMap(jobDetail,jobParam.getTarget(),jobParam.getTargetMethod()); this.scheduler.scheduleJob(jobDetail,createTrigger(jobParam.getJobKey(),jobParam.getCron())); } /** * 创建触发器 * @param jobKey * @param cron * @return */ private Trigger createTrigger(JobKey jobKey, String cron) { return TriggerBuilder.newTrigger().withIdentity(jobKey.getName(),jobKey.getGroup()) .withSchedule(CronScheduleBuilder.cronSchedule(cron) .withMisfireHandlingInstructionDoNothing()).build(); } /** * 添加作业map * @param jobDetail * @param target * @param targetMethod */ private void addJobDataMap(JobDetail jobDetail, Object target, Method targetMethod) { JobDataMap jobDataMap = jobDetail.getJobDataMap(); jobDataMap.put("executeJob",target); jobDataMap.put("executeMethod",targetMethod); } public Scheduler getScheduler() { return scheduler; } public void start() throws SchedulerException { this.scheduler.start(); } } 复制代码
/** * 作业抽象类 * @author baiyunpeng */ public abstract class ExecuteJob implements Job { protected Object executeJob; protected Method executeMethod; public void setExecuteJob(Object executeJob) { this.executeJob = executeJob; } public void setExecuteMethod(Method executeMethod) { this.executeMethod = executeMethod; } } /** * 非并发执行 * @author baiyunpeng */ @Slf4j public class SimpleJob extends ExecuteJob { @Override public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException { try { executeMethod.invoke(executeJob); } catch (IllegalAccessException | InvocationTargetException e) { log.error(VariableUtils.join(SystemConstants.Symbol.DELIMITER,this.getClass().getName(), ExceptionUtils.getRootCauseMessage(e))); } } } /** * 可并发执行 * @author baiyunpeng */ @Slf4j @PersistJobDataAfterExecution @DisallowConcurrentExecution public class ConcurrentJob extends ExecuteJob{ @Override public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException { try { executeMethod.invoke(executeJob); } catch (IllegalAccessException | InvocationTargetException e) { log.error(VariableUtils.join(SystemConstants.Symbol.DELIMITER,this.getClass().getName(), ExceptionUtils.getRootCauseMessage(e))); } } } 复制代码
/** * 作业配置解析 * @param scheduled * @param method * @param bean */ protected void processScheduled(SchedulerJob scheduled, Method method, Object bean) { Method invocableMethod = AopUtils.selectInvocableMethod(method, bean.getClass()); String cron = scheduled.cron(); if(StringUtils.hasText(cron)){ if(Objects.nonNull(this.embeddedValueResolver)){ cron = this.embeddedValueResolver.resolveStringValue(cron); } jobParams.add(new JobParam(scheduled,bean,invocableMethod,new JobKey(scheduled.value(),scheduled.group()),cron)); } } /** * 作业初始化 */ private void finishRegister() { if(Objects.isNull(this.schedulerBean)){ SchedulerBean schedulerBean = beanFactory.getBean(SCHEDULER_BEAN, SchedulerBean.class); AssertUtil.assertNull(schedulerBean, SystemErrorCode.NS000000,"the scheduler bean init error"); this.schedulerBean = schedulerBean; try { jobParams.parallelStream().forEach(jobParam -> { try { this.schedulerBean.createJob(jobParam); SchedulerJob schedulerJob = jobParam.getSchedulerJob(); if(!schedulerJob.control()){ if (schedulerJob.init()){ this.schedulerBean.getScheduler().triggerJob(jobParam.getJobKey()); } }else { this.schedulerBean.getScheduler().pauseJob(jobParam.getJobKey()); } } catch (SchedulerException e) { log.error(VariableUtils.join(SystemConstants.Symbol.DELIMITER,"the scheduler job init error", ExceptionUtils.getRootCauseMessage(e))); System.exit(1); } }); schedulerBean.start(); }catch (Exception e){ log.error(VariableUtils.join(SystemConstants.Symbol.DELIMITER,"the scheduler job init error", ExceptionUtils.getRootCauseMessage(e))); System.exit(1); } } } 复制代码