Merge remote-tracking branch 'origin/mqy20260511'

iot-platform/src/main/java/com/iot/platform/controller/IotController.java

@@ -0,0 +1,55 @@
+package com.iot.platform.controller;
+
+import com.iot.platform.common.AjaxResult;
+import com.iot.platform.service.TdEngineService;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.data.redis.core.StringRedisTemplate;
+import org.springframework.web.bind.annotation.*;
+
+import java.util.Map;
+
+@RestController
+@RequestMapping("/iot")
+public class IotController {
+
+    @Autowired
+    private StringRedisTemplate stringRedisTemplate;
+
+    @Autowired
+    private TdEngineService tdEngineService;
+
+    /**
+     * 根据 controllerId、deviceId 和 key 查询 Redis 中存储的数据
+     */
+    @GetMapping("/redis/query")
+    public AjaxResult queryRedisData(@RequestParam String controllerId,
+                                     @RequestParam String deviceId,
+                                     @RequestParam(required = false) String key) {
+        String redisKey = "pe_iot_" + controllerId + "_" + deviceId;
+
+        if (key == null || key.trim().isEmpty()) {
+            // 查询整个 hash
+            Map<Object, Object> data = stringRedisTemplate.opsForHash().entries(redisKey);
+            return AjaxResult.success(data);
+        } else {
+            // 查询指定 field
+            Object value = stringRedisTemplate.opsForHash().get(redisKey, key);
+            return AjaxResult.success(value);
+        }
+    }
+
+    /**
+     * 根据 controllerId、deviceId 和 key 查询 TDengine 中存储的数据
+     */
+    @GetMapping("/tdengine/query")
+    public AjaxResult queryTdEngineData(@RequestParam String controllerId,
+                                        @RequestParam String deviceId,
+                                        @RequestParam(required = false) String key) {
+        try {
+            Object result = tdEngineService.queryLatestData(controllerId, deviceId, key);
+            return AjaxResult.success(result);
+        } catch (Exception e) {
+            return AjaxResult.error("查询TDengine失败: " + e.getMessage());
+        }
+    }
+}

iot-platform/src/main/java/com/iot/platform/mapper/SysRealtimeMapper.java

@@ -9,8 +9,8 @@ import java.util.List;
 public interface SysRealtimeMapper {
     void createRealtime(@Param("tableName")String tableName);
     List<String> selectTables();
-    void insertTables(@Param("tableName")String tableName,@Param("createTime")String createTime,@Param("deviceId")String deviceId,@Param("timestamp")String timestamp,@Param("k")String k,@Param("v")Object v);
-    void updateTables(@Param("tableName")String tableName,@Param("createTime")String createTime,@Param("v")Object v,@Param("timestamp")String timestamp,@Param("k")String k,@Param("deviceId")String deviceId);
-    Integer selectKey(@Param("tableName")String tableName,@Param("k")String k);
-    List<String> selectAllKeys(@Param("tableName")String tableName);
+    void insertTables(@Param("tableName")String tableName, @Param("createTime")String createTime, @Param("controllerId")String controllerId, @Param("deviceId")String deviceId, @Param("timestamp")String timestamp, @Param("k")String k, @Param("v")Object v);
+    void updateTables(@Param("tableName")String tableName, @Param("createTime")String createTime, @Param("v")Object v, @Param("timestamp")String timestamp, @Param("k")String k, @Param("controllerId")String controllerId, @Param("deviceId")String deviceId);
+    Integer selectKey(@Param("tableName")String tableName, @Param("k")String k, @Param("controllerId")String controllerId);
+    List<String> selectAllKeys(@Param("tableName")String tableName, @Param("controllerId")String controllerId);
 }

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

@@ -105,18 +105,16 @@ public class MqttDynamicConsumer extends AbstractDynamicMqttConsumer {
         data.put("device_id", ctx.deviceId);
         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 dbNamePrefix = "pe_iot_"+ctx.controllerId.substring(0, 2);
+        String controllerId = ctx.controllerId;
+        tdengineService.insertBatch(dbNamePrefix, ctx.deviceId, controllerId, batch);
     }
 
     private void writeToRedis(MessageContext ctx) {
-        String redisKey = "DSB:" + ctx.dbName + ":" + ctx.controllerName;
+        String redisKey = "pe_iot_" + ctx.controllerId + "_" + ctx.deviceId;
 
         Map<String, String> hashData = new HashMap<>();
-        hashData.put("createTime", ctx.currentTime);
         hashData.put("timestamp", ctx.timestamp);
-        hashData.put("device_id", ctx.deviceId);
 
         for (Map.Entry<String, Object> entry : ctx.metricData.entrySet()) {
             if (entry.getValue() != null) {
@@ -125,25 +123,22 @@ public class MqttDynamicConsumer extends AbstractDynamicMqttConsumer {
         }
 
         stringRedisTemplate.opsForHash().putAll(redisKey, hashData);
-        stringRedisTemplate.expire(redisKey, 2, TimeUnit.HOURS);
-        stringRedisTemplate.opsForSet().add("DSB:active:devices", redisKey);
-        stringRedisTemplate.expire("DSB:active:devices", 2, TimeUnit.HOURS);
+        stringRedisTemplate.expire(redisKey, 24, TimeUnit.HOURS);
+        stringRedisTemplate.opsForSet().add("iot:dsb:active:devices", redisKey);
     }
 
     private static class MessageContext {
-        final String dbName;
-        final String superTable;
+        final String controllerId;
+        final String deviceId;
         final String controllerName;
         final JSONObject metricData;
         final String currentTime;
         final String timestamp;
-        final String deviceId;
 
-        MessageContext(String dbName, String superTable, String controllerName,
+        MessageContext(String controllerId, String devcieId, String controllerName,
                        JSONObject metricData, String currentTime,
                        String timestamp, String deviceId) {
-            this.dbName = dbName;
-            this.superTable = superTable;
+            this.controllerId = controllerId;
             this.controllerName = controllerName;
             this.metricData = metricData;
             this.currentTime = currentTime;

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

@@ -30,24 +30,24 @@ public class SysRealtimeService {
         return sysRealtimeMapper.selectTables();
     }
 
-    public void insertTables(String tableName, String createTime, String deviceId, String timestamp, String k, Object v) {
+    public void insertTables(String tableName, String createTime, String controllerId, String deviceId, String timestamp, String k, Object v) {
         validateTableName(tableName);
-        sysRealtimeMapper.insertTables(tableName, createTime, deviceId, timestamp, k, v);
+        sysRealtimeMapper.insertTables(tableName, createTime, controllerId, deviceId, timestamp, k, v);
     }
 
-    public void updateTables(String tableName, String createTime, Object v, String timestamp, String k, String deviceId) {
+    public void updateTables(String tableName, String createTime, Object v, String timestamp, String k, String controllerId, String deviceId) {
         validateTableName(tableName);
-        sysRealtimeMapper.updateTables(tableName, createTime, v, timestamp, k, deviceId);
+        sysRealtimeMapper.updateTables(tableName, createTime, v, timestamp, k, controllerId, deviceId);
     }
 
-    public Integer selectKey(String tableName, String k) {
+    public Integer selectKey(String tableName, String k, String controllerId) {
         validateTableName(tableName);
-        return sysRealtimeMapper.selectKey(tableName, k);
+        return sysRealtimeMapper.selectKey(tableName, k, controllerId);
     }
 
-    public List<String> selectAllKeys(String tableName) {
+    public List<String> selectAllKeys(String tableName, String controllerId) {
         validateTableName(tableName);
-        return sysRealtimeMapper.selectAllKeys(tableName);
+        return sysRealtimeMapper.selectAllKeys(tableName, controllerId);
     }
 
     private void validateTableName(String tableName) {

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

@@ -9,6 +9,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import java.nio.charset.StandardCharsets;
+import java.math.BigDecimal;
 import java.sql.*;
 import java.time.ZoneOffset;
 import java.util.*;
@@ -48,8 +49,11 @@ public class TdEngineService {
     // 每批次最大列数（防止 TDengine 行超限）
     private static final int MAX_COLUMNS_PER_INSERT = 100;
 
-    // 默认 VARCHAR 字段长度限制
-    private static final int DEFAULT_VARCHAR_LENGTH = 128;
+    // 默认 VARCHAR 字段长度限制（用于未提供实际长度时的兜底）
+    private static final int DEFAULT_VARCHAR_LENGTH = 16;
+
+    // TDengine 数据保留天数（超过此天数的数据自动删除）
+    private static final int DATA_RETENTION_DAYS = 180;
 
     // 东八区时区偏移（避免重复创建）
     private static final ZoneOffset ZONE_OFFSET_8 = ZoneOffset.of("+8");
@@ -126,7 +130,7 @@ public class TdEngineService {
         if (name == null || name.isEmpty()) {
             return "`unknown`";
         }
-        return "`" + name.replace("`", "") + "`";
+        return "`" + name.replaceAll("`", "") + "`";
     }
 
     private boolean isValidFieldName(String name) {
@@ -228,17 +232,17 @@ 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()) {
             stmt.setQueryTimeout(10);
 
-            stmt.executeUpdate("CREATE DATABASE IF NOT EXISTS " + wrapName(dbName));
+            stmt.executeUpdate("CREATE DATABASE IF NOT EXISTS " + wrapName(dbName) + " KEEP " + DATA_RETENTION_DAYS);
 
-            // 创建超级表：固定 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)
             );
@@ -248,17 +252,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);
         }
@@ -267,26 +271,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 {
 
@@ -304,7 +308,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());
@@ -313,7 +317,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());
@@ -374,11 +378,36 @@ public class TdEngineService {
     }
 
     /**
+     * 收集数据中所有动态列的最大字符串长度（用于动态 VARCHAR 长度）
+     */
+    private Map<String, Integer> collectColumnMaxLengths(List<Map<String, Object>> dataList) {
+        Map<String, Integer> maxLengths = new HashMap<>();
+        for (Map<String, Object> data : dataList) {
+            if (data == null) {
+                continue;
+            }
+            for (Map.Entry<String, Object> entry : data.entrySet()) {
+                String key = entry.getKey();
+                if (!isValidFieldName(key) || isReservedColumn(key)) {
+                    continue;
+                }
+                Object value = entry.getValue();
+                if (value != null) {
+                    int len = value.toString().length();
+                    maxLengths.merge(key, len, Math::max);
+                }
+            }
+        }
+        log.debug("收集到的列最大长度: {}", maxLengths);
+        return maxLengths;
+    }
+
+    /**
      * 构建批量插入 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);
     }
 
     /**
@@ -386,7 +415,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;
         }
@@ -399,7 +428,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));
@@ -412,7 +441,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);
@@ -435,15 +464,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()) {
@@ -451,6 +480,8 @@ public class TdEngineService {
             return;
         }
 
+        Map<String, Integer> columnMaxLengths = collectColumnMaxLengths(dataList);
+
         log.info("收集到的列类型: {}", columnTypes);
         Set<String> existingColumns = getStableColumns(dbName, superTableName);
         log.info("超级表已有列: {}", existingColumns);
@@ -468,7 +499,7 @@ public class TdEngineService {
             return;
         }
 
-        ensureColumnsExist(dbName, superTableName, columnTypes);
+        ensureColumnsExist(dbName, superTableName, columnTypes, columnMaxLengths);
 
         existingColumns = getStableColumns(dbName, superTableName);
         if (existingColumns.size() >= MAX_COLUMNS_PER_STABLE) {
@@ -477,7 +508,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;
@@ -503,7 +534,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();
@@ -517,8 +548,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);
@@ -526,10 +557,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);
     }
 
     /**
@@ -545,7 +576,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 等）
@@ -559,7 +590,7 @@ public class TdEngineService {
     /**
      * 根据 TdEngine 列类型获取创建列的 SQL 类型
      */
-    private String getColumnTypeForDDL(String tdType) {
+    private String getColumnTypeForDDL(String tdType, Integer maxLen) {
         switch (tdType) {
             case "BOOL":
             case "BIGINT":
@@ -568,7 +599,8 @@ public class TdEngineService {
                 return tdType;
             case "VARCHAR":
             default:
-                return "VARCHAR(" + DEFAULT_VARCHAR_LENGTH + ")";
+                int len = (maxLen != null && maxLen > 0) ? maxLen + 5 : DEFAULT_VARCHAR_LENGTH;
+                return "VARCHAR(" + len + ")";
         }
     }
 
@@ -612,20 +644,16 @@ public class TdEngineService {
             }
         }
 
-        // 字符串类型校验长度
-        if (strValue.length() > DEFAULT_VARCHAR_LENGTH) {
-            log.debug("字段值超长，截断存储 | 列: {} | 值长度: {} | 最大: {} | 截断后: {}...",
-                    columnName, strValue.length(), DEFAULT_VARCHAR_LENGTH, strValue.substring(0, DEFAULT_VARCHAR_LENGTH));
-            strValue = strValue.substring(0, DEFAULT_VARCHAR_LENGTH);
-        }
+        // 字符串类型：使用实际值长度 + 5 冗余，不截断
+        // （TDengine VARCHAR 实际限制在创建列时已保证充足）
         return "'" + escapeValue(strValue) + "'";
     }
 
     /**
      * 确保列存在，如有新列则 ALTER 添加（根据类型添加对应列）
      */
-    private void ensureColumnsExist(String dbName, String superTableName, Map<String, String> columnTypes)
-            throws SQLException {
+    private void ensureColumnsExist(String dbName, String superTableName, Map<String, String> columnTypes,
+                                     Map<String, Integer> columnMaxLengths) throws SQLException {
 
         Set<String> existingColumns = getStableColumns(dbName, superTableName);
 
@@ -641,6 +669,7 @@ public class TdEngineService {
             for (Map.Entry<String, String> entry : columnTypes.entrySet()) {
                 String col = entry.getKey();
                 String colType = entry.getValue();
+                Integer maxLen = columnMaxLengths != null ? columnMaxLengths.get(col) : null;
 
                 // 再次检查列数上限
                 if (existingColumns.size() >= MAX_COLUMNS_PER_STABLE) {
@@ -650,7 +679,7 @@ public class TdEngineService {
                 }
 
                 if (!existingColumns.contains(col)) {
-                    String ddlType = getColumnTypeForDDL(colType);
+                    String ddlType = getColumnTypeForDDL(colType, maxLen);
                     String alterSql = String.format(
                             "ALTER STABLE %s.%s ADD COLUMN %s %s",
                             wrapName(dbName),
@@ -684,7 +713,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)) {
@@ -702,7 +731,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;
                     }
                 }
@@ -747,6 +776,96 @@ public class TdEngineService {
         log.info("清除了 TdEngine 超级表结构缓存");
     }
 
+    /**
+     * 根据 controllerId、deviceId 查询 TDengine 最新数据
+     * @param controllerId 控制器ID
+     * @param deviceId 设备ID
+     * @param key 列名，不传则返回所有列
+     */
+    public Object queryLatestData(String controllerId, String deviceId, String key) throws SQLException {
+        // 构建 dbName 前缀
+        String dbNamePrefix = "pe_iot_" + controllerId.substring(0, 2);
+
+        // 遍历 10 个分组查找数据
+        for (int groupId = 0; groupId < 10; groupId++) {
+            // 表名格式: {deviceId}_{groupId}_{controllerId} (与 insertBatch 保持一致)
+            String tableName = deviceId + "_" + groupId + "_" + controllerId;
+
+            // 检查表是否存在
+            if (!isTableExists(dbNamePrefix, tableName)) {
+                continue;
+            }
+
+            // 构建查询 SQL
+            String colPart = (key != null && !key.trim().isEmpty()) ? wrapName(key) : "*";
+            String sql = String.format("SELECT %s FROM %s.%s ORDER BY ts DESC LIMIT 1",
+                    colPart, wrapName(dbNamePrefix), wrapName(tableName));
+
+            try (Connection conn = getConnection();
+                 Statement stmt = conn.createStatement();
+                 ResultSet rs = stmt.executeQuery(sql)) {
+
+                if (rs.next()) {
+                    if (key != null && !key.trim().isEmpty()) {
+                        // 返回指定列的值
+                        return rs.getObject(1);
+                    } else {
+                        // 返回整行数据
+                        java.util.Map<String, Object> row = new java.util.LinkedHashMap<>();
+                        ResultSetMetaData metaData = rs.getMetaData();
+                        int columnCount = metaData.getColumnCount();
+                        for (int i = 1; i <= columnCount; i++) {
+                            row.put(metaData.getColumnLabel(i), rs.getObject(i));
+                        }
+                        return row;
+                    }
+                }
+            } catch (SQLException e) {
+                // 列不存在，继续查找下一个分组
+                if (e.getMessage().contains("Invalid column")) {
+                    log.debug("表 {} 中无列 {}，继续查找其他分组", tableName, key);
+                    continue;
+                }
+                throw e;
+            }
+        }
+        return null;
+    }
+
+    /**
+     * 检查表是否存在
+     */
+    private boolean isTableExists(String dbName, String tableName) throws SQLException {
+        // 先检查数据库是否存在
+        String checkDbSql = "SHOW DATABASES";
+        try (Connection conn = getConnection();
+             Statement stmt = conn.createStatement()) {
+            stmt.setQueryTimeout(5);
+            boolean dbExists = false;
+            try (ResultSet rs = stmt.executeQuery(checkDbSql)) {
+                while (rs.next()) {
+                    if (dbName.equals(rs.getString(1))) {
+                        dbExists = true;
+                        break;
+                    }
+                }
+            }
+            if (!dbExists) {
+                return false;
+            }
+            // 检查表是否存在
+            String sql = String.format("SHOW %s.TABLES", dbName);
+            try (ResultSet rs = stmt.executeQuery(sql)) {
+                while (rs.next()) {
+                    if (tableName.equals(rs.getString(1))) {
+                        return true;
+                    }
+                }
+            }
+            return false;
+        }
+    }
+
     public void close() {
         log.info("关闭 TdEngine 服务...");
         if (dataSource != null) {

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

@@ -0,0 +1,135 @@
+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 StringRedisTemplate stringRedisTemplate;
+    private final SysRealtimeService sysrealtimeService;
+
+    public VehicleSyncTask(StringRedisTemplate stringRedisTemplate,
+                           SysRealtimeService sysrealtimeService) {
+
+        this.stringRedisTemplate = stringRedisTemplate;
+        this.sysrealtimeService = sysrealtimeService;
+    }
+
+    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("iot: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("iot:dsb:active:devices", redisKey);
+                    continue;
+                }
+
+                // redisKey 格式: pe_iot_<controllerId>_<deviceId>
+                // 从右往左取，避免 id 本身含下划线的问题
+                String[] parts = redisKey.split("_");
+                if (parts.length < 3) {
+                    log.warn("跳过非法 redis key: {}", redisKey);
+                    continue;
+                }
+                String controllerId = parts[parts.length - 2];
+                String deviceId = parts[parts.length - 1];
+                String tableName = "pe_iot_" + controllerId.substring(0, 2);
+
+                try {
+                    sysrealtimeService.createRealtime(tableName);
+                } catch (Exception e) {
+                    log.error("创建表失败: {} | {}", tableName, e.getMessage(), e);
+                    continue;
+                }
+
+                String createTime = getStringValue(dataMap, "createTime");
+                String timestamp = getStringValue(dataMap, "timestamp");
+
+                if (createTime == null || timestamp == null) {
+                    continue;
+                }
+
+                List<String> existingKeys = sysrealtimeService.selectAllKeys(tableName, 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) || "controller_id".equals(fieldKey)) {
+                        continue;
+                    }
+                    String fieldValue = getStringValue(dataMap, fieldKey);
+                    if (fieldValue == null) continue;
+
+                    if (existingKeySet.contains(fieldKey)) {
+                        sysrealtimeService.updateTables(tableName, createTime, fieldValue, timestamp, fieldKey, controllerId, deviceId);
+                    } else {
+                        sysrealtimeService.insertTables(tableName, createTime, controllerId, 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();
+    }
+}

iot-platform/src/main/resources/mapper/SysRealtimeMapper.xml

@@ -14,10 +14,14 @@
     <update id="createRealtime">
         CREATE TABLE IF NOT EXISTS `${tableName}` (
             create_time VARCHAR(255) NOT NULL COMMENT '时间戳',
+            controller_id VARCHAR(255) NOT NULL COMMENT '控制器id',
             device_id VARCHAR(255) NOT NULL COMMENT '设备id',
             timestamp VARCHAR(255) NOT NULL COMMENT '时间戳',
             k VARCHAR(255) NOT NULL COMMENT 'key',
             v VARCHAR(255) NOT NULL COMMENT '值',
+            INDEX idx_controller_id (controller_id),
+            INDEX idx_controller_device (controller_id, device_id),
+            INDEX idx_controller_k (controller_id, k),
             INDEX idx_device_id (device_id),
             INDEX idx_device_create_time (device_id, create_time),
             INDEX idx_k (k)
@@ -28,17 +32,17 @@
         SHOW TABLES;
     </select>
     <insert id="insertTables">
-        insert into `${tableName}`(create_time,device_id,timestamp,k,v) values(#{createTime},#{deviceId},#{timestamp},#{k},#{v})
+        insert into `${tableName}`(create_time,controller_id,device_id,timestamp,k,v) values(#{createTime},#{controllerId},#{deviceId},#{timestamp},#{k},#{v})
     </insert>
     <update id="updateTables">
-        update `${tableName}` set create_time=#{createTime},v=#{v},timestamp=#{timestamp} where k=#{k} and device_id=#{deviceId}
+        update `${tableName}` set create_time=#{createTime},v=#{v},timestamp=#{timestamp} where k=#{k} and device_id=#{deviceId} and controller_id=#{controllerId}
     </update>
 
     <select id="selectKey" resultType="Integer">
-        select COUNT(*) from `${tableName}` where k=#{k}
+        select COUNT(*) from `${tableName}` where k=#{k} and controller_id=#{controllerId}
     </select>
 
     <select id="selectAllKeys" resultType="String">
-        SELECT k FROM `${tableName}`
+        SELECT k FROM `${tableName}` WHERE controller_id=#{controllerId}
     </select>
 </mapper>