druid密码调整

pom.xml

@@ -227,6 +227,12 @@
             <version>2.5.0</version>
         </dependency>
 
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+            <version>1.2.5</version>
+        </dependency>
+
     </dependencies>
 
     <build>

src/main/java/com/lanhai/MybatisPlusGenerator.java

@@ -14,9 +14,9 @@ import com.baomidou.mybatisplus.generator.config.rules.NamingStrategy;
  */
 public class MybatisPlusGenerator {
 
-    public static final String database = "jdbc:mysql://8.130.9.244:13306/jeecg-boot";
+    public static final String database = "jdbc:mysql://172.27.17.3:13306/fx_nsp";
-    public static final String user = "root";
+    public static final String user = "user_fx";
-    public static final String passwd = "ZhiHang.20230629";
+    public static final String passwd = "user_fx";
 
     //生成代码地址
     public static final String outputdir = "D:/a5";
@@ -80,7 +80,7 @@ public class MybatisPlusGenerator {
 
     public static void main(String[] args) {
 
-        new MybatisPlusGenerator().generator("surv_station_info");//指标
+        new MybatisPlusGenerator().generator("surv_iot_manufacturer_config","surv_iot_manufacturer_info","surv_iot_protocol","surv_iot_virtual_device","surv_iot_virtual_device_group","surv_iot_virtual_device_module","surv_iot_virtual_device_net");//指标
 
     }
 

src/main/java/com/lanhai/config/DruidConfig.java

@@ -30,7 +30,7 @@ public class DruidConfig {
 
     //因为Springboot内置了servlet容器，所以没有web.xml，替代方法就是将ServletRegistrationBean注册进去
     //加入后台监控
-    @Bean  //这里其实就相当于servlet的web.xml
+    @Bean(name = "dataSource")  //这里其实就相当于servlet的web.xml
     public ServletRegistrationBean statViewServlet() {
         ServletRegistrationBean<StatViewServlet> bean =
                 new ServletRegistrationBean<StatViewServlet>(new StatViewServlet(), "/druid/*");

src/main/java/com/lanhai/controller/SurvIotManufacturerConfigController.java

@@ -0,0 +1,21 @@
+package com.lanhai.controller;
+
+
+import org.springframework.web.bind.annotation.RequestMapping;
+
+import org.springframework.stereotype.Controller;
+
+/**
+ * <p>
+ * 物联网厂家配置 前端控制器
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Controller
+@RequestMapping("/survIotManufacturerConfig")
+public class SurvIotManufacturerConfigController {
+
+}
+

src/main/java/com/lanhai/controller/SurvIotManufacturerInfoController.java

@@ -0,0 +1,21 @@
+package com.lanhai.controller;
+
+
+import org.springframework.web.bind.annotation.RequestMapping;
+
+import org.springframework.stereotype.Controller;
+
+/**
+ * <p>
+ * 物联网厂家 前端控制器
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Controller
+@RequestMapping("/survIotManufacturerInfo")
+public class SurvIotManufacturerInfoController {
+
+}
+

src/main/java/com/lanhai/controller/SurvIotProtocolController.java

@@ -0,0 +1,21 @@
+package com.lanhai.controller;
+
+
+import org.springframework.web.bind.annotation.RequestMapping;
+
+import org.springframework.stereotype.Controller;
+
+/**
+ * <p>
+ * 物联网设备协议 前端控制器
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Controller
+@RequestMapping("/survIotProtocol")
+public class SurvIotProtocolController {
+
+}
+

src/main/java/com/lanhai/controller/SurvIotVirtualDeviceController.java

@@ -0,0 +1,21 @@
+package com.lanhai.controller;
+
+
+import org.springframework.web.bind.annotation.RequestMapping;
+
+import org.springframework.stereotype.Controller;
+
+/**
+ * <p>
+ * 虚拟设备表(一般用于第三方设备对接) 前端控制器
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Controller
+@RequestMapping("/survIotVirtualDevice")
+public class SurvIotVirtualDeviceController {
+
+}
+

src/main/java/com/lanhai/controller/SurvIotVirtualDeviceGroupController.java

@@ -0,0 +1,21 @@
+package com.lanhai.controller;
+
+
+import org.springframework.web.bind.annotation.RequestMapping;
+
+import org.springframework.stereotype.Controller;
+
+/**
+ * <p>
+ * 设备模块分组 前端控制器
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Controller
+@RequestMapping("/survIotVirtualDeviceGroup")
+public class SurvIotVirtualDeviceGroupController {
+
+}
+

src/main/java/com/lanhai/controller/SurvIotVirtualDeviceModuleController.java

@@ -0,0 +1,21 @@
+package com.lanhai.controller;
+
+
+import org.springframework.web.bind.annotation.RequestMapping;
+
+import org.springframework.stereotype.Controller;
+
+/**
+ * <p>
+ * 虚拟设备模组 前端控制器
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Controller
+@RequestMapping("/survIotVirtualDeviceModule")
+public class SurvIotVirtualDeviceModuleController {
+
+}
+

src/main/java/com/lanhai/controller/SurvIotVirtualDeviceNetController.java

@@ -0,0 +1,21 @@
+package com.lanhai.controller;
+
+
+import org.springframework.web.bind.annotation.RequestMapping;
+
+import org.springframework.stereotype.Controller;
+
+/**
+ * <p>
+ * 虚拟设备网络模块 前端控制器
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Controller
+@RequestMapping("/survIotVirtualDeviceNet")
+public class SurvIotVirtualDeviceNetController {
+
+}
+

src/main/java/com/lanhai/entity/SurvIotManufacturerConfig.java

@@ -0,0 +1,106 @@
+package com.lanhai.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+import java.util.Date;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableField;
+import java.io.Serializable;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * <p>
+ * 物联网厂家配置
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class SurvIotManufacturerConfig extends Model<SurvIotManufacturerConfig> {
+
+    private static final long serialVersionUID=1L;
+
+    /**
+     * 主键
+     */
+    @TableId(value = "ID", type = IdType.ID_WORKER_STR)
+    private String id;
+
+    /**
+     * 厂家ID
+     */
+    @TableField("MA_ID")
+    private String maId;
+
+    /**
+     * 协议编号
+     */
+    @TableField("PROTOCOL_CODE")
+    private String protocolCode;
+
+    /**
+     * 配置类型;1=模型配置，2=其他
+     */
+    @TableField("CONFIG_TYPE")
+    private Integer configType;
+
+    /**
+     * 配置
+     */
+    @TableField("CONFIG_JSON")
+    private String configJson;
+
+    /**
+     * 租户号
+     */
+    @TableField("TENANT_ID")
+    private String tenantId;
+
+    /**
+     * 乐观锁
+     */
+    @TableField("RE_VISION")
+    private Integer reVision;
+
+    /**
+     * 创建人
+     */
+    @TableField("CREATE_BY")
+    private String createBy;
+
+    /**
+     * 创建时间
+     */
+    @TableField("CREATE_TIME")
+    private Date createTime;
+
+    /**
+     * 更新人
+     */
+    @TableField("UPDATE_BY")
+    private String updateBy;
+
+    /**
+     * 逻辑删除
+     */
+    @TableField("IS_DEL")
+    private Integer isDel;
+
+    /**
+     * 更新时间
+     */
+    @TableField("UPDATE_TIME")
+    private Date updateTime;
+
+
+    @Override
+    protected Serializable pkVal() {
+        return this.id;
+    }
+
+}

src/main/java/com/lanhai/entity/SurvIotManufacturerInfo.java

@@ -0,0 +1,115 @@
+package com.lanhai.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+import java.util.Date;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableField;
+import java.io.Serializable;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * <p>
+ * 物联网厂家
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class SurvIotManufacturerInfo extends Model<SurvIotManufacturerInfo> {
+
+    private static final long serialVersionUID=1L;
+
+    /**
+     * 主键
+     */
+    @TableId(value = "ID", type = IdType.ID_WORKER_STR)
+    private String id;
+
+    /**
+     * 厂家名称
+     */
+    @TableField("MA_NAME")
+    private String maName;
+
+    /**
+     * 厂家代号
+     */
+    @TableField("MA_CODE")
+    private String maCode;
+
+    /**
+     * 厂家名称地址
+     */
+    @TableField("MA_ADDR")
+    private String maAddr;
+
+    /**
+     * 联系电话
+     */
+    @TableField("MA_TEL")
+    private String maTel;
+
+    /**
+     * 备注
+     */
+    @TableField("MA_NOTES")
+    private String maNotes;
+
+    @TableField("SORT_NO")
+    private Integer sortNo;
+
+    /**
+     * 租户号
+     */
+    @TableField("TENANT_ID")
+    private String tenantId;
+
+    /**
+     * 乐观锁
+     */
+    @TableField("RE_VISION")
+    private Integer reVision;
+
+    /**
+     * 创建人
+     */
+    @TableField("CREATE_BY")
+    private String createBy;
+
+    /**
+     * 创建时间
+     */
+    @TableField("CREATE_TIME")
+    private Date createTime;
+
+    /**
+     * 更新人
+     */
+    @TableField("UPDATE_BY")
+    private String updateBy;
+
+    /**
+     * 逻辑删除
+     */
+    @TableField("IS_DEL")
+    private Integer isDel;
+
+    /**
+     * 更新时间
+     */
+    @TableField("UPDATE_TIME")
+    private Date updateTime;
+
+
+    @Override
+    protected Serializable pkVal() {
+        return this.id;
+    }
+
+}

src/main/java/com/lanhai/entity/SurvIotProtocol.java

@@ -0,0 +1,130 @@
+package com.lanhai.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+import java.util.Date;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableField;
+import java.io.Serializable;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * <p>
+ * 物联网设备协议
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class SurvIotProtocol extends Model<SurvIotProtocol> {
+
+    private static final long serialVersionUID=1L;
+
+    /**
+     * 主键
+     */
+    @TableId(value = "ID", type = IdType.ID_WORKER_STR)
+    private String id;
+
+    /**
+     * 协议名称
+     */
+    @TableField("PROTOCOL_NAME")
+    private String protocolName;
+
+    /**
+     * 所属厂家
+     */
+    @TableField("MANUFACTURER_ID")
+    private String manufacturerId;
+
+    /**
+     * 协议类型;1=设备协议
+     */
+    @TableField("PROTOCOL_TYPE")
+    private Integer protocolType;
+
+    /**
+     * 对接方式;mqtt,http,tcp
+     */
+    @TableField("INTEGRATE_TYPE")
+    private String integrateType;
+
+    /**
+     * 参数指令模板
+     */
+    @TableField("CMD_STR")
+    private String cmdStr;
+
+    /**
+     * 启动指令
+     */
+    @TableField("LAUNCH_CMD_STR")
+    private String launchCmdStr;
+
+    /**
+     * 停止指令
+     */
+    @TableField("STOP_CMD_STR")
+    private String stopCmdStr;
+
+    /**
+     * 排序
+     */
+    @TableField("SORT_NO")
+    private Integer sortNo;
+
+    /**
+     * 租户号
+     */
+    @TableField("TENANT_ID")
+    private String tenantId;
+
+    /**
+     * 乐观锁
+     */
+    @TableField("RE_VISION")
+    private Integer reVision;
+
+    /**
+     * 创建人
+     */
+    @TableField("CREATE_BY")
+    private String createBy;
+
+    /**
+     * 创建时间
+     */
+    @TableField("CREATE_TIME")
+    private Date createTime;
+
+    /**
+     * 更新人
+     */
+    @TableField("UPDATE_BY")
+    private String updateBy;
+
+    /**
+     * 逻辑删除
+     */
+    @TableField("IS_DEL")
+    private Integer isDel;
+
+    /**
+     * 更新时间
+     */
+    @TableField("UPDATE_TIME")
+    private Date updateTime;
+
+
+    @Override
+    protected Serializable pkVal() {
+        return this.id;
+    }
+
+}

src/main/java/com/lanhai/entity/SurvIotVirtualDevice.java

@@ -0,0 +1,220 @@
+package com.lanhai.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+import java.util.Date;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableField;
+import java.io.Serializable;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * <p>
+ * 虚拟设备表(一般用于第三方设备对接)
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class SurvIotVirtualDevice extends Model<SurvIotVirtualDevice> {
+
+    private static final long serialVersionUID=1L;
+
+    /**
+     * 主键
+     */
+    @TableId(value = "ID", type = IdType.ID_WORKER_STR)
+    private String id;
+
+    /**
+     * 企业id
+     */
+    @TableField("COMPANY_ID")
+    private String companyId;
+
+    /**
+     * 农场ID
+     */
+    @TableField("FARM_ID")
+    private String farmId;
+
+    /**
+     * 分组
+     */
+    @TableField("GROUP_ID")
+    private String groupId;
+
+    /**
+     * 业务关联id
+     */
+    @TableField("BUS_ID")
+    private String busId;
+
+    /**
+     * 设备物联网ID
+     */
+    @TableField("IOT_ID")
+    private String iotId;
+
+    /**
+     * 设备部署编号
+     */
+    @TableField("IOT_CODE")
+    private String iotCode;
+
+    /**
+     * 备用配置项
+     */
+    @TableField("IOT_PARAMS")
+    private String iotParams;
+
+    /**
+     * 设备名称
+     */
+    @TableField("DEVICE_NAME")
+    private String deviceName;
+
+    /**
+     * 设备主类型
+     */
+    @TableField("DEVICE_TYPE")
+    private String deviceType;
+
+    /**
+     * 设备副类型
+     */
+    @TableField("DEVICE_SEC_TYPE")
+    private String deviceSecType;
+
+    /**
+     * 协议ID
+     */
+    @TableField("PROTOCOL_ID")
+    private String protocolId;
+
+    /**
+     * 启动指令模板
+     */
+    @TableField("CMD_STR")
+    private String cmdStr;
+
+    /**
+     * 纬度
+     */
+    @TableField("DEVICE_LAT")
+    private String deviceLat;
+
+    /**
+     * 经度
+     */
+    @TableField("DEVICE_LONG")
+    private String deviceLong;
+
+    /**
+     * 经纬度
+     */
+    @TableField("DEVICE_LONGLAT")
+    private String deviceLonglat;
+
+    /**
+     * 备注
+     */
+    @TableField("DEVICE_NOTES")
+    private String deviceNotes;
+
+    /**
+     * 运行状态支持
+     */
+    @TableField("STATUS_SUPPORT")
+    private Integer statusSupport;
+
+    /**
+     * 控制支持
+     */
+    @TableField("CONTROL_SUPPORT")
+    private Integer controlSupport;
+
+    /**
+     * 灌溉记录支持
+     */
+    @TableField("RECORD_SUPPORT")
+    private Integer recordSupport;
+
+    /**
+     * 下行主题地址
+     */
+    @TableField("LOWER_URL")
+    private String lowerUrl;
+
+    /**
+     * 上行主题地址
+     */
+    @TableField("UPPER_URL")
+    private String upperUrl;
+
+    /**
+     * 排序值
+     */
+    @TableField("SORT_NO")
+    private Integer sortNo;
+
+    /**
+     * 是否启用
+     */
+    @TableField("IS_ENABLE")
+    private Integer isEnable;
+
+    /**
+     * 租户号
+     */
+    @TableField("TENANT_ID")
+    private String tenantId;
+
+    /**
+     * 乐观锁
+     */
+    @TableField("RE_VISION")
+    private Integer reVision;
+
+    /**
+     * 创建人
+     */
+    @TableField("CREATE_BY")
+    private String createBy;
+
+    /**
+     * 创建时间
+     */
+    @TableField("CREATE_TIME")
+    private Date createTime;
+
+    /**
+     * 更新人
+     */
+    @TableField("UPDATE_BY")
+    private String updateBy;
+
+    /**
+     * 逻辑删除
+     */
+    @TableField("IS_DEL")
+    private Integer isDel;
+
+    /**
+     * 更新时间
+     */
+    @TableField("UPDATE_TIME")
+    private Date updateTime;
+
+
+    @Override
+    protected Serializable pkVal() {
+        return this.id;
+    }
+
+}

src/main/java/com/lanhai/entity/SurvIotVirtualDeviceGroup.java

@@ -0,0 +1,124 @@
+package com.lanhai.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+import java.util.Date;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableField;
+import java.io.Serializable;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * <p>
+ * 设备模块分组
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class SurvIotVirtualDeviceGroup extends Model<SurvIotVirtualDeviceGroup> {
+
+    private static final long serialVersionUID=1L;
+
+    /**
+     * 主键
+     */
+    @TableId(value = "ID", type = IdType.ID_WORKER_STR)
+    private String id;
+
+    /**
+     * 企业id
+     */
+    @TableField("COMPANY_ID")
+    private String companyId;
+
+    /**
+     * 农场ID
+     */
+    @TableField("FARM_ID")
+    private String farmId;
+
+    /**
+     * 分组类型;module=模块分组
+     */
+    @TableField("GROUP_TYPE")
+    private String groupType;
+
+    /**
+     * 设备分组
+     */
+    @TableField("GROUP_NAME")
+    private String groupName;
+
+    /**
+     * 分组图标
+     */
+    @TableField("GROUP_ICON")
+    private String groupIcon;
+
+    /**
+     * 分组图标
+     */
+    @TableField("GROUP_NOTES")
+    private String groupNotes;
+
+    /**
+     * 排序值
+     */
+    @TableField("SORT_NO")
+    private Integer sortNo;
+
+    /**
+     * 租户号
+     */
+    @TableField("TENANT_ID")
+    private String tenantId;
+
+    /**
+     * 乐观锁
+     */
+    @TableField("RE_VISION")
+    private Integer reVision;
+
+    /**
+     * 创建人
+     */
+    @TableField("CREATE_BY")
+    private String createBy;
+
+    /**
+     * 创建时间
+     */
+    @TableField("CREATE_TIME")
+    private Date createTime;
+
+    /**
+     * 更新人
+     */
+    @TableField("UPDATE_BY")
+    private String updateBy;
+
+    /**
+     * 逻辑删除
+     */
+    @TableField("IS_DEL")
+    private Integer isDel;
+
+    /**
+     * 更新时间
+     */
+    @TableField("UPDATE_TIME")
+    private Date updateTime;
+
+
+    @Override
+    protected Serializable pkVal() {
+        return this.id;
+    }
+
+}

src/main/java/com/lanhai/entity/SurvIotVirtualDeviceModule.java

@@ -0,0 +1,229 @@
+package com.lanhai.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+import java.util.Date;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableField;
+import java.io.Serializable;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * <p>
+ * 虚拟设备模组
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class SurvIotVirtualDeviceModule extends Model<SurvIotVirtualDeviceModule> {
+
+    private static final long serialVersionUID=1L;
+
+    /**
+     * 主键
+     */
+    @TableId(value = "ID", type = IdType.ID_WORKER_STR)
+    private String id;
+
+    /**
+     * 企业id
+     */
+    @TableField("COMPANY_ID")
+    private String companyId;
+
+    /**
+     * 农场ID
+     */
+    @TableField("FARM_ID")
+    private String farmId;
+
+    /**
+     * 设备ID
+     */
+    @TableField("DEVICE_ID")
+    private String deviceId;
+
+    /**
+     * 模组编号
+     */
+    @TableField("MODULE_CODE")
+    private String moduleCode;
+
+    /**
+     * 模块序号
+     */
+    @TableField("MODULE_NUM")
+    private String moduleNum;
+
+    /**
+     * 模组业务ID
+     */
+    @TableField("MODULE_BUS_ID")
+    private String moduleBusId;
+
+    /**
+     * 模块分组id
+     */
+    @TableField("MODULE_GROUP_ID")
+    private String moduleGroupId;
+
+    /**
+     * 模块网络ID
+     */
+    @TableField("MODULE_NET_ID")
+    private String moduleNetId;
+
+    /**
+     * 模块主类型;sensor=传感器，meter=测量仪，control=控制器
+     */
+    @TableField("MODULE_TYPE")
+    private String moduleType;
+
+    /**
+     * 模块类型;flows=流量计，valve=电磁阀
+     */
+    @TableField("MODULE_SEC_TYPE")
+    private String moduleSecType;
+
+    /**
+     * 模块名称
+     */
+    @TableField("MODULE_NAME")
+    private String moduleName;
+
+    /**
+     * 模块备注
+     */
+    @TableField("MODULE_NOTES")
+    private String moduleNotes;
+
+    /**
+     * 1=虚拟模块，2=物理模块
+     */
+    @TableField("MODULE_MODE")
+    private String moduleMode;
+
+    /**
+     * 0=关，1=开
+     */
+    @TableField("MODULE_VIRTUAL_STATUS")
+    private String moduleVirtualStatus;
+
+    @TableField("MODULE_PATH")
+    private String modulePath;
+
+    /**
+     * 寄存器类型，4=保持寄存器，0=线圈，1=离散输入
+     */
+    @TableField("REGISTER_TYPE")
+    private Integer registerType;
+
+    /**
+     * 寄存器序号
+     */
+    @TableField("REGISTER_NUM")
+    private Integer registerNum;
+
+    /**
+     * 开指令
+     */
+    @TableField("REGISTER_CMD_ON")
+    private String registerCmdOn;
+
+    /**
+     * 关指令
+     */
+    @TableField("REGISTER_CMD_OFF")
+    private String registerCmdOff;
+
+    /**
+     * 停指令
+     */
+    @TableField("REGISTER_CMD_STOP")
+    private String registerCmdStop;
+
+    /**
+     * 回执状态开
+     */
+    @TableField("REGISTER_ON")
+    private String registerOn;
+
+    /**
+     * 回执状态关
+     */
+    @TableField("REGISTER_OFF")
+    private String registerOff;
+
+    /**
+     * 回执状态停
+     */
+    @TableField("REGISTER_STOP")
+    private String registerStop;
+
+    /**
+     * 排序值
+     */
+    @TableField("SORT_NO")
+    private Integer sortNo;
+
+    /**
+     * 是否启用
+     */
+    @TableField("IS_ENABLE")
+    private Integer isEnable;
+
+    /**
+     * 租户号
+     */
+    @TableField("TENANT_ID")
+    private String tenantId;
+
+    /**
+     * 乐观锁
+     */
+    @TableField("RE_VISION")
+    private Integer reVision;
+
+    /**
+     * 创建人
+     */
+    @TableField("CREATE_BY")
+    private String createBy;
+
+    /**
+     * 创建时间
+     */
+    @TableField("CREATE_TIME")
+    private Date createTime;
+
+    /**
+     * 更新人
+     */
+    @TableField("UPDATE_BY")
+    private String updateBy;
+
+    /**
+     * 逻辑删除
+     */
+    @TableField("IS_DEL")
+    private Integer isDel;
+
+    /**
+     * 更新时间
+     */
+    @TableField("UPDATE_TIME")
+    private Date updateTime;
+
+
+    @Override
+    protected Serializable pkVal() {
+        return this.id;
+    }
+
+}

src/main/java/com/lanhai/entity/SurvIotVirtualDeviceNet.java

@@ -0,0 +1,166 @@
+package com.lanhai.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.extension.activerecord.Model;
+import java.util.Date;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableField;
+import java.io.Serializable;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.experimental.Accessors;
+
+/**
+ * <p>
+ * 虚拟设备网络模块
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Data
+@EqualsAndHashCode(callSuper = false)
+@Accessors(chain = true)
+public class SurvIotVirtualDeviceNet extends Model<SurvIotVirtualDeviceNet> {
+
+    private static final long serialVersionUID=1L;
+
+    /**
+     * 主键
+     */
+    @TableId(value = "ID", type = IdType.ID_WORKER_STR)
+    private String id;
+
+    /**
+     * 企业id
+     */
+    @TableField("COMPANY_ID")
+    private String companyId;
+
+    /**
+     * 农场ID
+     */
+    @TableField("FARM_ID")
+    private String farmId;
+
+    /**
+     * 入网方式;dtu=数据传输单元，wifi=无线网络，cable=有线网络，mobile=移动网络
+     */
+    @TableField("NET_TYPE")
+    private String netType;
+
+    /**
+     * 网络名称
+     */
+    @TableField("NET_NAME")
+    private String netName;
+
+    /**
+     * 网络识别码
+     */
+    @TableField("NET_IDENT_CODE")
+    private String netIdentCode;
+
+    /**
+     * 备注
+     */
+    @TableField("NET_NOTES")
+    private String netNotes;
+
+    /**
+     * 上行mqtt主题
+     */
+    @TableField("NET_UP_TOPIC")
+    private String netUpTopic;
+
+    /**
+     * 下行mqtt主题
+     */
+    @TableField("NET_DOWN_TOPIC")
+    private String netDownTopic;
+
+    /**
+     * 传输口令
+     */
+    @TableField("NET_PASS")
+    private String netPass;
+
+    /**
+     * 传出密钥
+     */
+    @TableField("NET_SECRET")
+    private String netSecret;
+
+    /**
+     * 备用配置1
+     */
+    @TableField("NET_SET_ONE")
+    private String netSetOne;
+
+    /**
+     * 备用配置2
+     */
+    @TableField("NET_SET_TWO")
+    private String netSetTwo;
+
+    /**
+     * 排序
+     */
+    @TableField("SORT_NO")
+    private Integer sortNo;
+
+    /**
+     * 是否启用
+     */
+    @TableField("IS_ENABLE")
+    private Integer isEnable;
+
+    /**
+     * 租户号
+     */
+    @TableField("TENANT_ID")
+    private String tenantId;
+
+    /**
+     * 乐观锁
+     */
+    @TableField("RE_VISION")
+    private Integer reVision;
+
+    /**
+     * 创建人
+     */
+    @TableField("CREATE_BY")
+    private String createBy;
+
+    /**
+     * 创建时间
+     */
+    @TableField("CREATE_TIME")
+    private Date createTime;
+
+    /**
+     * 更新人
+     */
+    @TableField("UPDATE_BY")
+    private String updateBy;
+
+    /**
+     * 逻辑删除
+     */
+    @TableField("IS_DEL")
+    private Integer isDel;
+
+    /**
+     * 更新时间
+     */
+    @TableField("UPDATE_TIME")
+    private Date updateTime;
+
+
+    @Override
+    protected Serializable pkVal() {
+        return this.id;
+    }
+
+}

src/main/java/com/lanhai/enums/IotInerfaceTopicType.java

@@ -0,0 +1,52 @@
+package com.lanhai.enums;
+
+
+/**
+ * 物联网设备主题头类型枚举
+ */
+public enum IotInerfaceTopicType {
+    DTU_TOPIC("/pub-","厂家DTU模块头上行"),
+    LH_IOT_TOPIC_UP("/lh-iot-up","蓝海虚拟设备头上行"),
+    LH_IOT_TOPIC_DOWN("/lh-iot-down","蓝海虚拟设备头下行"),
+    API_TOPIC("/api-","厂家api头"),
+    OTHER("other","其他")
+    ;
+
+    IotInerfaceTopicType(String code, String desc) {
+        this.code = code;
+        this.desc = desc;
+    }
+
+    private String code;
+    private String desc;
+
+    public String getCode() {
+        return code;
+    }
+
+    public void setCode(String code) {
+        this.code = code;
+    }
+
+    public String getDesc() {
+        return desc;
+    }
+
+    public void setDesc(String desc) {
+        this.desc = desc;
+    }
+
+    /**
+     * 根据type获取枚举
+     *
+     * @return
+     */
+    public static IotInerfaceTopicType valueOfCode(String code) {
+        for (IotInerfaceTopicType e : IotInerfaceTopicType.values()) {
+            if (e.getCode().equals(code)) {
+                return e;
+            }
+        }
+        return OTHER;
+    }
+}

src/main/java/com/lanhai/enums/IotProtocolType.java

@@ -0,0 +1,51 @@
+package com.lanhai.enums;
+
+
+/**
+ * 物联网设备协议类型枚举
+ */
+public enum IotProtocolType {
+    MQTT("mqtt","mqtt协议"),
+    HTTP("http","http协议"),
+    BLANK("local","无协议"),
+    OTHER("other","其他")
+    ;
+
+    IotProtocolType(String code, String desc) {
+        this.code = code;
+        this.desc = desc;
+    }
+
+    private String code;
+    private String desc;
+
+    public String getCode() {
+        return code;
+    }
+
+    public void setCode(String code) {
+        this.code = code;
+    }
+
+    public String getDesc() {
+        return desc;
+    }
+
+    public void setDesc(String desc) {
+        this.desc = desc;
+    }
+
+    /**
+     * 根据type获取枚举
+     *
+     * @return
+     */
+    public static IotProtocolType valueOfCode(String code) {
+        for (IotProtocolType e : IotProtocolType.values()) {
+            if (e.getCode().equals(code)) {
+                return e;
+            }
+        }
+        return OTHER;
+    }
+}

src/main/java/com/lanhai/mapper/SurvIotManufacturerConfigMapper.java

@@ -0,0 +1,16 @@
+package com.lanhai.mapper;
+
+import com.lanhai.entity.SurvIotManufacturerConfig;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+/**
+ * <p>
+ * 物联网厂家配置 Mapper 接口
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface SurvIotManufacturerConfigMapper extends BaseMapper<SurvIotManufacturerConfig> {
+
+}

src/main/java/com/lanhai/mapper/SurvIotManufacturerInfoMapper.java

@@ -0,0 +1,16 @@
+package com.lanhai.mapper;
+
+import com.lanhai.entity.SurvIotManufacturerInfo;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+/**
+ * <p>
+ * 物联网厂家 Mapper 接口
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface SurvIotManufacturerInfoMapper extends BaseMapper<SurvIotManufacturerInfo> {
+
+}

src/main/java/com/lanhai/mapper/SurvIotProtocolMapper.java

@@ -0,0 +1,16 @@
+package com.lanhai.mapper;
+
+import com.lanhai.entity.SurvIotProtocol;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+/**
+ * <p>
+ * 物联网设备协议 Mapper 接口
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface SurvIotProtocolMapper extends BaseMapper<SurvIotProtocol> {
+
+}

src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceGroupMapper.java

@@ -0,0 +1,16 @@
+package com.lanhai.mapper;
+
+import com.lanhai.entity.SurvIotVirtualDeviceGroup;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+/**
+ * <p>
+ * 设备模块分组 Mapper 接口
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface SurvIotVirtualDeviceGroupMapper extends BaseMapper<SurvIotVirtualDeviceGroup> {
+
+}

src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceMapper.java

@@ -0,0 +1,19 @@
+package com.lanhai.mapper;
+
+import com.lanhai.entity.SurvIotVirtualDevice;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+import java.util.List;
+
+/**
+ * <p>
+ * 虚拟设备表(一般用于第三方设备对接) Mapper 接口
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface SurvIotVirtualDeviceMapper extends BaseMapper<SurvIotVirtualDevice> {
+
+    List<String> getAllTopic();
+}

src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceModuleMapper.java

@@ -0,0 +1,16 @@
+package com.lanhai.mapper;
+
+import com.lanhai.entity.SurvIotVirtualDeviceModule;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+/**
+ * <p>
+ * 虚拟设备模组 Mapper 接口
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface SurvIotVirtualDeviceModuleMapper extends BaseMapper<SurvIotVirtualDeviceModule> {
+
+}

src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceNetMapper.java

@@ -0,0 +1,16 @@
+package com.lanhai.mapper;
+
+import com.lanhai.entity.SurvIotVirtualDeviceNet;
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+
+/**
+ * <p>
+ * 虚拟设备网络模块 Mapper 接口
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface SurvIotVirtualDeviceNetMapper extends BaseMapper<SurvIotVirtualDeviceNet> {
+
+}

src/main/java/com/lanhai/mapper/xml/SurvIotManufacturerConfigMapper.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="com.lanhai.mapper.SurvIotManufacturerConfigMapper">
+
+</mapper>

src/main/java/com/lanhai/mapper/xml/SurvIotManufacturerInfoMapper.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="com.lanhai.mapper.SurvIotManufacturerInfoMapper">
+
+</mapper>

src/main/java/com/lanhai/mapper/xml/SurvIotProtocolMapper.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="com.lanhai.mapper.SurvIotProtocolMapper">
+
+</mapper>

src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceGroupMapper.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="com.lanhai.mapper.SurvIotVirtualDeviceGroupMapper">
+
+</mapper>

src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceMapper.xml

@@ -0,0 +1,8 @@
+<?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="com.lanhai.mapper.SurvIotVirtualDeviceMapper">
+
+    <select id="getAllTopic" resultType="java.lang.String">
+        select NET_UP_TOPIC from surv_iot_virtual_device_net;
+    </select>
+</mapper>

src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceModuleMapper.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="com.lanhai.mapper.SurvIotVirtualDeviceModuleMapper">
+
+</mapper>

src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceNetMapper.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="com.lanhai.mapper.SurvIotVirtualDeviceNetMapper">
+
+</mapper>

src/main/java/com/lanhai/mqtt/DTOMqtt.java

@@ -0,0 +1,13 @@
+package com.lanhai.mqtt;
+
+import io.swagger.annotations.ApiModelProperty;
+import lombok.Data;
+
+
+@Data
+public class DTOMqtt {
+    @ApiModelProperty("主题")
+    private String topic;
+    @ApiModelProperty("发送的内容")
+    private String message;
+}

src/main/java/com/lanhai/mqtt/LhMqttMsg.java

@@ -0,0 +1,34 @@
+package com.lanhai.mqtt;
+
+import lombok.Data;
+
+import java.io.Serializable;
+
+@Data
+public class LhMqttMsg implements Serializable {
+
+    private static final long serialVersionUID = -8303548938481407659L;
+
+    /**
+     * MD5值：MD5_lower(content + timestamp)
+     */
+    private String md5;
+
+    /**
+     * 消息内容
+     */
+    private String content = "";
+
+    /**
+     * 消息内容
+     */
+    private byte[] byteContent;
+
+    /**
+     * 时间戳
+     */
+    private Long timestamp;
+
+
+}
+

src/main/java/com/lanhai/mqtt/MqttConfig.java

@@ -0,0 +1,80 @@
+package com.lanhai.mqtt;
+
+import cn.hutool.core.lang.UUID;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.AutoConfigureOrder;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.DependsOn;
+import org.springframework.core.env.Environment;
+
+@Slf4j
+@Configuration
+@ConditionalOnProperty(
+        prefix = "iot.mq",
+        name = "enabled",
+        havingValue = "true",
+        matchIfMissing = true // 默认注册，除非显式设置为 false
+)
+@DependsOn({
+        "dataSource",                    // 数据源
+        "sqlSessionFactory",             // SQL 会话工厂
+        "sqlSessionTemplate",            // SQL 会话模板
+        "transactionManager",            // 事务管理器
+        "paginationInterceptor",        // 分页拦截器
+})
+public class MqttConfig {
+
+    @Value("${mqtt.host}")
+    public String host;
+    @Value("${mqtt.username}")
+    public String username;
+    @Value("${mqtt.password}")
+    public String password;
+//    @Value("${mqtt.clientId}")
+    public String clientId = "FxGather-" + UUID.fastUUID().toString().replaceAll("-","").substring(0,5);
+    @Value("${mqtt.timeout}")
+    public int timeOut;
+    @Value("${mqtt.keepalive}")
+    public int keepAlive;
+
+    @Value("${mqtt.clearSession}")
+    public boolean clearSession;
+
+    @Autowired
+    private Environment env;
+
+    @Bean//注入Spring
+    public MyMqttClient myMqttClient() {
+        MyMqttClient myMqttClient = new MyMqttClient(host, username, password, clientId, timeOut, keepAlive, clearSession);
+//        String activeProfile = env.getActiveProfiles()[0];
+        //正式环境才连接mqtt
+//        if(activeProfile.contains("prod")){
+            log.warn("-------------正式环境,MQTT启动初始化---------");
+            for (int i = 0; i < 10; i++) {
+                try {
+                    myMqttClient.connect();
+                    // 这里可以订阅主题，推荐放到 MqttCallbackExtended.connectComplete方法中
+                    //myMqttClient.subscribe("ABC", 1);
+                    return myMqttClient;
+                } catch (MqttException e) {
+                    log.error("== MqttConfig ==> MQTT connect exception, connect time = {}", i);
+                    log.error("=========mqtt连接异常："+e.getMessage());
+                    try {
+                        Thread.sleep(2000);
+                    } catch (InterruptedException e1) {
+                        e1.printStackTrace();
+                    }
+                }
+            }
+//        }else{
+//            log.warn("-------------非正式环境跳过Mqtt订阅---------");
+//        }
+        return myMqttClient;
+    }
+
+}

src/main/java/com/lanhai/mqtt/MqttService.java

@@ -0,0 +1,45 @@
+package com.lanhai.mqtt;
+
+
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+
+@ConditionalOnProperty(
+        prefix = "iot.mq",
+        name = "enabled",
+        havingValue = "true",
+        matchIfMissing = false // 默认注册，除非显式设置为 false
+)
+public interface MqttService {
+
+    /**
+     * 添加订阅主题
+     *
+     * @param topic 主题名称
+     */
+    void addTopic(String topic);
+
+    /**
+     * 取消订阅主题
+     *
+     * @param topic 主题名称
+     */
+    void removeTopic(String topic);
+
+    /**
+     * 发布主题消息内容
+     *
+     * @param msgContent
+     * @param topic
+     */
+    void publish(String topic,String msgContent);
+
+    /**
+     * 发布主题消息内容
+     *
+     * @param msgContent
+     * @param topic
+     */
+    void publish(String topic,byte[] msgContent);
+
+}
+

src/main/java/com/lanhai/mqtt/MyMqttCallback.java

@@ -0,0 +1,154 @@
+package com.lanhai.mqtt;
+
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.lanhai.entity.SurvAlertRecord;
+import com.lanhai.enums.IotInerfaceTopicType;
+import com.lanhai.mapper.SurvAlertRecordMapper;
+import com.lanhai.mapper.SurvIotVirtualDeviceMapper;
+import com.lanhai.service.ISurvIotVirtualDeviceService;
+import com.lanhai.service.Impl.SurvIotVirtualDeviceServiceImpl;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import com.lanhai.util.SpringContextUtil;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+import java.sql.Wrapper;
+import java.util.List;
+import java.util.Random;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class MyMqttCallback implements MqttCallbackExtended {
+
+    //手动注入
+//    private MqttConfig mqttConfig = SpringContextUtil.getBean(MqttConfig.class);
+
+    private SurvIotVirtualDeviceServiceImpl deviceService = SpringContextUtil.getBean("survIotVirtualDeviceServiceImpl", SurvIotVirtualDeviceServiceImpl.class);;
+
+
+//    private static RedisTemplate redisTemplate = SpringContextUtil.getBean("redisTemplate", RedisTemplate.class);
+
+
+    private MyMqttClient myMqttClient;
+
+    public MyMqttCallback(MyMqttClient myMqttClient) {
+        this.myMqttClient = myMqttClient;
+    }
+
+    /**
+     * MQTT Broker连接成功时被调用的方法。在该方法中可以执行 订阅系统约定的主题（推荐使用）。
+     * 如果 MQTT Broker断开连接之后又重新连接成功时，主题也需要再次订阅，将重新订阅主题放在连接成功后的回调方法中比较合理。
+     *
+     * @param reconnect
+     * @param serverURI MQTT Broker的url
+     */
+    @Override
+    public void connectComplete(boolean reconnect, String serverURI) {
+        try {
+            String connectMode = reconnect ? "重连" : "直连";
+            log.info("== MyMqttCallback ==> MQTT 连接成功，连接方式：{}，serverURI：{}", connectMode, serverURI);
+            //订阅主题
+            //查询所有需要订阅的mqtt主题
+            List<String> urlList = deviceService.getAllMqttTopic();
+            //去重
+            urlList = urlList.stream()
+                    .distinct()
+                    .collect(Collectors.toList());
+            for (String url : urlList) {//共享主题语法
+                myMqttClient.subscribe(url, 1);
+            }
+            log.info("== MyMqttCallback ==> 连接方式：{}，订阅主题成功，topic：{}", connectMode, String.join(",", urlList));
+        }catch (Exception e){
+            e.printStackTrace();
+        }
+    }
+
+
+    /**
+     * 丢失连接，可在这里做重连
+     * 只会调用一次
+     *
+     * @param throwable
+     */
+    @Override
+    public void connectionLost(Throwable throwable) {
+        log.error("== MyMqttCallback ==> connectionLost 连接断开，5S之后尝试重连: {}", throwable.getMessage());
+        long reconnectTimes = 1;
+        while (true) {
+            try {
+                if (MyMqttClient.getClient().isConnected()) {
+                    //判断已经重新连接成功  需要重新订阅主题 可以在这个if里面订阅主题  或者 connectComplete（方法里面）  看你们自己选择
+                    log.warn("== MyMqttCallback ==> mqtt reconnect success end  重新连接  重新订阅成功");
+                    return;
+                }
+                reconnectTimes += 1;
+                log.warn("== MyMqttCallback ==> mqtt reconnect times = {} try again...  mqtt重新连接时间 {}", reconnectTimes, reconnectTimes);
+                MyMqttClient.getClient().reconnect();
+            } catch (MqttException e) {
+                log.error("== MyMqttCallback ==> mqtt断连异常", e);
+            }
+            try {
+                Thread.sleep(5000);
+            } catch (InterruptedException e1) {
+            }
+        }
+    }
+
+    /**
+     * 接收到消息（subscribe订阅的主题消息）时被调用的方法
+     *
+     * @param topic
+     * @param mqttMessage
+     * @throws Exception 后得到的消息会执行到这里面
+     */
+    @Override
+    public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
+        String o1 = new String(mqttMessage.getPayload());
+        log.info("== MyMqttCallback ==> messageArrived 接收消息主题: {}，接收消息内容: {}", topic, o1);
+        try {
+                if (topic.startsWith(IotInerfaceTopicType.DTU_TOPIC.getCode())) {
+                    //todo 处理消息逻辑，解析数据最终入库
+//                    deviceService.processPayLoad(topic,mqttMessage.getPayload());
+                } else if (topic.startsWith(IotInerfaceTopicType.LH_IOT_TOPIC_DOWN.getCode())) {//蓝海虚拟设备下行逻辑
+                    //暂时无业务
+                }
+        }catch (Exception e){
+            e.printStackTrace();
+        }
+    }
+
+    /**
+     * 消息发送（publish）完成时被调用的方法
+     *
+     * @param iMqttDeliveryToken
+     */
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
+        log.info("== MyMqttCallback ==> deliveryComplete 消息发送完成，Complete= {}", iMqttDeliveryToken.isComplete());
+    }
+
+
+    // 指定最小和最大范围
+    public static void randomSleep(int minMs, int maxMs) {
+        try {
+            Random random = new Random();
+            if (minMs >= maxMs) {
+                throw new IllegalArgumentException("最小值必须小于最大值");
+            }
+            int sleepTime = minMs + random.nextInt(maxMs - minMs + 1);
+            System.out.println("随机休眠: " + sleepTime + "ms");
+            Thread.sleep(sleepTime);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+        }
+    }
+
+}
+

src/main/java/com/lanhai/mqtt/MyMqttClient.java

@@ -0,0 +1,209 @@
+package com.lanhai.mqtt;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+
+@Slf4j
+public class MyMqttClient {
+
+    /**
+     * MQTT Broker 基本连接参数，用户名、密码为非必选参数
+     */
+    private String host;
+    private String username;
+    private String password;
+    private String clientId;
+    private int timeout;
+    private int keepalive;
+    private boolean clearSession;
+
+    /**
+     * MQTT 客户端
+     */
+    private static MqttClient client;
+
+    public static MqttClient getClient() {
+        return client;
+    }
+
+    public static void setClient(MqttClient client) {
+        MyMqttClient.client = client;
+    }
+
+    public MyMqttClient(String host, String username, String password, String clientId, int timeOut, int keepAlive, boolean clearSession) {
+        this.host = host;
+        this.username = username;
+        this.password = password;
+        this.clientId = clientId;
+        this.timeout = timeOut;
+        this.keepalive = keepAlive;
+        this.clearSession = clearSession;
+    }
+
+    /**
+     * 设置 MQTT Broker 基本连接参数
+     *
+     * @param username
+     * @param password
+     * @param timeout
+     * @param keepalive
+     * @return
+     */
+    public MqttConnectOptions setMqttConnectOptions(String username, String password, int timeout, int keepalive) {
+        MqttConnectOptions options = new MqttConnectOptions();
+        options.setUserName(username);
+        options.setPassword(password.toCharArray());
+        options.setConnectionTimeout(timeout);
+        options.setKeepAliveInterval(keepalive);
+        options.setCleanSession(clearSession);
+        options.setAutomaticReconnect(true);
+        return options;
+    }
+
+    /**
+     * 连接 MQTT Broker，得到 MqttClient连接对象
+     */
+    public void connect() throws MqttException {
+        if (client == null) {
+            client = new MqttClient(host, clientId, new MemoryPersistence());
+            // 设置回调
+            client.setCallback(new MyMqttCallback(MyMqttClient.this));
+        }
+        // 连接参数
+        MqttConnectOptions mqttConnectOptions = setMqttConnectOptions(username, password, timeout, keepalive);
+        if (!client.isConnected()) {
+            client.connect(mqttConnectOptions);
+        } else {
+            client.disconnect();
+            client.connect(mqttConnectOptions);
+        }
+        log.info("== MyMqttClient ==> MQTT connect success");//未发生异常，则连接成功
+    }
+
+    /**
+     * 发布，默认qos为0，非持久化
+     *
+     * @param pushMessage
+     * @param topic
+     */
+    public void publish(String pushMessage, String topic) {
+        publish(pushMessage, topic, 2, false);
+    }
+
+    /**
+     * 发布，默认qos为0，非持久化
+     *
+     * @param pushMessage
+     * @param topic
+     */
+    public void publish(byte[] pushMessage, String topic) {
+        publish(pushMessage, topic, 2, false);//至少一次
+    }
+
+    /**
+     * 发布消息
+     *
+     * @param pushMessage
+     * @param topic
+     * @param qos
+     * @param retained:留存
+     */
+    public void publish(String pushMessage, String topic, int qos, boolean retained) {
+        MqttMessage message = new MqttMessage();
+        message.setPayload(pushMessage.getBytes());
+        message.setQos(qos);
+        message.setRetained(retained);
+        MqttTopic mqttTopic = MyMqttClient.getClient().getTopic(topic);
+        if (null == mqttTopic) {
+            log.error("== MyMqttClient ==> topic is not exist");
+        }
+        MqttDeliveryToken token;//Delivery:配送
+        synchronized (this) {//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁，分析见文章最后补充
+            try {
+                token = mqttTopic.publish(message);//也是发送到执行队列中，等待执行线程执行，将消息发送到消息中间件
+                token.waitForCompletion(3000L);
+            } catch (MqttPersistenceException e) {
+                e.printStackTrace();
+            } catch (MqttException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+
+    /**
+     * 发布消息
+     *
+     * @param pushMessage
+     * @param topic
+     * @param qos
+     * @param retained:留存
+     */
+    public void publish(byte[] pushMessage, String topic, int qos, boolean retained) {
+        MqttMessage message = new MqttMessage();
+        message.setPayload(pushMessage);
+        message.setQos(qos);
+        message.setRetained(retained);
+        MqttTopic mqttTopic = MyMqttClient.getClient().getTopic(topic);
+        if (null == mqttTopic) {
+            log.error("== byte MyMqttClient ==> topic is not exist");
+        }
+        MqttDeliveryToken token;//Delivery:配送
+        synchronized (this) {//注意：这里一定要同步，否则，在多线程publish的情况下，线程会发生死锁，分析见文章最后补充
+            try {
+                token = mqttTopic.publish(message);//也是发送到执行队列中，等待执行线程执行，将消息发送到消息中间件
+                token.waitForCompletion(1000L);
+            } catch (MqttPersistenceException e) {
+                e.printStackTrace();
+            } catch (MqttException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    /**
+     * 订阅某个主题，qos默认为0
+     *
+     * @param topic
+     */
+    public void subscribe(String topic) {
+        subscribe(topic, 2);
+    }
+
+    /**
+     * 订阅某个主题
+     *
+     * @param topic
+     * @param qos
+     */
+    public void subscribe(String topic, int qos) {
+        try {
+            MyMqttClient.getClient().subscribe(topic, qos);
+        } catch (MqttException e) {
+            e.printStackTrace();
+        }
+        log.info("== MyMqttClient ==> 订阅主题成功：topic = {}， qos = {}", topic, qos);
+    }
+
+
+    /**
+     * 取消订阅主题
+     *
+     * @param topic 主题名称
+     */
+    public void cleanTopic(String topic) {
+        if (client != null && client.isConnected()) {
+            try {
+                client.unsubscribe(topic);
+            } catch (MqttException e) {
+                e.printStackTrace();
+            }
+        } else {
+            log.error("== MyMqttClient ==> 取消订阅失败！");
+        }
+        log.info("== MyMqttClient ==> 取消订阅主题成功：topic = {}", topic);
+    }
+
+}
+

src/main/java/com/lanhai/mqtt/MyMqttController.java

@@ -0,0 +1,37 @@
+package com.lanhai.mqtt;
+
+import io.swagger.annotations.Api;
+import io.swagger.annotations.ApiOperation;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@RestController
+@RequestMapping("/mqtt")
+@Api(value = "MyMqttController", tags = {"MQTT相关操作接口"})
+public class MyMqttController {
+    @Autowired(required = false)
+    private MqttService mqttService;
+
+    @GetMapping("/addTopic")
+    @ApiOperation(value = "添加订阅主题接口")
+    public void addTopic(String topic) {
+        mqttService.addTopic(topic);
+    }
+
+    @GetMapping("/removeTopic")
+    @ApiOperation(value = "取消订阅主题接口")
+    public void removeTopic(String topic) {
+        mqttService.removeTopic(topic);
+    }
+
+    @PostMapping("/removeTopic")
+    @ApiOperation(value = "发布主题消息内容接口")
+    public void removeTopic(String msgContent, String topic) {
+        mqttService.publish(msgContent, topic);
+    }
+
+}
+

src/main/java/com/lanhai/mqtt/impl/MqttServiceImpl.java

@@ -0,0 +1,49 @@
+package com.lanhai.mqtt.impl;
+
+import com.lanhai.mqtt.MqttService;
+import com.lanhai.mqtt.MyMqttClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.stereotype.Service;
+
+@ConditionalOnProperty(
+        prefix = "iot.mq",
+        name = "enabled",
+        havingValue = "true",
+        matchIfMissing = false // 默认注册，除非显式设置为 false
+)
+@Service
+public class MqttServiceImpl implements MqttService {
+
+    @Autowired
+    private MyMqttClient myMqttClient;
+
+    @Override
+    public void addTopic(String topic) {
+        myMqttClient.subscribe(topic);
+    }
+
+    @Override
+    public void removeTopic(String topic) {
+        myMqttClient.cleanTopic(topic);
+    }
+
+    @Override
+    public void publish(String topic,String msgContent) {
+        //MyXxxMqttMsg 转Json
+//        LhMqttMsg myXxxMqttMsg = new LhMqttMsg();
+//        myXxxMqttMsg.setContent(msgContent);
+//        myXxxMqttMsg.setTimestamp(System.currentTimeMillis());
+//        // TODO Md5值
+//        myXxxMqttMsg.setMd5(UUID.randomUUID().toString());
+//        String msgJson = JSON.toJSONString(myXxxMqttMsg);
+
+        //发布消息
+        myMqttClient.publish(msgContent, topic);
+    }
+
+    @Override
+    public void publish(String topic,byte[] msgContent) {
+        myMqttClient.publish(msgContent, topic);
+    }
+}

src/main/java/com/lanhai/service/ISurvIotManufacturerConfigService.java

@@ -0,0 +1,16 @@
+package com.lanhai.service;
+
+import com.lanhai.entity.SurvIotManufacturerConfig;
+import com.baomidou.mybatisplus.extension.service.IService;
+
+/**
+ * <p>
+ * 物联网厂家配置 服务类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface ISurvIotManufacturerConfigService extends IService<SurvIotManufacturerConfig> {
+
+}

src/main/java/com/lanhai/service/ISurvIotManufacturerInfoService.java

@@ -0,0 +1,16 @@
+package com.lanhai.service;
+
+import com.lanhai.entity.SurvIotManufacturerInfo;
+import com.baomidou.mybatisplus.extension.service.IService;
+
+/**
+ * <p>
+ * 物联网厂家 服务类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface ISurvIotManufacturerInfoService extends IService<SurvIotManufacturerInfo> {
+
+}

src/main/java/com/lanhai/service/ISurvIotProtocolService.java

@@ -0,0 +1,16 @@
+package com.lanhai.service;
+
+import com.lanhai.entity.SurvIotProtocol;
+import com.baomidou.mybatisplus.extension.service.IService;
+
+/**
+ * <p>
+ * 物联网设备协议 服务类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface ISurvIotProtocolService extends IService<SurvIotProtocol> {
+
+}

src/main/java/com/lanhai/service/ISurvIotVirtualDeviceGroupService.java

@@ -0,0 +1,16 @@
+package com.lanhai.service;
+
+import com.lanhai.entity.SurvIotVirtualDeviceGroup;
+import com.baomidou.mybatisplus.extension.service.IService;
+
+/**
+ * <p>
+ * 设备模块分组 服务类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface ISurvIotVirtualDeviceGroupService extends IService<SurvIotVirtualDeviceGroup> {
+
+}

src/main/java/com/lanhai/service/ISurvIotVirtualDeviceModuleService.java

@@ -0,0 +1,16 @@
+package com.lanhai.service;
+
+import com.lanhai.entity.SurvIotVirtualDeviceModule;
+import com.baomidou.mybatisplus.extension.service.IService;
+
+/**
+ * <p>
+ * 虚拟设备模组 服务类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface ISurvIotVirtualDeviceModuleService extends IService<SurvIotVirtualDeviceModule> {
+
+}

src/main/java/com/lanhai/service/ISurvIotVirtualDeviceNetService.java

@@ -0,0 +1,16 @@
+package com.lanhai.service;
+
+import com.lanhai.entity.SurvIotVirtualDeviceNet;
+import com.baomidou.mybatisplus.extension.service.IService;
+
+/**
+ * <p>
+ * 虚拟设备网络模块 服务类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface ISurvIotVirtualDeviceNetService extends IService<SurvIotVirtualDeviceNet> {
+
+}

src/main/java/com/lanhai/service/ISurvIotVirtualDeviceService.java

@@ -0,0 +1,21 @@
+package com.lanhai.service;
+
+import com.lanhai.entity.SurvIotVirtualDevice;
+import com.baomidou.mybatisplus.extension.service.IService;
+
+import java.util.List;
+
+/**
+ * <p>
+ * 虚拟设备表(一般用于第三方设备对接) 服务类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+public interface ISurvIotVirtualDeviceService extends IService<SurvIotVirtualDevice> {
+
+    List<String> getAllMqttTopic();
+
+    void processPayLoad(String topic, byte[] payload);
+}

src/main/java/com/lanhai/service/Impl/SurvDeviceDeployServiceImpl.java

@@ -8,6 +8,7 @@ import com.lanhai.service.ISurvDeviceDeployService;
 import org.springframework.stereotype.Service;
 
 import java.util.Collections;
+import java.util.Date;
 import java.util.List;
 
 /**
@@ -45,4 +46,16 @@ public class SurvDeviceDeployServiceImpl extends ServiceImpl<SurvDeviceDeployMap
         List<SurvDeviceDeploy> list = lambdaQuery().in(SurvDeviceDeploy::getId,deployId).list();
         return list;
     }
+
+    /**
+     * 根据设备号刷新更新时间
+     * @param deploys
+     */
+    public void refreshLastSyncTimeByDeployCode(List<String> deploys) {
+        if (!deploys.isEmpty()) {
+            lambdaUpdate()
+                    .eq(SurvDeviceDeploy::getRunStatus, 0)
+                    .in(SurvDeviceDeploy::getDeployCode, deploys).set(SurvDeviceDeploy::getLastsyncTime, new Date()).update();
+        }
+    }
 }

src/main/java/com/lanhai/service/Impl/SurvHisdataOrientwaterServiceImpl.java

@@ -376,4 +376,6 @@ public class SurvHisdataOrientwaterServiceImpl extends ServiceImpl<SurvHisdataOr
 
         return oldWater;
     }
+
+
 }

src/main/java/com/lanhai/service/Impl/SurvIotManufacturerConfigServiceImpl.java

@@ -0,0 +1,20 @@
+package com.lanhai.service.Impl;
+
+import com.lanhai.entity.SurvIotManufacturerConfig;
+import com.lanhai.mapper.SurvIotManufacturerConfigMapper;
+import com.lanhai.service.ISurvIotManufacturerConfigService;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import org.springframework.stereotype.Service;
+
+/**
+ * <p>
+ * 物联网厂家配置 服务实现类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Service
+public class SurvIotManufacturerConfigServiceImpl extends ServiceImpl<SurvIotManufacturerConfigMapper, SurvIotManufacturerConfig> implements ISurvIotManufacturerConfigService {
+
+}

src/main/java/com/lanhai/service/Impl/SurvIotManufacturerInfoServiceImpl.java

@@ -0,0 +1,20 @@
+package com.lanhai.service.Impl;
+
+import com.lanhai.entity.SurvIotManufacturerInfo;
+import com.lanhai.mapper.SurvIotManufacturerInfoMapper;
+import com.lanhai.service.ISurvIotManufacturerInfoService;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import org.springframework.stereotype.Service;
+
+/**
+ * <p>
+ * 物联网厂家 服务实现类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Service
+public class SurvIotManufacturerInfoServiceImpl extends ServiceImpl<SurvIotManufacturerInfoMapper, SurvIotManufacturerInfo> implements ISurvIotManufacturerInfoService {
+
+}

src/main/java/com/lanhai/service/Impl/SurvIotProtocolServiceImpl.java

@@ -0,0 +1,20 @@
+package com.lanhai.service.Impl;
+
+import com.lanhai.entity.SurvIotProtocol;
+import com.lanhai.mapper.SurvIotProtocolMapper;
+import com.lanhai.service.ISurvIotProtocolService;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import org.springframework.stereotype.Service;
+
+/**
+ * <p>
+ * 物联网设备协议 服务实现类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Service
+public class SurvIotProtocolServiceImpl extends ServiceImpl<SurvIotProtocolMapper, SurvIotProtocol> implements ISurvIotProtocolService {
+
+}

src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceGroupServiceImpl.java

@@ -0,0 +1,20 @@
+package com.lanhai.service.Impl;
+
+import com.lanhai.entity.SurvIotVirtualDeviceGroup;
+import com.lanhai.mapper.SurvIotVirtualDeviceGroupMapper;
+import com.lanhai.service.ISurvIotVirtualDeviceGroupService;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import org.springframework.stereotype.Service;
+
+/**
+ * <p>
+ * 设备模块分组 服务实现类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Service
+public class SurvIotVirtualDeviceGroupServiceImpl extends ServiceImpl<SurvIotVirtualDeviceGroupMapper, SurvIotVirtualDeviceGroup> implements ISurvIotVirtualDeviceGroupService {
+
+}

src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceModuleServiceImpl.java

@@ -0,0 +1,32 @@
+package com.lanhai.service.Impl;
+
+import com.lanhai.entity.SurvIotVirtualDeviceModule;
+import com.lanhai.mapper.SurvIotVirtualDeviceModuleMapper;
+import com.lanhai.service.ISurvIotVirtualDeviceModuleService;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+
+/**
+ * <p>
+ * 虚拟设备模组 服务实现类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Service
+public class SurvIotVirtualDeviceModuleServiceImpl extends ServiceImpl<SurvIotVirtualDeviceModuleMapper, SurvIotVirtualDeviceModule> implements ISurvIotVirtualDeviceModuleService {
+
+    public List<SurvIotVirtualDeviceModule> getModulesByNet(String id) {
+        List<SurvIotVirtualDeviceModule> modules = lambdaQuery()
+                .eq(SurvIotVirtualDeviceModule::getModuleNetId,id)
+                .eq(SurvIotVirtualDeviceModule::getIsEnable,1)
+                .orderByAsc(SurvIotVirtualDeviceModule::getSortNo)
+                .orderByDesc(SurvIotVirtualDeviceModule::getCreateTime)
+                .list()
+                ;
+        return modules;
+    }
+}

src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceNetServiceImpl.java

@@ -0,0 +1,30 @@
+package com.lanhai.service.Impl;
+
+import com.lanhai.entity.SurvIotVirtualDeviceNet;
+import com.lanhai.mapper.SurvIotVirtualDeviceNetMapper;
+import com.lanhai.service.ISurvIotVirtualDeviceNetService;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import org.springframework.stereotype.Service;
+
+/**
+ * <p>
+ * 虚拟设备网络模块 服务实现类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Service
+public class SurvIotVirtualDeviceNetServiceImpl extends ServiceImpl<SurvIotVirtualDeviceNetMapper, SurvIotVirtualDeviceNet> implements ISurvIotVirtualDeviceNetService {
+
+    public SurvIotVirtualDeviceNet matchNetSet(String ident) {
+        SurvIotVirtualDeviceNet net = lambdaQuery()
+                .eq(SurvIotVirtualDeviceNet::getNetIdentCode, ident)
+                .eq(SurvIotVirtualDeviceNet::getIsEnable,1)
+                .orderByDesc(SurvIotVirtualDeviceNet::getCreateTime)
+                .last("limit 1")
+                .one()
+                ;
+        return net;
+    }
+}

src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceServiceImpl.java

@@ -0,0 +1,238 @@
+package com.lanhai.service.Impl;
+
+import cn.hutool.json.JSONUtil;
+import com.lanhai.constant.IotConstants;
+import com.lanhai.constant.PollutionConstants;
+import com.lanhai.entity.*;
+import com.lanhai.enums.IotInerfaceTopicType;
+import com.lanhai.enums.IotProtocolType;
+import com.lanhai.mapper.SurvIotVirtualDeviceMapper;
+import com.lanhai.service.ISurvIotVirtualDeviceService;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
+import org.springframework.stereotype.Service;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * <p>
+ * 虚拟设备表(一般用于第三方设备对接) 服务实现类
+ * </p>
+ *
+ * @author ${author}
+ * @since 2025-12-20
+ */
+@Service
+@Slf4j
+public class SurvIotVirtualDeviceServiceImpl extends ServiceImpl<SurvIotVirtualDeviceMapper, SurvIotVirtualDevice> implements ISurvIotVirtualDeviceService {
+
+    @Autowired
+    private SurvDeviceDeployServiceImpl deviceDeployService;
+    @Autowired
+    private SurvIotVirtualDeviceNetServiceImpl netService;
+    @Autowired
+    private SurvIotVirtualDeviceModuleServiceImpl moduleService;
+    @Autowired
+    private SurvTransdataOrientwaterServiceImpl orientwaterService;
+    @Autowired
+    private SurvTransdataLivestockwaterServiceImpl livestockwaterService;
+
+    @Override
+    public List<String> getAllMqttTopic() {
+        List<String> topics = new ArrayList<>();
+        //已部署设备中配置的主题
+        List<SurvDeviceDeploy> deploys = deviceDeployService.lambdaQuery()
+                .eq(SurvDeviceDeploy::getIsDel,0)
+                .eq(SurvDeviceDeploy::getRunStatus,0)
+                .eq(SurvDeviceDeploy::getProtocolType, IotProtocolType.MQTT.getCode())
+                .list();
+        if(!deploys.isEmpty()){
+            for (SurvDeviceDeploy deploy : deploys) {
+                if(StringUtils.isNotBlank(deploy.getDeviceUrl())){
+                    topics.add(deploy.getDeviceUrl());
+                }
+                if(StringUtils.isNotBlank(deploy.getDeviceIotUrl())){
+                    topics.add(deploy.getDeviceIotUrl());
+                }
+            }
+        }
+        //虚拟设备中需要订阅的主题
+        List<SurvIotVirtualDevice> devices =  lambdaQuery().nested(wrapper->
+                wrapper.isNotNull(SurvIotVirtualDevice::getUpperUrl)
+                        .or()
+                        .isNotNull(SurvIotVirtualDevice::getLowerUrl)
+        ).list();
+        if(!devices.isEmpty()){
+            for (SurvIotVirtualDevice device : devices) {
+                if(StringUtils.isNotBlank(device.getUpperUrl())){
+                    topics.add(device.getUpperUrl());
+                }
+                if(StringUtils.isNotBlank(device.getLowerUrl())){
+                    topics.add(device.getLowerUrl());
+                }
+            }
+        }
+        //物联网模块中的主题
+        List<SurvIotVirtualDeviceNet> nets =  netService.lambdaQuery()
+                .nested(wrapper->
+                        wrapper.isNotNull(SurvIotVirtualDeviceNet::getNetUpTopic)
+                                .or()
+                                .isNotNull(SurvIotVirtualDeviceNet::getNetDownTopic)
+                ).list();
+        ;
+        if(!nets.isEmpty()){
+            for (SurvIotVirtualDeviceNet net : nets) {
+                if(StringUtils.isNotBlank(net.getNetUpTopic())){
+                    topics.add(net.getNetUpTopic());
+                }
+                if(StringUtils.isNotBlank(net.getNetDownTopic())){
+                    topics.add(net.getNetDownTopic());
+                }
+            }
+        }
+        return topics;
+    }
+
+
+
+    @Override
+    public void processPayLoad(String topic, byte[] payload) {
+        try {
+            String ident = topic.replace(IotInerfaceTopicType.DTU_TOPIC.getCode(), "");
+            String orgStr = new String(payload);
+            log.warn("主题：{}，原文：{}", topic,orgStr);
+            if (StringUtils.isNotBlank(orgStr)) {
+                //todo 解析数据，入库
+
+                   //step 1 识别设备，
+                String deployId = "";//设备id
+                String deployCode = "";//设备编号
+                String deployName = "";//设备名称
+                String stationCode = "";//站点编号
+                String stationName = "";//站点名称
+                   //step 2 数据种类，
+                  String dataType = "";
+
+
+                  //step 3 组装入库,
+                  if(PollutionConstants.WATER_ORIENT.equals(dataType)){//面源数据
+                      SurvTransdataOrientwater orientwater = new SurvTransdataOrientwater();
+
+                      orientwater.setDeployCode(deployCode);
+                      orientwater.setDeviceId(deployId);
+                      orientwater.setDataDateTime(new Date());//采集时间
+                      orientwater.setDataGatherType("realTime");
+                      orientwater.setStationCode(stationCode);
+                      orientwater.setDeviceName(deployName);
+                      orientwater.setStationName(stationName);
+                      orientwater.setCreatedBy("mqtt");
+                      //总磷
+                      orientwater.setDataWaterTp("");
+                      orientwater.setDataWaterTpSy("");//水样
+                      orientwater.setDataWaterTpLd("");//零点
+                      //总氮
+                      orientwater.setDataWaterTn("");
+                      orientwater.setDataWaterTnSy("");//水样
+                      orientwater.setDataWaterTnLd("");//零点
+                      //化学需氧量
+                      orientwater.setDataWaterCod("");
+                      orientwater.setDataWaterCodSy("");//水样
+                      orientwater.setDataWaterCodLd("");//零点
+                      //阀门状态
+                      orientwater.setFirstValveStatus("");
+                      orientwater.setSecondValveStatus("");
+                      orientwater.setThirdValveStatus("");
+
+                      //实时数据入库
+                      orientwaterService.saveData(orientwater);
+                      //历史数据入库
+
+                  } else if (PollutionConstants.WATER_LIVE.equals(dataType)) {
+                      SurvTransdataLivestockwater livestockwater = new SurvTransdataLivestockwater();
+
+                      livestockwater.setDeployCode(deployCode);
+                      livestockwater.setDeviceId(deployId);
+                      livestockwater.setDataDateTime(new Date());//采集时间
+                      livestockwater.setDataGatherType("realTime");
+                      livestockwater.setStationCode(stationCode);
+                      livestockwater.setDeviceName(deployName);
+                      livestockwater.setStationName(stationName);
+                      livestockwater.setCreatedBy("mqtt");
+                      //总磷
+                      livestockwater.setDataWaterTp("");
+                      livestockwater.setDataWaterTpSy("");//水样
+                      livestockwater.setDataWaterTpLd("");//零点
+                      //总氮
+                      livestockwater.setDataWaterTn("");
+                      livestockwater.setDataWaterTnSy("");//水样
+                      livestockwater.setDataWaterTnLd("");//零点
+                      //化学需氧量
+                      livestockwater.setDataWaterCod("");
+                      livestockwater.setDataWaterCodSy("");//水样
+                      livestockwater.setDataWaterCodLd("");//零点
+                      //阀门状态
+                      livestockwater.setFirstValveStatus("");
+                      livestockwater.setSecondValveStatus("");
+                      livestockwater.setThirdValveStatus("");
+
+                      livestockwaterService.saveData(livestockwater);
+                  }
+
+                //查询网络模块
+                List<SurvIotVirtualDeviceModule> modules = getModulesByIdent(ident);//三个球阀
+                if (!modules.isEmpty()) {
+                    for (SurvIotVirtualDeviceModule module : modules) {
+                        module.getRegisterCmdOn();//开指令
+                    }
+                }
+            } else {
+                log.error("==================主题：{}，消息为空跳过===================", topic);
+            }
+            //如果是心跳，更新设备的时间
+            String infoStr = new String(payload);
+            log.error("========{}==========更新心跳：{}===================", infoStr, ident);
+            if ("www.usr.cn".equals(infoStr)) {
+                //查询网络模块
+                List<SurvIotVirtualDeviceModule> modules = getModulesByIdent(ident);
+                if (!modules.isEmpty()) {
+                    List<String> deviceId = modules.stream().map(SurvIotVirtualDeviceModule::getDeviceId).collect(Collectors.toList());
+                    List<SurvIotVirtualDevice> devices = lambdaQuery().in(SurvIotVirtualDevice::getId, deviceId).list();
+                    List<String> deploys = devices.stream().map(SurvIotVirtualDevice::getIotId).collect(Collectors.toList());
+                    deviceDeployService.refreshLastSyncTimeByDeployCode(deploys);
+                }
+            }
+        }catch (Exception e){
+            e.printStackTrace();
+            log.error("处理厂家mqtt报文失败，"+e.getMessage());
+        }
+
+    }
+
+
+    /**
+     * 根据主题获取设备模组
+     */
+    private List<SurvIotVirtualDeviceModule> getModulesByIdent(String identCode){
+        //查询网络模块
+        SurvIotVirtualDeviceNet net = netService.matchNetSet(identCode);
+        if (net != null) {
+            List<SurvIotVirtualDeviceModule> modules = moduleService.getModulesByNet(net.getId());
+            if (!modules.isEmpty()) {
+                //模组关联的设备
+                return modules;
+            }else{
+                log.error("==================网络模块为空：{}:{}===================", identCode,net.getId());
+            }
+        } else {
+            log.error("==================网络模块未识别：{}===================", identCode);
+        }
+        return new ArrayList<>();
+    }
+}

src/main/java/com/lanhai/service/Impl/SurvTransdataLivestockwaterServiceImpl.java

@@ -150,4 +150,8 @@ public class SurvTransdataLivestockwaterServiceImpl extends ServiceImpl<SurvTran
             log.info("解析错误",e);
         }
     }
+
+    public void saveData(SurvTransdataLivestockwater livestockwater) {
+
+    }
 }

src/main/java/com/lanhai/service/Impl/SurvTransdataOrientwaterServiceImpl.java

@@ -152,4 +152,7 @@ public class SurvTransdataOrientwaterServiceImpl extends ServiceImpl<SurvTransda
 
 
     }
+
+    public void saveData(SurvTransdataOrientwater orientwater) {
+    }
 }

src/main/resources/application-local.yml

@@ -7,7 +7,7 @@ server:
 spring:
   datasource:
     driver-class-name: com.mysql.cj.jdbc.Driver
-    url: jdbc:mysql://192.168.2.3:3306/fx_nsp?useUnicode=true&useSSL=false&characterEncoding=utf8&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true
+    url: jdbc:mysql://172.27.17.3:13306/fx_nsp?useUnicode=true&useSSL=false&characterEncoding=utf8&serverTimezone=Asia/Shanghai&allowPublicKeyRetrieval=true
     username: user_fx
     password: user_fx
     type: com.alibaba.druid.pool.DruidDataSource
@@ -76,6 +76,14 @@ xxl:
       port: 10004
       logpath: /logs/xxl-job/jobhandler
       logretentiondays: 30
-
+mqtt:
+  host: tcp://mqtt.ilhzn.cn:10007
+  userName: lhzn.mqtt
+  passWord: lhzn.2025
+  qos: 1
+  clientId: fxgather-client-${random.uuid} #ClientId_local必须唯一。
+  timeout: 10 # 超时时间
+  keepalive: 30 # 保持连接时间
+  clearSession: false   # 清除会话(设置为false,断开连接，重连后使用原来的会话 保留订阅的主题，能接收离线期间的消息)
 
 

src/main/resources/application-prod.yml

@@ -76,7 +76,15 @@ xxl:
       port: 10007
       logpath: /logs/xxl-job/jobhandler
       logretentiondays: 30
-
+mqtt:
+  host: tcp://mqtt.ilhzn.cn:10007
+  userName: lhzn.mqtt
+  passWord: lhzn.2025
+  qos: 1
+  clientId: fxgather-client-${random.uuid} #ClientId_local必须唯一。
+  timeout: 10 # 超时时间
+  keepalive: 30 # 保持连接时间
+  clearSession: false   # 清除会话(设置为false,断开连接，重连后使用原来的会话 保留订阅的主题，能接收离线期间的消息)