spring boot整合quartz存储到数据库

首先是pom.xml依赖

 <!-- MQTT依赖 -->
        <dependency>
            <groupId>org.eclipse.paho</groupId>
            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
            <version>1.0.2</version>
        </dependency>
        
             <!--quartz依赖-->
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-quartz</artifactId>
        </dependency>

就一个controller就可以，剩下的自己去写存储数据库的方法

package com.family.Switch.controller;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;
import java.util.concurrent.TimeUnit;

import org.jboss.logging.Logger;
import org.quartz.CronScheduleBuilder;
import org.quartz.CronTrigger;
import org.quartz.Job;
import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.TriggerKey;
import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.scheduling.Trigger;
import org.springframework.stereotype.Component;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import com.family.Switch.job.CrawlNewsJob;
import com.family.Switch.model.ScheduledTask;
import com.family.Switch.service.IScheduledTaskService;
import com.family.util.mqtt.ServerMQTT;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
 

@Component
@Api(value = "手动定时开关",description ="手动定时开关")
@RestController
@RequestMapping(value = "/trigger")
public class DynamicTaskController implements Job {

    @Autowired
    private ServerMQTT servermqtt;
    
    @Autowired
    private IScheduledTaskService ischeduledtaskservice;
    
    @Autowired
    private Scheduler scheduler;
    
    public final Logger log = Logger.getLogger(this.getClass());
    
    //全局变量
    //private String cronStr = "0/5 * * * * ?";
 
    @ApiOperation(value="添加定时", notes="添加定时")
    @RequestMapping(value = "/startCronadd", method = RequestMethod.POST)
    @ApiImplicitParams({
        @ApiImplicitParam(name="cronStr",value="corn值",dataType="string", paramType = "query"),
        @ApiImplicitParam(name="news",value="传给传感器的参数控制开关",dataType="string", paramType = "query"),
        @ApiImplicitParam(name="logger",value="log",dataType="string", paramType = "query"),
        @ApiImplicitParam(name="key",value="唯一的",dataType="string", paramType = "query"),
        @ApiImplicitParam(name="group",value="唯一的",dataType="string", paramType = "query")
        })
    @ResponseBody
    public String addcon(@RequestParam("cronStr") String cronStr,@RequestParam("news") String news,@RequestParam("logger") String logger,@RequestParam("key") String key,@RequestParam("group") String group) {
        
        
        try {
            //Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
//            scheduler.start();
            // 启动
            if (!scheduler.isShutdown()) {
                
                ScheduledTask scheduledtask=new ScheduledTask();
                
                SimpleDateFormat df= new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                
                scheduledtask.setCreateTime(df.format(new Date()));
                
                scheduledtask.setTaskKey(key);
                
                scheduledtask.setTaskDesc(group);
                
                scheduledtask.setTaskCron(cronStr);
                
                scheduledtask.setInitStartFlag(1);
                
                int selectcount = ischeduledtaskservice.selectcount(scheduledtask);
                
                if (selectcount>0) {
                    log.info("已有数据");
                }else {
                        scheduler.start();
                        System.out.println("Quartz Start !");
                        //具体任务
                        JobDetail job = JobBuilder.newJob(DynamicTaskController.class).withIdentity(key,group).build();

                        //触发器
                       // SimpleScheduleBuilder simpleScheduleBuilder = SimpleScheduleBuilder.simpleSchedule().withIntervalInMinutes(60).repeatForever();
                        Trigger trigger = TriggerBuilder.newTrigger().withIdentity(key,group).startNow().withSchedule(CronScheduleBuilder.cronSchedule(cronStr)).build();

                        scheduler.scheduleJob(job,trigger);
                    
                    
                    ischeduledtaskservice.insert(scheduledtask);
                    log.info("添加完毕");
                }
                
            }
            //睡眠
            TimeUnit.MINUTES.sleep(1);
            scheduler.shutdown(true);
            
            if(scheduler.isShutdown()) {
                  ScheduledTask scheduledtask=new ScheduledTask();
                  SimpleDateFormat df= new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                  scheduledtask.setUpdateTime(df.format(new Date()));
                  scheduledtask.setInitStartFlag(0);
                  scheduledtask.setTaskCron(cronStr);
                int update = ischeduledtaskservice.update(scheduledtask);
                if (update>0) {
                    log.info("状态修改完成");
                }
            }
            System.out.println("scheduler shutdown ! ");
        }catch(Exception e){
            e.printStackTrace();
      }
        
        
//        future = threadPoolTaskScheduler.schedule(new MyRunnable(), new Trigger(){
//
//
//            @Override
//            public Date nextExecutionTime(TriggerContext triggerContext) {
//                try {
//                    servermqtt.mqtt(news, "test");
//                } catch (MqttException e) {
//                    // TODO Auto-generated catch block
//                    e.printStackTrace();
//                }
//                return new CronTrigger(cronStr).nextExecutionTime(triggerContext);
//            }
//
//    });
        
        return "startCronadd";
    }
    
    
    @ApiOperation(value="查询所有定时任务", notes="查询所有定时任务")
    @RequestMapping(value = "/stopCronselect", method = RequestMethod.GET)
    public List stopCronselect() {
       return ischeduledtaskservice.select();
    }
    
    @ApiOperation(value="更新定时", notes="更新定时")
    @RequestMapping(value = "/updatecron", method = RequestMethod.POST)
    @ApiImplicitParams({
        @ApiImplicitParam(name="cronStr",value="corn值",dataType="string", paramType = "query"),
        @ApiImplicitParam(name="key",value="log",dataType="string", paramType = "query"),
        @ApiImplicitParam(name="group",value="log",dataType="string", paramType = "query")
        })
    @ResponseBody
    public String updatecron(@RequestParam("cronStr") String cronStr,@RequestParam("key") String key,@RequestParam("group") String group) throws SchedulerException {
        
         TriggerKey triggerKey = TriggerKey.triggerKey(key, group);
         
         CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
  
         CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(cronStr);
  
         trigger = trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(scheduleBuilder).build();
  
         scheduler.rescheduleJob(triggerKey, trigger);
         
         ScheduledTask scheduledtask=new ScheduledTask();
             SimpleDateFormat df= new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
             scheduledtask.setUpdateTime(df.format(new Date()));
             scheduledtask.setInitStartFlag(1);
             scheduledtask.setTaskCron(cronStr);
           int update = ischeduledtaskservice.update(scheduledtask);
           if (update>0) {
                log.info("状态修改完成");
            }
        
        return "updatecron";
    }
    
    

//    @RequestMapping("/startCron")
//    public String startCron() {
//
//       future = threadPoolTaskScheduler.schedule(new MyRunnable(), new CronTrigger("0/5 * * * * ?"));
//        
//       System.out.println("DynamicTask.startCron()");
//
//       return "startCron";
//
//    }
    
    @ApiOperation(value="删除定时", notes="删除定时")
    @RequestMapping(value = "/deletestopCron", method = RequestMethod.POST)
    @ApiImplicitParams({
        @ApiImplicitParam(name="key",value="log",dataType="string", paramType = "query"),
        @ApiImplicitParam(name="group",value="log",dataType="string", paramType = "query")
        })
    @ResponseBody
    public String deletestopCron(@RequestParam("key") String key,@RequestParam("group") String group) throws SchedulerException {
        JobKey jobKey = JobKey.jobKey(key,group);
        scheduler.deleteJob(jobKey);
        ScheduledTask scheduledtask=new ScheduledTask();
        scheduledtask.setTaskKey(key);
        int delete = ischeduledtaskservice.delete(scheduledtask);
        if(scheduler.isShutdown() && delete>0) {
            log.info("已关闭");
        }
        
       return "deletestopCron";
    }


        @Override
        public void execute(JobExecutionContext context) throws JobExecutionException {
            System.out.println("111111111111111111111");
        }

 

//    @ApiOperation(value="停止", notes="停止")
//    @RequestMapping(value = "/stopCron", method = RequestMethod.GET)
//    public String stopCron() {
//       if (future != null) {
//           future.cancel(true);
//       }
//       System.out.println("DynamicTask.stopCron()");
//       return "stopCron";
//    }

 

//    @RequestMapping("/changeCron10")
//
//    public String startCron10() {
//
//       stopCron();// 先停止，在开启.
//
//       future = threadPoolTaskScheduler.schedule(new MyRunnable(), new CronTrigger("0/2 * * * * ?"));
//
//       System.out.println("DynamicTask.startCron10()");
//
//       return "changeCron10";
//
//    }
    
    

//    private class MyRunnable implements Runnable {
//
//       @Override
//
//       public void run() {
//
//           System.out.println("DynamicTask.MyRunnable.run()，" + new Date());
//
//       }
//
//    }

 

}

目前又需求改了，又写出一版一个接口实现添加和修改的job和数据库实时同步

package com.family.Switch.controller;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.jboss.logging.Logger;
import org.quartz.CronScheduleBuilder;
import org.quartz.CronTrigger;
import org.quartz.JobBuilder;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.TriggerKey;
import org.springframework.beans.factory.annotation.Autowired;
//import org.springframework.scheduling.Trigger;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.ResponseBody;
import org.springframework.web.bind.annotation.RestController;

import com.family.Switch.job.EndNewsJob;
import com.family.Switch.job.StaNewsJob;
import com.family.Switch.model.ScheduledTask;
import com.family.Switch.service.IScheduledTaskService;
import com.family.query.model.GreenhouseDevice;
import com.family.query.service.QueryService;

import io.swagger.annotations.Api;
import io.swagger.annotations.ApiImplicitParam;
import io.swagger.annotations.ApiImplicitParams;
import io.swagger.annotations.ApiOperation;
 

@CrossOrigin
@Api(value = "手动定时开关",description ="手动定时开关")
@RestController
@RequestMapping(value = "/trigger")
public class DynamicTaskController {

    @Autowired
    private IScheduledTaskService ischeduledtaskservice;
    
    @Autowired
    private Scheduler scheduler;
    
    @Autowired
    private QueryService queryservice;
    
    public final Logger log = Logger.getLogger(this.getClass());
 
    @ApiOperation(value="添加定时", notes="添加定时")
    @RequestMapping(value = "/startCronadd", method = RequestMethod.POST)
    @ApiImplicitParams({
        @ApiImplicitParam(name="map",value="corn值",dataType="Map", paramType = "add")
        })
    @ResponseBody
    public Map<String,String> addcon(@RequestBody Map<String,String> map) {
        Map<String,String> ds=new HashMap<String, String>();
        try {
            //Scheduler scheduler = StdSchedulerFactory.getDefaultScheduler();
            // 启动
            if (!scheduler.isShutdown()) {
                
                ScheduledTask scheduledtask=new ScheduledTask();
                
                SimpleDateFormat df= new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                
                scheduledtask.setCreateTime(df.format(new Date()));
                
                scheduledtask.setTaskKeySta(map.get("taskKeySta"));
                
                scheduledtask.setTaskKeyEnd(map.get("taskKeyEnd"));
                
                scheduledtask.setTaskDescSta(map.get("taskDescSta"));
                
                scheduledtask.setTaskDescEnd(map.get("taskDescEnd"));
                
                scheduledtask.setTaskCronSta(map.get("taskCronSta"));
                
                scheduledtask.setTaskCronEnd(map.get("taskCronEnd"));
                
                scheduledtask.setStaTime(map.get("staTime"));
                
                scheduledtask.setEndTime(map.get("endTime"));
                
                scheduledtask.setInitStartFlag(1);
                
                int selectcount = ischeduledtaskservice.selectcount(scheduledtask);
                
                if (selectcount>0) {
                    log.info("已有数据---执行修改");
                     //修改开始
                     TriggerKey triggerKeysta = TriggerKey.triggerKey(map.get("taskKeySta"), map.get("taskDescSta"));
                       CronTrigger triggersta = (CronTrigger) scheduler.getTrigger(triggerKeysta);
                       CronScheduleBuilder scheduleBuildersta = CronScheduleBuilder.cronSchedule(map.get("taskCronSta")).withMisfireHandlingInstructionDoNothing();
                       triggersta = triggersta.getTriggerBuilder().withIdentity(triggerKeysta).withSchedule(scheduleBuildersta).build();
                       scheduler.rescheduleJob(triggerKeysta, triggersta);
                     //修改结束
                       TriggerKey triggerKeyend = TriggerKey.triggerKey(map.get("taskKeyEnd"), map.get("taskDescEnd"));
                     CronTrigger triggerend = (CronTrigger) scheduler.getTrigger(triggerKeyend);
                     //可通过下面来进行设置不立即执行，按照corn表达式来执行 CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(job.getCronExpression()).withMisfireHandlingInstructionDoNothing();
                     CronScheduleBuilder scheduleBuilderend = CronScheduleBuilder.cronSchedule(map.get("taskCronEnd")).withMisfireHandlingInstructionDoNothing();
                     triggerend = triggerend.getTriggerBuilder().withIdentity(triggerKeyend).withSchedule(scheduleBuilderend).build();
                     scheduler.rescheduleJob(triggerKeyend, triggerend);
                       
                   int update = ischeduledtaskservice.update(scheduledtask);
                   if (update>0) {
                       log.info(df.format(new Date()));
                       log.info("定时修改完成");
                       ds.put("rtnCode", "1");
                    ds.put("rtnMsg","修改成功启动！");
                       return ds;
                   }
                }else {
                        scheduler.start();
                        log.info("Quartz Start !");
                        //具体开始任务
                        StaNewsJob.sta=map.get("sta");
                        StaNewsJob.gatelog=map.get("gatelog");
                        JobDetail jobsta = JobBuilder.newJob(StaNewsJob.class).withIdentity(map.get("taskKeySta"),map.get("taskDescSta")).build();
                        //触发器
                        // SimpleScheduleBuilder simpleScheduleBuilder = SimpleScheduleBuilder.simpleSchedule().withIntervalInMinutes(starttime-endtime).withRepeatCount(1).repeatForever();
                        Trigger triggersta = TriggerBuilder.newTrigger().withIdentity(map.get("taskKeySta"),map.get("taskDescSta")).startNow().withSchedule(CronScheduleBuilder.cronSchedule(map.get("taskCronSta"))).build();
                        scheduler.scheduleJob(jobsta,triggersta);
                        
                        
                        //具体结束任务
                        EndNewsJob.end =map.get("end");
                        EndNewsJob.gatelog = map.get("gatelog");
                        JobDetail jobend = JobBuilder.newJob(EndNewsJob.class).withIdentity(map.get("taskKeyEnd"),map.get("taskDescEnd")).build();
                        //触发器-毫秒
                        Trigger triggerend = TriggerBuilder.newTrigger().withIdentity(map.get("taskKeyEnd"),map.get("taskDescEnd")).startNow().withSchedule(CronScheduleBuilder.cronSchedule(map.get("taskCronEnd"))).build();
                        scheduler.scheduleJob(jobend,triggerend);
                    
                    
                    int insert = ischeduledtaskservice.insert(scheduledtask);
                    if (insert>0) {
                        
                        GreenhouseDevice greenhousedevice=new GreenhouseDevice();
                        greenhousedevice.setUpdateTtime(df.format(new Date()));
                        greenhousedevice.setModel(map.get("taskKeySta"));
                        greenhousedevice.setDeviceStatus(1);
                        queryservice.update(greenhousedevice);
                        
                        ds.put("rtnCode", "1");
                        ds.put("rtnMsg","启动！");
                        log.info("添加完毕");
                        return ds;
                    }
                }
            }
            //睡眠
            //TimeUnit.MINUTES.sleep(1);
            //Thread.sleep(starttime-endtime+2000);
            //scheduler.shutdown(true);
           
        }catch(Exception e){
            e.printStackTrace();
      }
        ds.put("rtnCode", "0");
        ds.put("rtnMsg","启动失败？");
        return ds;
    }
    
    
    @ApiOperation(value="查询所有定时任务", notes="查询所有定时任务")
    @RequestMapping(value = "/stopCronselect", method = RequestMethod.GET)
    public List<ScheduledTask> stopCronselect() {
       return ischeduledtaskservice.select();
    }
    
    @ApiOperation(value="删除定时", notes="删除定时")
    @RequestMapping(value = "/deletestopCron", method = RequestMethod.POST)
    @ApiImplicitParams({
        @ApiImplicitParam(name="map",value="map",dataType="map", paramType = "delete")
        })
    @ResponseBody
    public Map<String,String> deletestopCron(@RequestBody Map<String,String> map) throws SchedulerException {
        Map<String,String> ds=new HashMap<String, String>();
        JobKey jobKeysta = JobKey.jobKey(map.get("taskKeySta"), map.get("taskDescSta"));
        scheduler.deleteJob(jobKeysta);
        
        JobKey jobKeyend = JobKey.jobKey(map.get("taskKeyEnd"),map.get("taskDescEnd"));
        scheduler.deleteJob(jobKeyend);
        
        ScheduledTask scheduledtask=new ScheduledTask();
        scheduledtask.setTaskKeySta(map.get("taskKeySta"));
        scheduledtask.setTaskKeyEnd(map.get("taskKeyEnd"));
        int delete = ischeduledtaskservice.delete(scheduledtask);
        if(delete>0) {
            log.info("已关闭!删除");
            ds.put("rtnCode", "1");
            ds.put("rtnMsg","切换手动成功!");
            return ds;
        }
        ds.put("rtnCode", "0");
        ds.put("rtnMsg","切换手动失败?!");
       return ds;
    }
}

下面是job-定时风扇关闭指令

package com.family.Switch.job;

import java.text.SimpleDateFormat;
import java.util.Date;

import javax.annotation.PostConstruct;

import org.eclipse.paho.client.mqttv3.MqttException;
import org.jboss.logging.Logger;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.family.util.mqtt.ServerMQTT;

@Component
public class EndNewsJob implements Job {

        private static Logger logger = Logger.getLogger(EndNewsJob.class);
        
        public static String gatelog;
        
        public static String end;
        
        @Autowired
        private ServerMQTT servermqtt;
        
        public static EndNewsJob tokenUtil; 
        
        public final Logger log = Logger.getLogger(this.getClass());
        
        @PostConstruct
        public void init() {
            tokenUtil = this;
            tokenUtil.servermqtt = this.servermqtt;
        }
        
        @Override
        public void execute(JobExecutionContext context){
                SimpleDateFormat df= new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                log.info(df.format(new Date()));
                
                
                try {
                    //MQTT发送关闭指令
                    tokenUtil.servermqtt.TOPIC=gatelog;
                    tokenUtil.servermqtt.mqtt(end, gatelog);
                } catch (MqttException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
        }
}

接下来是job--风扇开的指令

package com.family.Switch.job;

import java.text.SimpleDateFormat;
import java.util.Date;

import javax.annotation.PostConstruct;

import org.eclipse.paho.client.mqttv3.MqttException;
import org.jboss.logging.Logger;
import org.quartz.Job;
import org.quartz.JobExecutionContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.family.util.mqtt.ServerMQTT;

@Component
public class StaNewsJob implements Job {

        private static Logger logger = Logger.getLogger(StaNewsJob.class);
        
        public static String gatelog;
        
        public static String sta;
        
        @Autowired
        private ServerMQTT servermqtt;
        
        public static StaNewsJob tokenUtil; 
        
        public final Logger log = Logger.getLogger(this.getClass());
        
        @PostConstruct
        public void init() {
            tokenUtil = this;
            tokenUtil.servermqtt = this.servermqtt;
        }
        
        @Override
        public void execute(JobExecutionContext context){
                SimpleDateFormat df= new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                log.info(df.format(new Date()));
                
                try {
                    //MQTT发送开
                    tokenUtil.servermqtt.TOPIC=gatelog;
                    tokenUtil.servermqtt.mqtt(sta, gatelog);
                } catch (MqttException e) {
                    // TODO Auto-generated catch block
                    e.printStackTrace();
                }
        }
}

还有一个MQTT的发送

package com.family.util.mqtt;

import java.io.UnsupportedEncodingException;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.MqttPersistenceException;
import org.eclipse.paho.client.mqttv3.MqttTopic;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.stereotype.Component;
import org.springframework.stereotype.Service;

@Component
public class ServerMQTT {
    
     //MQTT安装的服务器地址和端口号
    public static final String HOST = "tcp:/ip地址:1883";
    //定义一个主题
    public static String TOPIC;
    //定义MQTT的ID，可以在MQTT服务配置中指定
    private static final String clientid = "client-1";
    

        
        private MqttClient client;
        private MqttTopic topic;
            //账号
        private String userName = "zhny";
            //密码
        private String passWord = "zhny2020";
        private MqttMessage message;
        /**
         * g构造函数
         */
        public ServerMQTT() throws MqttException {
            // MemoryPersistence设置clientid的保存形式，默认为以内存保存
            client = new MqttClient(HOST, clientid, new MemoryPersistence());
            
            if (null!=TOPIC && !"".equals(TOPIC)) {
                connect();
            }
            
        }
        
        public void mqtt(String news,String top) throws MqttException  {
            ServerMQTT server = new ServerMQTT();
            TOPIC=top;
            server.message = new MqttMessage();
            server.message.setQos(2);
            //保留信息最后一次是否
            server.message.setRetained(false);
            //编辑消息
            try {
                server.message.setPayload(news.getBytes("GBK"));
            } catch (UnsupportedEncodingException e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            server.publish(server.topic , server.message);
            System.out.println(server.message.isRetained() + "------ratained状态");
        }
        
        
        /**
         * l连接服务器
         */
        private void connect() {
                 MqttConnectOptions options = new MqttConnectOptions();
                 options.setCleanSession(false);
                 options.setUserName(userName);
                 options.setPassword(passWord.toCharArray());
                 // 设置超时时间
                 options.setConnectionTimeout(10);
                 // 设置会话心跳时间
                 options.setKeepAliveInterval(20);
                 try {
                     client.setCallback(new MqttCallback() {
                         public void connectionLost(Throwable cause) {
                             // 连接丢失后，一般在这里面进行重连
                             System.out.println("连接断开……（可以做重连）");
                         }
                         
                         public void deliveryComplete(IMqttDeliveryToken token) {
                             System.out.println("deliveryComplete---------" + token.isComplete());
                         }
                         
                         public void messageArrived(String topic, MqttMessage message) throws Exception {
                             // subscribe后得到的消息会执行到这里面
                             System.out.println("接收消息主题:" + topic + "  接收消息Qos:" + message.getQos() + "接收消息内容:" + new String(message.getPayload()));
                         }
                     });
                     client.connect(options);

                     topic = client.getTopic(TOPIC);
                 } catch (Exception e) {
                     e.printStackTrace();
                 }
        }

        /**
         * t推送消息
         */
        public void publish(MqttTopic topic , MqttMessage message) throws MqttPersistenceException, MqttException {
            MqttDeliveryToken token = topic.publish(message);
            token.waitForCompletion();
            System.out.println("测试: " + token.isComplete());
        }


    
}

最后附赠一个关于开启，关闭，添加，恢复可自行添加到自己的代码里面,这个下面的用不到只用到其中看那些方法能用就自己的方法里面就可以

package com.xlt.xfzb.util;
 
 
import com.xlt.xfzb.entity.Dingshi;
import org.apache.log4j.Logger;
import org.quartz.*;
import org.quartz.DateBuilder.IntervalUnit;
import org.quartz.impl.matchers.GroupMatcher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
 
 
import java.util.ArrayList;
import java.util.List;
import java.util.Set;
/**
 * @Classname QuartzManager
 * @Description TODO
 * @Date 2019/12/2 11:04
 * @Created by xm
 */
@Service
public class QuartzManager {
 
    public final Logger log = Logger.getLogger(this.getClass());
    @Autowired
    private Scheduler scheduler;
 
    /**
     * 添加任务
     *
     * @param
     * @throws SchedulerException
     */
    @SuppressWarnings("unchecked")
    public void addJob(Dingshi task) {
        try {
            // 创建jobDetail实例，绑定Job实现类
            // 指明job的名称，所在组的名称，以及绑定job类
 
            Class<? extends Job> jobClass = (Class<? extends Job>) (Class.forName(task.getBean_name()).newInstance()
                    .getClass());
            JobDetail jobDetail = JobBuilder.newJob(jobClass).withIdentity(task.getJob_name(), task.getJobGroup())// 任务名称和组构成任务key
                    .build();
            // 定义调度触发规则
            // 使用cornTrigger规则
            Trigger trigger = TriggerBuilder.newTrigger().withIdentity(task.getJob_name(), task.getJobGroup())// 触发器key
                    .startAt(DateBuilder.futureDate(1, IntervalUnit.SECOND))
                    .withSchedule(CronScheduleBuilder.cronSchedule(task.getCron())).startNow().build();
            // 把作业和触发器注册到任务调度中
            scheduler.scheduleJob(jobDetail, trigger);
            // 启动
            if (!scheduler.isShutdown()) {
                scheduler.start();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
 
    /**
     * 获取所有计划中的任务列表
     *
     * @return
     * @throws SchedulerException
     */
    public List<Dingshi> getAllJob() throws SchedulerException {
        GroupMatcher<JobKey> matcher = GroupMatcher.anyJobGroup();
        Set<JobKey> jobKeys = scheduler.getJobKeys(matcher);
        List<Dingshi> jobList = new ArrayList<Dingshi>();
        for (JobKey jobKey : jobKeys) {
            List<? extends Trigger> triggers = scheduler.getTriggersOfJob(jobKey);
            for (Trigger trigger : triggers) {
                Dingshi job = new Dingshi();
                job.setJob_name(jobKey.getName());
                job.setJobGroup(jobKey.getGroup());
                job.setRemark("触发器:" + trigger.getKey());
                Trigger.TriggerState triggerState = scheduler.getTriggerState(trigger.getKey());
                job.setStatus(triggerState.name());
                if (trigger instanceof CronTrigger) {
                    CronTrigger cronTrigger = (CronTrigger) trigger;
                    String cronExpression = cronTrigger.getCronExpression();
                    job.setCron(cronExpression);
                }
                jobList.add(job);
            }
        }
        return jobList;
    }
 
    /**
     * 所有正在运行的job
     *
     * @return
     * @throws SchedulerException
     */
    public List<Dingshi> getRunningJob() throws SchedulerException {
        List<JobExecutionContext> executingJobs = scheduler.getCurrentlyExecutingJobs();
        List<Dingshi> jobList = new ArrayList<Dingshi>(executingJobs.size());
        for (JobExecutionContext executingJob : executingJobs) {
            Dingshi job = new Dingshi();
            JobDetail jobDetail = executingJob.getJobDetail();
            JobKey jobKey = jobDetail.getKey();
            Trigger trigger = executingJob.getTrigger();
            job.setJob_name(jobKey.getName());
            job.setJobGroup(jobKey.getGroup());
            job.setRemark("触发器:" + trigger.getKey());
            Trigger.TriggerState triggerState = scheduler.getTriggerState(trigger.getKey());
            job.setStatus(triggerState.name());
            if (trigger instanceof CronTrigger) {
                CronTrigger cronTrigger = (CronTrigger) trigger;
                String cronExpression = cronTrigger.getCronExpression();
                job.setCron(cronExpression);
            }
            jobList.add(job);
        }
        return jobList;
    }
 
    /**
     * 暂停一个job
     *
     * @param dingshi
     * @throws SchedulerException
     */
    public void pauseJob(Dingshi dingshi) throws SchedulerException {
        JobKey jobKey = JobKey.jobKey(dingshi.getJob_name(), dingshi.getJobGroup());
        scheduler.pauseJob(jobKey);
    }
 
    /**
     * 恢复一个job
     *
     * @param task
     * @throws SchedulerException
     */
    public void resumeJob(Dingshi task) throws SchedulerException {
        JobKey jobKey = JobKey.jobKey(task.getJob_name(), task.getJobGroup());
        scheduler.resumeJob(jobKey);
    }
 
    /**
     * 删除一个job
     *
     * @param task
     * @throws SchedulerException
     */
    public void deleteJob(Dingshi task) throws SchedulerException {
        JobKey jobKey = JobKey.jobKey(task.getJob_name(), task.getJobGroup());
        scheduler.deleteJob(jobKey);
 
    }
 
    /**
     * 立即执行job
     *
     * @param task
     * @throws SchedulerException
     */
    public void runJobNow(Dingshi task) throws SchedulerException {
        JobKey jobKey = JobKey.jobKey(task.getJob_name(), task.getJobGroup());
        scheduler.triggerJob(jobKey);
    }
 
    /**
     * 更新job时间表达式
     *
     * @param task
     * @throws SchedulerException
     */
    public void updateJobCron(Dingshi task) throws SchedulerException {
 
        TriggerKey triggerKey = TriggerKey.triggerKey(task.getJob_name(), task.getJobGroup());
 
        CronTrigger trigger = (CronTrigger) scheduler.getTrigger(triggerKey);
 
        CronScheduleBuilder scheduleBuilder = CronScheduleBuilder.cronSchedule(task.getCron());
 
        trigger = trigger.getTriggerBuilder().withIdentity(triggerKey).withSchedule(scheduleBuilder).build();
 
        scheduler.rescheduleJob(triggerKey, trigger);
    }
}