因为Quartz的用法网上还是很多的很容易找到。难的是如何和Spring-boot结合起来是比较麻烦的。所以我可能重点会放在这个上面。
具体实现步骤:
1、首先在项目的Gradle里面添加Quartz 依赖(Maven的话就自己找一个依赖的源)
compile('org.quartz-scheduler:quartz:2.3.0')
一、实现观察者模式。实现一个Listener去观察Service的动向,减少耦合关系。
1、先让被观察者继承Java.util.Observable对象。表示可以拥有给观察者发送消息的能力
@Service public class OutlierServiceImpl extends Observable { @Override public void addTask(OutlierDetection outlierDetection){ MessageObject messageObject =new MessageObject(); messageObject.setOperate(OutlierOperate.ADD); messageObject.setOutlierDetection(outlierDetection); setChanged(); notifyObservers(messageObject); } @Override public void removeTask(String assetId,String outlierName){ setChanged(); notifyObservers(messageObject); } @Override public void updateTask(OutlierDetection outlierDetection){ MessageObject messageObject=new MessageObject(); messageObject.setOperate(OutlierOperate.UPDATE); messageObject.setOutlierDetection(outlierDetection); setChanged(); notifyObservers(messageObject); }
2、创建一个观察者实现Observer的接口。表示可以拥有观察的能力
@Component public class OutlierServiceListener implements Observer{ //private Observable ob; private String name; private Scheduler scheduler; @Autowired OutlierServiceImpl outlierServiceImpl; @Autowired @Qualifier("schedulerFactoryBean") private SchedulerFactoryBean schedulerFactoryBean; @Autowired JobFactory jobFactory; @PostConstruct public void registryOutlier(){ outlierServiceImpl.addObserver(this); scheduler = schedulerFactoryBean.getScheduler(); } @Override public void update(Observable o, java.lang.Object arg) { MessageObject messageObject =(MessageObject) arg; if(messageObject.getOperate().equals(OutlierOperate.ADD)){ OutlierDetection outlier = messageObject.getOutlierDetection(); addJob(outlier); }else if(messageObject.getOperate().equals(OutlierOperate.REMOVE)){ String outlierName = messageObject.getOutlierName(); String assetId = messageObject.getAssetId(); removeJob(assetId,outlierName); }else if(messageObject.getOperate().equals(OutlierOperate.UPDATE)){ OutlierDetection outlier = messageObject.getOutlierDetection(); updateJob(outlier); } }
3.建立观察者与被观察者之间的联系
通过Spring的注解将outlierService注入到outlierListener里面。然后设置Outlier的观察者是自己。这样观察者就可以与被观察者联系起来了。(注解PostConstruct的意思是在初始化完整个类后会执行这个函数)
4.被观察者发出通知消息。观察者接收到通知消息然后执行相应的操作。
绿色方框里面是我自己定义的MessageObject。用于在观察者和被观察者之前传递消息。被观察者先配置好消息,然后调用SetChanged()表示被观察者已经有了变化。然后调用notifyObservers(messageObject)函数。通知所有的观察者。并将消息发送过去。
二、在Listener里面利用Quartz实现周期性执行任务
Quartz简单介绍一下:Quartz是一个完全由java编写的开源作业调度框架、可以按照计划创建简单或者复杂的几十,几百。甚至数十万的作业数。
重点:如何和Spring集成(Spring-boot这种没有xml配置的项目)如何集成。Quartz的使用方法网上介绍挺多的,也挺详细的。
1、 创建一个自己的job继承于Quartz的job,Overrideexecute函数。这个函数就是每一个job执行一次所要做的功能。把一个线程应该做的工作就写在里面就好了。
public class OutlierJob implements Job{ @Autowired OutlierService outlierService; OutlierReport lastReport; @Override public void execute(JobExecutionContext context){ JobDataMap dataMap = context.getJobDetail().getJobDataMap(); String name = getName(dataMap); String assetId = getAssetId(dataMap); System.out.println("Outlier task 【"+name+"】is running"); OutlierDetection outlierDetection=outlierService.getOutlierDetectionCfg(assetId,name); if (lastReport==null){ lastReport = outlierService.findLastReportForTask(assetId,name); } lastReport = outlierService.runOutlierDetection(lastReport,outlierDetection); } public String getAssetId(JobDataMap dataMap){ return dataMap.getString("assetId"); } public String getName(JobDataMap dataMap){ return dataMap.getString("name"); } }
2、 创建Scheduler。给Scheduler添加job和job的trigger。然后将job加入到Scheduler里面。Job就会开始按照trigger设置的周期定时的执行了。
private void addJob(OutlierDetection outlierDetection){ String taskName = outlierDetection.getTaskName(); String assetId = outlierDetection.getAssetId(); int interval = new Long(outlierDetection.getInterval()).intValue(); JobDetail job = JobBuilder.newJob(OutlierJob.class) .withIdentity(taskName,assetId).build(); job.getJobDataMap().put("name",taskName); job.getJobDataMap().put("assetId",assetId); Trigger trigger = TriggerBuilder.newTrigger() .withIdentity(taskName,assetId) .withSchedule(SimpleScheduleBuilder.simpleSchedule().withIntervalInSeconds(interval).repeatForever()) .build(); try { scheduler.scheduleJob(job,trigger); scheduler.start(); } catch (SchedulerException e) { e.printStackTrace(); } }3、 停止一个Job。当job执行起来以后如何停止是应该需要考虑到的。不然就会一直执行了。先暂停Trigger。然后将Trigger移出Scheduler。最后deleteJob
private void removeJob(String assetId,String jobName){ System.out.println("Job【"+jobName+"】exit"); try { JobKey jobKey = JobKey.jobKey(jobName,assetId); TriggerKey triggerKey = TriggerKey.triggerKey(jobName,assetId); scheduler.pauseTrigger(triggerKey); scheduler.unscheduleJob(triggerKey); scheduler.deleteJob(jobKey); outlierServiceImpl.deleteOutlierDetectionTask(assetId,jobName); } catch (SchedulerException e) { e.printStackTrace(); } }
小结一下:在这里。我们已经完成了创建一个Listener去观察Service。当Service发出需要添加一个task的时候,Listener就能得到消息。然后在Quartz的Scheduler里面添加一个job。Job就会按照时间和周期定时的去执行我们之前写好在execute里面的代码。
问题:但是我们在job里面如何去调用由Spring管理的bean。这个问题就很麻烦了。因为我们每次添加一个task都是手动去new一个job。那么在new job的时候不是由Spring的容器在管理。所以在这种情况下,使用@autowired依赖注入Spring的bean类会出现注入不进来的情况。outlierService为空。这样我们就没有办法调用Spring的bean。经过各种查找,找到了解决这个问题的方法。4、创建一个JobFactory类继承于AdaptableJobFactory。注入AutowireCapableBeansFactory.这样就完成Spring对Job的注入功能。
@Component public class JobFactory extends AdaptableJobFactory{ @Autowired private AutowireCapableBeanFactory capableBeanFactory; @Override protected Object createJobInstance(TriggerFiredBundle bundle) throws Exception { //调用父类的方法 Object jobInstance = super.createJobInstance(bundle); //进行注入 capableBeanFactory.autowireBean(jobInstance); return jobInstance; } }5、 添加一个QuartzConfig类。对Scheduer进行重新配置设置Scheduler的JobFactory使用我们自己创建的JobFactory
@Configuration public class QuartzConfig { @Autowired private JobFactory jobFactory; @Bean(name = "schedulerFactoryBean") public SchedulerFactoryBean createSchedulerFactoryBean(){ SchedulerFactoryBean schedulerFactoryBean=new SchedulerFactoryBean(); schedulerFactoryBean.setOverwriteExistingJobs(true); schedulerFactoryBean.setJobFactory(jobFactory); return schedulerFactoryBean; } @Bean public JobDetailImpl createJobDetailsImpl(){ return new JobDetailImpl(); } }6、 构造Scheduler使用我们自己创建的SchedulerFactoryBean
@PostConstruct public void registryOutlier(){ outlierServiceImpl.addObserver(this); scheduler = schedulerFactoryBean.getScheduler(); }
到现在为止。这样在Job里面就可以注入Spring的bean类了。并且将线程的处理逻辑分开。线程负责调度和逻辑的跳转。