清淤锁

aijinhui

2023-11-22 af0829b98126c3680ceefdc17621473fa3f97750

清淤锁

 ard-work/src/main/java/com/ruoyi/health/controller/HealthController.java

@@ -84,6 +84,7 @@
        List<TreeDeptWell> treeDeptWells = new ArrayList<>();
        if ((int) mapResult.get("code") == 200) {
            List<Map<String, Object>> listResult = (List<Map<String, Object>>) JSON.parse((String) mapResult.get("resdata"));
//            System.out.println(String.valueOf(listResult));
            //循环得到权限下匹配数据
            for (int i = 0; i < listResult.size(); i++) {
                String equipNumber = (String) listResult.get(i).get("EquipNumber");

 ard-work/src/main/java/com/ruoyi/sy/controller/ArdSyCarLockController.java

@@ -188,7 +188,7 @@

    @PostMapping("/onOff")
    @ApiOperation("远程开锁")
    public AjaxResult onOff(@RequestBody LockTypeParam lockTypeParam) {
    public AjaxResult onOff(@RequestBody LockTypeParam lockTypeParam) throws MqttException {
        return AjaxResult.success(ardSyCarLockService.onOff(lockTypeParam));
    }


 ard-work/src/main/java/com/ruoyi/sy/gps31/PushClientImplPosition.java

@@ -8,7 +8,6 @@
package com.ruoyi.sy.gps31;

import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.gps31.push.netty.PushClient;
import com.gps31.push.netty.PushMsg;
import com.gps31.push.netty.client.TcpClient;
@@ -18,14 +17,12 @@
import com.ruoyi.sy.service.*;
import com.ruoyi.sy.service.impl.*;
import com.ruoyi.utils.gps.GeoTools;
import com.ruoyi.utils.mqtt.MqttConsumer;
import com.ruoyi.utils.mqtt.MqttOnce;
import com.ruoyi.utils.qymqtt.MqttCustomerClient;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.eclipse.paho.client.mqttv3.MqttException;

import java.awt.geom.Point2D;
import java.math.BigDecimal;
import java.text.SimpleDateFormat;
import java.util.*;

@@ -93,6 +90,7 @@
                String deptName = (String) result.get("deptName");//获取部门名称
                String carModel = (String) result.get("carModel");//获取车辆模型
                Integer drct = (Integer) gpsMap.get("drct");//获取车辆模型
                Integer speed = (Integer) gpsMap.get("speed");
                //定义2维空间
                Point2D.Double point = new Point2D.Double(Double.parseDouble(lng), Double.parseDouble(lat));
                List<Point2D.Double> pts = new ArrayList<Point2D.Double>();
@@ -106,12 +104,18 @@
                ArdSyCarRtuService carRtuService = SpringUtils.getBean(ArdSyCarRtuService.class);
                //注入锁Bean
                ArdSyCarLockService lockService = SpringUtils.getBean(ArdSyCarLockService.class);
                //创建mqtt工具类对象
                MqttOnce mqttOnce = new MqttOnce();
                //查询定位油车表中是否有该车辆
                ArdAccessFence ardAccessFence = ardAccessFenceService.queryAccess(carId);
                //查询所有装油点围栏
                List<ArdOilWall> ardOilWalls = ardOilWallService.accessAll();
                //车速标记
                Boolean stop = true;
                //判断车速
                if(speed>5){
                    //mqtt发送既定Rtu状态
                    feng(carId,"施封",carRtuService,lockService);
                    stop = false;
                }
                //如果没有说明车辆未进入装油点，需要匹配所有装油点围栏
                if(ardAccessFence==null){
                    //车辆未进入装油点
@@ -141,8 +145,10 @@
                            ardAccessFence1.setDrct(drct);
                            ardAccessFence1.setIsAlarm(0);
                            ardAccessFenceService.access(ardAccessFence1);
                            //mqtt发送既定Rtu状态
                            feng(carId,"装油点",carRtuService,lockService,mqttOnce);
                            if(stop){
                                //mqtt发送既定Rtu状态
                                feng(carId,"装油点",carRtuService,lockService);
                            }
                            break;
                        }
                    }
@@ -201,8 +207,10 @@
                                    ardAccessFence.setIntoShipmentTime(time);
                                    ardAccessFence.setNowWallName(ardOilWall.getName());
                                    ardAccessFence.setLastTime(time);
                                    //mqtt发送既定Rtu状态
                                    feng(carId,"装油点",carRtuService,lockService,mqttOnce);
                                    if(stop){
                                        //mqtt发送既定Rtu状态
                                        feng(carId,"装油点",carRtuService,lockService);
                                    }
                                    break;
                                }
                            }
@@ -230,7 +238,7 @@
                                    ardAccessFence.setLat(lat);
                                    ardAccessFenceService.updAccess(ardAccessFence);
                                    //开出装油点围栏，发送上锁及施封命令
                                    feng(carId,"施封",carRtuService,lockService,mqttOnce);
                                    feng(carId,"施封",carRtuService,lockService);
                                }
                            }else {
                                //车辆已经开出装油点，查看是否有卸油点ID
@@ -260,8 +268,10 @@
                                            ardAccessFenceService.updAccess(ardAccessFence);
                                            //已经进入卸油点，不在路上不需要停车报警
                                            isRoad = false;
                                            //mqtt发送既定Rtu状态
                                            feng(carId,"卸油点",carRtuService,lockService,mqttOnce);
                                            if(stop){
                                                //mqtt发送既定Rtu状态
                                                feng(carId,"卸油点",carRtuService,lockService);
                                            }
                                            break;
                                        }
                                    }
@@ -304,7 +314,7 @@
                                    }
                                    if (!GeoTools.IsPtInPoly(point, pts)) {
                                        //开出了卸油点围栏，发送上锁及施解封指令
                                        feng(carId,"施封",carRtuService,lockService,mqttOnce);
                                        feng(carId,"施封",carRtuService,lockService);
                                        //删除该条车辆数据
                                        ardAccessFenceService.delAccess(carId);
                                    }else {
@@ -420,9 +430,11 @@
        sendMassage();
    }

    private void feng(String carId,String type,ArdSyCarRtuService carRtuService,ArdSyCarLockService lockService,MqttOnce mqttOnce) throws MqttException {
    private void feng(String carId,String type,ArdSyCarRtuService carRtuService,ArdSyCarLockService lockService) throws MqttException {
        //查询装油点既定Rtu状态
        ArdSyCarRtu ardSyCarRtu = carRtuService.one(carId);
        //查询锁信息
        List<ArdSyCarLock> ardSyCarLocks = lockService.carLock(ardSyCarRtu.getId());
        Integer fengType = null;
        if("装油点".equals(type)){
            fengType = ardSyCarRtu.getInstallType();
@@ -433,52 +445,70 @@
        }
        //1解封
        if(fengType==1){
            Map<String,Object> map = new HashMap<>();
            map.put("A01", 110000);//导通第一路继电器
            map.put("A02", 110000);//导通第二路继电器
            map.put("res", String.valueOf(System.currentTimeMillis()));
            mqttOnce.publishCar(carId,ardSyCarRtu.getTopicPublish(),String.valueOf(new JSONObject(map)),"解封");
            //如果不是解封封状态，则发送施封命令
            if(ardSyCarLocks.get(0).getRelayInfo()==0){
                Map<String,Object> map = new HashMap<>();
                map.put("A01", 110000);//导通第一路继电器
                map.put("A02", 110000);//导通第二路继电器
                map.put("res", String.valueOf(System.currentTimeMillis()));
                String mapString = com.alibaba.fastjson.JSON.toJSONString(map);
                MqttCustomerClient mqttCustomerClient = new MqttCustomerClient();
                mqttCustomerClient.connect(carId+"change");
                mqttCustomerClient.pushlish(ardSyCarRtu.getTopicPublish(),mapString);
                log.info("车辆"+carId+"解封命令发送成功！");
//                mqttOnce.publishCar(carId,ardSyCarRtu.getTopicPublish(),String.valueOf(new JSONObject(map)),"解封");
            }
        }
        //0施封
        if(fengType==0){
            //锁开关状态
            int lockCloseNum = 0;
            //查询锁信息
            List<ArdSyCarLock> ardSyCarLocks = lockService.carLock(ardSyCarRtu.getId());
            for (int j = 0; j < ardSyCarLocks.size(); j++) {
                ArdSyCarLock ardSyCarLock = ardSyCarLocks.get(j);
                String currents = ardSyCarLock.getCurrents();
                Double currentInfo = ardSyCarLock.getCurrentInfo();
                //电流值为4.0是关锁状态
                if(currentInfo.equals(4.00) || currentInfo.equals(4)){
                    lockCloseNum += 1;
                }else {
                    String key = "";
                    if(currents.equals("C01")){
                        key = "D03";
                    }else if(currents.equals("C02")){
                        key = "D04";
                    }
                    Map<String,Object> map = new HashMap();
                    map.put(key, 0.01);
                    map.put("res", String.valueOf(System.currentTimeMillis()));
                    mqttOnce.publishCar("carId",ardSyCarRtu.getTopicPublish(),String.valueOf(new JSONObject(map)),"开锁："+currents);
                    while(true){
                        ArdSyCarLock ardSyCarLock1 = lockService.getCurrentInfo(ardSyCarRtu.getId(),currents);
                        Double currentInfo1 = ardSyCarLock1.getCurrentInfo();
                        if(currentInfo1.equals(4.00)){
                            lockCloseNum +=1;
                            break;
            //如果不是施封状态，则发送施封命令
            if(ardSyCarLocks.get(0).getRelayInfo()==1){
                for (int j = 0; j < ardSyCarLocks.size(); j++) {
                    ArdSyCarLock ardSyCarLock = ardSyCarLocks.get(j);
                    String currents = ardSyCarLock.getCurrents();
                    Double currentInfo = ardSyCarLock.getCurrentInfo();
                    //电流值为4.0是关锁状态
                    if(currentInfo.equals(4.00) || currentInfo.equals(4)){
                        lockCloseNum += 1;
                    }else {
                        String key = "";
                        if(currents.equals("C01")){
                            key = "D03";
                        }else if(currents.equals("C02")){
                            key = "D04";
                        }
                        Map<String,Object> map = new HashMap();
                        map.put(key, 0.01);
                        map.put("res", String.valueOf(System.currentTimeMillis()));
                        String mapString = com.alibaba.fastjson.JSON.toJSONString(map);
                        MqttCustomerClient mqttCustomerClient = new MqttCustomerClient();
                        mqttCustomerClient.connect(carId+"change");
                        mqttCustomerClient.pushlish(ardSyCarRtu.getTopicPublish(),mapString);
//                        mqttOnce.publishCar("carId",ardSyCarRtu.getTopicPublish(),String.valueOf(new JSONObject(map)),"开锁："+currents);
                        while(true){
                            ArdSyCarLock ardSyCarLock1 = lockService.getCurrentInfo(ardSyCarRtu.getId(),currents);
                            Double currentInfo1 = ardSyCarLock1.getCurrentInfo();
                            if(currentInfo1.equals(4.00) || currentInfo1.equals(4)){
                                lockCloseNum +=1;
                                break;
                            }
                        }
                    }
                }
            }
            if(lockCloseNum == 2){
                Map<String,Object> map = new HashMap<>();
                map.put("A01", 100000);//断开第一路继电器
                map.put("A02", 100000);//断开第二路继电器
                map.put("res", String.valueOf(System.currentTimeMillis()));
                mqttOnce.publishCar(carId,ardSyCarRtu.getTopicPublish(),String.valueOf(new JSONObject(map)),"施封");
                if(lockCloseNum == 2){
                    Map<String,Object> map = new HashMap<>();
                    map.put("A01", 100000);//断开第一路继电器
                    map.put("A02", 100000);//断开第二路继电器
                    map.put("res", String.valueOf(System.currentTimeMillis()));
                    String mapString = com.alibaba.fastjson.JSON.toJSONString(map);
                    MqttCustomerClient mqttCustomerClient = new MqttCustomerClient();
                    mqttCustomerClient.connect(carId+"change");
                    mqttCustomerClient.pushlish(ardSyCarRtu.getTopicPublish(),mapString);
                    log.info("车辆"+carId+"施封命令发送成功！");
//                    mqttOnce.publishCar(carId,ardSyCarRtu.getTopicPublish(),String.valueOf(new JSONObject(map)),"施封");
                }
            }
        }
    }

 ard-work/src/main/java/com/ruoyi/sy/service/ArdSyCarLockService.java

@@ -49,7 +49,7 @@

    AjaxResult seal(RtuTypeParam rtuTypeParam);

    AjaxResult onOff(LockTypeParam lockTypeParam);
    AjaxResult onOff(LockTypeParam lockTypeParam) throws MqttException;

    List<ArdSyCarLock> lockIng(String carId);


 ard-work/src/main/java/com/ruoyi/sy/service/impl/ArdSyCarLockServiceImpl.java

@@ -17,14 +17,14 @@
import com.ruoyi.sy.service.ArdSyCarLockService;
import com.ruoyi.sy.vo.CarVo;
import com.ruoyi.sy.vo.RtuVo;
import com.ruoyi.system.mapper.SysConfigMapper;
import com.ruoyi.system.mapper.SysDeptMapper;
import com.ruoyi.utils.mqtt.MqttOnce;
import com.ruoyi.utils.qymqtt.MqttCustomerClient;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.text.DecimalFormat;
import java.util.*;

/**
@@ -42,6 +42,8 @@
    private ArdSyCarRtuMapper rtuMapper;
    @Autowired
    private ArdSyCarLockMapper lockMapper;
    @Autowired
    private SysConfigMapper configMapper;


    @Override
@@ -75,8 +77,9 @@
        ArdSyCar ardSyCar = ardSyCarMapper.selectOne(queryWrapper);
        ardSyCar.setReserved1(ardSyCarRtu.getId());
        ardSyCarMapper.updateById(ardSyCar);
        MqttOnce mqttOnce = new MqttOnce();
        mqttOnce.subscribeCar(ardSyCarRtu);
        MqttCustomerClient mqttCustomerClient = new MqttCustomerClient();
        mqttCustomerClient.connect(addRtuParam.getCarId()+"mqtt");
        mqttCustomerClient.subscribe(ardSyCarRtu.getTopicSubscribe());
        return num;
    }

@@ -90,6 +93,11 @@

    @Override
    public int delCar(String id) {
        ArdSyCarRtu ardSyCarRtu = rtuMapper.selectById(id);
        String subscribe = ardSyCarRtu.getTopicSubscribe();
        MqttCustomerClient mqttCustomerClient = new MqttCustomerClient();
        mqttCustomerClient.connect(ardSyCarRtu.getCarId()+"del");
        mqttCustomerClient.unSubscribe(subscribe);
        rtuMapper.deleteById(id);
        QueryWrapper<ArdSyCarLock> queryWrapper1 = new QueryWrapper<>();
        queryWrapper1.eq("rtu_id",id);
@@ -236,15 +244,19 @@
        QueryWrapper<ArdSyCarRtu> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("carId",rtuTypeParam.getCarId());
        ArdSyCarRtu ardSyCarRtu = rtuMapper.selectOne(queryWrapper);
        MqttOnce mqttOnce = new MqttOnce();
        //1解封
        if(rtuTypeParam.getType()==1){
            Map<String,Object> map = new HashMap<>();
            map.put("A01", 110000);//导通第一路继电器
            map.put("A02", 110000);//导通第二路继电器
            map.put("res", String.valueOf(System.currentTimeMillis()));
            String mapString = JSON.toJSONString(map);
            try {
                return AjaxResult.success(mqttOnce.publishCar(rtuTypeParam.getCarId(),ardSyCarRtu.getTopicPublish(),String.valueOf(new com.alibaba.fastjson2.JSONObject(map)),"解封"));
                MqttCustomerClient mqttCustomerClient = new MqttCustomerClient();
                mqttCustomerClient.connect(rtuTypeParam.getCarId()+"change");
                mqttCustomerClient.pushlish(ardSyCarRtu.getTopicPublish(),mapString);
                return AjaxResult.success("解封命令发送成功！");
//                return AjaxResult.success(mqttOnce.publishCar(rtuTypeParam.getCarId(),ardSyCarRtu.getTopicPublish(),String.valueOf(new com.alibaba.fastjson2.JSONObject(map)),"解封"));
            } catch (MqttException e) {
                log.debug("解封指令发生失败"+e);
            }
@@ -272,15 +284,19 @@
                    Map<String,Object> map = new HashMap();
                    map.put(key, 0.01);
                    map.put("res", String.valueOf(System.currentTimeMillis()));
                    String mapString = JSON.toJSONString(map);
                    try {
                        mqttOnce.publishCar("carId",ardSyCarRtu.getTopicPublish(),String.valueOf(new com.alibaba.fastjson2.JSONObject(map)),"开锁："+currents);
                        MqttCustomerClient mqttCustomerClient = new MqttCustomerClient();
                        mqttCustomerClient.connect(rtuTypeParam.getCarId()+"change");
                        mqttCustomerClient.pushlish(ardSyCarRtu.getTopicPublish(),mapString);
//                        mqttOnce.publishCar("carId",ardSyCarRtu.getTopicPublish(),String.valueOf(new com.alibaba.fastjson2.JSONObject(map)),"开锁："+currents);
                    } catch (MqttException e) {
                        log.debug("开锁指令发生失败"+e);
                    }
                    while(true){
                        ArdSyCarLock ardSyCarLock1 = getCurrentInfo(ardSyCarRtu.getId(),currents);
                        Double currentInfo1 = ardSyCarLock1.getCurrentInfo();
                        if(currentInfo1.equals(4.00)){
                        if(currentInfo1.equals(4.00) || currentInfo1.equals(4)){
                            lockCloseNum +=1;
                            break;
                        }
@@ -292,8 +308,13 @@
                map.put("A01", 100000);//断开第一路继电器
                map.put("A02", 100000);//断开第二路继电器
                map.put("res", String.valueOf(System.currentTimeMillis()));
                String mapString = JSON.toJSONString(map);
                try {
                    return AjaxResult.success(mqttOnce.publishCar(rtuTypeParam.getCarId(),ardSyCarRtu.getTopicPublish(),String.valueOf(new com.alibaba.fastjson2.JSONObject(map)),"施封"));
                    MqttCustomerClient mqttCustomerClient = new MqttCustomerClient();
                    mqttCustomerClient.connect(rtuTypeParam.getCarId()+"change");
                    mqttCustomerClient.pushlish(ardSyCarRtu.getTopicPublish(),mapString);
                    return AjaxResult.success("施封命令发生成功！");
//                    return AjaxResult.success(mqttOnce.publishCar(rtuTypeParam.getCarId(),ardSyCarRtu.getTopicPublish(),String.valueOf(new com.alibaba.fastjson2.JSONObject(map)),"施封"));
                } catch (MqttException e) {
                    log.debug("施封指令发生失败"+e);
                }
@@ -303,7 +324,7 @@
    }

    @Override
    public AjaxResult onOff(LockTypeParam lockTypeParam) {
    public AjaxResult onOff(LockTypeParam lockTypeParam) throws MqttException {
        QueryWrapper<ArdSyCarRtu> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("car_id",lockTypeParam.getCarId());
        ArdSyCarRtu ardSyCarRtu = rtuMapper.selectOne(queryWrapper);
@@ -332,13 +353,50 @@
        map.put("res", String.valueOf(System.currentTimeMillis()));
        String mapString = JSON.toJSONString(map);
        System.out.println(mapString);
        MqttOnce mqttOnce = new MqttOnce();
        try {
            return AjaxResult.success(mqttOnce.publishCar("carId",ardSyCarRtu.getTopicPublish(), mapString,"开关锁指令"+lockTypeParam.getNum()));
        } catch (MqttException e) {
            log.debug("开关锁指令发送失败"+e);
        }
        return AjaxResult.success();
        MqttCustomerClient mqttCustomerClient = new MqttCustomerClient();
        mqttCustomerClient.connect(lockTypeParam.getCarId()+"change");
        mqttCustomerClient.pushlish(ardSyCarRtu.getTopicPublish(),mapString);
        return AjaxResult.success("开关锁命令发送成功,发送进度为"+lockTypeParam.getNum());
//        QueryWrapper<SysConfig> queryWrapper1 = new QueryWrapper<>();
//        queryWrapper1.eq("config_key","ElectromagneticLockUrl");
//        SysConfig sysConfig = configMapper.selectOne(queryWrapper1);
//        String url = "";
//        if(sysConfig!=null){
//            url = sysConfig.getConfigValue();
//        }else {
//            return AjaxResult.error("配置表无地址数据");
//        }
//        MqttConsumer mqttConsumer = new MqttConsumer();
//        mqttConsumer.getClientCar(url,lockTypeParam.getCarId());
//        mqttConsumer.publishCar(1,false,ardSyCarRtu.getTopicPublish(),mapString);
//        mqttConsumer.disConnect();
//        return AjaxResult.success();
        //MqttOnce客户端-----开始
//        MqttOnce mqttOnce = new MqttOnce();
//        return AjaxResult.success(mqttOnce.publishCar("carId",ardSyCarRtu.getTopicPublish(), mapString,"开关锁指令"+lockTypeParam.getNum()));
        //MqttOnce客户端-----结束
        //EmqClient客户端-----开始
//        EmqClient emqClient = new EmqClient("tcp://112.98.126.2:1883",String.valueOf(UUID.randomUUID()));
//        emqClient.connect("admin","xzx12345");
//        log.info("主题："+ardSyCarRtu.getTopicPublish());
//        emqClient.publish(ardSyCarRtu.getTopicPublish(),mapString, QosEnum.QoS2,false);
//        System.out.println("////////////////////");
//        System.out.println(mapString);
//        emqClient.disConnect();
//        return AjaxResult.success(ardSyCarRtu.getTopicPublish()+"开关锁指令"+lockTypeParam.getNum()+"发布成功！");
        //EmqClient客户端-----结束
        //demo------开始
//        EmqClient emqClient = new EmqClient("tcp://112.98.126.2:1883","mqttTestp");
//        emqClient.connect("admin","xzx12345");
//        Map<String,Object> map = new HashMap();
//        map.put("D03", 99.99);//范围是0.01到99.99
//        map.put("res", String.valueOf(new Date().getTime()));
//        System.out.println(map);
//        String msg = JSON.toJSONString(map);
//        emqClient.publish("/p14/set/",msg,QosEnum.QoS2,false);
//        emqClient.disConnect();
//        return AjaxResult.success();
        //demo----结束
    }

    @Override
@@ -346,7 +404,14 @@
        QueryWrapper<ArdSyCarRtu> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("car_id",carId);
        ArdSyCarRtu ardSyCarRtu = rtuMapper.selectOne(queryWrapper);
        return lockMapper.lockList(ardSyCarRtu.getId());
        List<ArdSyCarLock> locks = lockMapper.lockList(ardSyCarRtu.getId());
        for (int i = 0; i < locks.size(); i++) {
            ArdSyCarLock ardSyCarLock = locks.get(i);
            double info = ardSyCarLock.getCurrentInfo();
            int d = (int) ((int)(info*150/138.3 - 4)/0.16);
            ardSyCarLock.setCurrentInfo((double) d);
        }
        return locks;
    }



 ard-work/src/main/java/com/ruoyi/utils/mqtt/MqttOnce.java

ÎļþÒÑɾ³ý

 ard-work/src/main/java/com/ruoyi/utils/mqtt/MqttOnceCallback.java

ÎļþÒÑɾ³ý

 ard-work/src/main/java/com/ruoyi/utils/qymqtt/AutoMqtt.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,38 @@
package com.ruoyi.utils.qymqtt;

import com.ruoyi.sy.domain.ArdSyCarRtu;
import com.ruoyi.sy.service.ArdSyCarRtuService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.util.List;

@Slf4j
@Component
public class AutoMqtt {

    @Autowired
    private ArdSyCarRtuService ardSyCarRtuService;

    @PostConstruct
    public void init() {
//        ArdSyCarRtuService carRtuService = SpringUtils.getBean(ArdSyCarRtuService.class);
//        //获取电磁锁地址
//        String mqttUrl = carRtuService.url("ElectromagneticLockUrl");
        List<ArdSyCarRtu> rtus = ardSyCarRtuService.allCarTopicList();
        for (int i = 0; i < rtus.size(); i++) {
            ArdSyCarRtu rtu = rtus.get(i);
            String carId = rtu.getCarId();
            String subscribe = rtu.getTopicSubscribe();
            try {
                MqttCustomerClient mqttCustomerClient = new MqttCustomerClient();
                mqttCustomerClient.connect(carId+"mqtt");
                mqttCustomerClient.subscribe(subscribe);
            }catch (Exception e) {
                e.printStackTrace();
            }
        }
    }
}

 ard-work/src/main/java/com/ruoyi/utils/qymqtt/MqttCustomerClient.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,141 @@
package com.ruoyi.utils.qymqtt;

import com.ruoyi.common.utils.spring.SpringUtils;
import com.ruoyi.sy.service.ArdSyCarRtuService;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;

/**
 *  mqtt客户端
 */
@Slf4j
@Component
public class MqttCustomerClient {

    private static MqttClient client;

    private static MqttConnectOptions options;

    private PushCallback pushCallback;

    public  static MqttClient getClient(){
        return  client;
    }

    public static void setClient(MqttClient client){
        MqttCustomerClient.client=client;
    }

    public static MqttConnectOptions getOptions() {
        return options;
    }

    public static void setOptions(MqttConnectOptions options) {
        MqttCustomerClient.options = options;
    }

    /**
     * å®¢æˆ·ç«¯è¿žæŽ¥
     *
     * @param clientID  å®¢æˆ·ç«¯Id
     */
    public void connect(String clientID){
        MqttClient client;
        try {
            ArdSyCarRtuService carRtuService = SpringUtils.getBean(ArdSyCarRtuService.class);
            //获取电磁锁地址
            String mqttUrl = carRtuService.url("ElectromagneticLockUrl");
            client=new MqttClient(mqttUrl,clientID,new MemoryPersistence());
            MqttConnectOptions options = new MqttConnectOptions();
            options.setCleanSession(true);
            options.setUserName("admin");
            options.setPassword("xzx12345".toCharArray());
            options.setConnectionTimeout(100);
            options.setKeepAliveInterval(60);
            options.setAutomaticReconnect(true);
            MqttCustomerClient.setClient(client);
            MqttCustomerClient.setOptions(options);
            try {
                client.setCallback(new PushCallback());
                client.connect(options);
            }catch (Exception e){
                e.printStackTrace();
            }
        }catch (Exception e){
            e.printStackTrace();
        }
    }

    /**
     * å‘布
     * @param topic
     * @param pushMessage
     */
    public void pushlish(String topic,String pushMessage) throws MqttException {
        pushlish(2,false,topic,pushMessage);
    }

    /**
     * å‘布
     *
     * @param qos         è¿žæŽ¥æ–¹å¼
     * @param retained    æ˜¯å¦ä¿ç•™
     * @param topic       ä¸»é¢˜
     * @param pushMessage æ¶ˆæ¯ä½“
     */
    public void pushlish(int qos,boolean retained,String topic,String pushMessage) throws MqttException {
        MqttMessage message=new MqttMessage();
        message.setQos(qos);
        message.setRetained(retained);
        message.setPayload(pushMessage.getBytes());
        try{
            client.publish(topic,message);
        }catch (Exception e){
            log.error(topic+"发布失败！");
        }
        client.disconnect();
    }

    /**
     * è®¢é˜…某个主题
     * @param topic
     */
    public void subscribe(String topic){
        log.info("开始订阅主题" + topic);
        subscribe(topic,2);
    }

    public void subscribe(String topic,int qos){
        try {
            MqttCustomerClient.getClient().subscribe(topic,qos);
        }catch (MqttException e){
            e.printStackTrace();
        }
    }

    //取消订阅
    public void unSubscribe(String topicFilter){
        try {
            client.unsubscribe(topicFilter);
        } catch (MqttException e) {
            System.out.println("取消订阅主题"+topicFilter+"失败");
            e.getMessage();
        }
    }

    //断开连接
    @PreDestroy//该注解对象销毁后触发
    public void disConnect(){
        try {
            client.disconnect();
        } catch (MqttException e) {
            System.out.println("断开连接产生异常");
            e.getMessage();
        }
    }
}

 ard-work/src/main/java/com/ruoyi/utils/qymqtt/PushCallback.java

¶Ô±ÈÐÂÎļþ
@@ -0,0 +1,108 @@
package com.ruoyi.utils.qymqtt;

import com.ruoyi.common.utils.spring.SpringUtils;
import com.ruoyi.sy.domain.ArdSyCarLock;
import com.ruoyi.sy.domain.ArdSyCarRtu;
import com.ruoyi.sy.service.ArdSyCarLockService;
import com.ruoyi.sy.service.ArdSyCarRtuService;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;

import java.nio.charset.StandardCharsets;
import java.util.List;

/**
 *  æ¶ˆè´¹ç›‘听
 */
@Slf4j
public class PushCallback implements MqttCallback {

    private MqttClient client;

    //注入电子锁Bean
    ArdSyCarRtuService carRtuService = SpringUtils.getBean(ArdSyCarRtuService.class);
    ArdSyCarLockService carLockService = SpringUtils.getBean(ArdSyCarLockService.class);

//    private MqttClient client;
//    private MqttConnectOptions options;
//    private String topic;
//    private int qos;

//    public PushCallback(MqttClient client, MqttConnectOptions options, String topic, int qos) {
//        this.client = client;
//        this.options = options;
//        this.topic = topic;
//        this.qos = qos;
//    }

    @Override
    public void connectionLost(Throwable throwable) {
        if (client == null || !client.isConnected()) {
            System.out.println("连接断开，正在重连....");
        }
    }

    @Override
    public void messageArrived(String topic, MqttMessage message) throws Exception {
        try {
            System.out.println("【车辆锁主题】："+topic+"【车辆锁信息】："+new String(message.getPayload(), StandardCharsets.UTF_8));
            //出来回调数据
            List<ArdSyCarLock> ardSyCarLocks = carLockService.data(new String(message.getPayload(), StandardCharsets.UTF_8));
            //根据topic查询RTUID
            ArdSyCarRtu ardSyCarRtu = carRtuService.subscribeByCarId(topic);
            String RTU = ardSyCarRtu.getId();
            //根据RTUID查询锁
            List<ArdSyCarLock> ardSyCarLockList = carLockService.carLock(RTU);
            for (int i = 0; i < ardSyCarLocks.size(); i++) {
                ArdSyCarLock ardSyCarLockData = ardSyCarLocks.get(i);
                String relay = ardSyCarLockData.getRelay();
                Integer relayInfo = ardSyCarLockData.getRelayInfo();
                String current = ardSyCarLockData.getCurrents();
                Double currentInfo = ardSyCarLockData.getCurrentInfo();
                for (int j = 0; j < ardSyCarLockList.size(); j++) {
                    ArdSyCarLock lock = ardSyCarLockList.get(j);
                    String relay1 = lock.getRelay();
                    String current1 = lock.getCurrents();
                    Integer relayInfo1 = lock.getRelayInfo();
                    Double currentInfo1 = lock.getCurrentInfo();
                    Boolean upd = false;
                    if(relay.equals(relay1)){
                        if(!relayInfo.equals(relayInfo1)){
                            lock.setRelayInfo(relayInfo);
                            upd = true;
                        }
                    }
                    if(current.equals(current1)){
                        if(!currentInfo.equals(currentInfo1)){
                            lock.setCurrentInfo(currentInfo);
                            upd = true;
                        }
                    }
                    if(upd){
                        int num = carLockService.updLock(lock);
//                        log.debug("修改"+num+"数据");
                    }
                }
            }
        } catch (Exception e) {
            log.debug("车辆电磁锁处理mqtt消息异常:" + e);
        }
    }

    @SneakyThrows
    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {
        System.out.println("deliveryComplete---------" + token.isComplete());
        int messageId = token.getMessageId();
        String[] topics = token.getTopics();
//        byte[] msg = token.getMessage().getPayload();
        String topicStr = "";
        for(String topic : topics){
            topicStr = topicStr + topic + ",";
        }
        topicStr = topicStr.substring(0, topicStr.length() - 1);
        System.out.println("消息发布完成,messageId="+messageId+",topics="+topicStr);
    }
}