From 0758f2de3acae22a8e371c6cf39b2a0c8b4cfaeb Mon Sep 17 00:00:00 2001 From: HuangHuiKang Date: Wed, 1 Jul 2026 11:34:29 +0800 Subject: [PATCH] =?UTF-8?q?fix:=E4=BF=AE=E5=A4=8Diot=E9=87=87=E9=9B=86?= =?UTF-8?q?=E8=AE=BE=E5=A4=87=E9=97=AE=E9=A2=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../DeviceModelAttributeMapper.java | 2 +- .../iot/service/device/DeviceServiceImpl.java | 97 ++++++++++++++----- .../iot/service/device/TDengineService.java | 8 +- 3 files changed, 80 insertions(+), 27 deletions(-) diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/devicemodelattribute/DeviceModelAttributeMapper.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/devicemodelattribute/DeviceModelAttributeMapper.java index b0d111d3f..bccbd81e3 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/devicemodelattribute/DeviceModelAttributeMapper.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/dal/mysql/devicemodelattribute/DeviceModelAttributeMapper.java @@ -47,7 +47,7 @@ public interface DeviceModelAttributeMapper extends BaseMapperXlambdaQuery().eq(DeviceDO::getTopic, topic)); + private void validTopicExists(String topic, Long currentDeviceId) { + if (StringUtils.isBlank(topic)) { + return; + } + boolean exists = deviceMapper.exists(Wrappers.lambdaQuery() + .eq(DeviceDO::getTopic, topic) + .ne(currentDeviceId != null, DeviceDO::getId, currentDeviceId)); if (exists){ throw exception(DEVICE_MQTT_TOPIC_ALREADY_EXIST); } } + private void disableGatewayByDevice(DeviceDO device) { + if (device == null) { + return; + } + LambdaUpdateWrapper updateWrapper = new LambdaUpdateWrapper() + .set(GatewayDO::getIsEnable, false) + .set(GatewayDO::getUpdateTime, LocalDateTime.now()); + if (StringUtils.isNotBlank(device.getTopic())) { + updateWrapper.eq(GatewayDO::getTopic, device.getTopic()); + } else if (StringUtils.isNotBlank(device.getDeviceCode())) { + updateWrapper.eq(GatewayDO::getGatewayCode, device.getDeviceCode()); + } else { + return; + } + if (StringUtils.isNotBlank(device.getDeviceCode())) { + updateWrapper.or().eq(GatewayDO::getGatewayCode, device.getDeviceCode()); + } + gatewayMapper.update(null, updateWrapper); + } + + private void removeRuntimeTopic(String topic) { + if (StringUtils.isBlank(topic) || defaultBizTopicSet == null || defaultBizTopicSet.getTopicMap() == null) { + return; + } + synchronized (defaultBizTopicSet.getTopicMap()) { + defaultBizTopicSet.getTopicMap().removeIf(item -> topic.equals(item.getSubTopic())); + } + } + @Override @Transactional(rollbackFor = Exception.class) public void deleteDevice(List ids) { @@ -323,10 +362,12 @@ public class DeviceServiceImpl implements DeviceService { deviceContactModelMapper.delete(deviceContactModelDOLambdaQueryWrapper); } - private void validateDeviceExists(Long id) { - if (deviceMapper.selectById(id) == null) { + private DeviceDO validateDeviceExists(Long id) { + DeviceDO device = deviceMapper.selectById(id); + if (device == null) { throw exception(DEVICE_NOT_EXISTS); } + return device; } @Override @@ -795,6 +836,7 @@ public class DeviceServiceImpl implements DeviceService { } @Override +// @Transactional(rollbackFor = Exception.class) public Long copyDevice(Long id) { if (id == null){ throw exception(DEVICE_ID_MODEL_NOT_EXISTS); @@ -811,6 +853,7 @@ public class DeviceServiceImpl implements DeviceService { int randomNumber = random.nextInt(9000) + 1000; newDevice.setDeviceCode(deviceDO.getDeviceCode()+ "-" + randomNumber); newDevice.setDeviceName(deviceDO.getDeviceName() + "-副本"); + newDevice.setTopic(null); deviceMapper.insert(newDevice); //复制关联表 @@ -818,14 +861,20 @@ public class DeviceServiceImpl implements DeviceService { lambdaQueryWrapper.eq(DeviceContactModelDO::getDeviceId,deviceDO.getId()); List deviceContactModelDOS = deviceContactModelMapper.selectList(lambdaQueryWrapper); - + List newContactModelList = new ArrayList<>(); if (deviceContactModelDOS != null && !deviceContactModelDOS.isEmpty()){ - for (DeviceContactModelDO deviceModelAttributeDOS : deviceContactModelDOS) { - deviceModelAttributeDOS.setId(null); - deviceModelAttributeDOS.setDeviceId(newDevice.getId()); + for (DeviceContactModelDO deviceContactModelDO : deviceContactModelDOS) { + DeviceContactModelDO newContactModel = new DeviceContactModelDO(); + BeanUtils.copyProperties(deviceContactModelDO, newContactModel); + newContactModel.setId(null); + newContactModel.setDeviceId(newDevice.getId()); + newContactModel.setCreateTime(LocalDateTime.now()); + newContactModel.setUpdateTime(LocalDateTime.now()); + newContactModelList.add(newContactModel); } - deviceContactModelMapper.insertBatch(deviceContactModelDOS); + deviceContactModelMapper.insertBatch(newContactModelList); } + tdengineService.createTdengineTable(newDevice.getId(), newContactModelList); return newDevice.getId(); @@ -2032,12 +2081,7 @@ public class DeviceServiceImpl implements DeviceService { throw exception(DEVICE_MQTT_TOPIC_EXIST); } - // 2. 如果状态没有变化，直接返回 - if (Objects.equals(deviceDO.getIsEnable(), updateEnabledReqVO.getEnabled())) { - return; - } - - // 3. 执行状态更新操作 + // 2. 执行状态同步操作：重复启停也要同步 MQTT 与 gateway，保证幂等 executeEnableUpdate(deviceDO, updateEnabledReqVO.getEnabled()); @@ -2051,6 +2095,8 @@ public class DeviceServiceImpl implements DeviceService { private void executeEnableUpdate(DeviceDO deviceDO, Boolean enabled) { + MqttException syncException = null; + try { // 1. 更新设备启用状态 deviceDO.setIsEnable(enabled); @@ -2066,6 +2112,7 @@ public class DeviceServiceImpl implements DeviceService { mqttService.subscribeTopic(topic); log.info("MQTT订阅成功: {}", topic); } catch (MqttException e) { + syncException = e; log.error("MQTT订阅失败: {}", topic, e); } @@ -2091,6 +2138,8 @@ public class DeviceServiceImpl implements DeviceService { } else { //更新gateway状态 + gateway.setGatewayName(deviceDO.getDeviceName()); + gateway.setGatewayCode(deviceDO.getDeviceCode()); gateway.setIsEnable(true); gateway.setUpdateTime(LocalDateTime.now()); @@ -2109,17 +2158,14 @@ public class DeviceServiceImpl implements DeviceService { mqttService.unsubscribeTopic(topic); log.info("MQTT取消订阅成功: {}", topic); } catch (MqttException e) { + syncException = e; log.error("MQTT取消订阅失败: {}", topic, e); + } finally { + removeRuntimeTopic(topic); } // 3.2 更新gateway状态为禁用 - gatewayMapper.update( - null, - new LambdaUpdateWrapper() - .eq(GatewayDO::getTopic, topic) - .set(GatewayDO::getIsEnable, false) - .set(GatewayDO::getUpdateTime, LocalDateTime.now()) - ); + disableGatewayByDevice(deviceDO); log.info("gateway订阅记录已禁用 topic={}", topic); @@ -2130,9 +2176,14 @@ public class DeviceServiceImpl implements DeviceService { } + if (syncException != null) { + throw new RuntimeException("MQTT订阅状态同步失败", syncException); + } + } catch (Exception e) { log.error("更新设备状态失败 deviceCode={}", deviceDO.getDeviceCode(), e); + throw new RuntimeException("更新设备状态失败", e); } } diff --git a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/TDengineService.java b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/TDengineService.java index e2654f0b6..26b1972af 100644 --- a/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/TDengineService.java +++ b/yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/TDengineService.java @@ -779,6 +779,8 @@ public class TDengineService { continue; } + validateColumnName(attributeCode); + String tdType = JavaToTdengineTypeEnum.getTdTypeByJavaType(dataType); if (tdType == null) { tdType = "DOUBLE"; @@ -1292,7 +1294,7 @@ public class TDengineService { Set reservedWords = new HashSet<>(Arrays.asList( "value", "timestamp", "current", "database", "table", "user", "password", "select", "insert", "update", "delete", - "create", "drop", "alter", "show", "describe", "use", "ts" + "create", "drop", "alter", "show", "describe", "use", "ts", "state" )); return reservedWords.contains(columnName.toLowerCase()); @@ -1312,7 +1314,7 @@ public class TDengineService { Set reservedKeywords = new HashSet<>(Arrays.asList( "value", "timestamp", "current", "database", "table", "user", "password", "select", "insert", "update", "delete", "create", "drop", "alter", - "show", "describe", "use", "ts", "now", "current_timestamp" + "show", "describe", "use", "ts", "now", "current_timestamp", "state" )); if (reservedKeywords.contains(columnName.toLowerCase())) { @@ -1359,7 +1361,7 @@ public class TDengineService { Set reservedKeywords = new HashSet<>(Arrays.asList( "value", "timestamp", "current", "database", "table", "user", "password", "select", "insert", "update", "delete", "create", "drop", "alter", - "show", "describe", "use", "ts", "now", "current_timestamp" + "show", "describe", "use", "ts", "now", "current_timestamp", "state" )); if (reservedKeywords.contains(tableName.toLowerCase())) {