Merge branch 'mqy20260511' of http://114.215.146.132:3000/Mqy/Wisdom-Data into mqy20260511

 Conflicts:

	iot-platform/src/main/java/com/iot/platform/mqtt/MqttDynamicConsumer.java

	iot-platform/src/main/java/com/iot/platform/service/TdEngineService.java

iot-platform/src/main/java/com/iot/platform/mqtt/MqttChargeStationConsumer.java

@@ -79,6 +79,6 @@ public class MqttChargeStationConsumer extends AbstractDynamicMqttConsumer {
 //        superTable + "_" +
 //        int i = ((int) "g".getBytes("UTF-8")[0]) % 10;
 
-        tdengineService.insertBatch(dbNamePrefix, superTable,tableName, batchToInsert);
+        tdengineService.insertBatch(dbNamePrefix, superTable, tableName, batchToInsert);
     }
 }

iot-platform/src/main/java/com/iot/platform/mqtt/MqttDynamicConsumer.java

@@ -107,8 +107,8 @@ public class MqttDynamicConsumer extends AbstractDynamicMqttConsumer {
         List<Map<String, Object>> batch = Collections.singletonList(data);
 
         String dbNamePrefix = "pe_iot_"+ctx.dbName.substring(0, 2);
-        String tableName = ctx.dbName;
-        tdengineService.insertBatch(dbNamePrefix, ctx.superTable,tableName, batch);
+        String controllerId = ctx.dbName;
+        tdengineService.insertBatch(dbNamePrefix, ctx.superTable, controllerId, batch);
     }
 
     private void writeToRedis(MessageContext ctx) {

iot-platform/src/main/java/com/iot/platform/service/TdEngineService.java

@@ -8,6 +8,7 @@ import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.math.BigDecimal;
 import java.sql.*;
 import java.time.ZoneOffset;
 import java.util.*;
@@ -230,7 +231,7 @@ public class TdEngineService {
     // ==========================================
     // 初始化表结构（按列存储，无 ext_data）
     // ==========================================
-    private void initTableStructure(String dbName, String superTableName, String table)
+    private void initTableStructure(String dbName, String superTableName, String table, String controllerId)
             throws SQLException {
 
         try (Connection conn = getConnection(); Statement stmt = conn.createStatement()) {
@@ -238,9 +239,9 @@ public class TdEngineService {
 
             stmt.executeUpdate("CREATE DATABASE IF NOT EXISTS " + wrapName(dbName));
 
-            // 创建超级表：固定 ts + surfacename，无 ext_data 列
+            // 创建超级表：固定 ts + controller_id 列（TDengine 要求至少 2 列）
             String stableSql = String.format(
-                    "CREATE STABLE IF NOT EXISTS %s.%s (ts TIMESTAMP, surfacename VARCHAR(64)) TAGS (location VARCHAR(255))",
+                    "CREATE STABLE IF NOT EXISTS %s.%s (ts TIMESTAMP, controller_id BIGINT) TAGS (location VARCHAR(64))",
                     wrapName(dbName),
                     wrapName(superTableName)
             );
@@ -250,17 +251,17 @@ public class TdEngineService {
             String key = getStableKey(dbName, superTableName);
             Set<String> fixedCols = new HashSet<>();
             fixedCols.add("ts");
-            fixedCols.add("surfacename");
+            fixedCols.add("controller_id");
             stableColumnCache.put(key, fixedCols);
 
-            // 创建子表
+            // 创建子表，使用 controllerId 作为 tag
             String tableSql = String.format(
                     "CREATE TABLE IF NOT EXISTS %s.%s USING %s.%s TAGS ('%s')",
                     wrapName(dbName),
                     wrapName(table),
                     wrapName(dbName),
                     wrapName(superTableName),
-                    escapeValue(superTableName)
+                    escapeValue(controllerId)
             );
             stmt.executeUpdate(tableSql);
         }
@@ -269,26 +270,26 @@ public class TdEngineService {
     // ==========================================
     // 批量插入（按列存储）
     // ==========================================
-    public void insertBatch(String dbName, String superTable, String table, List<Map<String, Object>> dataList)
+    public void insertBatch(String dbName, String superTable, String controllerId, List<Map<String, Object>> dataList)
             throws SQLException {
 
         if (dataList == null || dataList.isEmpty()) {
-            log.debug("insertBatch 收到空数据，直接返回 | dbName={}, table={}", dbName, table);
+            log.debug("insertBatch 收到空数据，直接返回 | dbName={}, table={}", dbName, controllerId);
             return;
         }
 
         Map<String, String> columnTypes = collectColumnTypes(dataList);
 
         // 所有数据都走拆分逻辑：按首字符 UTF-8 值模 10 分组到不同的超级表
-        splitAndInsertToMultipleStables(dbName, superTable, table, dataList, columnTypes);
+        splitAndInsertToMultipleStables(dbName, superTable, controllerId, dataList, columnTypes);
 
-        log.info("批量写入完成: {} | 条数: {}", table, dataList.size());
+        log.info("批量写入完成: {} | 条数: {}", controllerId, dataList.size());
     }
 
     /**
      * 根据列 key 首字符的 UTF-8 值拆分到多个超级表
      */
-    private void splitAndInsertToMultipleStables(String dbName, String stableName, String tableName,
+    private void splitAndInsertToMultipleStables(String dbName, String stableName, String controllerId,
                                                  List<Map<String, Object>> dataList,
                                                  Map<String, String> columnTypes) throws SQLException {
         // 按首字符 UTF-8 值分组
@@ -305,7 +306,7 @@ public class TdEngineService {
             int groupId = group.getKey();
             // 超级表名 = superTableName + groupId (例如 charge + 3 = charge3)
             String newStableName = stableName+"_"+groupId;
-            String newTableName = stableName+"_"+groupId+"_"+tableName;
+            String newTableName = stableName+"_"+groupId+"_"+controllerId;
 
             // 筛选出该组列的数据
             List<Map<String, Object>> filteredData = filterDataByColumnGroup(dataList, group.getValue().keySet());
@@ -314,7 +315,7 @@ public class TdEngineService {
             int batchSize = DEFAULT_BATCH_SIZE;
             for (int i = 0; i < filteredData.size(); i += batchSize) {
                 List<Map<String, Object>> batch = filteredData.subList(i, Math.min(i + batchSize, filteredData.size()));
-                insertBatchInternal(dbName, newStableName, newTableName, batch);
+                insertBatchInternal(dbName, newStableName, newTableName, controllerId, batch);
             }
             log.info("分组插入完成: stable={}, table={}, 列数={}, 数据条数={}",
                     newStableName, newTableName, group.getValue().size(), filteredData.size());
@@ -381,8 +382,8 @@ public class TdEngineService {
      * 构建批量插入 SQL
      */
     private String buildInsertSql(String dbName, String table, String superTableName,
-                                   Map<String, String> columnTypes, List<Map<String, Object>> dataList) {
-        return buildInsertSql(dbName, table, superTableName, columnTypes, dataList, null);
+                                   Map<String, String> columnTypes, List<Map<String, Object>> dataList, String controllerId) {
+        return buildInsertSql(dbName, table, superTableName, columnTypes, dataList, null, controllerId);
     }
 
     /**
@@ -390,7 +391,7 @@ public class TdEngineService {
      */
     private String buildInsertSql(String dbName, String table, String superTableName,
                                    Map<String, String> columnTypes, List<Map<String, Object>> dataList,
-                                   Long customTs) {
+                                   Long customTs, String controllerId) {
         if (columnTypes.isEmpty()) {
             return null;
         }
@@ -403,7 +404,7 @@ public class TdEngineService {
 
         StringBuilder sql = new StringBuilder();
         sql.append("INSERT INTO ").append(wrapName(dbName)).append(".").append(wrapName(table))
-           .append(" (ts, surfacename");
+           .append(" (ts, controller_id");
 
         for (String col : columnTypes.keySet()) {
             sql.append(", ").append(wrapName(col));
@@ -416,7 +417,7 @@ public class TdEngineService {
                 continue;
             }
 
-            sql.append("(").append(tsValue).append(", '").append(escapeValue(superTableName)).append("'");
+            sql.append("(").append(tsValue).append(", '").append(escapeValue(controllerId)).append("'");
             for (Map.Entry<String, String> entry : columnTypes.entrySet()) {
                 String col = entry.getKey();
                 Object value = data.get(col);
@@ -439,15 +440,15 @@ public class TdEngineService {
     /**
      * 内部方法：插入一批数据（支持一次重试）
      */
-    private void insertBatchInternal(String dbName, String superTableName, String table,
+    private void insertBatchInternal(String dbName, String superTableName, String table,String controllerId,
                                      List<Map<String, Object>> dataList) throws SQLException {
-        insertBatchInternal(dbName, superTableName, table, dataList, false);
+        insertBatchInternal(dbName, superTableName, table, controllerId,dataList,false);
     }
 
-    private void insertBatchInternal(String dbName, String superTableName, String table,
+    private void insertBatchInternal(String dbName, String superTableName, String table,String controllerId,
                                      List<Map<String, Object>> dataList, boolean isRetry) throws SQLException {
 
-        ensureTableExists(dbName, superTableName, table);
+        ensureTableExists(dbName, superTableName, table,controllerId);
 
         Map<String, String> columnTypes = collectColumnTypes(dataList);
         if (columnTypes.isEmpty()) {
@@ -481,7 +482,7 @@ public class TdEngineService {
             return;
         }
 
-        String sql = buildInsertSql(dbName, table, superTableName, columnTypes, dataList);
+        String sql = buildInsertSql(dbName, table, superTableName, columnTypes, dataList, controllerId);
         if (sql == null) {
             log.warn("insertBatchInternal SQL 构建为空，跳过插入 | dbName={}, table={}", dbName, table);
             return;
@@ -507,7 +508,7 @@ public class TdEngineService {
                         subColumnTypes.put(col, columnTypes.get(col));
                     }
 
-                    String subSql = buildInsertSql(dbName, table, superTableName, subColumnTypes, dataList, unifiedTs);
+                    String subSql = buildInsertSql(dbName, table, superTableName, subColumnTypes, dataList, unifiedTs, controllerId);
                     if (subSql != null) {
                         stmt.executeUpdate(subSql);
                         insertedCount += dataList.size();
@@ -521,8 +522,8 @@ public class TdEngineService {
             if (!isRetry && e.getMessage().contains("Table does not exist")) {
                 log.warn("表不存在，重建并重试: {}", table);
                 clearStableColumnCache();
-                initTableStructure(dbName, superTableName, table);
-                insertBatchInternal(dbName, superTableName, table, dataList, true);
+                initTableStructure(dbName, superTableName, table, controllerId);
+                insertBatchInternal(dbName, superTableName, table, controllerId, dataList, true);
                 return;
             }
             throw new SQLException("批量写入 SQL 失败: " + table + " | 错误: " + e.getMessage(), e);
@@ -530,10 +531,10 @@ public class TdEngineService {
     }
 
     /**
-     * 判断是否为保留列（ts, surfacename）
+     * 判断是否为保留列（ts）
      */
     private boolean isReservedColumn(String columnName) {
-        return "ts".equalsIgnoreCase(columnName) || "surfacename".equalsIgnoreCase(columnName);
+        return "ts".equalsIgnoreCase(columnName) || "controller_id".equalsIgnoreCase(columnName);
     }
 
     /**
@@ -549,7 +550,7 @@ public class TdEngineService {
         if (value instanceof Integer || value instanceof Long) {
             return "BIGINT";
         }
-        if (value instanceof Float || value instanceof Double) {
+        if (value instanceof Float || value instanceof Double || value instanceof BigDecimal) {
             return "DOUBLE";
         }
         // 时间类型（Date, Timestamp, LocalDateTime 等）
@@ -688,7 +689,7 @@ public class TdEngineService {
     /**
      * 确保表存在
      */
-    private void ensureTableExists(String dbName, String superTableName, String table)
+    private void ensureTableExists(String dbName, String superTableName, String table,String controllerId)
             throws SQLException {
 
         if (!isValidTableName(dbName) || !isValidTableName(superTableName) || !isValidTableName(table)) {
@@ -706,7 +707,7 @@ public class TdEngineService {
                 try (ResultSet rs = pStmt.executeQuery()) {
                     if (!rs.next()) {
                         log.info("超级表不存在，创建: {}.{}", dbName, superTableName);
-                        initTableStructure(dbName, superTableName, table);
+                        initTableStructure(dbName, superTableName, table, controllerId);
                         return;
                     }
                 }

iot-platform/src/main/java/com/iot/platform/task/VehicleSyncTask.java

@@ -0,0 +1,149 @@
+package com.iot.platform.task;
+
+import com.iot.platform.domain.SysCar;
+import com.iot.platform.domain.SysDevice;
+import com.iot.platform.domain.SysDeviceControl;
+import com.iot.platform.config.IotProperties;
+import com.iot.platform.service.*;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.dao.DataAccessException;
+import org.springframework.data.redis.RedisConnectionFailureException;
+import org.springframework.data.redis.core.RedisCallback;
+import org.springframework.data.redis.core.ScanOptions;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestClientException;
+import org.springframework.web.client.RestTemplate;
+
+import java.util.*;
+import java.util.concurrent.TimeUnit;
+
+@Component
+public class VehicleSyncTask {
+
+    private static final Logger log = LoggerFactory.getLogger(VehicleSyncTask.class);
+
+    private final SysCarService sysCarService;
+    private final SysDeviceService sysDeviceService;
+    private final StringRedisTemplate stringRedisTemplate;
+    private final SysRealtimeService sysrealtimeService;
+
+    private final SysDeviceVoService sysDeviceVoService;
+    private final SysDeviceControlService sysDeviceControlService;
+    private final RestTemplate restTemplate;
+    private final IotProperties iotProperties;
+
+    public VehicleSyncTask(SysCarService sysCarService,
+                           SysDeviceService sysDeviceService,
+                           StringRedisTemplate stringRedisTemplate,
+                           SysRealtimeService sysrealtimeService,
+                           SysDeviceVoService sysDeviceVoService,
+                           SysDeviceControlService sysDeviceControlService,
+                           RestTemplate restTemplate,
+                           IotProperties iotProperties) {
+        this.sysCarService = sysCarService;
+        this.sysDeviceService = sysDeviceService;
+        this.stringRedisTemplate = stringRedisTemplate;
+        this.sysrealtimeService = sysrealtimeService;
+        this.sysDeviceVoService = sysDeviceVoService;
+        this.sysDeviceControlService = sysDeviceControlService;
+        this.restTemplate = restTemplate;
+        this.iotProperties = iotProperties;
+    }
+
+    private boolean tryLock(String lockKey, long expireSeconds) {
+        Boolean acquired = stringRedisTemplate.opsForValue().setIfAbsent(lockKey, "1", expireSeconds, TimeUnit.SECONDS);
+        return Boolean.TRUE.equals(acquired);
+    }
+
+    private void unlock(String lockKey) {
+        Boolean deleted = stringRedisTemplate.delete(lockKey);
+        if (!Boolean.TRUE.equals(deleted)) {
+            log.warn("分布式锁释放失败: {}", lockKey);
+        }
+    }
+
+    /**
+     * 更新数据库实时数据
+     */
+    @Scheduled(fixedDelay = 30000)
+    public void syncRedisToMySQL() {
+        String lockKey = "lock:vehicle-sync:syncRedisToMySQL";
+        if (!tryLock(lockKey, 60)) {
+            log.debug("获取锁失败，跳过本次执行: {}", lockKey);
+            return;
+        }
+        try {
+            doSyncRedisToMySQL();
+        } finally {
+            unlock(lockKey);
+        }
+    }
+
+    private void doSyncRedisToMySQL() {
+        Set<String> activeKeys = stringRedisTemplate.opsForSet().members("DSB:active:devices");
+        if (activeKeys == null || activeKeys.isEmpty()) return;
+
+        for (String redisKey : activeKeys) {
+            try {
+                Map<Object, Object> dataMap = stringRedisTemplate.opsForHash().entries(redisKey);
+                if (dataMap == null || dataMap.isEmpty()) {
+                    stringRedisTemplate.opsForSet().remove("DSB:active:devices", redisKey);
+                    continue;
+                }
+
+                String[] parts = redisKey.split(":", 3);
+                if (parts.length != 3 || !"DSB".equals(parts[0])) {
+                    log.warn("跳过非法 key: {}", redisKey);
+                    continue;
+                }
+                String controllerId = parts[1];
+
+                try {
+                    sysrealtimeService.createRealtime(controllerId);
+                } catch (Exception e) {
+                    log.error("创建表失败: {} | {}", controllerId, e.getMessage(), e);
+                    continue;
+                }
+
+                String createTime = getStringValue(dataMap, "createTime");
+                String timestamp = getStringValue(dataMap, "timestamp");
+                String deviceId = getStringValue(dataMap, "device_id");
+                if (createTime == null || timestamp == null || deviceId == null) {
+                    continue;
+                }
+
+                List<String> existingKeys = sysrealtimeService.selectAllKeys(controllerId);
+                Set<String> existingKeySet = existingKeys != null ? new HashSet<>(existingKeys) : Collections.emptySet();
+
+                for (Map.Entry<Object, Object> entry : dataMap.entrySet()) {
+                    String fieldKey = entry.getKey().toString();
+                    if ("createTime".equals(fieldKey) || "timestamp".equals(fieldKey) || "device_id".equals(fieldKey)) {
+                        continue;
+                    }
+                    String fieldValue = getStringValue(dataMap, fieldKey);
+                    if (fieldValue == null) continue;
+
+                    if (existingKeySet.contains(fieldKey)) {
+                        sysrealtimeService.updateTables(controllerId, createTime, fieldValue, timestamp, fieldKey, deviceId);
+                    } else {
+                        sysrealtimeService.insertTables(controllerId, createTime, deviceId, timestamp, fieldKey, fieldValue);
+                    }
+                }
+            } catch (RedisConnectionFailureException e) {
+                log.error("Redis 连接失败: {} | {}", redisKey, e.getMessage());
+            } catch (DataAccessException e) {
+                log.error("数据库操作失败: {} | {}", redisKey, e.getMessage(), e);
+            } catch (Exception e) {
+                log.error("同步设备失败: {} | {}", redisKey, e.getMessage(), e);
+            }
+        }
+    }
+
+    private String getStringValue(Map<Object, Object> map, String key) {
+        Object val = map.get(key);
+        return val == null ? null : val.toString().trim();
+    }
+}