mqttx插入优化

yudao-module-iot/yudao-module-iot-biz/src/main/java/cn/iocoder/yudao/module/iot/service/device/TDengineService.java

@@ -892,6 +892,8 @@ public class TDengineService {
         // TDengine 必须字段
         params.add(new Timestamp(System.currentTimeMillis()));
 
+        Set<String> insertedColumns = new HashSet<>();
+
         for (DeviceContactModelDO model : dataList) {
 
             if (model == null) {
@@ -917,6 +919,16 @@ public class TDengineService {
                 continue;
             }
 
+            if (!insertedColumns.add(column)) {
+                log.warn("Duplicate TDengine column in one insert, skip: deviceId={}, column={}", deviceId, column);
+                continue;
+            }
+
+            if (!ensureTDColumnExists(deviceId, model)) {
+                log.warn("TDengine column missing and auto add failed, skip: deviceId={}, column={}", deviceId, column);
+                continue;
+            }
+
             try {
 
                 // 2. 自动新增列
@@ -992,6 +1004,25 @@ public class TDengineService {
      * @param oldColumnName 原列名
      * @param newColumnName 新列名
      */
+    private boolean ensureTDColumnExists(Long deviceId, DeviceContactModelDO model) {
+        String column = model.getAttributeCode();
+        if (columnExists(deviceId, column)) {
+            return true;
+        }
+
+        try {
+            AddTDDatabaseColumn(deviceId, model);
+            return true;
+        } catch (Exception e) {
+            if (columnExists(deviceId, column)) {
+                return true;
+            }
+            log.warn("Auto add TDengine column failed: deviceId={}, column={}, reason={}",
+                    deviceId, column, e.getMessage());
+            return false;
+        }
+    }
+
     @DS("tdengine")
     public void renameTDColumn(Long deviceId, String oldColumnName, String newColumnName) {
         if (deviceId == null || StrUtil.isBlank(oldColumnName) || StrUtil.isBlank(newColumnName)) {