mqtt 数据上报功能

iot-framework/iot-starter-redis/src/main/java/com/middle/platform/redis/constant/CacheConstant.java

@@ -9,4 +9,12 @@ public interface CacheConstant {
     String DICT_CACHE = "dict:%s";
     String DICT_ITEM_CACHE = "dictItem:%s:%s";
     String TOPIC_CACHE = "topic:%s";
+    String PRODUCT_CACHE = "product:%s:%s";
+    String MOD_CACHE = "mod:%s";
+    String CLOUD_CACHE = "cloud:%s";
+    //td源数据，是否建子表缓存
+    String TD_ORG_CACHE = "td:org:%s:%s";
+    //td转换后的数据，是否建子表缓存
+    String TD_CACHE = "td:%s:%s";
+
 }

iot-framework/iot-starter-redis/src/main/java/com/middle/platform/redis/service/CacheService.java

@@ -0,0 +1,33 @@
+package com.middle.platform.redis.service;
+
+import com.middle.platform.redis.constant.CacheConstant;
+import jakarta.annotation.Resource;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * 产品相关缓存
+ *
+ * @author xucaiqin
+ * @date 2023-12-23 15:56:57
+ */
+@Component
+public class CacheService {
+    @Resource
+    private RedisTemplate<String, Object> redisTemplate;
+
+    /**
+     * CacheConstant.TOPIC_CACHE 保存
+     *
+     * @param key
+     * @param val
+     */
+    public void setTopic(String key, String val) {
+        redisTemplate.opsForValue().set(String.format(CacheConstant.TOPIC_CACHE, key), val);
+    }
+    public void delTopic(String key){
+        redisTemplate.delete(String.format(CacheConstant.TOPIC_CACHE, key));
+    }
+
+
+}

iot-framework/iot-starter-redis/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -1 +1,2 @@
 com.middle.platform.redis.config.RedisAutoConfiguration
+com.middle.platform.redis.service.CacheService

iot-module/iot-module-auth/iot-module-auth-api/src/main/java/com/middle/platform/auth/api/feign/AuthApi.java

@@ -12,7 +12,7 @@ import org.springframework.web.bind.annotation.RequestParam;
  */
 @FeignClient(name = ApiConstants.NAME)
 public interface AuthApi {
-    String prefix = RpcConstants.RPC_API_PREFIX+"auth";
+    String prefix = RpcConstants.RPC_API_PREFIX+"/auth";
 
     @GetMapping(prefix + "/get")
     String queryAccount(@RequestParam("account") String account);

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/constant/TopicConstant.java → iot-module/iot-module-data/iot-module-data-api/src/main/java/com/middle/platform/data/api/constant/TopicConstant.java

@@ -1,4 +1,4 @@
-package com.middle.platform.data.biz.constant;
+package com.middle.platform.data.api.constant;
 
 /**
  * @author xucaiqin
@@ -6,12 +6,12 @@ package com.middle.platform.data.biz.constant;
  */
 public interface TopicConstant {
     //属性上报-发布
-    String ATTR_PUB = "1-1";
+    String ATTR_PUB = "1"+TopicType.PUB;
     //云端响应属性上报-订阅
-    String ATTR_SUB = "1-2";
+    String ATTR_SUB = "1"+TopicType.SUB;
     //事件上报
-    String EVENT_PUB = "2-1";
+    String EVENT_PUB = "2"+TopicType.PUB;
     //云端响应事件上报-订阅
-    String EVENT_SUB = "2-2";
+    String EVENT_SUB = "2"+TopicType.SUB;
 
 }

iot-module/iot-module-data/iot-module-data-api/src/main/java/com/middle/platform/data/api/constant/TopicType.java

@@ -0,0 +1,11 @@
+package com.middle.platform.data.api.constant;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-22 16:22:37
+ */
+public interface TopicType {
+    Integer PUB = 1;
+    Integer SUB = 2;
+    Integer BOTH = 3;
+}

iot-module/iot-module-data/iot-module-data-api/src/main/java/com/middle/platform/data/api/feign/DataApi.java

@@ -4,5 +4,5 @@ package com.middle.platform.data.api.feign;
  * @author xucaiqin
  * @date 2023-12-17 16:05:08
  */
-public class DataApi {
+public interface DataApi {
 }

iot-module/iot-module-data/iot-module-data-api/src/main/java/com/middle/platform/data/api/feign/DynamicTopicApi.java

@@ -15,6 +15,9 @@ import org.springframework.web.bind.annotation.RequestBody;
 public interface DynamicTopicApi {
     String PREFIX = RpcConstants.RPC_API_PREFIX + "/dynamic-topic";
 
+    @PostMapping(PREFIX + "/removeTopic")
+    Boolean removeTopic(@RequestBody TopicDto topicDto);
+
     @PostMapping(PREFIX + "/saveTopic")
     Boolean saveTopic(@RequestBody TopicDto topicDto);
 

iot-module/iot-module-data/iot-module-data-api/src/main/java/com/middle/platform/data/api/feign/TdApi.java

@@ -0,0 +1,21 @@
+package com.middle.platform.data.api.feign;
+
+import com.middle.platform.common.constant.RpcConstants;
+import com.middle.platform.data.api.enums.ApiConstants;
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+
+/**
+ * td相关接口
+ *
+ * @author xucaiqin
+ * @date 2023-12-17 16:05:08
+ */
+@FeignClient(name = ApiConstants.NAME, contextId = "tdApi")
+public interface TdApi {
+    String PREFIX = RpcConstants.RPC_API_PREFIX + "/taos";
+
+    @PostMapping(PREFIX + "/createProduct")
+    void createProduct(@RequestParam("code") String code);
+}

iot-module/iot-module-data/iot-module-data-api/src/main/java/com/middle/platform/data/api/pojo/TopicDto.java

@@ -1,6 +1,8 @@
 package com.middle.platform.data.api.pojo;
 
+import lombok.AllArgsConstructor;
 import lombok.Getter;
+import lombok.NoArgsConstructor;
 import lombok.Setter;
 
 /**
@@ -9,5 +11,9 @@ import lombok.Setter;
  */
 @Getter
 @Setter
+@AllArgsConstructor
+@NoArgsConstructor
 public class TopicDto {
+    private String topic;
+    private Integer qos;
 }

iot-module/iot-module-data/iot-module-data-biz/pom.xml

@@ -17,6 +17,10 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
     <dependencies>
+        <dependency>
+            <groupId>com.middle.platform</groupId>
+            <artifactId>iot-common</artifactId>
+        </dependency>
         <!-- Spring Cloud 基础 -->
         <dependency>
             <groupId>org.springframework.cloud</groupId>
@@ -64,6 +68,10 @@
             <groupId>com.middle.platform</groupId>
             <artifactId>iot-module-data-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.middle.platform</groupId>
+            <artifactId>iot-module-manage-api</artifactId>
+        </dependency>
 
     </dependencies>
 </project>

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/api/DynamicTopicApiImpl.java

@@ -1,6 +1,7 @@
 package com.middle.platform.data.biz.api;
 
 import com.middle.platform.data.api.feign.DynamicTopicApi;
+import com.middle.platform.data.api.pojo.TopicDto;
 import com.middle.platform.data.biz.service.mqtt.impl.DynamicTopic;
 import lombok.RequiredArgsConstructor;
 import org.springframework.validation.annotation.Validated;
@@ -15,4 +16,16 @@ import org.springframework.web.bind.annotation.RestController;
 @RequiredArgsConstructor
 public class DynamicTopicApiImpl implements DynamicTopicApi {
     private final DynamicTopic dynamicTopic;
+
+    @Override
+    public Boolean removeTopic(TopicDto topicDto) {
+        dynamicTopic.removeTopic(topicDto.getTopic());
+        return true;
+    }
+
+    @Override
+    public Boolean saveTopic(TopicDto topicDto) {
+        dynamicTopic.addTopic(topicDto.getTopic(), topicDto.getQos());
+        return true;
+    }
 }

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/api/TdApiImpl.java

@@ -0,0 +1,24 @@
+package com.middle.platform.data.biz.api;
+
+import com.middle.platform.data.api.feign.TdApi;
+import com.middle.platform.data.biz.mapper.TaosMapper;
+import lombok.RequiredArgsConstructor;
+import org.springframework.validation.annotation.Validated;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-24 16:22:04
+ */
+@RestController
+@Validated
+@RequiredArgsConstructor
+public class TdApiImpl implements TdApi {
+    private final TaosMapper taosMapper;
+
+    @Override
+    public void createProduct(String code) {
+        taosMapper.createProduct(code);
+        taosMapper.createOriginal(code);
+    }
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/config/FeignConfig.java

@@ -0,0 +1,18 @@
+package com.middle.platform.data.biz.config;
+
+import com.middle.platform.manage.api.feign.CloudApi;
+import com.middle.platform.manage.api.feign.ModApi;
+import com.middle.platform.manage.api.feign.ProductApi;
+import org.springframework.cloud.openfeign.EnableFeignClients;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * 配置feign接口
+ *
+ * @author xucaiqin
+ * @date 2023-12-19 11:00:25
+ */
+@Configuration
+@EnableFeignClients(clients = {CloudApi.class, ModApi.class, ProductApi.class})
+public class FeignConfig {
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/config/MqttConfig.java

@@ -53,7 +53,7 @@ public class MqttConfig {
     @Bean("mqttPahoMessageDrivenChannelAdapter")
     public MessageProducer inbound(MqttPahoClientFactory mqttPahoClientFactory) {
         MqttPahoMessageDrivenChannelAdapter adapter = new MqttPahoMessageDrivenChannelAdapter(
-                mqttProperties.getClientId(), mqttPahoClientFactory,"/iot/device");
+                mqttProperties.getClientId(), mqttPahoClientFactory);
         adapter.setCompletionTimeout(5000);
         adapter.setConverter(new DefaultPahoMessageConverter());
         adapter.setQos(1);

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/controller/DataController.java

@@ -0,0 +1,22 @@
+package com.middle.platform.data.biz.controller;
+
+import com.middle.platform.data.biz.pojo.ProductPara;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-15 08:46:49
+ */
+@RestController
+@RequestMapping("/data")
+public class DataController {
+
+
+    @PostMapping("/up")
+    public void test(@RequestBody ProductPara productPara) {
+    }
+
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/controller/IndexController.java

@@ -2,7 +2,7 @@ package com.middle.platform.data.biz.controller;
 
 import com.middle.platform.common.utils.DateTimeUtil;
 import com.middle.platform.common.utils.Result;
-import com.middle.platform.data.biz.mapper.ProductMapper;
+import com.middle.platform.data.biz.mapper.TaosMapper;
 import com.middle.platform.data.biz.pojo.ProductPara;
 import jakarta.annotation.Resource;
 import org.springframework.beans.factory.annotation.Value;
@@ -21,7 +21,7 @@ public class IndexController {
     @Value("${spring.application.name}")
     private String name;
     @Resource
-    private ProductMapper productMapper;
+    private TaosMapper productMapper;
     @Resource
     private MqttPahoMessageDrivenChannelAdapter mqttPahoMessageDrivenChannelAdapter;
 
@@ -36,4 +36,5 @@ public class IndexController {
     public void test(@RequestBody ProductPara productPara) {
         productMapper.insert(productPara);
     }
+
 }

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/event/DataEvent.java

@@ -0,0 +1,19 @@
+package com.middle.platform.data.biz.event;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.ApplicationEvent;
+import org.springframework.context.ApplicationListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-24 11:07:47
+ */
+@Component
+@Slf4j
+public class DataEvent implements ApplicationListener<ApplicationEvent> {
+    @Override
+    public void onApplicationEvent(ApplicationEvent event) {
+        log.info("事件：{}", event.getClass().getName());
+    }
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/mapper/ProductMapper.java

@@ -1,18 +0,0 @@
-package com.middle.platform.data.biz.mapper;
-
-import com.baomidou.dynamic.datasource.annotation.DS;
-import com.middle.platform.data.biz.pojo.ProductPara;
-import org.apache.ibatis.annotations.Insert;
-import org.apache.ibatis.annotations.Mapper;
-
-/**
- * @author xucaiqin
- * @date 2023-12-20 09:22:38
- */
-@Mapper
-@DS("td")
-public interface ProductMapper {
-    @Insert("INSERT INTO device_gnss_${deviceId} (create_time,point_value,raw_value,origin_time) VALUES (#{createTime},#{pointValue},#{rawValue},#{originTime})")
-    void insert(ProductPara productPara);
-
-}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/mapper/TaosMapper.java

@@ -0,0 +1,84 @@
+package com.middle.platform.data.biz.mapper;
+
+import com.baomidou.dynamic.datasource.annotation.DS;
+import com.middle.platform.data.biz.pojo.OriginalPara;
+import com.middle.platform.data.biz.pojo.ProductPara;
+import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-20 09:22:38
+ */
+@Mapper
+@DS("td")
+public interface TaosMapper {
+    /**
+     * 产品（超级表）
+     *
+     * @param code 产品编码
+     * @return
+     */
+    void createProduct(@Param("code") String code);
+
+    /**
+     * 创建设备表（子表）
+     *
+     * @param guid 设备guid
+     * @param code 产品编码
+     * @return
+     */
+    int createDeviceTable(@Param("guid") String guid, @Param("code") String code);
+
+    /**
+     * 写入设备表数据
+     *
+     * @param productPara
+     */
+    void insert(ProductPara productPara);
+
+    /**
+     * 批量写入设备表数据
+     *
+     * @param collect
+     * @return
+     */
+    int batchInsert(@Param("list") List<ProductPara> collect, @Param("guid") String guid, @Param("code") String code);
+
+
+    /*原始数据*/
+
+    /**
+     * 原始数据表（超级表）
+     *
+     * @param code
+     * @return
+     */
+    void createOriginal(@Param("code") String code);
+
+    /**
+     * 原始数据表（子表）
+     *
+     * @param guid
+     * @param code 产品编码
+     * @return
+     */
+    void createOriginalTable(@Param("guid") String guid, @Param("code") String code);
+
+    /**
+     * 写入原始数据表数据
+     *
+     * @param originalPara
+     */
+    int insertOriginal(OriginalPara originalPara);
+
+    /**
+     * 批量写入设备表数据
+     *
+     * @param collect
+     * @return
+     */
+    int batchInsertOriginal(@Param("list") List<OriginalPara> collect, @Param("guid") String guid, @Param("code") String code);
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/pojo/OriginalPara.java

@@ -0,0 +1,27 @@
+package com.middle.platform.data.biz.pojo;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.Date;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-20 09:26:56
+ */
+@Getter
+@Setter
+public class OriginalPara {
+    private String code;
+    private String guid;
+    /**
+     * 源数据
+     */
+    private String rawStr;
+    /**
+     * 创建时间
+     */
+    private Date ts;
+    private String msgId;
+
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/pojo/ProductPara.java

@@ -12,19 +12,28 @@ import java.util.Date;
 @Getter
 @Setter
 public class ProductPara {
-    private String deviceId;
-    private String pointId;
+    private String code;
+
+    private String guid;
+    /**
+     * 字段
+     */
+    private String line;
     /**
-     * 处理值，进行过缩放、格式化等操作
+     * 预留字段
      */
-    private String pointValue;
+    private String reserve;
 
     /**
-     * 原始值
+     * 值
+     */
+    private String val;
+    /**
+     * 消息唯一id
      */
-    private String rawValue;
+    private String msgId;
 
-    private Date createTime;
+    private Long ts;
 
     private Date originTime;
 }

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/pojo/mod/DataTypeDto.java

@@ -0,0 +1,15 @@
+package com.middle.platform.data.biz.pojo.mod;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-24 11:31:39
+ */
+@Getter
+@Setter
+public class DataTypeDto {
+    private String type;
+    private SpecsDto specs;
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/pojo/mod/ModDto.java

@@ -0,0 +1,22 @@
+package com.middle.platform.data.biz.pojo.mod;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.List;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-24 11:21:47
+ */
+@Getter
+@Setter
+public class ModDto {
+    /**
+     * 属性
+     */
+    private List<Property> properties;
+
+    private List<Property> event;
+
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/pojo/mod/Property.java

@@ -0,0 +1,41 @@
+package com.middle.platform.data.biz.pojo.mod;
+
+import lombok.Data;
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-24 11:22:21
+ */
+@NoArgsConstructor
+@Data
+@Getter
+@Setter
+public class Property {
+    /**
+     * 标识符
+     */
+    private String identifier;
+    /**
+     * 名称
+     */
+    private String name;
+    /**
+     * 读写/只读
+     */
+    private String accessMode;
+    /**
+     * 是否必填
+     */
+    private Boolean required;
+    /**
+     * 数据类型
+     */
+    private DataTypeDto dataTypeDto;
+    /**
+     * 描述
+     */
+    private String desc;
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/pojo/mod/SpecsDto.java

@@ -0,0 +1,20 @@
+package com.middle.platform.data.biz.pojo.mod;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-24 11:32:09
+ */
+@Getter
+@Setter
+public class SpecsDto {
+    private String min;
+    private String max;
+    private String step;
+    private String unit;
+    private String unitName;
+
+
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/DataService.java

@@ -0,0 +1,88 @@
+package com.middle.platform.data.biz.service;
+
+import cn.hutool.core.collection.CollUtil;
+import com.alibaba.fastjson.JSONObject;
+import com.middle.platform.data.biz.mapper.TaosMapper;
+import com.middle.platform.data.biz.pojo.OriginalPara;
+import com.middle.platform.data.biz.pojo.ProductPara;
+import com.middle.platform.data.biz.utils.TsUtil;
+import com.middle.platform.manage.api.pojo.ProductVo;
+import com.middle.platform.redis.constant.CacheConstant;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-24 15:33:31
+ */
+@Component
+@Slf4j
+public class DataService {
+    @Resource
+    private TaosMapper taosMapper;
+    @Resource
+    private RedisTemplate<String, Object> redisTemplate;
+
+    /**
+     * 源数据处理
+     *
+     * @param productVo
+     * @param payload
+     */
+    public void rawData(String msgId, ProductVo productVo, Object payload) {
+        Object o = redisTemplate.opsForValue().get(String.format(CacheConstant.TD_ORG_CACHE, productVo.getCode(), productVo.getGuid()));
+        if (Objects.isNull(o)) {
+            taosMapper.createOriginalTable(productVo.getGuid(), productVo.getCode());
+            redisTemplate.opsForValue().set(String.format(CacheConstant.TD_ORG_CACHE, productVo.getCode(), productVo.getGuid()), true);
+        }
+        OriginalPara originalPara = new OriginalPara();
+        originalPara.setCode(productVo.getCode());
+        originalPara.setGuid(productVo.getGuid());
+        originalPara.setRawStr(payload.toString());
+        originalPara.setTs(new Date());
+        originalPara.setMsgId(msgId);
+
+        taosMapper.insertOriginal(originalPara);
+    }
+
+    /**
+     * 解析后的数据
+     *
+     * @param msgId
+     * @param productVo
+     * @param modData
+     */
+    public void analyzeData(String msgId, ProductVo productVo, Object modData) {
+        log.info("解析后的数据：{}", modData);
+        Object o = redisTemplate.opsForValue().get(String.format(CacheConstant.TD_CACHE, productVo.getCode(), productVo.getGuid()));
+        if (Objects.isNull(o)) {
+            taosMapper.createDeviceTable(productVo.getGuid(), productVo.getCode());
+            redisTemplate.opsForValue().set(String.format(CacheConstant.TD_CACHE, productVo.getCode(), productVo.getGuid()), true);
+        }
+        if (modData instanceof JSONObject jsonObject) {
+            List<ProductPara> list = new ArrayList<>();
+            ProductPara productPara;
+            for (Map.Entry<String, Object> map : jsonObject.entrySet()) {
+                productPara = new ProductPara();
+                productPara.setCode(productVo.getCode());
+                productPara.setGuid(productVo.getGuid());
+                productPara.setLine(map.getKey());
+                productPara.setReserve("");
+                productPara.setVal(String.valueOf(map.getValue()));
+                productPara.setMsgId(msgId);
+                productPara.setTs(TsUtil.getAndIncrement());
+                productPara.setOriginTime(new Date());//todo 有则获取
+                list.add(productPara);
+            }
+            if (CollUtil.isNotEmpty(list)) {
+                taosMapper.batchInsert(list, productVo.getGuid(), productVo.getCode());
+            }
+        }
+
+    }
+
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/MqttService.java

@@ -3,6 +3,7 @@ package com.middle.platform.data.biz.service;
 import com.middle.platform.data.biz.config.MqttConfig;
 import com.middle.platform.data.biz.service.mqtt.MqttStrategyFactory;
 import com.middle.platform.data.biz.service.mqtt.MqttTopicStrategy;
+import com.middle.platform.data.biz.utils.MqttTopicUtil;
 import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.context.annotation.Bean;
@@ -21,16 +22,17 @@ public class MqttService {
 
     @ServiceActivator(inputChannel = MqttConfig.input)
     @Bean
-    public MessageHandler test() {
+    public MessageHandler data() {
         return message -> {
             MessageHeaders headers = message.getHeaders();
             Object payload = message.getPayload();
             String topic = (String) headers.get("mqtt_receivedTopic");
             log.info("topic {}", topic);
             log.info("payload {}", payload);
-            MqttTopicStrategy mqttTopicStrategy = mqttStrategyFactory.getByTopic(topic);
+
+            MqttTopicStrategy mqttTopicStrategy = mqttStrategyFactory.getByTopic(MqttTopicUtil.replaceTopic(topic));
             if (Objects.nonNull(mqttTopicStrategy)) {
-                mqttTopicStrategy.dealMsg(payload);
+                mqttTopicStrategy.dealMsg(topic, payload);
             }
         };
     }

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/ProductAnalyse.java

@@ -0,0 +1,128 @@
+package com.middle.platform.data.biz.service;
+
+import cn.hutool.core.collection.CollUtil;
+import com.alibaba.fastjson.JSONObject;
+import com.middle.platform.data.biz.pojo.mod.ModDto;
+import com.middle.platform.data.biz.pojo.mod.Property;
+import com.middle.platform.manage.api.enums.DataFormatConstant;
+import com.middle.platform.manage.api.feign.CloudApi;
+import com.middle.platform.manage.api.feign.ModApi;
+import com.middle.platform.manage.api.feign.ProductApi;
+import com.middle.platform.manage.api.pojo.IotCloudVo;
+import com.middle.platform.manage.api.pojo.ProductVo;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.stereotype.Component;
+
+import java.util.*;
+
+/**
+ * 产品解析
+ *
+ * @author xucaiqin
+ * @date 2023-12-23 13:56:23
+ */
+@Component
+@Slf4j
+public class ProductAnalyse {
+    @Resource
+    private CloudApi cloudApi;
+    @Resource
+    private ModApi modApi;
+    @Resource
+    private ProductApi productApi;
+
+    private Object objStrToJSON(Object object) {
+        try {
+            return JSONObject.parseObject(JSONObject.toJSONString(object));
+        } catch (Exception e) {
+            return object;
+        }
+    }
+
+    /**
+     * 通过产品key查询产品信息
+     *
+     * @param productKey 产品key
+     * @return
+     */
+    public ProductVo getProduct(String productKey, String deviceSn) {
+        return productApi.queryProduct(productKey, deviceSn);
+    }
+
+
+    /**
+     * 云函数转换数据
+     *
+     * @param productVo 产品
+     * @param payload   上报数据
+     * @return 云函数转换后的数据
+     */
+    public Object dataConvert(ProductVo productVo, Object payload) {
+        log.info("源数据：{}", payload);
+        IotCloudVo iotCloudVo = cloudApi.queryCloud(productVo.getId());
+        if (Objects.isNull(iotCloudVo)) {
+            log.warn("未找到云函数");
+            return objStrToJSON(payload);
+        }
+        //通过云函数转换 todo
+        if (productVo.getDataFormat().equals(DataFormatConstant.json)) {
+
+        }
+        return objStrToJSON(payload);
+    }
+
+    /**
+     * 物模型转换
+     *
+     * @param payload 云函数解析后的数据
+     * @return 解析为物模型对应属性数据
+     */
+    public Object modConvert(ProductVo productVo, Object payload) {
+        log.info("物模型数据转换：{}", payload);
+        String mod = modApi.queryMod(productVo.getId());
+        if (StringUtils.isNotBlank(mod)) {
+            ModDto modDto = JSONObject.parseObject(mod, ModDto.class);
+            return Optional.ofNullable(modDto)
+                    .map(ModDto::getProperties)
+                    .map(properties -> {
+                        if (payload instanceof String tmp) {
+                            JSONObject jsonObject = JSONObject.parseObject(tmp);
+                            return change(properties, jsonObject);
+                        }
+                        if (payload instanceof JSONObject jsonObject) {
+                            return change(properties, jsonObject);
+                        }
+                        return new JSONObject();
+                    }).orElse(new JSONObject());
+        }
+        return null;
+    }
+
+    /**
+     * 过滤物模型中的属性，仅保留物模型中的属性字段
+     *
+     * @param properties 物模型属性
+     * @param jsonObject 上报的json数据
+     * @return
+     */
+    private JSONObject change(List<Property> properties, JSONObject jsonObject) {
+        if (CollUtil.isEmpty(properties)) {
+            return jsonObject;
+        }
+        //物模型中需要保留的属性字段
+        List<String> key = properties.stream().map(Property::getIdentifier).toList();
+        //上报数据中，所有的字段
+        Set<String> keyAll = jsonObject.keySet();
+
+        // 仅保留需要的属性字段
+        List<String> collect = keyAll.stream().filter(one -> !key.contains(one)).toList();
+
+        // 使用 removeAll 方法一次性移除所有需要删除的字段
+        collect.forEach(jsonObject.keySet()::remove);
+        return jsonObject;
+    }
+
+
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/FuncPerEnum.java

@@ -1,6 +1,6 @@
 package com.middle.platform.data.biz.service.mqtt;
 
-import com.middle.platform.data.biz.constant.TopicConstant;
+import com.middle.platform.data.api.constant.TopicConstant;
 import com.middle.platform.data.biz.service.mqtt.strategy.AttrReportStrategy;
 import com.middle.platform.data.biz.service.mqtt.strategy.EventReportStrategy;
 import lombok.AllArgsConstructor;

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/MqttStrategyFactory.java

@@ -15,7 +15,7 @@ public class MqttStrategyFactory {
     /**
      * 通过topic名称获取对应解析策略
      *
-     * @param topic mqtt主题
+     * @param topic mqtt主题 如：/iot/VGSogVDr1StQ/+/att
      * @return
      */
     public MqttTopicStrategy getByTopic(String topic) {

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/MqttTopicInit.java

@@ -1,29 +1,39 @@
 package com.middle.platform.data.biz.service.mqtt;
 
+import com.middle.platform.data.biz.service.mqtt.impl.DynamicTopic;
+import com.middle.platform.manage.api.feign.ProductApi;
+import com.middle.platform.manage.api.pojo.TopicVo;
+import com.middle.platform.redis.constant.CacheConstant;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+@Component
+@Slf4j
 public class MqttTopicInit {
-    private String topic;
-    private String name;
+    @Resource
+    private ProductApi productApi;
+    @Resource
+    private DynamicTopic dynamicTopic;
+    @Resource
+    private RedisTemplate<String, Object> redisTemplate;
+
+    public void init() {
+        List<TopicVo> topicVos = productApi.productTopic();
 
+        if (!CollectionUtils.isEmpty(topicVos)) {
+            log.info("初始化订阅：{}", topicVos);
+            Map<String, String> map = topicVos.stream().collect(Collectors.toMap(a -> String.format(CacheConstant.TOPIC_CACHE, a.getUrl()), b -> b.getFunc() + String.valueOf(b.getPermission()), (a, b) -> a));
+            redisTemplate.opsForValue().multiSet(map);
+            dynamicTopic.addTopic(topicVos.stream().map(TopicVo::getUrl).collect(Collectors.toList()));
+        }
+    }
 
-//
-//    public static String strategy(String topic) {
-//        for (MqttTopicEnum value : MqttTopicEnum.values()) {
-//            if (StringUtils.equals(value.getTopic(), topic)) {
-//                String beanName = value.getName();
-//                char[] chars = beanName.toCharArray();
-//                chars[0] += 32;
-//                return String.valueOf(chars);
-//            }
-//        }
-//        return null;
-//    }
-//
-//    public static String[] getTopics() {
-//        List<String> res = new ArrayList<>();
-//        for (MqttTopicEnum value : MqttTopicEnum.values()) {
-//            res.add(value.topic);
-//        }
-//        return res.toArray(String[]::new);
-//    }
 
 }

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/MqttTopicStrategy.java

@@ -1,6 +1,12 @@
 package com.middle.platform.data.biz.service.mqtt;
 
 public interface MqttTopicStrategy {
-    void dealMsg(Object msg);
+    /**
+     * 处理消息
+     *
+     * @param topic topic
+     * @param msg   消息
+     */
+    void dealMsg(String topic, Object msg);
 
 }

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/impl/DynamicTopic.java

@@ -4,6 +4,8 @@ import jakarta.annotation.Resource;
 import org.springframework.integration.mqtt.inbound.MqttPahoMessageDrivenChannelAdapter;
 import org.springframework.stereotype.Component;
 
+import java.util.List;
+
 /**
  * @author xucaiqin
  * @date 2023-12-22 13:52:20
@@ -31,4 +33,13 @@ public class DynamicTopic {
     public void removeTopic(String topic) {
         mqttPahoMessageDrivenChannelAdapter.removeTopic(topic);
     }
+
+    /**
+     * 批量订阅topic
+     *
+     * @param topic
+     */
+    public void addTopic(List<String> topic) {
+        mqttPahoMessageDrivenChannelAdapter.addTopic(topic.toArray(String[]::new));
+    }
 }

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/strategy/AttrReportStrategy.java

@@ -1,19 +1,45 @@
 package com.middle.platform.data.biz.service.mqtt.strategy;
 
+import cn.hutool.core.util.IdUtil;
+import com.middle.platform.data.biz.service.DataService;
+import com.middle.platform.data.biz.service.ProductAnalyse;
 import com.middle.platform.data.biz.service.mqtt.MqttTopicStrategy;
+import com.middle.platform.data.biz.utils.MqttTopicUtil;
+import com.middle.platform.manage.api.pojo.ProductVo;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
+import java.util.Objects;
+
 /**
  * 属性上报
  */
 @Service
 @Slf4j
+@RequiredArgsConstructor
 public class AttrReportStrategy implements MqttTopicStrategy {
+    private final ProductAnalyse productAnalyse;
+    private final DataService dataService;
 
     @Override
-    public void dealMsg(Object msg) {
-        log.debug("属性上报数据：{}", msg);
-
+    public void dealMsg(String topic, Object msg) {
+        log.info("属性[{}]上报数据：{}", topic, msg);
+        String deviceSn = MqttTopicUtil.getDeviceSn(topic);
+        String productKey = MqttTopicUtil.getProductKey(topic);
+        //获取产品信息
+        ProductVo productVo = productAnalyse.getProduct(productKey, deviceSn);
+        if (Objects.isNull(productVo)) {
+            return;
+        }
+        String msgId = IdUtil.fastSimpleUUID();
+        //1.源数据处理
+        dataService.rawData(msgId, productVo, msg);
+        //2.云函数解析
+        Object cloudData = productAnalyse.dataConvert(productVo, msg);
+        //3.物模型解析
+        Object modData = productAnalyse.modConvert(productVo, cloudData);
+        //4.存储解析后的数据
+        dataService.analyzeData(msgId, productVo, modData);
     }
 }

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/mqtt/strategy/EventReportStrategy.java

@@ -12,7 +12,7 @@ import org.springframework.stereotype.Service;
 public class EventReportStrategy implements MqttTopicStrategy {
 
     @Override
-    public void dealMsg(Object msg) {
+    public void dealMsg(String guid,Object msg) {
         log.debug("属性上报数据：{}", msg);
 
     }

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/service/runner/TopicRunner.java

@@ -0,0 +1,22 @@
+package com.middle.platform.data.biz.service.runner;
+
+import com.middle.platform.data.biz.service.mqtt.MqttTopicInit;
+import jakarta.annotation.Resource;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.stereotype.Component;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-23 15:59:53
+ */
+@Component
+public class TopicRunner implements ApplicationRunner {
+    @Resource
+    private MqttTopicInit mqttTopicInit;
+
+    @Override
+    public void run(ApplicationArguments args) throws Exception {
+        mqttTopicInit.init();
+    }
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/utils/MqttTopicUtil.java

@@ -0,0 +1,50 @@
+package com.middle.platform.data.biz.utils;
+
+import org.apache.commons.lang3.StringUtils;
+
+import java.util.Optional;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-24 15:18:14
+ */
+public class MqttTopicUtil {
+    public static String getProductKey(String input) {
+        String pattern = "/iot/(.*?)/.*";
+        Pattern regex = Pattern.compile(pattern);
+        Matcher matcher = regex.matcher(input);
+
+        if (matcher.find()) {
+            // 提取匹配到的数据
+            return matcher.group(1);
+        }
+        return null;
+    }
+
+    /**
+     * 设备上报的sn
+     *
+     * @param input
+     * @return
+     */
+    public static String getDeviceSn(String input) {
+        String pattern = "/iot/[^/]+/([^/]+)/.*";
+        Pattern regex = Pattern.compile(pattern);
+        Matcher matcher = regex.matcher(input);
+
+        if (matcher.find()) {
+            // 提取匹配到的数据
+            return matcher.group(1);
+        }
+        return null;
+    }
+
+    public static String replaceTopic(String topic) {
+        return Optional.ofNullable(topic).map(t -> {
+            String s = StringUtils.startsWith(t, "/") ? "(/[^/]+/[^/]+)/[^/]+" : "([^/]+/[^/]+)/[^/]+";
+            return t.replaceFirst(s, "$1/+");
+        }).orElse(topic);
+    }
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/java/com/middle/platform/data/biz/utils/TsUtil.java

@@ -0,0 +1,28 @@
+package com.middle.platform.data.biz.utils;
+
+import java.util.Date;
+import java.util.Objects;
+import java.util.Optional;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-25 14:31:39
+ */
+public class TsUtil {
+    private static final ThreadLocal<Long> threadLocal = new ThreadLocal<>();
+
+    /**
+     * td用ts作为主键id时，批量插入时间相同只会写入一条数据的解决方法
+     *
+     * @return
+     */
+    public static Long getAndIncrement() {
+        if (Objects.isNull(threadLocal.get())) {
+            long time = new Date().getTime();
+            threadLocal.set(time);
+            return time;
+        }
+        Optional.ofNullable(threadLocal.get()).ifPresent(g -> threadLocal.set(g + 1));
+        return threadLocal.get();
+    }
+}

iot-module/iot-module-data/iot-module-data-biz/src/main/resources/bootstrap.yaml

@@ -18,5 +18,8 @@ spring:
       write-date-timestamps-as-nanoseconds: false # 设置不使用 nanoseconds 的格式。例如说 1611460870.401，而是直接 1611460870401
       write-durations-as-timestamps: true # 设置 Duration 的格式，使用时间戳
       fail-on-empty-beans: false # 允许序列化无属性的 Bean
-
+logging:
+  level:
+    root: info
+    com.middle.platform: debug
 

iot-module/iot-module-data/iot-module-data-biz/src/main/resources/mapper/ProductMapper.xml

@@ -1,5 +0,0 @@
-<?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.middle.platform.data.biz.mapper.ProductMapper">
-
-</mapper>

iot-module/iot-module-data/iot-module-data-biz/src/main/resources/mapper/TaosMapper.xml

@@ -0,0 +1,43 @@
+<?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.middle.platform.data.biz.mapper.TaosMapper">
+    <update id="createProduct">
+        CREATE STABLE IF NOT EXISTS product_${code} (ts TIMESTAMP, line NCHAR(32),val NCHAR(32),msg_id
+        NCHAR(64),reserve NCHAR(64),origin_time TIMESTAMP) TAGS (guid NCHAR(32))
+    </update>
+    <update id="createDeviceTable">
+        CREATE TABLE IF NOT EXISTS device_${code}_${guid} using product_${code} TAGS (#{guid})
+    </update>
+
+    <insert id="insert">
+        INSERT INTO device_${code}_${guid} (ts, line, val, msg_id,reserve, origin_time)
+        VALUES (#{ts}, #{line}, #{val}, #{msgId}, #{reserve}, #{originTime})
+    </insert>
+    <insert id="batchInsert">
+        INSERT INTO device_${code}_${guid} (ts, line, val, msg_id, reserve, origin_time) VALUES
+        <foreach collection='list' item='item' index='index' separator=','>
+            (#{item.ts}, #{item.line}, #{item.val}, #{item.msgId}, #{item.reserve}, #{item.originTime})
+        </foreach>
+    </insert>
+
+    <!--源数据-->
+    <select id="createOriginal" >
+        create STABLE IF NOT EXISTS original_${code} (ts TIMESTAMP, raw_str varchar(10000),msg_id NCHAR(64)) tags (guid varchar(64));
+    </select>
+
+    <select id="createOriginalTable" >
+        CREATE TABLE IF NOT EXISTS original_${code}_${guid} using original_${code} TAGS (#{guid})
+    </select>
+
+    <insert id="insertOriginal">
+        INSERT INTO original_${code}_${guid} (ts, raw_str,msg_id)
+        VALUES (#{ts}, #{rawStr}, #{msgId})
+    </insert>
+
+    <select id="batchInsertOriginal" resultType="int">
+        INSERT INTO original_${code}_${guid} (ts, raw_str,msg_id) VALUES
+        <foreach collection='list' item='item' index='index' separator=',' >
+            (#{item.ts}, #{item.rawStr}, #{item.msgId})
+        </foreach>
+    </select>
+</mapper>

iot-module/iot-module-manage/iot-module-manage-api/pom.xml

@@ -17,5 +17,14 @@
         <maven.compiler.target>17</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
-
+    <dependencies>
+        <dependency>
+            <groupId>com.middle.platform</groupId>
+            <artifactId>iot-starter-feign</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.middle.platform</groupId>
+            <artifactId>iot-common</artifactId>
+        </dependency>
+    </dependencies>
 </project>

iot-module/iot-module-manage/iot-module-manage-api/src/main/java/com/middle/platform/manage/api/Test.java

@@ -1,8 +0,0 @@
-package com.middle.platform.manage.api;
-
-/**
- * @author xucaiqin
- * @date 2023-12-17 10:50:04
- */
-public class Test {
-}

iot-module/iot-module-manage/iot-module-manage-api/src/main/java/com/middle/platform/manage/api/enums/ApiConstants.java

@@ -0,0 +1,12 @@
+package com.middle.platform.manage.api.enums;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-03 10:28:58
+ */
+public class ApiConstants {
+    public static final String NAME = "iot-manage";
+
+    public static final String VERSION = "1.0.0";
+
+}

iot-module/iot-module-manage/iot-module-manage-api/src/main/java/com/middle/platform/manage/api/enums/DataFormatConstant.java

@@ -0,0 +1,11 @@
+package com.middle.platform.manage.api.enums;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-23 14:24:44
+ */
+public interface DataFormatConstant {
+    Integer json = 1;
+
+    Integer xml = 2;
+}

iot-module/iot-module-manage/iot-module-manage-api/src/main/java/com/middle/platform/manage/api/feign/CloudApi.java

@@ -0,0 +1,29 @@
+package com.middle.platform.manage.api.feign;
+
+
+import com.middle.platform.common.constant.RpcConstants;
+import com.middle.platform.manage.api.enums.ApiConstants;
+import com.middle.platform.manage.api.pojo.IotCloudVo;
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-17 11:09:05
+ */
+@FeignClient(name = ApiConstants.NAME, contextId = "cloudApi")
+public interface CloudApi {
+    String prefix = RpcConstants.RPC_API_PREFIX + "/cloud";
+
+    /**
+     * 通过topic获取云函数
+     *
+     * @param productId
+     * @return
+     */
+    @GetMapping(prefix + "/get")
+    IotCloudVo queryCloud(@RequestParam("productId") Long productId);
+
+
+}

iot-module/iot-module-manage/iot-module-manage-api/src/main/java/com/middle/platform/manage/api/feign/ModApi.java

@@ -0,0 +1,20 @@
+package com.middle.platform.manage.api.feign;
+
+import com.middle.platform.common.constant.RpcConstants;
+import com.middle.platform.manage.api.enums.ApiConstants;
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-17 11:09:05
+ */
+@FeignClient(name = ApiConstants.NAME, contextId = "modApi")
+public interface ModApi {
+    String prefix = RpcConstants.RPC_API_PREFIX + "/mod";
+
+    @GetMapping(prefix + "/get")
+    String queryMod(@RequestParam("productId") Long productId);
+
+}

iot-module/iot-module-manage/iot-module-manage-api/src/main/java/com/middle/platform/manage/api/feign/ProductApi.java

@@ -0,0 +1,38 @@
+package com.middle.platform.manage.api.feign;
+
+import com.middle.platform.common.constant.RpcConstants;
+import com.middle.platform.manage.api.enums.ApiConstants;
+import com.middle.platform.manage.api.pojo.ProductVo;
+import com.middle.platform.manage.api.pojo.TopicVo;
+import org.springframework.cloud.openfeign.FeignClient;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestParam;
+
+import java.util.List;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-17 11:09:05
+ */
+@FeignClient(name = ApiConstants.NAME, contextId = "productApi")
+public interface ProductApi {
+    String prefix = RpcConstants.RPC_API_PREFIX + "/product";
+
+    /**
+     * 查询产品
+     *
+     * @param productKey
+     * @return
+     */
+    @GetMapping(prefix + "/get")
+    ProductVo queryProduct(@RequestParam("productKey") String productKey,@RequestParam("deviceSn")String deviceSn);
+
+    /**
+     * 获取所有产品的mqtt，需要订阅的topic
+     *
+     * @return 需要订阅的topic
+     */
+    @GetMapping(prefix + "/productTopic")
+    List<TopicVo> productTopic();
+
+}

iot-module/iot-module-manage/iot-module-manage-api/src/main/java/com/middle/platform/manage/api/pojo/IotCloudVo.java

@@ -0,0 +1,26 @@
+package com.middle.platform.manage.api.pojo;
+
+import lombok.Getter;
+import lombok.Setter;
+
+/**
+ * 云函数表
+ *
+ * @author xucaiqin
+ * @date 2023-12-19 13:43:38
+ */
+@Getter
+@Setter
+public class IotCloudVo  {
+    private Long id;
+
+    /**
+     * 脚本语言 1-js 2-php
+     */
+    private Integer lang;
+
+    /**
+     * 云函数
+     */
+    private String cloudText;
+}

iot-module/iot-module-manage/iot-module-manage-api/src/main/java/com/middle/platform/manage/api/pojo/ProductVo.java

@@ -0,0 +1,31 @@
+package com.middle.platform.manage.api.pojo;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.io.Serial;
+import java.io.Serializable;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-23 14:04:27
+ */
+@Getter
+@Setter
+public class ProductVo implements Serializable {
+    @Serial
+    private static final long serialVersionUID = -8359264859254253425L;
+    /**
+     * 产品id
+     */
+    private Long id;
+    /**
+     * 产品编码
+     */
+    private String code;
+    /**
+     * 设备guid
+     */
+    private String guid;
+    private Integer dataFormat;
+}

iot-module/iot-module-manage/iot-module-manage-api/src/main/java/com/middle/platform/manage/api/pojo/TopicVo.java

@@ -0,0 +1,21 @@
+package com.middle.platform.manage.api.pojo;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.io.Serial;
+import java.io.Serializable;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-23 14:04:27
+ */
+@Getter
+@Setter
+public class TopicVo implements Serializable {
+    @Serial
+    private static final long serialVersionUID = -8359264859254253425L;
+    private String url;
+    private Integer func;
+    private Integer permission;
+}

iot-module/iot-module-manage/iot-module-manage-biz/pom.xml

@@ -70,5 +70,9 @@
             <groupId>com.middle.platform</groupId>
             <artifactId>iot-module-data-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.middle.platform</groupId>
+            <artifactId>iot-module-manage-api</artifactId>
+        </dependency>
     </dependencies>
 </project>

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/api/CloudApiImpl.java

@@ -0,0 +1,22 @@
+package com.middle.platform.manage.biz.api;
+
+import com.middle.platform.manage.api.feign.CloudApi;
+import com.middle.platform.manage.api.pojo.IotCloudVo;
+import com.middle.platform.manage.biz.service.IotCloudService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-23 13:41:30
+ */
+@RestController
+@RequiredArgsConstructor
+public class CloudApiImpl implements CloudApi {
+    private final IotCloudService iotCloudService;
+
+    @Override
+    public IotCloudVo queryCloud(Long productId) {
+        return iotCloudService.cacheCloud(productId);
+    }
+}

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/api/ModApiImpl.java

@@ -0,0 +1,21 @@
+package com.middle.platform.manage.biz.api;
+
+import com.middle.platform.manage.api.feign.ModApi;
+import com.middle.platform.manage.biz.service.IotModService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.RestController;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-23 13:42:39
+ */
+@RestController
+@RequiredArgsConstructor
+public class ModApiImpl implements ModApi {
+    private final IotModService iotModService;
+
+    @Override
+    public String queryMod(Long productId) {
+         return iotModService.cacheMod(productId);
+    }
+}

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/api/ProductApiImpl.java

@@ -0,0 +1,30 @@
+package com.middle.platform.manage.biz.api;
+
+import com.middle.platform.manage.api.feign.ProductApi;
+import com.middle.platform.manage.api.pojo.ProductVo;
+import com.middle.platform.manage.api.pojo.TopicVo;
+import com.middle.platform.manage.biz.service.IotProductService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.List;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-23 14:12:44
+ */
+@RequiredArgsConstructor
+@RestController
+public class ProductApiImpl implements ProductApi {
+    private final IotProductService iotProductService;
+
+    @Override
+    public List<TopicVo> productTopic() {
+        return iotProductService.queryList();
+    }
+
+    @Override
+    public ProductVo queryProduct(String productKey, String deviceSn) {
+        return iotProductService.cacheProduct(productKey, deviceSn);
+    }
+}

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/config/FeignConfig.java

@@ -1,6 +1,7 @@
 package com.middle.platform.manage.biz.config;
 
 import com.middle.platform.data.api.feign.DynamicTopicApi;
+import com.middle.platform.data.api.feign.TdApi;
 import com.middle.platform.system.api.feign.UserApi;
 import org.springframework.cloud.openfeign.EnableFeignClients;
 import org.springframework.context.annotation.Configuration;
@@ -12,6 +13,6 @@ import org.springframework.context.annotation.Configuration;
  * @date 2023-12-19 11:00:25
  */
 @Configuration
-@EnableFeignClients(clients = {UserApi.class, DynamicTopicApi.class})
+@EnableFeignClients(clients = {UserApi.class, DynamicTopicApi.class, TdApi.class})
 public class FeignConfig {
 }

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/constant/UrlInit.java

@@ -5,8 +5,8 @@ package com.middle.platform.manage.biz.constant;
  * @date 2023-12-21 09:25:19
  */
 public enum UrlInit {
-    ATTR_UP("/product/%s/${deviceName}/property/post", "设备属性上报", 1, 1),
-    ATTR_DOWN("/product/%s/${deviceName}/property/post_reply", "云端响应属性上报", 2, 2),
+    ATTR_UP("/iot/%s/+/property/post", "设备属性上报", 1, 1),
+    ATTR_DOWN("/iot/%s/+/property/post_reply", "云端响应属性上报", 2, 2),
     ;
     private final String prefix;
     private final String desc;

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/controller/IotProductController.java

@@ -78,7 +78,7 @@ public class IotProductController {
      * @return
      */
     @GetMapping("/detail")
-    public Result<Object> detail(@RequestParam("id") @NotBlank(message = "id不能为空") Long id) {
+    public Result<Object> detail(@RequestParam("id") @NotNull(message = "id不能为空") Long id) {
         return Result.ok(iotProductService.detail(id), "查询成功");
     }
 
@@ -91,7 +91,7 @@ public class IotProductController {
      * @return
      */
     @GetMapping("/getMod")
-    public Result<Object> getMod(@RequestParam("productId") @NotBlank(message = "id不能为空") Long productId) {
+    public Result<Object> getMod(@RequestParam("productId") @NotNull(message = "id不能为空") Long productId) {
         return Result.ok(iotModService.queryMod(productId), "查询成功");
     }
 

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/mapper/IotCloudMapper.java

@@ -1,13 +1,22 @@
 package com.middle.platform.manage.biz.mapper;
 
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.middle.platform.manage.api.pojo.IotCloudVo;
 import com.middle.platform.manage.biz.entity.IotCloud;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
 
 /**
-* @date 2023-12-19 13:43:38
-* @author xucaiqin
-*/
+ * @author xucaiqin
+ * @date 2023-12-19 13:43:38
+ */
 @Mapper
 public interface IotCloudMapper extends BaseMapper<IotCloud> {
-}
+    /**
+     * 查询指定产品的云函数
+     *
+     * @param productId
+     * @return
+     */
+    IotCloudVo selectCloud(@Param("productId") Long productId);
+}

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/mapper/IotProductMapper.java

@@ -1,11 +1,14 @@
 package com.middle.platform.manage.biz.mapper;
 
 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.middle.platform.manage.api.pojo.ProductVo;
+import com.middle.platform.manage.api.pojo.TopicVo;
 import com.middle.platform.manage.biz.domain.req.ProductPage;
 import com.middle.platform.manage.biz.domain.vo.IotProductDownVo;
 import com.middle.platform.manage.biz.domain.vo.IotProductVo;
 import com.middle.platform.manage.biz.entity.IotProduct;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
 
 import java.util.List;
 
@@ -24,4 +27,14 @@ public interface IotProductMapper extends BaseMapper<IotProduct> {
     List<IotProductVo> pageQuery(ProductPage devicePage);
 
     List<IotProductDownVo> list();
+
+    ProductVo selectProduct(@Param("productKey") String productKey,@Param("deviceSn")String deviceSn);
+
+    /**
+     * 根据类型查询对应的地址
+     *
+     * @param type 地址类型 1-mqtt 2-http 3-coap
+     * @return
+     */
+    List<TopicVo> queryProductTopic(@Param("type") Integer type);
 }

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/service/IotCloudService.java

@@ -5,13 +5,18 @@ import com.middle.platform.common.constant.Global;
 import com.middle.platform.common.exception.BusinessException;
 import com.middle.platform.manage.biz.domain.req.IotCloudPara;
 import com.middle.platform.manage.biz.domain.req.IotCloudTest;
+import com.middle.platform.manage.api.pojo.IotCloudVo;
 import com.middle.platform.manage.biz.entity.IotCloud;
 import com.middle.platform.manage.biz.mapper.IotCloudMapper;
+import com.middle.platform.redis.constant.CacheConstant;
+import jakarta.annotation.Resource;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.StringUtils;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
 import java.util.Objects;
+import java.util.Optional;
 
 /**
  * @author xucaiqin
@@ -21,6 +26,8 @@ import java.util.Objects;
 @RequiredArgsConstructor
 public class IotCloudService {
     private final IotCloudMapper iotCloudMapper;
+    @Resource
+    private RedisTemplate<String, Object> redisTemplate;
 
     /**
      * 新增或修改云函数
@@ -93,6 +100,21 @@ public class IotCloudService {
      */
     public void removeProduct(Long id) {
         iotCloudMapper.delete(new LambdaQueryWrapper<IotCloud>().eq(IotCloud::getProductId, id).eq(IotCloud::getDelFlag, Global.UN_DEL));
+    }
 
+    /**
+     * 查询指定产品的云函数
+     *
+     * @param productId
+     * @return
+     */
+    public IotCloudVo cacheCloud(Long productId) {
+        IotCloudVo cloud = (IotCloudVo) redisTemplate.opsForValue().get(String.format(CacheConstant.CLOUD_CACHE, productId));
+        if (Objects.nonNull(cloud)) {
+            return cloud;
+        }
+        IotCloudVo iotCloudVo = iotCloudMapper.selectCloud(productId);
+        Optional.ofNullable(iotCloudVo).ifPresent(iotCloud -> redisTemplate.opsForValue().set(String.format(CacheConstant.CLOUD_CACHE, productId), iotCloud));
+        return iotCloudVo;
     }
 }

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/service/IotModService.java

@@ -5,10 +5,15 @@ import com.middle.platform.common.constant.Global;
 import com.middle.platform.manage.biz.domain.req.IotModPara;
 import com.middle.platform.manage.biz.entity.IotMod;
 import com.middle.platform.manage.biz.mapper.IotModMapper;
+import com.middle.platform.redis.constant.CacheConstant;
+import jakarta.annotation.Resource;
 import lombok.RequiredArgsConstructor;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 
 import java.util.Objects;
+import java.util.Optional;
 
 /**
  * @author xucaiqin
@@ -18,6 +23,8 @@ import java.util.Objects;
 @RequiredArgsConstructor
 public class IotModService {
     private final IotModMapper iotModMapper;
+    @Resource
+    private RedisTemplate<String, Object> redisTemplate;
 
     /**
      * 保存物模型数据
@@ -54,6 +61,7 @@ public class IotModService {
     public Object queryMod(Long productId) {
         return query(productId);
     }
+
     /**
      * 通过产品id删除关联的物模型
      *
@@ -62,4 +70,20 @@ public class IotModService {
     public void removeProduct(Long id) {
         iotModMapper.delete(new LambdaQueryWrapper<IotMod>().eq(IotMod::getProductId, id).eq(IotMod::getDelFlag, Global.UN_DEL));
     }
+
+    /**
+     * 查询物模型json数据
+     *
+     * @param productId
+     * @return
+     */
+    public String cacheMod(Long productId) {
+        String tslJson = (String) redisTemplate.opsForValue().get(String.format(CacheConstant.MOD_CACHE, productId));
+        if (StringUtils.isNotBlank(tslJson)) {
+            return tslJson;
+        }
+        IotMod query = query(productId);
+        Optional.ofNullable(query).ifPresent(m -> redisTemplate.opsForValue().set(String.format(CacheConstant.MOD_CACHE, productId), m.getTslJson()));
+        return Objects.nonNull(query) ? query.getTslJson() : null;
+    }
 }

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/service/IotProductService.java

@@ -5,24 +5,30 @@ import com.github.pagehelper.PageHelper;
 import com.github.pagehelper.PageInfo;
 import com.middle.platform.common.exception.BusinessException;
 import com.middle.platform.common.utils.PageRes;
+import com.middle.platform.data.api.feign.TdApi;
+import com.middle.platform.manage.api.pojo.ProductVo;
+import com.middle.platform.manage.api.pojo.TopicVo;
+import com.middle.platform.manage.biz.constant.UrlType;
 import com.middle.platform.manage.biz.domain.req.IotProductPara;
 import com.middle.platform.manage.biz.domain.req.ProductPage;
 import com.middle.platform.manage.biz.domain.vo.IotProductDetailVo;
 import com.middle.platform.manage.biz.domain.vo.IotProductVo;
-import com.middle.platform.manage.biz.entity.IotDevice;
 import com.middle.platform.manage.biz.entity.IotProduct;
 import com.middle.platform.manage.biz.mapper.IotDeviceMapper;
 import com.middle.platform.manage.biz.mapper.IotProductMapper;
+import com.middle.platform.redis.constant.CacheConstant;
 import com.middle.platform.system.api.enums.DictType;
 import com.middle.platform.system.api.feign.DictApi;
 import com.middle.platform.system.api.feign.UserApi;
 import jakarta.annotation.Resource;
 import lombok.RequiredArgsConstructor;
 import org.springframework.beans.BeanUtils;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 
 import java.util.List;
+import java.util.Objects;
 import java.util.Optional;
 
 /**
@@ -38,9 +44,12 @@ public class IotProductService {
     private IotUrlService iotUrlService;
     private final IotModService iotModService;
     private final IotCloudService iotCloudService;
+    private final TdApi tdApi;
 
     private final UserApi userApi;
     private final DictApi dictApi;
+    @Resource
+    private RedisTemplate<String, Object> redisTemplate;
 
     /**
      * 分页查询产品数据
@@ -90,6 +99,8 @@ public class IotProductService {
         iotProductMapper.insert(iotProduct);
         /*保存默认topic*/
         iotUrlService.save(iotProduct);
+        /*创建产品*/
+        tdApi.createProduct(iotProduct.getCode());
         return null;
     }
 
@@ -126,9 +137,12 @@ public class IotProductService {
      * @return
      */
     public Object detail(Long id) {
-        IotDevice iotDevice = iotDeviceMapper.selectById(id);
+        IotProduct iotProduct = iotProductMapper.selectById(id);
+        if (Objects.isNull(iotProduct)) {
+            throw new BusinessException("当前产品不存在");
+        }
         IotProductDetailVo iotProductDetailVo = new IotProductDetailVo();
-        BeanUtils.copyProperties(iotDevice, iotProductDetailVo);
+        BeanUtils.copyProperties(iotProduct, iotProductDetailVo);
         return iotProductDetailVo;
     }
 
@@ -150,4 +164,30 @@ public class IotProductService {
     public Object list() {
         return iotProductMapper.list();
     }
+
+    /**
+     * 查询产品的缓存
+     *
+     * @param productKey 产品key
+     * @param deviceSn   设备sn
+     * @return
+     */
+    public ProductVo cacheProduct(String productKey, String deviceSn) {
+        ProductVo productVo = (ProductVo) redisTemplate.opsForValue().get(String.format(CacheConstant.PRODUCT_CACHE, productKey, deviceSn));
+        if (Objects.nonNull(productVo)) {
+            return productVo;
+        }
+        ProductVo productVoC = iotProductMapper.selectProduct(productKey, deviceSn);
+        Optional.ofNullable(productVoC).ifPresent(product -> redisTemplate.opsForValue().set(String.format(CacheConstant.PRODUCT_CACHE, productKey, deviceSn), product));
+        return productVoC;
+    }
+
+    /**
+     * 通过产品关联的通信地址
+     *
+     * @return
+     */
+    public List<TopicVo> queryList() {
+        return iotProductMapper.queryProductTopic(UrlType.MQTT);
+    }
 }

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/service/IotUrlService.java

@@ -14,6 +14,7 @@ import com.middle.platform.manage.biz.domain.req.IotUrlPara;
 import com.middle.platform.manage.biz.entity.IotProduct;
 import com.middle.platform.manage.biz.entity.IotUrl;
 import com.middle.platform.manage.biz.mapper.IotUrlMapper;
+import com.middle.platform.redis.service.CacheService;
 import jakarta.annotation.Resource;
 import lombok.RequiredArgsConstructor;
 import org.apache.commons.lang3.StringUtils;
@@ -34,7 +35,57 @@ public class IotUrlService {
     @Resource
     private IotProductService iotProductService;
     private final DynamicTopicApi dynamicTopicApi;
-    private String topicPrefix = "/device/%s/*/";
+    private final CacheService cacheService;
+    private String topicPrefix = "/iot/%s/+/";
+
+    /**
+     * 校验topic
+     *
+     * @param iotUrlPara
+     */
+    private void checkTopic(IotUrlPara iotUrlPara) {
+        //校验前缀是否为 /iot/{productKey}/*/
+        IotProduct query = iotProductService.query(iotUrlPara.getProductId());
+        if (Objects.isNull(query)) {
+            throw new BusinessException("产品不存在");
+        }
+        //产品key
+        String code = query.getCode();
+        if (!StringUtils.startsWith(iotUrlPara.getUrl(), String.format(topicPrefix, code))) {
+            throw new BusinessException("topic必须以+" + String.format(topicPrefix, code) + "为前缀");
+        }
+        //todo topic合规校验
+
+    }
+
+    /**
+     * 订阅地址，添加缓存
+     *
+     * @param url  topic地址
+     * @param func func+permission
+     */
+    private void subUrl(String url, String func) {
+        ThreadTask.addJob(() -> {
+            //订阅topic
+            dynamicTopicApi.saveTopic(new TopicDto(url, 0));
+            //写入 CacheConstant.TOPIC_CACHE缓存
+            cacheService.setTopic(url, func);
+        });
+    }
+
+    /**
+     * 取消订阅的topic,并删除缓存的topic
+     *
+     * @param url
+     */
+    private void removeUrl(String url) {
+        ThreadTask.addJob(() -> {
+            dynamicTopicApi.removeTopic(new TopicDto(url, 0));
+            //删除原topic
+            cacheService.delTopic(url);
+        });
+
+    }
 
     /**
      * 新增topic
@@ -42,6 +93,7 @@ public class IotUrlService {
      * @param iotUrlPara
      * @return
      */
+    @Transactional(rollbackFor = Exception.class)
     public Object saveUrl(IotUrlPara iotUrlPara) {
         //topi规则校验
         checkTopic(iotUrlPara);
@@ -55,31 +107,13 @@ public class IotUrlService {
         iotUrl.setRemark(iotUrlPara.getRemark());
         iotUrlMapper.insert(iotUrl);
         //新增订阅的topic
-        if (Objects.equals(iotUrlPara.getType(), UrlType.MQTT) && (Objects.equals(iotUrlPara.getPermission(), TopicType.PUB) || Objects.equals(iotUrlPara.getPermission(), TopicType.BOTH))) {
-            dynamicTopicApi.saveTopic(new TopicDto(iotUrlPara.getUrl(), 0));
+        if (Objects.equals(iotUrlPara.getType(), UrlType.MQTT)
+                && (Objects.equals(iotUrlPara.getPermission(), TopicType.PUB) || Objects.equals(iotUrlPara.getPermission(), TopicType.BOTH))) {
+            subUrl(iotUrl.getUrl(), iotUrl.getFunc() + String.valueOf(iotUrl.getPermission()));
         }
         return true;
     }
 
-    /**
-     * 校验topic
-     *
-     * @param iotUrlPara
-     */
-    private void checkTopic(IotUrlPara iotUrlPara) {
-        //校验前缀是否为 /device/{productKey}/*/
-        IotProduct query = iotProductService.query(iotUrlPara.getProductId());
-        if (Objects.isNull(query)) {
-            throw new BusinessException("产品不存在");
-        }
-        //产品key
-        String code = query.getCode();
-        if (!StringUtils.startsWith(iotUrlPara.getUrl(), String.format(topicPrefix, code))) {
-            throw new BusinessException("topic必须以+" + String.format(topicPrefix, code) + "为前缀");
-        }
-        //todo topic合规校验
-
-    }
 
     /**
      * 修改topic
@@ -109,9 +143,8 @@ public class IotUrlService {
         iotUrlMapper.updateById(iotUrl);
         //取消订阅原topic 订阅topic
         if (Objects.equals(iotUrlPara.getType(), UrlType.MQTT) && (Objects.equals(iotUrlPara.getPermission(), TopicType.PUB) || Objects.equals(iotUrlPara.getPermission(), TopicType.BOTH))) {
-            dynamicTopicApi.removeTopic(new TopicDto(iotUrlPara.getUrl(), 0));
-            dynamicTopicApi.saveTopic(new TopicDto(iotUrlPara.getUrl(), 0));
-
+            removeUrl(iotUrlCheck.getUrl());
+            subUrl(iotUrl.getUrl(), iotUrl.getFunc() + String.valueOf(iotUrl.getPermission()));
         }
         return true;
     }
@@ -133,7 +166,7 @@ public class IotUrlService {
         iotUrlMapper.update(iotUrl, new LambdaQueryWrapper<IotUrl>().eq(IotUrl::getId, id).eq(IotUrl::getDelFlag, Global.UN_DEL));
         //取消订阅
         if (Objects.equals(iotUrlCheck.getType(), UrlType.MQTT) && (Objects.equals(iotUrlCheck.getPermission(), TopicType.PUB) || Objects.equals(iotUrlCheck.getPermission(), TopicType.BOTH))) {
-            dynamicTopicApi.removeTopic(new TopicDto(iotUrlCheck.getUrl(), 0));
+            removeUrl(iotUrlCheck.getUrl());
         }
         return true;
     }
@@ -167,9 +200,13 @@ public class IotUrlService {
                 iotUrl.setType(iotProduct.getReportProtocol());
                 iotUrl.setRemark(value.getDesc());
                 iotUrlMapper.insert(iotUrl);
+                subUrl(iotUrl.getUrl(), iotUrl.getFunc() + String.valueOf(iotUrl.getPermission()));
             }
         }
+        //http
+        if (Objects.equals(iotProduct.getReportProtocol(), UrlProtocol.HTTP)) {
 
+        }
     }
 
     /**

iot-module/iot-module-manage/iot-module-manage-biz/src/main/java/com/middle/platform/manage/biz/service/ThreadTask.java

@@ -0,0 +1,25 @@
+package com.middle.platform.manage.biz.service;
+
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author xucaiqin
+ * @date 2023-12-25 09:03:26
+ */
+public class ThreadTask {
+
+    private ThreadTask() {
+    }
+
+    public static class Task {
+        private final static ThreadPoolExecutor THREAD_POOL_EXECUTOR =
+                new ThreadPoolExecutor(1, 2, 1, TimeUnit.MINUTES, new LinkedBlockingQueue<>(12));
+
+    }
+
+    public static void addJob(Runnable runnable) {
+        Task.THREAD_POOL_EXECUTOR.execute(runnable);
+    }
+}

iot-module/iot-module-manage/iot-module-manage-biz/src/main/resources/mapper/IotCloudMapper.xml

@@ -17,7 +17,21 @@
   </resultMap>
   <sql id="Base_Column_List">
     <!--@mbg.generated-->
-    id, product_id, lang, cloud_text, create_by, create_time, update_by, update_time, 
+    id, product_id, lang, cloud_text, create_by, create_time, update_by, update_time,
     delete_time, del_flag
   </sql>
-</mapper>
+
+  <select id="selectCloud" resultType="com.middle.platform.manage.api.pojo.IotCloudVo">
+      select ic.id,
+             ic.cloud_text,
+             ic.lang
+      from iot_cloud ic
+      <where>
+          <if test="productId != null">
+              and ic.product_id = #{productId,jdbcType=BIGINT}
+          </if>
+          and ic.del_flag = 0
+          limit 1
+      </where>
+  </select>
+</mapper>

iot-module/iot-module-manage/iot-module-manage-biz/src/main/resources/mapper/IotProductMapper.xml

@@ -1,52 +1,99 @@
 <?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.middle.platform.manage.biz.mapper.IotProductMapper">
-  <resultMap id="BaseResultMap" type="com.middle.platform.manage.biz.entity.IotProduct">
-    <!--@mbg.generated-->
-    <!--@Table iot_product-->
-    <id column="id" jdbcType="BIGINT" property="id" />
-    <result column="code" jdbcType="VARCHAR" property="code" />
-    <result column="name" jdbcType="VARCHAR" property="name" />
-    <result column="secret" jdbcType="VARCHAR" property="secret" />
-    <result column="node_type" jdbcType="TINYINT" property="nodeType" />
-    <result column="network_type" jdbcType="TINYINT" property="networkType" />
-    <result column="report_protocol" jdbcType="TINYINT" property="reportProtocol" />
-    <result column="vendors" jdbcType="TINYINT" property="vendors" />
-    <result column="data_format" jdbcType="TINYINT" property="dataFormat" />
-    <result column="auth_type" jdbcType="TINYINT" property="authType" />
-    <result column="tag" jdbcType="VARCHAR" property="tag" />
-    <result column="size" jdbcType="INTEGER" property="size" />
-    <result column="remark" jdbcType="VARCHAR" property="remark" />
-    <result column="create_by" jdbcType="BIGINT" property="createBy" />
-    <result column="create_time" jdbcType="TIMESTAMP" property="createTime" />
-    <result column="update_by" jdbcType="BIGINT" property="updateBy" />
-    <result column="update_time" jdbcType="TIMESTAMP" property="updateTime" />
-    <result column="delete_time" jdbcType="TIMESTAMP" property="deleteTime" />
-    <result column="del_flag" jdbcType="TINYINT" property="delFlag" />
-  </resultMap>
-  <sql id="Base_Column_List">
-    <!--@mbg.generated-->
-    id, code, `name`, secret, node_type, network_type, report_protocol, data_format, vendors,
-    auth_type, tag, `size`, remark, create_by, create_time, update_by, update_time, delete_time,
-    del_flag
-  </sql>
+    <resultMap id="BaseResultMap" type="com.middle.platform.manage.biz.entity.IotProduct">
+        <!--@mbg.generated-->
+        <!--@Table iot_product-->
+        <id column="id" jdbcType="BIGINT" property="id"/>
+        <result column="code" jdbcType="VARCHAR" property="code"/>
+        <result column="name" jdbcType="VARCHAR" property="name"/>
+        <result column="secret" jdbcType="VARCHAR" property="secret"/>
+        <result column="node_type" jdbcType="TINYINT" property="nodeType"/>
+        <result column="network_type" jdbcType="TINYINT" property="networkType"/>
+        <result column="report_protocol" jdbcType="TINYINT" property="reportProtocol"/>
+        <result column="vendors" jdbcType="TINYINT" property="vendors"/>
+        <result column="data_format" jdbcType="TINYINT" property="dataFormat"/>
+        <result column="auth_type" jdbcType="TINYINT" property="authType"/>
+        <result column="tag" jdbcType="VARCHAR" property="tag"/>
+        <result column="size" jdbcType="INTEGER" property="size"/>
+        <result column="remark" jdbcType="VARCHAR" property="remark"/>
+        <result column="create_by" jdbcType="BIGINT" property="createBy"/>
+        <result column="create_time" jdbcType="TIMESTAMP" property="createTime"/>
+        <result column="update_by" jdbcType="BIGINT" property="updateBy"/>
+        <result column="update_time" jdbcType="TIMESTAMP" property="updateTime"/>
+        <result column="delete_time" jdbcType="TIMESTAMP" property="deleteTime"/>
+        <result column="del_flag" jdbcType="TINYINT" property="delFlag"/>
+    </resultMap>
+    <sql id="Base_Column_List">
+        <!--@mbg.generated-->
+        id,
+        code,
+        `name`,
+        secret,
+        node_type,
+        network_type,
+        report_protocol,
+        data_format,
+        vendors,
+        auth_type,
+        tag,
+        `size`,
+        remark,
+        create_by,
+        create_time,
+        update_by,
+        update_time,
+        delete_time,
+        del_flag
+    </sql>
 
-  <select id="pageQuery" resultType="com.middle.platform.manage.biz.domain.vo.IotProductVo">
-      select *
-      from iot_product ip
-      <where>
-          <if test="keywords != null and keywords != ''">
-              and ip.name like concat('%', #{keywords,jdbcType=VARCHAR}, '%')
-          </if>
-          and ip.del_flag = 0
-      </where>
-  </select>
+    <select id="pageQuery" resultType="com.middle.platform.manage.biz.domain.vo.IotProductVo">
+        select *
+        from iot_product ip
+        <where>
+            <if test="keywords != null and keywords != ''">
+                and ip.name like concat('%', #{keywords,jdbcType=VARCHAR}, '%')
+            </if>
+            and ip.del_flag = 0
+        </where>
+    </select>
 
-  <select id="list" resultType="com.middle.platform.manage.biz.domain.vo.IotProductDownVo">
-      select ip.id, ip.name, ip.code
-      from iot_product ip
-      <where>
-          and ip.del_flag = 0
-      </where>
-  </select>
+    <select id="list" resultType="com.middle.platform.manage.biz.domain.vo.IotProductDownVo">
+        select ip.id, ip.name, ip.code
+        from iot_product ip
+        <where>
+            and ip.del_flag = 0
+        </where>
+    </select>
+
+    <select id="selectProduct" resultType="com.middle.platform.manage.api.pojo.ProductVo">
+        select ip.id,
+               ip.code,
+               ip.data_format,
+               id.guid
+        from iot_product ip
+                 inner join iot_device id on ip.id = id.product_id and id.del_flag = 0
+        <where>
+            <if test="productKey != null and productKey != ''">
+                and ip.code = #{productKey,jdbcType=VARCHAR}
+            </if>
+            <if test="deviceSn != null and deviceSn != ''">
+                and id.sn = #{deviceSn,jdbcType=VARCHAR}
+            </if>
+            and ip.del_flag = 0
+        </where>
+    </select>
+
+    <select id="queryProductTopic" resultType="com.middle.platform.manage.api.pojo.TopicVo">
+        select iu.url, iu.permission, iu.func
+        from iot_product ip
+                 inner join iot_url iu on ip.id = iu.product_id and iu.del_flag = 0
+        <where>
+            <if test="type != null">
+                and iu.type = #{type,jdbcType=INTEGER}
+            </if>
+            and iu.permission in (1, 3)
+            and ip.del_flag = 0
+        </where>
+    </select>
 </mapper>