diff --git a/pom.xml b/pom.xml index 759d41e..dfa9c12 100644 --- a/pom.xml +++ b/pom.xml @@ -227,6 +227,12 @@ 2.5.0 + + org.eclipse.paho + org.eclipse.paho.client.mqttv3 + 1.2.5 + + diff --git a/src/main/java/com/lanhai/MybatisPlusGenerator.java b/src/main/java/com/lanhai/MybatisPlusGenerator.java index bef4acf..0153f63 100644 --- a/src/main/java/com/lanhai/MybatisPlusGenerator.java +++ b/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 user = "root"; - public static final String passwd = "ZhiHang.20230629"; + public static final String database = "jdbc:mysql://172.27.17.3:13306/fx_nsp"; + public static final String user = "user_fx"; + 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");//指标 } diff --git a/src/main/java/com/lanhai/config/DruidConfig.java b/src/main/java/com/lanhai/config/DruidConfig.java index de477f3..ff7bef9 100644 --- a/src/main/java/com/lanhai/config/DruidConfig.java +++ b/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 bean = new ServletRegistrationBean(new StatViewServlet(), "/druid/*"); diff --git a/src/main/java/com/lanhai/controller/SurvIotManufacturerConfigController.java b/src/main/java/com/lanhai/controller/SurvIotManufacturerConfigController.java new file mode 100644 index 0000000..73d5e83 --- /dev/null +++ b/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; + +/** + *

+ * 物联网厂家配置 前端控制器 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Controller +@RequestMapping("/survIotManufacturerConfig") +public class SurvIotManufacturerConfigController { + +} + diff --git a/src/main/java/com/lanhai/controller/SurvIotManufacturerInfoController.java b/src/main/java/com/lanhai/controller/SurvIotManufacturerInfoController.java new file mode 100644 index 0000000..9f6b5c8 --- /dev/null +++ b/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; + +/** + *

+ * 物联网厂家 前端控制器 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Controller +@RequestMapping("/survIotManufacturerInfo") +public class SurvIotManufacturerInfoController { + +} + diff --git a/src/main/java/com/lanhai/controller/SurvIotProtocolController.java b/src/main/java/com/lanhai/controller/SurvIotProtocolController.java new file mode 100644 index 0000000..13fab34 --- /dev/null +++ b/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; + +/** + *

+ * 物联网设备协议 前端控制器 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Controller +@RequestMapping("/survIotProtocol") +public class SurvIotProtocolController { + +} + diff --git a/src/main/java/com/lanhai/controller/SurvIotVirtualDeviceController.java b/src/main/java/com/lanhai/controller/SurvIotVirtualDeviceController.java new file mode 100644 index 0000000..bead50d --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备表(一般用于第三方设备对接) 前端控制器 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Controller +@RequestMapping("/survIotVirtualDevice") +public class SurvIotVirtualDeviceController { + +} + diff --git a/src/main/java/com/lanhai/controller/SurvIotVirtualDeviceGroupController.java b/src/main/java/com/lanhai/controller/SurvIotVirtualDeviceGroupController.java new file mode 100644 index 0000000..5b29c79 --- /dev/null +++ b/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; + +/** + *

+ * 设备模块分组 前端控制器 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Controller +@RequestMapping("/survIotVirtualDeviceGroup") +public class SurvIotVirtualDeviceGroupController { + +} + diff --git a/src/main/java/com/lanhai/controller/SurvIotVirtualDeviceModuleController.java b/src/main/java/com/lanhai/controller/SurvIotVirtualDeviceModuleController.java new file mode 100644 index 0000000..031c84a --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备模组 前端控制器 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Controller +@RequestMapping("/survIotVirtualDeviceModule") +public class SurvIotVirtualDeviceModuleController { + +} + diff --git a/src/main/java/com/lanhai/controller/SurvIotVirtualDeviceNetController.java b/src/main/java/com/lanhai/controller/SurvIotVirtualDeviceNetController.java new file mode 100644 index 0000000..d37787b --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备网络模块 前端控制器 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Controller +@RequestMapping("/survIotVirtualDeviceNet") +public class SurvIotVirtualDeviceNetController { + +} + diff --git a/src/main/java/com/lanhai/entity/SurvIotManufacturerConfig.java b/src/main/java/com/lanhai/entity/SurvIotManufacturerConfig.java new file mode 100644 index 0000000..0e256d7 --- /dev/null +++ b/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; + +/** + *

+ * 物联网厂家配置 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Data +@EqualsAndHashCode(callSuper = false) +@Accessors(chain = true) +public class SurvIotManufacturerConfig extends Model { + + 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; + } + +} diff --git a/src/main/java/com/lanhai/entity/SurvIotManufacturerInfo.java b/src/main/java/com/lanhai/entity/SurvIotManufacturerInfo.java new file mode 100644 index 0000000..9ee0979 --- /dev/null +++ b/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; + +/** + *

+ * 物联网厂家 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Data +@EqualsAndHashCode(callSuper = false) +@Accessors(chain = true) +public class SurvIotManufacturerInfo extends Model { + + 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; + } + +} diff --git a/src/main/java/com/lanhai/entity/SurvIotProtocol.java b/src/main/java/com/lanhai/entity/SurvIotProtocol.java new file mode 100644 index 0000000..a7f5a85 --- /dev/null +++ b/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; + +/** + *

+ * 物联网设备协议 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Data +@EqualsAndHashCode(callSuper = false) +@Accessors(chain = true) +public class SurvIotProtocol extends Model { + + 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; + } + +} diff --git a/src/main/java/com/lanhai/entity/SurvIotVirtualDevice.java b/src/main/java/com/lanhai/entity/SurvIotVirtualDevice.java new file mode 100644 index 0000000..70f34c7 --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备表(一般用于第三方设备对接) + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Data +@EqualsAndHashCode(callSuper = false) +@Accessors(chain = true) +public class SurvIotVirtualDevice extends Model { + + 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; + } + +} diff --git a/src/main/java/com/lanhai/entity/SurvIotVirtualDeviceGroup.java b/src/main/java/com/lanhai/entity/SurvIotVirtualDeviceGroup.java new file mode 100644 index 0000000..916928c --- /dev/null +++ b/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; + +/** + *

+ * 设备模块分组 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Data +@EqualsAndHashCode(callSuper = false) +@Accessors(chain = true) +public class SurvIotVirtualDeviceGroup extends Model { + + 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; + } + +} diff --git a/src/main/java/com/lanhai/entity/SurvIotVirtualDeviceModule.java b/src/main/java/com/lanhai/entity/SurvIotVirtualDeviceModule.java new file mode 100644 index 0000000..48bc461 --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备模组 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Data +@EqualsAndHashCode(callSuper = false) +@Accessors(chain = true) +public class SurvIotVirtualDeviceModule extends Model { + + 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; + } + +} diff --git a/src/main/java/com/lanhai/entity/SurvIotVirtualDeviceNet.java b/src/main/java/com/lanhai/entity/SurvIotVirtualDeviceNet.java new file mode 100644 index 0000000..6d31c7e --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备网络模块 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Data +@EqualsAndHashCode(callSuper = false) +@Accessors(chain = true) +public class SurvIotVirtualDeviceNet extends Model { + + 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; + } + +} diff --git a/src/main/java/com/lanhai/enums/IotInerfaceTopicType.java b/src/main/java/com/lanhai/enums/IotInerfaceTopicType.java new file mode 100644 index 0000000..ecf7ab9 --- /dev/null +++ b/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; + } +} \ No newline at end of file diff --git a/src/main/java/com/lanhai/enums/IotProtocolType.java b/src/main/java/com/lanhai/enums/IotProtocolType.java new file mode 100644 index 0000000..724d02e --- /dev/null +++ b/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; + } +} \ No newline at end of file diff --git a/src/main/java/com/lanhai/mapper/SurvIotManufacturerConfigMapper.java b/src/main/java/com/lanhai/mapper/SurvIotManufacturerConfigMapper.java new file mode 100644 index 0000000..232017e --- /dev/null +++ b/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; + +/** + *

+ * 物联网厂家配置 Mapper 接口 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface SurvIotManufacturerConfigMapper extends BaseMapper { + +} diff --git a/src/main/java/com/lanhai/mapper/SurvIotManufacturerInfoMapper.java b/src/main/java/com/lanhai/mapper/SurvIotManufacturerInfoMapper.java new file mode 100644 index 0000000..c6b14b2 --- /dev/null +++ b/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; + +/** + *

+ * 物联网厂家 Mapper 接口 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface SurvIotManufacturerInfoMapper extends BaseMapper { + +} diff --git a/src/main/java/com/lanhai/mapper/SurvIotProtocolMapper.java b/src/main/java/com/lanhai/mapper/SurvIotProtocolMapper.java new file mode 100644 index 0000000..06b9484 --- /dev/null +++ b/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; + +/** + *

+ * 物联网设备协议 Mapper 接口 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface SurvIotProtocolMapper extends BaseMapper { + +} diff --git a/src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceGroupMapper.java b/src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceGroupMapper.java new file mode 100644 index 0000000..f0f2ad0 --- /dev/null +++ b/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; + +/** + *

+ * 设备模块分组 Mapper 接口 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface SurvIotVirtualDeviceGroupMapper extends BaseMapper { + +} diff --git a/src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceMapper.java b/src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceMapper.java new file mode 100644 index 0000000..93f5c8a --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备表(一般用于第三方设备对接) Mapper 接口 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface SurvIotVirtualDeviceMapper extends BaseMapper { + + List getAllTopic(); +} diff --git a/src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceModuleMapper.java b/src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceModuleMapper.java new file mode 100644 index 0000000..4d453d1 --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备模组 Mapper 接口 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface SurvIotVirtualDeviceModuleMapper extends BaseMapper { + +} diff --git a/src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceNetMapper.java b/src/main/java/com/lanhai/mapper/SurvIotVirtualDeviceNetMapper.java new file mode 100644 index 0000000..44f75f4 --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备网络模块 Mapper 接口 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface SurvIotVirtualDeviceNetMapper extends BaseMapper { + +} diff --git a/src/main/java/com/lanhai/mapper/xml/SurvIotManufacturerConfigMapper.xml b/src/main/java/com/lanhai/mapper/xml/SurvIotManufacturerConfigMapper.xml new file mode 100644 index 0000000..a28498c --- /dev/null +++ b/src/main/java/com/lanhai/mapper/xml/SurvIotManufacturerConfigMapper.xml @@ -0,0 +1,5 @@ + + + + + diff --git a/src/main/java/com/lanhai/mapper/xml/SurvIotManufacturerInfoMapper.xml b/src/main/java/com/lanhai/mapper/xml/SurvIotManufacturerInfoMapper.xml new file mode 100644 index 0000000..a8cb02e --- /dev/null +++ b/src/main/java/com/lanhai/mapper/xml/SurvIotManufacturerInfoMapper.xml @@ -0,0 +1,5 @@ + + + + + diff --git a/src/main/java/com/lanhai/mapper/xml/SurvIotProtocolMapper.xml b/src/main/java/com/lanhai/mapper/xml/SurvIotProtocolMapper.xml new file mode 100644 index 0000000..3250937 --- /dev/null +++ b/src/main/java/com/lanhai/mapper/xml/SurvIotProtocolMapper.xml @@ -0,0 +1,5 @@ + + + + + diff --git a/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceGroupMapper.xml b/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceGroupMapper.xml new file mode 100644 index 0000000..b947468 --- /dev/null +++ b/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceGroupMapper.xml @@ -0,0 +1,5 @@ + + + + + diff --git a/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceMapper.xml b/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceMapper.xml new file mode 100644 index 0000000..6a57ce4 --- /dev/null +++ b/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceMapper.xml @@ -0,0 +1,8 @@ + + + + + + diff --git a/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceModuleMapper.xml b/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceModuleMapper.xml new file mode 100644 index 0000000..9af94ba --- /dev/null +++ b/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceModuleMapper.xml @@ -0,0 +1,5 @@ + + + + + diff --git a/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceNetMapper.xml b/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceNetMapper.xml new file mode 100644 index 0000000..a9d62fa --- /dev/null +++ b/src/main/java/com/lanhai/mapper/xml/SurvIotVirtualDeviceNetMapper.xml @@ -0,0 +1,5 @@ + + + + + diff --git a/src/main/java/com/lanhai/mqtt/DTOMqtt.java b/src/main/java/com/lanhai/mqtt/DTOMqtt.java new file mode 100644 index 0000000..5e812d1 --- /dev/null +++ b/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; +} diff --git a/src/main/java/com/lanhai/mqtt/LhMqttMsg.java b/src/main/java/com/lanhai/mqtt/LhMqttMsg.java new file mode 100644 index 0000000..b44da5b --- /dev/null +++ b/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; + + +} + diff --git a/src/main/java/com/lanhai/mqtt/MqttConfig.java b/src/main/java/com/lanhai/mqtt/MqttConfig.java new file mode 100644 index 0000000..619b3a6 --- /dev/null +++ b/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; + } + +} diff --git a/src/main/java/com/lanhai/mqtt/MqttService.java b/src/main/java/com/lanhai/mqtt/MqttService.java new file mode 100644 index 0000000..613087b --- /dev/null +++ b/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); + +} + diff --git a/src/main/java/com/lanhai/mqtt/MyMqttCallback.java b/src/main/java/com/lanhai/mqtt/MyMqttCallback.java new file mode 100644 index 0000000..df7909b --- /dev/null +++ b/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 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(); + } + } + +} + diff --git a/src/main/java/com/lanhai/mqtt/MyMqttClient.java b/src/main/java/com/lanhai/mqtt/MyMqttClient.java new file mode 100644 index 0000000..30be926 --- /dev/null +++ b/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); + } + +} + diff --git a/src/main/java/com/lanhai/mqtt/MyMqttController.java b/src/main/java/com/lanhai/mqtt/MyMqttController.java new file mode 100644 index 0000000..2d54498 --- /dev/null +++ b/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); + } + +} + diff --git a/src/main/java/com/lanhai/mqtt/impl/MqttServiceImpl.java b/src/main/java/com/lanhai/mqtt/impl/MqttServiceImpl.java new file mode 100644 index 0000000..425bc67 --- /dev/null +++ b/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); + } +} diff --git a/src/main/java/com/lanhai/service/ISurvIotManufacturerConfigService.java b/src/main/java/com/lanhai/service/ISurvIotManufacturerConfigService.java new file mode 100644 index 0000000..5a989d1 --- /dev/null +++ b/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; + +/** + *

+ * 物联网厂家配置 服务类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface ISurvIotManufacturerConfigService extends IService { + +} diff --git a/src/main/java/com/lanhai/service/ISurvIotManufacturerInfoService.java b/src/main/java/com/lanhai/service/ISurvIotManufacturerInfoService.java new file mode 100644 index 0000000..b85cc35 --- /dev/null +++ b/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; + +/** + *

+ * 物联网厂家 服务类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface ISurvIotManufacturerInfoService extends IService { + +} diff --git a/src/main/java/com/lanhai/service/ISurvIotProtocolService.java b/src/main/java/com/lanhai/service/ISurvIotProtocolService.java new file mode 100644 index 0000000..3edd593 --- /dev/null +++ b/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; + +/** + *

+ * 物联网设备协议 服务类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface ISurvIotProtocolService extends IService { + +} diff --git a/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceGroupService.java b/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceGroupService.java new file mode 100644 index 0000000..94dd123 --- /dev/null +++ b/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; + +/** + *

+ * 设备模块分组 服务类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface ISurvIotVirtualDeviceGroupService extends IService { + +} diff --git a/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceModuleService.java b/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceModuleService.java new file mode 100644 index 0000000..334d9af --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备模组 服务类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface ISurvIotVirtualDeviceModuleService extends IService { + +} diff --git a/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceNetService.java b/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceNetService.java new file mode 100644 index 0000000..0aafef2 --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备网络模块 服务类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface ISurvIotVirtualDeviceNetService extends IService { + +} diff --git a/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceService.java b/src/main/java/com/lanhai/service/ISurvIotVirtualDeviceService.java new file mode 100644 index 0000000..ed9e288 --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备表(一般用于第三方设备对接) 服务类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +public interface ISurvIotVirtualDeviceService extends IService { + + List getAllMqttTopic(); + + void processPayLoad(String topic, byte[] payload); +} diff --git a/src/main/java/com/lanhai/service/Impl/SurvDeviceDeployServiceImpl.java b/src/main/java/com/lanhai/service/Impl/SurvDeviceDeployServiceImpl.java index e2ef995..af2a483 100644 --- a/src/main/java/com/lanhai/service/Impl/SurvDeviceDeployServiceImpl.java +++ b/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 list = lambdaQuery().in(SurvDeviceDeploy::getId,deployId).list(); return list; } + + /** + * 根据设备号刷新更新时间 + * @param deploys + */ + public void refreshLastSyncTimeByDeployCode(List deploys) { + if (!deploys.isEmpty()) { + lambdaUpdate() + .eq(SurvDeviceDeploy::getRunStatus, 0) + .in(SurvDeviceDeploy::getDeployCode, deploys).set(SurvDeviceDeploy::getLastsyncTime, new Date()).update(); + } + } } diff --git a/src/main/java/com/lanhai/service/Impl/SurvHisdataOrientwaterServiceImpl.java b/src/main/java/com/lanhai/service/Impl/SurvHisdataOrientwaterServiceImpl.java index fa6fc95..32df1f1 100644 --- a/src/main/java/com/lanhai/service/Impl/SurvHisdataOrientwaterServiceImpl.java +++ b/src/main/java/com/lanhai/service/Impl/SurvHisdataOrientwaterServiceImpl.java @@ -376,4 +376,6 @@ public class SurvHisdataOrientwaterServiceImpl extends ServiceImpl + * 物联网厂家配置 服务实现类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Service +public class SurvIotManufacturerConfigServiceImpl extends ServiceImpl implements ISurvIotManufacturerConfigService { + +} diff --git a/src/main/java/com/lanhai/service/Impl/SurvIotManufacturerInfoServiceImpl.java b/src/main/java/com/lanhai/service/Impl/SurvIotManufacturerInfoServiceImpl.java new file mode 100644 index 0000000..274ffa2 --- /dev/null +++ b/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; + +/** + *

+ * 物联网厂家 服务实现类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Service +public class SurvIotManufacturerInfoServiceImpl extends ServiceImpl implements ISurvIotManufacturerInfoService { + +} diff --git a/src/main/java/com/lanhai/service/Impl/SurvIotProtocolServiceImpl.java b/src/main/java/com/lanhai/service/Impl/SurvIotProtocolServiceImpl.java new file mode 100644 index 0000000..dc99455 --- /dev/null +++ b/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; + +/** + *

+ * 物联网设备协议 服务实现类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Service +public class SurvIotProtocolServiceImpl extends ServiceImpl implements ISurvIotProtocolService { + +} diff --git a/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceGroupServiceImpl.java b/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceGroupServiceImpl.java new file mode 100644 index 0000000..ace640e --- /dev/null +++ b/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; + +/** + *

+ * 设备模块分组 服务实现类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Service +public class SurvIotVirtualDeviceGroupServiceImpl extends ServiceImpl implements ISurvIotVirtualDeviceGroupService { + +} diff --git a/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceModuleServiceImpl.java b/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceModuleServiceImpl.java new file mode 100644 index 0000000..88c6639 --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备模组 服务实现类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Service +public class SurvIotVirtualDeviceModuleServiceImpl extends ServiceImpl implements ISurvIotVirtualDeviceModuleService { + + public List getModulesByNet(String id) { + List modules = lambdaQuery() + .eq(SurvIotVirtualDeviceModule::getModuleNetId,id) + .eq(SurvIotVirtualDeviceModule::getIsEnable,1) + .orderByAsc(SurvIotVirtualDeviceModule::getSortNo) + .orderByDesc(SurvIotVirtualDeviceModule::getCreateTime) + .list() + ; + return modules; + } +} diff --git a/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceNetServiceImpl.java b/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceNetServiceImpl.java new file mode 100644 index 0000000..451edb7 --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备网络模块 服务实现类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Service +public class SurvIotVirtualDeviceNetServiceImpl extends ServiceImpl 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; + } +} diff --git a/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceServiceImpl.java b/src/main/java/com/lanhai/service/Impl/SurvIotVirtualDeviceServiceImpl.java new file mode 100644 index 0000000..18b5d80 --- /dev/null +++ b/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; + +/** + *

+ * 虚拟设备表(一般用于第三方设备对接) 服务实现类 + *

+ * + * @author ${author} + * @since 2025-12-20 + */ +@Service +@Slf4j +public class SurvIotVirtualDeviceServiceImpl extends ServiceImpl 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 getAllMqttTopic() { + List topics = new ArrayList<>(); + //已部署设备中配置的主题 + List 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 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 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 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 modules = getModulesByIdent(ident); + if (!modules.isEmpty()) { + List deviceId = modules.stream().map(SurvIotVirtualDeviceModule::getDeviceId).collect(Collectors.toList()); + List devices = lambdaQuery().in(SurvIotVirtualDevice::getId, deviceId).list(); + List deploys = devices.stream().map(SurvIotVirtualDevice::getIotId).collect(Collectors.toList()); + deviceDeployService.refreshLastSyncTimeByDeployCode(deploys); + } + } + }catch (Exception e){ + e.printStackTrace(); + log.error("处理厂家mqtt报文失败，"+e.getMessage()); + } + + } + + + /** + * 根据主题获取设备模组 + */ + private List getModulesByIdent(String identCode){ + //查询网络模块 + SurvIotVirtualDeviceNet net = netService.matchNetSet(identCode); + if (net != null) { + List modules = moduleService.getModulesByNet(net.getId()); + if (!modules.isEmpty()) { + //模组关联的设备 + return modules; + }else{ + log.error("==================网络模块为空：{}:{}===================", identCode,net.getId()); + } + } else { + log.error("==================网络模块未识别：{}===================", identCode); + } + return new ArrayList<>(); + } +} diff --git a/src/main/java/com/lanhai/service/Impl/SurvTransdataLivestockwaterServiceImpl.java b/src/main/java/com/lanhai/service/Impl/SurvTransdataLivestockwaterServiceImpl.java index f73e25b..cc92e3d 100644 --- a/src/main/java/com/lanhai/service/Impl/SurvTransdataLivestockwaterServiceImpl.java +++ b/src/main/java/com/lanhai/service/Impl/SurvTransdataLivestockwaterServiceImpl.java @@ -150,4 +150,8 @@ public class SurvTransdataLivestockwaterServiceImpl extends ServiceImpl