修改redis 过期事件通知

device-manager/src/main/java/com/mortals/xhx/busiz/web/DeviceApiController.java

@@ -274,7 +274,7 @@ public class DeviceApiController {
                 TopicPartitionInfo info = TopicPartitionInfo.builder().exchangeName(exchangeName).topic(Constant.UPLOAD_TOPIC + deviceEntity.getDeviceMac()).build();
                 TbQueueMsgHeaders header = new DefaultTbQueueMsgHeaders();
                 header.put(MessageHeader.MESSAGETYPE, Constant.MESSAGETYPE_HEARTBEAT);
-                for (int i = 0; i < 1000; i++) {
+                for (int i = 0; i < 10; i++) {
                     deviceService.sendDeviceMessage(deviceEntity, info, header, JSON.toJSONString(req), null);
                     // Thread.sleep(50);
                 }

device-manager/src/main/java/com/mortals/xhx/daemon/applicationservice/DeviceMsgComsumerStartedService.java

@@ -74,7 +74,7 @@ public class DeviceMsgComsumerStartedService implements IApplicationStartedServi
         log.info("服务端消费消息服务开始..");
         TbQueueConsumer<TbQueueMsg> mainConsumer = consumerService.getMainConsumer();
         if (!ObjectUtils.isEmpty(mainConsumer)) {
-            //订阅所有已几快活设备
+            //订阅所有已激活设备
             Set<TopicPartitionInfo> topicPartitionInfoSet = deviceService.find(new DeviceQuery().active(ActiveEnum.已激活.getValue()).status(StatusEnum.启用.getValue())).stream()
                     .filter(f -> !ObjectUtils.isEmpty(platformService.get(f.getPlatformId())))
                     .filter(f -> !ObjectUtils.isEmpty(productService.get(f.getProductId())))

device-manager/src/main/java/com/mortals/xhx/module/device/service/impl/DeviceServiceImpl.java

@@ -114,6 +114,13 @@ public class DeviceServiceImpl extends AbstractCRUDCacheServiceImpl<DeviceDao, D
         if (!ObjectUtils.isEmpty(platformEntity) && !ObjectUtils.isEmpty(productEntity)) {
             //注册rabbmit相关队列与绑定
             registerRabbitQueue(deviceEntity, platformEntity, productEntity);
+
+            TbQueueConsumer<TbQueueMsg> mainConsumer =consumerService.getMainConsumer();
+            String exchangeName = platformEntity.getPlatformSn() + Constant.EXCHANGE_SPLIT + productEntity.getProductCode();
+            TopicPartitionInfo topicPartitionInfo = new TopicPartitionInfo(Constant.UPLOAD_TOPIC + deviceCode, null, exchangeName);
+            Set<TopicPartitionInfo> topicPartitionInfos = new HashSet<>();
+            topicPartitionInfos.add(topicPartitionInfo);
+            mainConsumer.subscribe(topicPartitionInfos);
             //新增设备通知第三方平台
             sendThirdParty(deviceEntity, productEntity, platformEntity, DeviceStatusEnum.ACTIVE);
         }