Spring-boot使用Quartz实现多线程调度任务

以你之姓@ 2022-06-05 09:45 944阅读 0赞

个人开发的基于内存的请求监控系统。欢迎star和一同参与。[https://github.com/standup-jb/tortoise][https_github.com_standup-jb_tortoise]

背景：项目的背景还原一下，项目的App需要去调用第三方的接口获取数据。我们要创建很多个任务（Task），每个Task都要去获取数据，但是每个Task的执行周期和频率不一样。所以使用多线程来解决。为每一个Task启动一个线程。然后循环不停的去获取数据。刚开始的时候打算自己写Thread然后对Thread进行管理。后面发现有一个框架Quartz可以很Spring-boot进行集成。非常好用，但是使用的过程中还是遇到了很多的问题。自己网上查资料找结果才将问题最后都解决了。在这个工程中深感查询资料的不容易，很多人写的博客表述不清楚。所以自己写一个希望可以帮助到更多的人。

因为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类了。并且将线程的处理逻辑分开。线程负责调度和逻辑的跳转。

[https_github.com_standup-jb_tortoise]: https://github.com/standup-jb/tortoise