Quartz动态调控任务管理

男娘i 2022-06-07 00:26 141阅读 0赞

任务中存储的信息,我们创建一个实体类,来存储我们要传递到任务管理控制类中

package com.yssoft.fs.supervise.comm;
    
    import com.yssoft.fs.repository.entity.comm.Company;
    import com.yssoft.fs.repository.entity.comm.EarlyWarningTask;
    import com.yssoft.fs.repository.entity.comm.EarlyWarningWhitelist;
    import com.yssoft.fs.supervise.service.riskWarning.RiskWarningTaskService;
    import com.yssoft.fs.supervise.utils.SpringUtil;
    import org.apache.log4j.spi.LoggerFactory;
    import org.quartz.*;
    import org.quartz.impl.StdSchedulerFactory;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
    import org.springframework.context.ApplicationContext;
    import org.springframework.stereotype.Component;
    
    import java.util.ArrayList;
    import java.util.HashMap;
    import java.util.List;
    import java.util.Map;
    
    @Component
    public class QuartzJobFactory implements Job{ 
    
    
        private static Map<String, ScheduleJob> jobMap = new HashMap<String, ScheduleJob>();
    
        private static  SchedulerFactory schedulerFactoryBean=new StdSchedulerFactory();
    
        private  RiskWarningTaskService riskWarningTaskService=null;
        /** * 添加任务 * @param job */
        public static void addJob(ScheduleJob job) {
            try {
                jobMap.put(job.getJobGroup() + "_" + job.getJobName(), job);
                Scheduler scheduler = schedulerFactoryBean.getScheduler();
                //按新的cronExpression表达式构建一个新的trigger
                TriggerKey triggerKey = TriggerKey.triggerKey(job.getJobName(), job.getJobGroup());
                //获取trigger，即在spring配置文件中定义的 bean id="myTrigger"
                CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
                //不存在，创建一个
                if (null == trigger) {
                    JobDetail jobDetail = JobBuilder.newJob(QuartzJobFactory.class).withIdentity(job.getJobName(), job.getJobGroup()).build();
                    jobDetail.getJobDataMap().put("scheduleJob", job);
                    //表达式调度构建器
                    CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(job .getCronExpression());
                    //按新的cronExpression表达式构建一个新的trigger
                    trigger = TriggerBuilder.newTrigger().withIdentity(job.getJobName(), job.getJobGroup()).withSchedule(scheduleBuilder).build();
                    scheduler.start();
                    scheduler.scheduleJob(jobDetail, trigger);
                } else {
                    // Trigger已存在，那么更新相应的定时设置
                    //表达式调度构建器
                    CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(job  .getCronExpression());
                    //按新的cronExpression表达式重新构建trigger
                    trigger = trigger.getTriggerBuilder().withIdentity(triggerKey) .withSchedule(scheduleBuilder).build();
                    //按新的trigger重新设置job执行
                    scheduler.rescheduleJob(triggerKey, trigger);
                }
            } catch (SchedulerException e) {
                e.printStackTrace();
            }
        }
    
        /** * map中的任务重新装配 */
        public  static void resetJob(){
            try {
                //schedulerFactoryBean 由spring创建注入
                Scheduler scheduler = schedulerFactoryBean.getScheduler();
                //这里获取任务信息数据
                List<ScheduleJob> jobList = getAllJob();
                for (ScheduleJob job : jobList) {
                    TriggerKey triggerKey = TriggerKey.triggerKey(job.getJobName(), job.getJobGroup());
                    //获取trigger，即在spring配置文件中定义的 bean id="myTrigger"
                    CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
                    //不存在，创建一个
                    if (null == trigger) {
                        JobDetail jobDetail = JobBuilder.newJob(QuartzJobFactory.class).withIdentity(job.getJobName(), job.getJobGroup()).build();
                        jobDetail.getJobDataMap().put("scheduleJob", job);
                        //表达式调度构建器
                        CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(job .getCronExpression());
                        //按新的cronExpression表达式构建一个新的trigger
                        trigger = TriggerBuilder.newTrigger().withIdentity(job.getJobName(), job.getJobGroup()).withSchedule(scheduleBuilder).build();
                        scheduler.start();
                        scheduler.scheduleJob(jobDetail, trigger);
                    } else {
                        // Trigger已存在，那么更新相应的定时设置
                        //表达式调度构建器
                        CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(job  .getCronExpression());
                        //按新的cronExpression表达式重新构建trigger
                        trigger = trigger.getTriggerBuilder().withIdentity(triggerKey) .withSchedule(scheduleBuilder).build();
                        //按新的trigger重新设置job执行
                        scheduler.rescheduleJob(triggerKey, trigger);
                    }
                }
            } catch (SchedulerException e) {
                e.printStackTrace();
            }
        }
        /** * 获取任务 * */
        public  static ScheduleJob getJob(String jobId) {
            return (ScheduleJob) jobMap.get(jobId);
        }
        /** * 获取所有任务 * */
        public  static List<ScheduleJob> getAllJob() {
            List jobList = new ArrayList(jobMap.size());
            for (Map.Entry entry : jobMap.entrySet()) {
                jobList.add(entry.getValue());
            }
            return jobList;
        }
    
        /** * 运行中的任务 */
        public   static  List<ScheduleJob> isRunTask(){
            try {
                Scheduler scheduler = schedulerFactoryBean.getScheduler();
                List<JobExecutionContext> executingJobs = scheduler.getCurrentlyExecutingJobs();
                List<ScheduleJob> jobList = new ArrayList<ScheduleJob>(executingJobs.size());
                for (JobExecutionContext executingJob : executingJobs) {
                    ScheduleJob job = new ScheduleJob();
                    JobDetail jobDetail = executingJob.getJobDetail();
                    JobKey jobKey = jobDetail.getKey();
                    Trigger trigger = executingJob.getTrigger();
                    job.setJobName(jobKey.getName());
                    job.setJobGroup(jobKey.getGroup());
                    job.setDesc("触发器:" + trigger.getKey());
                    Trigger.TriggerState triggerState = scheduler.getTriggerState(trigger.getKey());
                    //job.setJobStatus(triggerState.name());
                    if (trigger instanceof CronTrigger) {
                        CronTrigger cronTrigger = (CronTrigger) trigger;
                        String cronExpression = cronTrigger.getCronExpression();
                        job.setCronExpression(cronExpression);
                    }
                    jobList.add(job);
                }
                return jobList;
            } catch (SchedulerException e) {
                e.printStackTrace();
                return null;
            }
        }
    
        /** * 暂停任务 */
        public static  void pauseTask(String jobName,String jobGroup){
            try {
                Scheduler scheduler = schedulerFactoryBean.getScheduler();
                JobKey jobKey = JobKey.jobKey(jobName, jobGroup);
                scheduler.pauseJob(jobKey);
            } catch (SchedulerException e) {
                e.printStackTrace();
            }
        }
    
        /** * 恢复任务 */
        public static  void resumeTask(String jobName,String jobGroup){
            try {
                Scheduler scheduler = schedulerFactoryBean.getScheduler();
                JobKey jobKey = JobKey.jobKey(jobName, jobGroup);
                scheduler.resumeJob(jobKey);
            } catch (SchedulerException e) {
                e.printStackTrace();
            }
        }
    
        /** * 删除任务 */
        public static  void removeTask(String jobName,String jobGroup){
            try {
                Scheduler scheduler = schedulerFactoryBean.getScheduler();
                JobKey jobKey = JobKey.jobKey(jobName, jobGroup);
                scheduler.deleteJob(jobKey);
            } catch (SchedulerException e) {
                e.printStackTrace();
            }
        }
    
        /** * 立即运行任务 */
        public static  void triggerTask(String jobName,String jobGroup){
            try {
                Scheduler scheduler =schedulerFactoryBean.getScheduler();
                JobKey jobKey = JobKey.jobKey(jobName, jobGroup);
                scheduler.start();
                //scheduler.triggerJob(jobKey);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        /** * 立即更新任务设置的时间 */
        public  static void resetTaskTime(ScheduleJob job){
            try {
                Scheduler scheduler = schedulerFactoryBean.getScheduler();
                TriggerKey triggerKey = TriggerKey.triggerKey(job.getJobName(), job.getJobGroup());
                //获取trigger，即在spring配置文件中定义的 bean id="myTrigger"
                CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
                //表达式调度构建器
                CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(job.getCronExpression());
                //按新的cronExpression表达式重新构建trigger
                trigger = trigger.getTriggerBuilder().withIdentity(triggerKey)  .withSchedule(scheduleBuilder).build();
                //按新的trigger重新设置job执行
                scheduler.rescheduleJob(triggerKey, trigger);
            } catch ( Exception e) {
                e.printStackTrace();
            }
        }
    
        /** * 任务中执行的内容 * @param jobExecutionContext * @throws JobExecutionException */
        @Override
        public void execute(JobExecutionContext jobExecutionContext) throws JobExecutionException {
            try {
                //获取Spring中的上下文
                ApplicationContext appCtx = SpringUtil.getApplicationContext();
                riskWarningTaskService= (RiskWarningTaskService)appCtx.getBean("riskWarningTaskService");
                JobDetail c = jobExecutionContext.getJobDetail();//获取工作队列中的任务
                ScheduleJob scheduleJob = (ScheduleJob)c.getJobDataMap().get("scheduleJob");
                Long takeId = scheduleJob.getEarlyWarningTaskId();//在任务实体中获取任务实体类
                //白名单中的企业
                List<Map<String,Object>> whiteList = riskWarningTaskService.findListBySQL(new SQLGenerator() {
                    @Override
                    public Map<String, Object[]> generate() {
                        StringBuffer sb=new StringBuffer();
                        Map<String,Object[]> result=new HashMap<>();
                        sb.append("select t.title,w.creator from ts_early_warning_whitelist w left join ts_early_warning_task t on t.id =w.ew_task_id and t.delete_flag=0 where w.delete_flag=0 ");
                        result.put(sb.toString(),new Object[]{});
                        return result;
                    }
                },null);
    
                List<Map<String,Object>> companyAll=null;
                if("经营许可证检查".equals(scheduleJob.getTaskType())){
                     companyAll =riskWarningTaskService.findListBySQL(new SQLGenerator() {
                        @Override
                        public Map<String, Object[]> generate() {
                            StringBuffer sb=new StringBuffer();
                            Map<String,Object[]> result=new HashMap<>();
                            sb.append("select * from tc_company c where c.delete_flag=0 and ( " );
                            sb.append("SELECT TO_DATE(to_char(c.license_expire_date,'yyyy-MM-dd'),'YYYY-MM-DD') -TO_DATE(to_char(sysdate,'yyyy-MM-dd'),'YYYY-MM-DD') FROM DUAL ");
                            sb.append(      ") <=20");
                            result.put(sb.toString(),new Object[]{});
                            return result;
                        }
                    },null);//查询不合格的企业
    
                }else if("最近三天无消洗检查".equals(scheduleJob.getTaskType())){
                    companyAll =riskWarningTaskService.findListBySQL(new SQLGenerator() {
                        @Override
                        public Map<String, Object[]> generate() {
                            StringBuffer sb=new StringBuffer();
                            Map<String,Object[]> result=new HashMap<>();
                            sb.append("select c.* from tc_company c ");
                            sb.append("left join tb_tableware_disf_record t on t.company_id=c.id and t.delete_flag=0 ");
                            sb.append("where c.delete_flag=0 and ( ");
                            sb.append("SELECT TO_DATE(to_char(sysdate,'yyyy-MM-dd'),'YYYY-MM-DD')-TO_DATE(to_char( t.end_time,'yyyy-MM-dd'),'YYYY-MM-DD') FROM DUAL ) >=3 " );
                            result.put(sb.toString(),new Object[]{});
                            return result;
                        }
                    },null);//查询不合格的企业
    
                }else if("最近三天无垃圾处理检查".equals(scheduleJob.getTaskType())){
                    companyAll =riskWarningTaskService.findListBySQL(new SQLGenerator() {
                        @Override
                        public Map<String, Object[]> generate() {
                            StringBuffer sb=new StringBuffer();
                            Map<String,Object[]> result=new HashMap<>();
                            sb.append("select * from tc_company c left join tb_garbage_collection g on g.company_id=c.id and g.delete_flag=0");
                            sb.append("where c.delete_flag=0 and (");
                            sb.append("SELECT TO_DATE(to_char(sysdate,'yyyy-MM-dd'),'YYYY-MM-DD')-TO_DATE(to_char( g.recovery_time,'yyyy-MM-dd'),'YYYY-MM-DD') FROM DUAL");
                            sb.append(" ) >=3");
                            result.put(sb.toString(),new Object[]{});
                            return result;
                        }
                    },null);//查询不合格的企业
                }
                //根据任务类型,检查企业不同的内容,查询白名单中这个企业这个检查项是否在白名单中,如果在就不检查
                String taskType = scheduleJob.getTaskType();//获取任务类型 0,经营许可证检查 1最近三天无消洗检查 2最近三天无垃圾处理检查
                for (Map<String,Object> company : companyAll) {
                    if(whiteList.size()>0){
                        for (Map<String,Object> whitelist : whiteList) {
                            String whiteListTitle= whitelist.get("title").toString();
                            if(!taskType.equals(whiteListTitle) && company.get("id").toString().equals(whitelist.get("creator").toString())){
                                riskWarningTaskService.createRiskWarningInfo(takeId,Long.parseLong(company.get("id").toString()));
                            }
                        }
                    }else{
                        riskWarningTaskService.createRiskWarningInfo(takeId,Long.parseLong(company.get("id").toString()));
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    
        public RiskWarningTaskService getRiskWarningTaskService() {
            return riskWarningTaskService;
        }
    
        public void setRiskWarningTaskService(RiskWarningTaskService riskWarningTaskService) {
            this.riskWarningTaskService = riskWarningTaskService;
        }
    
    }

任务管理控制类

下面是调用创建任务的方法

public void editState(String tast, String id,String title) {
            EarlyWarningTask eat = earlyWarningTaskDao.findOne(Long.parseLong(id));
            Date date=eat.getTaskTime();
            SimpleDateFormat sdf=new SimpleDateFormat("HH:mm");
            String time = sdf.format(date);
            String[] times = time.split(":");
            ScheduleJob scheduleJob=new ScheduleJob();
           // System.out.println("0 "+times[1]+" "+times[0]+" 1-3 * ? *" );
          // scheduleJob.setCronExpression("0 "+times[1]+" "+times[0]+" 1-3 * ? *");
            scheduleJob.setCronExpression("*/5 * * * * ?");//设置一秒进行运行一次任务
            scheduleJob.setEarlyWarningTaskId(Long.parseLong(id));
            if("0".equals(tast)){
                //打开
                eat.setStat("1");
                if("经营许可证检查".equals(title)){
                    scheduleJob.setJobId("licence");
                    scheduleJob.setJobName("licenceName");
                    scheduleJob.setJobGroup("inspection");
                    scheduleJob.setJobStatus("1");
                    scheduleJob.setDesc("经营许可证定时检查");
                    scheduleJob.setTaskType("经营许可证检查");
                    QuartzJobFactory.addJob(scheduleJob);
                    QuartzJobFactory.triggerTask("licenceName","inspection");
                }
                if("最近三天无消洗检查".equals(title)){
                    scheduleJob.setJobId("disinfection");
                    scheduleJob.setJobName("disinfectionName");
                    scheduleJob.setJobGroup("inspection");
                    scheduleJob.setJobStatus("1");
                    scheduleJob.setDesc("最近三天无消洗检查");
                    scheduleJob.setTaskType("最近三天无消洗检查");
                    QuartzJobFactory.addJob(scheduleJob);
                    QuartzJobFactory.triggerTask("disinfectionName","inspection");
                }
                if("最近三天无垃圾处理检查".equals(title)){
                    scheduleJob.setJobId("garbage");
                    scheduleJob.setJobName("garbageName");
                    scheduleJob.setJobGroup("inspection");
                    scheduleJob.setJobStatus("1");
                    scheduleJob.setDesc("最近三天无垃圾处理检查");
                    scheduleJob.setTaskType("最近三天无垃圾处理检查");
                    QuartzJobFactory.addJob(scheduleJob);
                    QuartzJobFactory.triggerTask("garbageName","inspection");
                }
            }else{
                //关闭
                eat.setStat("0");
                if("经营许可证检查".equals(title)) {
                    QuartzJobFactory.removeTask("licenceName","inspection");
                }
                if("最近三天无消洗检查".equals(title)) {
                    QuartzJobFactory.removeTask("disinfectionName","inspection");
                }
                if("最近三天无垃圾处理检查".equals(title)) {
                    QuartzJobFactory.removeTask("garbageName","inspection");
                }
            }
            earlyWarningTaskDao.save(eat);
        }