棒一 高线 补偿机制

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/actualControl/billetActual/billetActual/controller/BilletBasicInfoController.java

@@ -1,5 +1,7 @@
 package org.jeecg.modules.actualControl.billetActual.billetActual.controller;
 
+import cn.hutool.core.date.DateTime;
+import com.alibaba.fastjson.JSON;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.core.metadata.IPage;
 import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -15,15 +17,18 @@ import org.jeecg.modules.actualControl.billetActual.billetActual.entity.BilletBa
 import org.jeecg.modules.actualControl.billetActual.billetActual.service.IBilletBasicInfoService;
 import org.jeecg.modules.billet.billetHotsendConfig.entity.BilletHotsendTypeConfig;
 import org.jeecg.modules.billet.billetHotsendConfig.service.IBilletHotsendTypeConfigService;
+import org.jeecg.modules.connConfig.mapper.ConfigMqttMapper;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.web.bind.annotation.*;
 import org.springframework.web.servlet.ModelAndView;
 
 import javax.servlet.http.HttpServletRequest;
 import javax.servlet.http.HttpServletResponse;
+import java.lang.reflect.InvocationTargetException;
 import java.util.Arrays;
 import java.util.List;
 
+
 /**
  * @Description: 钢坯基础信息
  * @Author: jeecg-boot
@@ -41,6 +46,9 @@ public class BilletBasicInfoController extends JeecgController<BilletBasicInfo,
 	@Autowired
 	private IBilletHotsendTypeConfigService billetHotsendTypeConfigService;
 
+	@Autowired
+	private ConfigMqttMapper configMqttMapper;
+
 	/**
 	 * 分页列表查询
 	 *
@@ -86,42 +94,11 @@ public class BilletBasicInfoController extends JeecgController<BilletBasicInfo,
 	@ApiOperation(value="钢坯基础信息-添加", notes="钢坯基础信息-添加")
 	@PostMapping(value = "/add")
 	public Result<?> add(@RequestBody BilletBasicInfo billetBasicInfo,
-						 @RequestParam(name="number", defaultValue="1") Integer number) {
-		// 查询该炉号最后一条钢坯记录
-		BilletBasicInfo lastBilletBasicInfo = billetBasicInfoService.getOne(new QueryWrapper<BilletBasicInfo>().eq("ccm_no", billetBasicInfo.getCcmNo()).eq("heat_no", billetBasicInfo.getHeatNo()).orderByDesc("create_time"));
-		if (number == 1) {
-			if (oConvertUtils.isNotEmpty(lastBilletBasicInfo)) {
-				billetBasicInfo.setHeatnoIndex(lastBilletBasicInfo.getHeatnoIndex() + 1);
-				billetBasicInfo.setStrandnoIndex(lastBilletBasicInfo.getStrandnoIndex() + 1);
-			} else {
-				billetBasicInfo.setHeatnoIndex(1);
-				// 随机生成坯号
-				String billetNo = generateUniqueBilletNo(billetBasicInfo.getCcmNo(), billetBasicInfo.getHeatNo());
-				billetBasicInfo.setBilletNo(billetNo);
-			}
-			billetBasicInfoService.save(billetBasicInfo);
-		} else { // 多条补录
-			for (int i = 0; i < number; i++) {
-				billetBasicInfo.setId(null);
-				billetBasicInfoService.save(billetBasicInfo);
-			}
-		}
+						 @RequestParam(name="number", defaultValue="1") Integer number) throws InvocationTargetException, IllegalAccessException, NoSuchMethodException {
+		billetBasicInfoService.addB(billetBasicInfo, number);
 		return Result.OK("添加成功！");
 	}
 
-	/**
-	 * 生成唯一的坯号
-	 *
-	 * @return 坯号
-	 */
-	private String generateUniqueBilletNo(Integer ccmNo, String heatNo) {
-		String billetNo;
-		do {
-			billetNo = heatNo +  (int)(Math.random() * 10000);
-		} while (billetBasicInfoService.exists(new QueryWrapper<BilletBasicInfo>().eq("billet_no", billetNo). eq("ccm_no", ccmNo)));
-		return billetNo;
-	}
-
 	/**
 	 *  编辑
 	 *

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/actualControl/billetActual/billetActual/entity/BilletBasicInfo.java

@@ -110,8 +110,8 @@ public class BilletBasicInfo implements Serializable {
     @ApiModelProperty(value = "坯号",required = true)
     private String billetNo;
     /**坯号*/
-    @Excel(name = "坯号", width = 15)
-    @ApiModelProperty(value = "坯号",required = true)
+    @Excel(name = "流内顺序号", width = 15)
+    @ApiModelProperty(value = "流内顺序号",required = true)
     private Integer strandnoIndex;
     /**开始切割时间*/
     @Excel(name = "开始切割时间", width = 20, format = "yyyy-MM-dd HH:mm:ss")

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/actualControl/billetActual/billetActual/service/IBilletBasicInfoService.java

@@ -24,5 +24,7 @@ public interface IBilletBasicInfoService extends IService<BilletBasicInfo> {
 
     List<BilletHotsendTypeConfig> queryBilletNameList(String ccmNo);
 
-    boolean exists(QueryWrapper<BilletBasicInfo> billetNo);
+    boolean exists(QueryWrapper<BilletBasicInfo> eq);
+
+    void addB(BilletBasicInfo billetBasicInfo, Integer num);
 }

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/actualControl/billetActual/billetActual/service/impl/BilletBasicInfoServiceImpl.java

@@ -1,19 +1,29 @@
 package org.jeecg.modules.actualControl.billetActual.billetActual.service.impl;
 
+import cn.hutool.core.date.DateTime;
+import com.alibaba.fastjson.JSON;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
 import org.jeecg.common.util.oConvertUtils;
 import org.jeecg.modules.actualControl.billetActual.billetActual.entity.BilletBasicInfo;
+import org.jeecg.modules.actualControl.billetActual.billetActual.entity.BilletRulerConfig;
 import org.jeecg.modules.actualControl.billetActual.billetActual.mapper.BilletBasicInfoMapper;
+import org.jeecg.modules.actualControl.billetActual.billetActual.mapper.BilletRulerConfigMapper;
 import org.jeecg.modules.actualControl.billetActual.billetActual.service.IBilletBasicInfoService;
+import org.jeecg.modules.actualControl.billetActual.billetActual.utils.MqttClientUtil;
+import org.jeecg.modules.actualControl.billetActual.billetAssemblyNumber.entity.BilletAssemblyNumber;
+import org.jeecg.modules.actualControl.billetActual.billetAssemblyNumber.mapper.BilletAssemblyNumberMapper;
+import org.jeecg.modules.actualControl.heatsActuals.entity.HeatsActuals;
 import org.jeecg.modules.billet.billetHotsend.entity.RulerDefaultConfig;
 import org.jeecg.modules.billet.billetHotsendConfig.entity.BilletHotsendTypeConfig;
 import org.jeecg.modules.billet.billetHotsendConfig.service.IBilletHotsendTypeConfigService;
+import org.jeecg.modules.connConfig.mapper.ConfigMqttMapper;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.math.BigDecimal;
+import java.util.*;
 import java.util.stream.Collectors;
 
 /**
@@ -28,6 +38,18 @@ public class BilletBasicInfoServiceImpl extends ServiceImpl<BilletBasicInfoMappe
     @Autowired
     private IBilletHotsendTypeConfigService billetHotsendTypeConfigService;
 
+    @Autowired
+    private BilletBasicInfoMapper billetBasicInfoMapper;
+
+    @Autowired
+    private ConfigMqttMapper configMqttMapper;
+
+    @Autowired
+    BilletRulerConfigMapper billetRulerConfigMapper;
+
+    @Autowired
+    BilletAssemblyNumberMapper billetAssemblyNumberMapper;
+
     @Override
     public void insertBatch(String sid, RulerDefaultConfig rulerDefaultConfig) {
         for (int i = 0; i < 4; i++) {
@@ -38,6 +60,227 @@ public class BilletBasicInfoServiceImpl extends ServiceImpl<BilletBasicInfoMappe
         }
     }
 
+    @Override
+    public void addB(BilletBasicInfo billetBasicInfo, Integer number) {
+        // 查询该炉号最后一条钢坯记录
+        BilletBasicInfo lastBilletBasicInfo = billetBasicInfoMapper.selectOne(
+                new QueryWrapper<BilletBasicInfo>()
+                        .eq("ccm_no", billetBasicInfo.getCcmNo())
+                        .eq("heat_no", billetBasicInfo.getHeatNo())
+                        .orderByDesc("create_time").last("limit 1")
+        );
+        // 棒线标识
+        String belongTable = billetBasicInfo.getBelongTable();
+        String btchid = billetBasicInfo.getBhtcId();
+        if (oConvertUtils.isNotEmpty(lastBilletBasicInfo)) {
+            billetBasicInfo = lastBilletBasicInfo;
+            billetBasicInfo.setId(null);
+            billetBasicInfo.setCreateTime(null);
+            billetBasicInfo.setUpdateBy(null);
+            billetBasicInfo.setUpdateTime(null);
+            billetBasicInfo.setCreateBy(null);
+            billetBasicInfo.setCutStartTime(null);
+            billetBasicInfo.setCutStopTime(null);
+            billetBasicInfo.setBelongTable(null);
+            billetBasicInfo.setBhtcId(null);
+        } else {
+            // 处理炉次实绩
+            sendHeatsActuals(billetBasicInfo);
+            billetBasicInfo.setId(null);
+            billetBasicInfo.setHeatnoIndex(0);
+            billetBasicInfo.setBelongTable(null);
+            billetBasicInfo.setBhtcId(null);
+            // 随机生成坯号
+            String billetNo = generateUniqueBilletNo(billetBasicInfo.getCcmNo(), billetBasicInfo.getHeatNo());
+            billetBasicInfo.setBilletNo(billetNo);
+        }
+        // 查询定尺规则
+        LambdaQueryWrapper<BilletRulerConfig> queryWrapperbilletRulerConfig = new LambdaQueryWrapper<BilletRulerConfig>().eq(BilletRulerConfig::getLength, billetBasicInfo.getLength());
+        BilletRulerConfig billetRulerConfig = billetRulerConfigMapper.selectOne(queryWrapperbilletRulerConfig);
+        Double weight = 0.0;
+        if(oConvertUtils.isNotEmpty(billetRulerConfig)) {
+            weight = billetRulerConfig.getWeight();
+        }
+        billetBasicInfo.setBilletWeight(weight);
+        billetBasicInfo.setWeight(weight);
+        if (billetBasicInfo.getCcmNo() == 5) { // 5#
+            // 循环处理number
+            for (int i = 0; i < number; i++) {
+                BigDecimal nextBilletNo = new BigDecimal(billetBasicInfo.getBilletNo());
+                billetBasicInfo.setBilletNo(String.valueOf(nextBilletNo.add(BigDecimal.ONE)));
+                billetBasicInfo.setHeatnoIndex(billetBasicInfo.getHeatnoIndex() + 1);
+                // 将对象转换为 Map
+                Map<String, Object> map = JSON.parseObject(JSON.toJSONString(billetBasicInfo), Map.class);
+                // 创建 MqttClientUtil 实例并调用非静态方法 pushCData
+                MqttClientUtil mqttClientUtil = new MqttClientUtil();
+                // 钢坯实绩
+                mqttClientUtil.pushCData(configMqttMapper, map, "trace/performance/billet/add");
+                // #棒一数据处理
+                if ("roll_club_one".equals(belongTable) && oConvertUtils.isNotEmpty(btchid)){ // 5号机去棒一
+                    sendRollClubOne(billetBasicInfo, btchid);
+                }
+            }
+        } else if (billetBasicInfo.getCcmNo() == 6) {
+            Integer heatnoIndex = billetBasicInfo.getHeatnoIndex();
+            // 确保每次循环正确更新 i 的值，并处理边界条件
+            for (int i = 0; i < number; i += 4) {
+                // 在此处添加具体的业务逻辑
+                String billetsNos = "";
+                String assemblyNumber = "";
+                Double weightAll = 0.0;
+                // 组坯号
+                assemblyNumber = generateUniqueAssemblyNumber(billetBasicInfo.getCcmNo(), billetBasicInfo.getHeatNo());
+                for (int j = 0; j < 4 && i + j < number ; j++) { // 4支处理
+                    System.out.println("执行钢坯实绩：" + i + "-" + j);
+                    BigDecimal nextBilletNo = new BigDecimal(billetBasicInfo.getBilletNo());
+                    billetBasicInfo.setBilletNo(String.valueOf(nextBilletNo.add(BigDecimal.ONE)));
+                    billetBasicInfo.setHeatnoIndex(heatnoIndex + 1);
+                    billetBasicInfo.setAssemblyNumber(assemblyNumber);
+                    // 将对象转换为 Map
+                    Map<String, Object> map = JSON.parseObject(JSON.toJSONString(billetBasicInfo), Map.class);
+                    // 创建 MqttClientUtil 实例并调用非静态方法 pushCData
+                    MqttClientUtil mqttClientUtil = new MqttClientUtil();
+                    // 钢坯实绩
+                    mqttClientUtil.pushCData(configMqttMapper, map, "trace/performance/billet/add");
+                    // 拼接坯号
+                    billetsNos += billetBasicInfo.getBilletNo() + ",";
+                    // 重量累加
+                    weightAll += billetBasicInfo.getBilletWeight();
+                    heatnoIndex = billetBasicInfo.getHeatnoIndex();
+                }
+                // #高线数据处理
+                if ("roll_height".equals(belongTable) && oConvertUtils.isNotEmpty(btchid)){ // 6号机去高线
+                    // 处理字符串尾部的,
+                    billetsNos = billetsNos.substring(0, billetsNos.length() - 1);
+                    // 先处理组坯
+                    sendAssemblyNumber(billetBasicInfo, billetsNos, assemblyNumber, weightAll);
+                    // 在送往高线
+                    sendRollHeight(billetBasicInfo, btchid, billetsNos);
+                }
+            }
+        }
+    }
+
+    /**
+     * 生成高线逻辑
+     * @param billetBasicInfo
+     */
+    private void sendRollHeight(BilletBasicInfo billetBasicInfo, String  btchid, String billetsNos){
+//         延迟1秒执行
+        try {
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+        // 定义一个空 map
+        Map<String, Object> mapSendInfo = new HashMap<>();
+        mapSendInfo.put("ccmNo", billetBasicInfo.getCcmNo()); // 铸机号
+        mapSendInfo.put("billetNos", billetsNos); // 坯号集合
+        mapSendInfo.put("destination", "高线"); // 高线
+        mapSendInfo.put("billetHotsendTypeConfigId", btchid); // 目的地id
+        Map<String, Object> mapBe = JSON.parseObject(JSON.toJSONString(mapSendInfo), Map.class);
+        // 创建 MqttClientUtil 实例并调用非静态方法 pushCData
+        MqttClientUtil mqttClientUtilBe = new MqttClientUtil();
+        mqttClientUtilBe.pushCData(configMqttMapper, mapBe, "syn/billetHotsendBase/save");
+    }
+
+    /**
+     * 生成组坯记录
+     * @param billetBasicInfo
+     */
+    private void sendAssemblyNumber(BilletBasicInfo billetBasicInfo, String billetsNos, String assemblyNumber, Double weightAll){
+        // 延迟1秒执行
+        try {
+            Thread.sleep(2000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+        // 定义一个空 map
+        Map<String, Object> mapSendInfo = new HashMap<>();
+        mapSendInfo.put("heatNo", billetBasicInfo.getHeatNo()); // 炉号
+        mapSendInfo.put("ccmNo", billetBasicInfo.getCcmNo()); // 铸机号
+        mapSendInfo.put("billetsNo", billetsNos); // 坯号集合
+        mapSendInfo.put("assemblyNumber", assemblyNumber); // 组坯号
+        // 获取当前时间 2025-04-22 14：56：28
+        mapSendInfo.put("assemblyTime", new Date()); // 组坯时间
+        mapSendInfo.put("length", billetBasicInfo.getLength()); // 定尺
+        mapSendInfo.put("billetsNum", 4); // 数量
+        mapSendInfo.put("billetWeight", weightAll); // 重量
+        Map<String, Object> mapBe = JSON.parseObject(JSON.toJSONString(mapSendInfo), Map.class);
+        // 创建 MqttClientUtil 实例并调用非静态方法 pushCData
+        MqttClientUtil mqttClientUtilBe = new MqttClientUtil();
+        mqttClientUtilBe.pushCData(configMqttMapper, mapBe, "trace/billet/billetAssemblyNumber/add");
+    }
+
+    /**
+     * 生成棒一逻辑
+     * @param billetBasicInfo
+     */
+    private void sendRollClubOne(BilletBasicInfo billetBasicInfo, String  btchid){
+//         延迟1秒执行
+        try {
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+        // 定义一个空 map
+        Map<String, Object> mapSendInfo = new HashMap<>();
+        mapSendInfo.put("ccmNo", billetBasicInfo.getCcmNo()); // 铸机号
+        mapSendInfo.put("billetNos", billetBasicInfo.getBilletNo()); // 坯号
+        mapSendInfo.put("destination", "棒一"); // 棒一
+        mapSendInfo.put("billetHotsendTypeConfigId", btchid); // 目的地id
+        Map<String, Object> mapBe = JSON.parseObject(JSON.toJSONString(mapSendInfo), Map.class);
+        // 创建 MqttClientUtil 实例并调用非静态方法 pushCData
+        MqttClientUtil mqttClientUtilBe = new MqttClientUtil();
+        mqttClientUtilBe.pushCData(configMqttMapper, mapBe, "syn/billetHotsendBase/save");
+    }
+
+    /**
+     * 生成炉次实绩
+     * @param billetBasicInfo
+     */
+    private void sendHeatsActuals(BilletBasicInfo billetBasicInfo){
+        HeatsActuals heatsActuals = new HeatsActuals();
+        heatsActuals.setHeatsCode(billetBasicInfo.getHeatNo());
+        heatsActuals.setCasterCode(String.valueOf(billetBasicInfo.getCcmNo()));
+        heatsActuals.setStartPourTime(DateTime.now());
+        heatsActuals.setFullLadleWeight(0.00);
+        heatsActuals.setEmptyLadleWeight(0.00);
+        heatsActuals.setMoltenSteelWeight(0.00);
+        heatsActuals.setOptype("0");
+        // 将对象转换为 Map
+        Map<String, Object> map = JSON.parseObject(JSON.toJSONString(heatsActuals), Map.class);
+        // 创建 MqttClientUtil 实例并调用非静态方法 pushCData
+        MqttClientUtil mqttClientUtil = new MqttClientUtil();
+        mqttClientUtil.pushCData(configMqttMapper, map, "trace/performance/converter/add");
+    }
+
+    /**
+     * 生成唯一的坯号
+     *
+     * @return 坯号
+     */
+    private String generateUniqueBilletNo(Integer ccmNo, String heatNo) {
+        String billetNo;
+        do {
+            billetNo = heatNo +  (int)(Math.random() * 10000);
+        } while (billetBasicInfoMapper.exists(new QueryWrapper<BilletBasicInfo>().eq("billet_no", billetNo). eq("ccm_no", ccmNo)));
+        return billetNo;
+    }
+
+    /**
+     * 生成唯一的组坯号
+     *
+     * @return 坯号
+     */
+    private String generateUniqueAssemblyNumber(Integer ccmNo, String heatNo) {
+        String assembly_number;
+        do {
+            assembly_number = heatNo +  (int)(Math.random() * 1000000);
+        } while (billetAssemblyNumberMapper.exists(new QueryWrapper<BilletAssemblyNumber>().eq("assembly_number", assembly_number). eq("ccm_no", ccmNo)));
+        return assembly_number;
+    }
+
     @Override
     public void addC(BilletBasicInfo billetBasicInfo) {
         LambdaQueryWrapper<BilletBasicInfo> queryWrapper = new LambdaQueryWrapper<BilletBasicInfo>().eq(BilletBasicInfo::getBilletNo, billetBasicInfo.getBilletNo());
@@ -69,4 +312,9 @@ public class BilletBasicInfoServiceImpl extends ServiceImpl<BilletBasicInfoMappe
         queryWrapper.eq(BilletHotsendTypeConfig::getCastMachine, ccmNo);
         return billetHotsendTypeConfigService.list(queryWrapper);
     }
+
+    @Override
+    public boolean exists(QueryWrapper<BilletBasicInfo> eq) {
+        return baseMapper.selectCount(eq) > 0;
+    }
 }

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/actualControl/billetActual/billetActual/utils/MqttClientUtil.java

@@ -0,0 +1,120 @@
+package org.jeecg.modules.actualControl.billetActual.billetActual.utils;
+
+import com.alibaba.fastjson.JSON;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.ObjectUtils;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.jeecg.modules.connConfig.entity.ConfigMqtt;
+import org.jeecg.modules.connConfig.mapper.ConfigMqttMapper;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Slf4j
+@Component
+public class MqttClientUtil implements ApplicationRunner {
+
+    public ConcurrentHashMap<String, MqttClient> mqttClients = new ConcurrentHashMap();
+
+    public MqttClient getMqttClient(ConfigMqtt configMqtt) throws MqttException {
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setUserName(configMqtt.getUsername());
+        options.setPassword(configMqtt.getPassword().toCharArray());
+        options.setConnectionTimeout(100);
+        options.setKeepAliveInterval(60);
+        options.setCleanSession(true);
+        options.setAutomaticReconnect(true);
+        StringBuffer url = new StringBuffer();
+        url.append("tcp://").append(configMqtt.getIp()).append(":").append(configMqtt.getHost());
+        MqttClient client = new MqttClient(url.toString(), configMqtt.getId(), new MemoryPersistence());
+        client.connect(options);
+        return client;
+    }
+
+    public void run(ApplicationArguments args){
+        log.info("mqtt开始进行连接！");
+//        LambdaQueryWrapper<ConfigMqtt> eq = new LambdaQueryWrapper<ConfigMqtt>().ne(ConfigMqtt::getPushOrSub,"0");
+//        List<ConfigMqtt> configMqtts = configMqttMapper.selectList(eq);
+//        for (ConfigMqtt configMqtt : configMqtts) {
+//            log.info("当前mqtt IP配置为: {}端口配置为: {}, 主题配置为: {},用户配置为: {}",configMqtt.getIp(),configMqtt.getHost(),configMqtt.getTopic(), configMqtt.getUsername());
+//            sub(configMqtt);
+//        }
+    }
+
+    public void pushCData(ConfigMqttMapper configMqttMapper, Map<String, Object> map, String topicInfo){
+        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+        map.put("time",formatter.format(new Date()));
+        ConfigMqtt configMqtt = configMqttMapper.selectOne(new LambdaQueryWrapper<ConfigMqtt>().like(ConfigMqtt::getTopic,topicInfo).eq(ConfigMqtt::getPushOrSub,"0"));
+        MqttClient mqttClient = null;
+        try {
+            mqttClient = getMqttClient(configMqtt);
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+        MqttTopic topic = mqttClient.getTopic(topicInfo);
+        MqttMessage message = new MqttMessage();
+        message.setPayload(JSON.toJSON(map).toString().getBytes());
+        message.setQos(0);
+        message.setRetained(true);
+        if (null == topic) {
+            log.error("topic is not exist");
+        }else{
+            MqttDeliveryToken token;//Delivery:配送
+            synchronized (topic) {//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁，分析见文章最后补充
+                try {
+                    token = topic.publish(message);//也是发送到执行队列中，等待执行线程执行，将消息发送到消息中间件
+                    token.waitForCompletion(1000L);
+                } catch (MqttPersistenceException e) {
+                    e.printStackTrace();
+                } catch (MqttException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+    }
+
+    public Boolean testConn(ConfigMqtt configMqtt){
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setUserName(configMqtt.getUsername());
+        options.setPassword(configMqtt.getPassword().toCharArray());
+        options.setConnectionTimeout(100);
+        options.setKeepAliveInterval(100);
+        options.setCleanSession(true);
+        options.setAutomaticReconnect(true);
+        StringBuffer url = new StringBuffer();
+        url.append("tcp://").append(configMqtt.getIp()).append(":").append(configMqtt.getHost());
+        MqttClient client = null;
+        try {
+            client = new MqttClient(url.toString(), configMqtt.getId(), new MemoryPersistence());
+            client.connect(options);
+            MqttTopic mqttTopic = client.getTopic(configMqtt.getTopic());
+            if(ObjectUtils.isEmpty(mqttTopic)){
+                return false;
+            }
+        } catch (Exception e) {
+            return false;
+        }finally {
+            if(client!=null) {
+                try {
+                    client.disconnect();
+                } catch (MqttException e) {
+                    e.printStackTrace();
+                }
+            }
+        }
+        return true;
+    }
+
+
+    public void sub(ConfigMqtt configMqtt) {
+    }
+}

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/connConfig/controller/ConfigMqttController.java

@@ -0,0 +1,195 @@
+package org.jeecg.modules.connConfig.controller;
+
+import java.util.Arrays;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+
+import org.apache.commons.lang3.ObjectUtils;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.jeecg.common.api.vo.Result;
+import org.jeecg.common.system.query.QueryGenerator;
+import org.jeecg.modules.actualControl.billetActual.billetActual.utils.MqttClientUtil;
+import org.jeecg.modules.connConfig.entity.ConfigMqtt;
+import org.jeecg.modules.connConfig.entity.MqttMsg;
+import org.jeecg.modules.connConfig.service.IConfigMqttService;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import lombok.extern.slf4j.Slf4j;
+
+import org.jeecg.common.system.base.controller.JeecgController;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.*;
+import org.springframework.web.servlet.ModelAndView;
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import org.jeecg.common.aspect.annotation.AutoLog;
+
+/**
+ * @Description: mqtt配置
+ * @Author: jeecg-boot
+ * @Date:   2023-11-15
+ * @Version: V1.0
+ */
+@Api(tags="mqtt配置")
+@RestController
+@RequestMapping("/connConfig/configMqtt/configMqtt")
+@Slf4j
+public class ConfigMqttController extends JeecgController<ConfigMqtt, IConfigMqttService> {
+	@Autowired
+	private IConfigMqttService configMqttService;
+	@Autowired
+	private MqttClientUtil mqttClientUtil;
+	
+	/**
+	 * 分页列表查询
+	 *
+	 * @param configMqtt
+	 * @param pageNo
+	 * @param pageSize
+	 * @param req
+	 * @return
+	 */
+	//@AutoLog(value = "mqtt配置-分页列表查询")
+	@ApiOperation(value="mqtt配置-分页列表查询", notes="mqtt配置-分页列表查询")
+	@GetMapping(value = "/list")
+	public Result<IPage<ConfigMqtt>> queryPageList(ConfigMqtt configMqtt,
+								   @RequestParam(name="pageNo", defaultValue="1") Integer pageNo,
+								   @RequestParam(name="pageSize", defaultValue="10") Integer pageSize,
+								   HttpServletRequest req) {
+		QueryWrapper<ConfigMqtt> queryWrapper = QueryGenerator.initQueryWrapper(configMqtt, req.getParameterMap());
+		Page<ConfigMqtt> page = new Page<ConfigMqtt>(pageNo, pageSize);
+		IPage<ConfigMqtt> pageList = configMqttService.page(page, queryWrapper);
+		return Result.OK(pageList);
+	}
+	
+	/**
+	 *   添加
+	 *
+	 * @param configMqtt
+	 * @return
+	 */
+	@AutoLog(value = "mqtt配置-添加")
+	@ApiOperation(value="mqtt配置-添加", notes="mqtt配置-添加")
+	@PostMapping(value = "/add")
+	public Result<String> add(@RequestBody ConfigMqtt configMqtt) {
+		configMqttService.save(configMqtt);
+		return Result.OK("添加成功！");
+	}
+	
+	/**
+	 *  编辑
+	 *
+	 * @param configMqtt
+	 * @return
+	 */
+	@AutoLog(value = "mqtt配置-编辑")
+	@ApiOperation(value="mqtt配置-编辑", notes="mqtt配置-编辑")
+	@RequestMapping(value = "/edit", method = {RequestMethod.PUT,RequestMethod.POST})
+	public Result<String> edit(@RequestBody ConfigMqtt configMqtt) {
+		configMqttService.updateById(configMqtt);
+		return Result.OK("编辑成功!");
+	}
+	
+	/**
+	 *   通过id删除
+	 *
+	 * @param id
+	 * @return
+	 */
+	@AutoLog(value = "mqtt配置-通过id删除")
+	@ApiOperation(value="mqtt配置-通过id删除", notes="mqtt配置-通过id删除")
+	@DeleteMapping(value = "/delete")
+	public Result<String> delete(@RequestParam(name="id",required=true) String id) {
+		configMqttService.removeById(id);
+		return Result.OK("删除成功!");
+	}
+	
+	/**
+	 *  批量删除
+	 *
+	 * @param ids
+	 * @return
+	 */
+	@AutoLog(value = "mqtt配置-批量删除")
+	@ApiOperation(value="mqtt配置-批量删除", notes="mqtt配置-批量删除")
+	@DeleteMapping(value = "/deleteBatch")
+	public Result<String> deleteBatch(@RequestParam(name="ids",required=true) String ids) {
+		this.configMqttService.removeByIds(Arrays.asList(ids.split(",")));
+		return Result.OK("批量删除成功!");
+	}
+	
+	/**
+	 * 通过id查询
+	 *
+	 * @param id
+	 * @return
+	 */
+	//@AutoLog(value = "mqtt配置-通过id查询")
+	@ApiOperation(value="mqtt配置-通过id查询", notes="mqtt配置-通过id查询")
+	@GetMapping(value = "/queryById")
+	public Result<ConfigMqtt> queryById(@RequestParam(name="id",required=true) String id) {
+		ConfigMqtt configMqtt = configMqttService.getById(id);
+		if(configMqtt==null) {
+			return Result.error("未找到对应数据");
+		}
+		return Result.OK(configMqtt);
+	}
+
+    /**
+    * 导出excel
+    *
+    * @param request
+    * @param configMqtt
+    */
+    @RequestMapping(value = "/exportXls")
+    public ModelAndView exportXls(HttpServletRequest request, ConfigMqtt configMqtt) {
+        return super.exportXls(request, configMqtt, ConfigMqtt.class, "mqtt配置");
+    }
+
+    /**
+      * 通过excel导入数据
+    *
+    * @param request
+    * @param response
+    * @return
+    */
+    @RequestMapping(value = "/importExcel", method = RequestMethod.POST)
+    public Result<?> importExcel(HttpServletRequest request, HttpServletResponse response) {
+        return super.importExcel(request, response, ConfigMqtt.class);
+    }
+
+	 @PostMapping(value = "/testConn")
+	 public Result<String> testConn(@RequestBody ConfigMqtt configMqtt){
+		 Boolean connStatus = mqttClientUtil.testConn(configMqtt);
+		 if(connStatus){
+			 return Result.OK("连接成功");
+		 }else{
+			 return Result.error("连接失败");
+		 }
+	 }
+
+	 @GetMapping(value = "/getTopics")
+	 public Result<?> getTopics(String id) {
+		 ConfigMqtt configMqtt = configMqttService.getById(id);
+		 String[] topics = null;
+		 if(ObjectUtils.isNotEmpty(configMqtt)){
+			 topics = configMqtt.getTopic().split(",");
+		 }
+		 return Result.OK(topics);
+	 }
+
+
+	 @ApiOperation(value="mqtt配置-分页列表查询", notes="mqtt配置-分页列表查询")
+	 @GetMapping(value = "/msgList")
+	 public Result<IPage<MqttMsg>> queryPageList(MqttMsg mqttMsg,
+                                                 @RequestParam(name="pageNo", defaultValue="1") Integer pageNo,
+                                                 @RequestParam(name="pageSize", defaultValue="10") Integer pageSize,
+                                                 HttpServletRequest req) {
+		 QueryWrapper<MqttMsg> queryWrapper = QueryGenerator.initQueryWrapper(mqttMsg, req.getParameterMap());
+		 Page<MqttMsg> page = new Page<MqttMsg>(pageNo, pageSize);
+		 IPage<MqttMsg> pageList = configMqttService.pageMsg(page, queryWrapper);
+		 return Result.OK(pageList);
+	 }
+}

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/connConfig/entity/ConfigMqtt.java

@@ -0,0 +1,85 @@
+package org.jeecg.modules.connConfig.entity;
+
+import java.io.Serializable;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+import com.fasterxml.jackson.annotation.JsonFormat;
+import org.springframework.format.annotation.DateTimeFormat;
+import org.jeecgframework.poi.excel.annotation.Excel;
+import org.jeecg.common.aspect.annotation.Dict;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * @Description: mqtt配置
+ * @Author: jeecg-boot
+ * @Date:   2023-11-15
+ * @Version: V1.0
+ */
+@Data
+@TableName("config_mqtt")
+@Accessors(chain = true)
+@EqualsAndHashCode(callSuper = false)
+@ApiModel(value="config_mqtt对象", description="mqtt配置")
+public class ConfigMqtt implements Serializable {
+    private static final long serialVersionUID = 1L;
+
+	/**id*/
+	@TableId(type = IdType.ASSIGN_ID)
+    @ApiModelProperty(value = "id")
+    private String id;
+	/**ip地址*/
+	@Excel(name = "ip地址", width = 15)
+    @ApiModelProperty(value = "ip地址")
+    private String ip;
+	/**端口号*/
+	@Excel(name = "端口号", width = 15)
+    @ApiModelProperty(value = "端口号")
+    private String host;
+	/**用户名*/
+	@Excel(name = "用户名", width = 15)
+    @ApiModelProperty(value = "用户名")
+    private String username;
+	/**密码*/
+	@Excel(name = "密码", width = 15)
+    @ApiModelProperty(value = "密码")
+    private String password;
+	/**主题*/
+	@Excel(name = "主题", width = 15)
+    @ApiModelProperty(value = "主题")
+    private String topic;
+	/**状态  已连接0/未连接1*/
+	@Excel(name = "状态  已连接0/未连接1", width = 15, dicCode = "connectStatus")
+	@Dict(dicCode = "connectStatus")
+    @ApiModelProperty(value = "状态  已连接0/未连接1")
+    private Integer status;
+    /**推送/订阅*/
+    @Excel(name = "推送/订阅", width = 15, dicCode = "pushOrSub")
+    @Dict(dicCode = "pushOrSub")
+    @ApiModelProperty(value = "推送/订阅")
+    private String pushOrSub;
+	/**创建人*/
+    @ApiModelProperty(value = "创建人")
+    private String createBy;
+	/**创建日期*/
+	@JsonFormat(timezone = "GMT+8",pattern = "yyyy-MM-dd HH:mm:ss")
+    @DateTimeFormat(pattern="yyyy-MM-dd HH:mm:ss")
+    @ApiModelProperty(value = "创建日期")
+    private java.util.Date createTime;
+	/**更新人*/
+    @ApiModelProperty(value = "更新人")
+    private String updateBy;
+	/**更新日期*/
+	@JsonFormat(timezone = "GMT+8",pattern = "yyyy-MM-dd HH:mm:ss")
+    @DateTimeFormat(pattern="yyyy-MM-dd HH:mm:ss")
+    @ApiModelProperty(value = "更新日期")
+    private java.util.Date updateTime;
+	/**所属部门*/
+    @ApiModelProperty(value = "所属部门")
+    private String sysOrgCode;
+}

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/connConfig/entity/MqttMsg.java

@@ -0,0 +1,38 @@
+package org.jeecg.modules.connConfig.entity;
+
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import io.swagger.annotations.ApiModel;
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+import java.util.Date;
+
+@Data
+@TableName("mqtt_msg")
+@Accessors(chain = true)
+@EqualsAndHashCode(callSuper = false)
+@ApiModel(value="mqtt消息对象", description="mqtt消息对象")
+public class MqttMsg {
+
+    @TableId(type = IdType.ASSIGN_ID)
+    @ApiModelProperty(value = "id")
+    private String id;
+
+    @ApiModelProperty(value = "mqtt连接id")
+    private String clientId;
+
+    @ApiModelProperty(value = "创建时间")
+    private Date createDate;
+
+    @ApiModelProperty(value = "信息内容")
+    private String dataValue;
+
+    @ApiModelProperty(value = "主题")
+    private String topic;
+
+}

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/connConfig/mapper/ConfigMqttMapper.java

@@ -0,0 +1,15 @@
+package org.jeecg.modules.connConfig.mapper;
+
+import org.jeecg.modules.connConfig.entity.ConfigMqtt;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+/**
+ * @Description: mqtt配置
+ * @Author: jeecg-boot
+ * @Date:   2023-11-15
+ * @Version: V1.0
+ */
+public interface ConfigMqttMapper extends BaseMapper<ConfigMqtt> {
+
+    ConfigMqtt selectByTopic(String topic);
+}

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/connConfig/mapper/MqttMsgMapper.java

@@ -0,0 +1,7 @@
+package org.jeecg.modules.connConfig.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import org.jeecg.modules.connConfig.entity.MqttMsg;
+
+public interface MqttMsgMapper extends BaseMapper<MqttMsg> {
+}

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/connConfig/mapper/xml/ConfigMqttMapper.xml

@@ -0,0 +1,11 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="org.jeecg.modules.connConfig.mapper.ConfigMqttMapper">
+
+    <select id="selectByTopic"
+            resultType="org.jeecg.modules.connConfig.entity.ConfigMqtt">
+        select *
+        from config_mqtt
+        where topic = #{topic}
+    </select>
+</mapper>

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/connConfig/mapper/xml/MqttMsgMapper.xml

@@ -0,0 +1,5 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
+<mapper namespace="org.jeecg.modules.connConfig.mapper.MqttMsgMapper">
+
+</mapper>

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/connConfig/service/IConfigMqttService.java

@@ -0,0 +1,19 @@
+package org.jeecg.modules.connConfig.service;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import org.jeecg.modules.connConfig.entity.ConfigMqtt;
+import com.baomidou.mybatisplus.extension.service.IService;
+import org.jeecg.modules.connConfig.entity.MqttMsg;
+
+/**
+ * @Description: mqtt配置
+ * @Author: jeecg-boot
+ * @Date:   2023-11-15
+ * @Version: V1.0
+ */
+public interface IConfigMqttService extends IService<ConfigMqtt> {
+
+    IPage<MqttMsg> pageMsg(Page<MqttMsg> page, QueryWrapper<MqttMsg> queryWrapper);
+}

zgztBus/jeecg-module-sbm/src/main/java/org/jeecg/modules/connConfig/service/impl/ConfigMqttServiceImpl.java

@@ -0,0 +1,43 @@
+package org.jeecg.modules.connConfig.service.impl;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import org.jeecg.modules.actualControl.billetActual.billetActual.utils.MqttClientUtil;
+import org.jeecg.modules.connConfig.entity.ConfigMqtt;
+import org.jeecg.modules.connConfig.entity.MqttMsg;
+import org.jeecg.modules.connConfig.mapper.ConfigMqttMapper;
+import org.jeecg.modules.connConfig.mapper.MqttMsgMapper;
+import org.jeecg.modules.connConfig.service.IConfigMqttService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+/**
+ * @Description: mqtt配置
+ * @Author: jeecg-boot
+ * @Date:   2023-11-15
+ * @Version: V1.0
+ */
+@Service
+public class ConfigMqttServiceImpl extends ServiceImpl<ConfigMqttMapper, ConfigMqtt> implements IConfigMqttService {
+
+    @Autowired
+    MqttClientUtil mqttClientUtil;
+
+    @Autowired
+    MqttMsgMapper mqttMsgMapper;
+
+    @Override
+    public boolean save(ConfigMqtt entity) {
+        entity.setStatus(1);
+        if(mqttClientUtil.testConn(entity))entity.setStatus(0);
+        return super.save(entity);
+    }
+
+    @Override
+    public IPage<MqttMsg> pageMsg(Page<MqttMsg> page, QueryWrapper<MqttMsg> queryWrapper) {
+        Page<MqttMsg> mqttMsgPage = mqttMsgMapper.selectPage(page, queryWrapper);
+        return mqttMsgPage;
+    }
+}