feat：【IoT 物联网】增加（重构） IotDevicePropertyMessageSubscriber 记录设备属性

YunaiV
1 parent 0bb01eae
Showing 8 changed files with 59 additions and 39 deletions
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/controller/admin/device/vo/data/IotDevicePropertyHistoryPageReqVO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/RedisKeyConstants.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/redis/device/DevicePropertyRedisDAO.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/tdengine/IotDevicePropertyMapper.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/mq/consumer/rule/IotRuleSceneMessageHandler.java
yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/data/IotDevicePropertyServiceImpl.java
yudao-module-iot/yudao-module-iot-biz/src/main/resources/mapper/device/IotDevicePropertyMapper.xml
@@ -20,8 +20,11 @@ public class IotDevicePropertyHistoryPageReqVO extends PageParam {
     @NotNull(message = "设备编号不能为空")
     private Long deviceId;
-    @Schema(description = "设备 Key", hidden = true)
-    private String deviceKey; // 非前端传递，后端自己查询设置
+    @Schema(description = "产品 Key", hidden = true)
+    private String productKey; // 非前端传递，后端自己查询设置
+
+    @Schema(description = "设备名称", hidden = true)
+    private String deviceName; // 非前端传递，后端自己查询设置
     @Schema(description = "属性标识符", requiredMode = Schema.RequiredMode.REQUIRED)
     @NotEmpty(message = "属性标识符不能为空")
@@ -9,15 +9,14 @@ import cn.iocoder.yudao.module.iot.dal.dataobject.device.IotDevicePropertyDO;
  */
 public interface RedisKeyConstants {
-    // TODO @芋艿：弱化 deviceKey；使用 product_key + device_name 替代
     /**
      * 设备属性的数据缓存，采用 HASH 结构
      * <p>
-     * KEY 格式：device_property:{deviceKey}
+     * KEY 格式：device_property:{productKey},${deviceName}
      * HASH KEY：identifier 属性标识
      * VALUE 数据类型：String(JSON) {@link IotDevicePropertyDO}
      */
-    String DEVICE_PROPERTY = "iot:device_property:%s";
+    String DEVICE_PROPERTY = "iot:device_property:%s,%s";
     /**
      * 设备的最后上报时间，采用 ZSET 结构
@@ -22,8 +22,8 @@ public class DevicePropertyRedisDAO {
     @Resource
     private StringRedisTemplate stringRedisTemplate;
-    public Map<String, IotDevicePropertyDO> get(String deviceKey) {
-        String redisKey = formatKey(deviceKey);
+    public Map<String, IotDevicePropertyDO> get(String productKey, String deviceName) {
+        String redisKey = formatKey(productKey, deviceName);
         Map<Object, Object> entries = stringRedisTemplate.opsForHash().entries(redisKey);
         if (CollUtil.isEmpty(entries)) {
             return Collections.emptyMap();
@@ -33,18 +33,18 @@ public class DevicePropertyRedisDAO {
                 entry -> JsonUtils.parseObject((String) entry.getValue(), IotDevicePropertyDO.class));
     }
-    public void putAll(String deviceKey, Map<String, IotDevicePropertyDO> properties) {
+    public void putAll(String productKey, String deviceName, Map<String, IotDevicePropertyDO> properties) {
         if (CollUtil.isEmpty(properties)) {
             return;
         }
-        String redisKey = formatKey(deviceKey);
+        String redisKey = formatKey(productKey, deviceName);
         stringRedisTemplate.opsForHash().putAll(redisKey, convertMap(properties.entrySet(),
                 Map.Entry::getKey,
                 entry -> JsonUtils.toJsonString(entry.getValue())));
     }
-    private static String formatKey(String deviceKey) {
-        return String.format(DEVICE_PROPERTY, deviceKey);
+    private static String formatKey(String productKey, String deviceName) {
+        return String.format(DEVICE_PROPERTY, productKey, deviceName);
     }
 }
@@ -33,7 +33,7 @@ public interface IotDevicePropertyMapper {
                                             List<TDengineTableField> oldFields,
                                             List<TDengineTableField> newFields) {
         oldFields.removeIf(field -> StrUtil.equalsAny(field.getField(),
-                TDengineTableField.FIELD_TS, "report_time", "device_key"));
+                TDengineTableField.FIELD_TS, "report_time", "device_name"));
         List<TDengineTableField> addFields = newFields.stream().filter( // 新增的字段
                         newField -> oldFields.stream().noneMatch(oldField -> oldField.getField().equals(newField.getField())))
                 .collect(Collectors.toList());
 package cn.iocoder.yudao.module.iot.mq.consumer.device;
-import cn.hutool.core.util.ObjectUtil;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageBus;
+import cn.iocoder.yudao.module.iot.core.messagebus.core.IotMessageSubscriber;
 import cn.iocoder.yudao.module.iot.core.mq.message.IotDeviceMessage;
 import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageIdentifierEnum;
 import cn.iocoder.yudao.module.iot.enums.device.IotDeviceMessageTypeEnum;
 import cn.iocoder.yudao.module.iot.service.device.data.IotDevicePropertyService;
+import com.google.common.base.Objects;
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.Resource;
 import lombok.extern.slf4j.Slf4j;
-import org.springframework.context.event.EventListener;
-import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
-import jakarta.annotation.Resource;
-
 /**
- * 针对 {@link IotDeviceMessage} 的消费者，记录设备属性
+ * 针对 {@link IotDeviceMessage} 的消费者：记录设备属性
  *
  * @author alwayssuper
  */
 @Component
 @Slf4j
-public class IotDevicePropertyMessageConsumer {
+public class IotDevicePropertyMessageSubscriber implements IotMessageSubscriber<IotDeviceMessage> {
     @Resource
     private IotDevicePropertyService deviceDataService;
-    @EventListener
-    @Async
+    @Resource
+    private IotMessageBus messageBus;
+
+    @PostConstruct
+    public void init() {
+        messageBus.register(this);
+    }
+
+    @Override
+    public String getTopic() {
+        return IotDeviceMessage.MESSAGE_BUS_DEVICE_MESSAGE_TOPIC;
+    }
+
+    @Override
+    public String getGroup() {
+        return "iot_device_property_consumer";
+    }
+
+    @Override
     public void onMessage(IotDeviceMessage message) {
-        if (ObjectUtil.notEqual(message.getType(), IotDeviceMessageTypeEnum.PROPERTY.getType())
-                || ObjectUtil.notEqual(message.getIdentifier(), IotDeviceMessageIdentifierEnum.PROPERTY_REPORT.getIdentifier())) {
-            return;
+        if (Objects.equal(message.getType(), IotDeviceMessageTypeEnum.PROPERTY.getType())
+                && Objects.equal(message.getIdentifier(), IotDeviceMessageIdentifierEnum.PROPERTY_REPORT.getIdentifier())) {
+            // 保存设备属性
+            deviceDataService.saveDeviceProperty(message);
         }
-        log.info("[onMessage][消息内容({})]", message);
-
-        // 保存设备属性
-        deviceDataService.saveDeviceProperty(message);
     }
 }
@@ -8,6 +8,7 @@ import org.springframework.context.event.EventListener;
 import org.springframework.scheduling.annotation.Async;
 import org.springframework.stereotype.Component;
+// TODO @puhui999：后面重构哈
 /**
  * 针对 {@link IotDeviceMessage} 的消费者，处理规则场景
  *
@@ -128,7 +128,8 @@ public class IotDevicePropertyServiceImpl implements IotDevicePropertyService {
             return;
         }
         // 1. 获得设备信息
-        IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceNameFromCache(message.getProductKey(), message.getDeviceName());
+        IotDeviceDO device = deviceService.getDeviceByProductKeyAndDeviceNameFromCache(
+                message.getProductKey(), message.getDeviceName());
         if (device == null) {
             log.error("[saveDeviceProperty][消息({}) 对应的设备不存在]", message);
             return;
@@ -155,9 +156,9 @@ public class IotDevicePropertyServiceImpl implements IotDevicePropertyService {
                 LocalDateTimeUtil.toEpochMilli(message.getReportTime()));
         // 3.2 保存设备属性【日志】
-        // TODO @芋艿：这里要调整下；
-        deviceDataRedisDAO.putAll(device.getDeviceKey(), convertMap(properties.entrySet(), Map.Entry::getKey,
-                entry -> IotDevicePropertyDO.builder().value(entry.getValue()).updateTime(message.getReportTime()).build()));
+        Map<String, IotDevicePropertyDO> properties2 = convertMap(properties.entrySet(), Map.Entry::getKey, entry ->
+                IotDevicePropertyDO.builder().value(entry.getValue()).updateTime(message.getReportTime()).build());
+        deviceDataRedisDAO.putAll(device.getProductKey(), device.getDeviceName(), properties2);
     }
     @Override
@@ -166,15 +167,16 @@ public class IotDevicePropertyServiceImpl implements IotDevicePropertyService {
         IotDeviceDO device = deviceService.validateDeviceExists(deviceId);
         // 获得设备属性
-        return deviceDataRedisDAO.get(device.getDeviceKey());
+        return deviceDataRedisDAO.get(device.getProductKey(), device.getDeviceName());
     }
     @Override
     public PageResult<IotDevicePropertyRespVO> getHistoryDevicePropertyPage(IotDevicePropertyHistoryPageReqVO pageReqVO) {
         // 获取设备信息
         IotDeviceDO device = deviceService.validateDeviceExists(pageReqVO.getDeviceId());
-        pageReqVO.setDeviceKey(device.getDeviceKey());
+        pageReqVO.setProductKey(device.getProductKey()).setDeviceName(device.getDeviceName());
+        // 分页查询
         try {
             IPage<IotDevicePropertyRespVO> page = devicePropertyMapper.selectPageByHistory(
                     new Page<>(pageReqVO.getPageNo(), pageReqVO.getPageSize()), pageReqVO);
@@ -20,7 +20,7 @@
         </foreach>
         )
         TAGS (
-            device_key NCHAR(50)
+            device_name NCHAR(50)
         )
     </update>
@@ -46,9 +46,9 @@
     </update>
     <insert id="insert">
-        INSERT INTO device_property_${device.deviceKey}
+        INSERT INTO device_property_${device.productKey}_${device.deviceName}
         USING product_property_${device.productKey}
-        TAGS ('${device.deviceKey}')
+        TAGS ('${device.deviceName}')
         (ts, report_time,
         <foreach item="key" collection="properties.keys" separator=",">
             ${@cn.hutool.core.util.StrUtil@toUnderlineCase(key)}
@@ -66,9 +66,10 @@
         DESCRIBE product_property_${productKey}
     </select>
-    <select id="selectPageByHistory" resultType="cn.iocoder.yudao.module.iot.controller.admin.device.vo.data.IotDevicePropertyRespVO">
+    <select id="selectPageByHistory"
+            resultType="cn.iocoder.yudao.module.iot.controller.admin.device.vo.data.IotDevicePropertyRespVO">
         SELECT ${@cn.hutool.core.util.StrUtil@toUnderlineCase(reqVO.identifier)} AS `value`, ts AS update_time
-        FROM device_property_${reqVO.deviceKey}
+        FROM device_property_${reqVO.productKey}_${reqVO.deviceName}
         WHERE ${@cn.hutool.core.util.StrUtil@toUnderlineCase(reqVO.identifier)} IS NOT NULL
           AND ts BETWEEN ${@cn.hutool.core.date.LocalDateTimeUtil@toEpochMilli(reqVO.times[0])}
               AND ${@cn.hutool.core.date.LocalDateTimeUtil@toEpochMilli(reqVO.times[1])}