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

 Conflicts:

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

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/MqttDynamicConsumer.java

@@ -12,7 +12,6 @@ import org.springframework.context.annotation.DependsOn;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
-import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
@@ -106,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 controllerId = ctx.dbName;
-        tdengineService.insertBatch(dbNamePrefix, ctx.superTable, controllerId, 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) {
@@ -126,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

@@ -8,6 +8,7 @@ import org.slf4j.LoggerFactory;
 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;
@@ -48,11 +49,11 @@ public class TdEngineService {
     // 每批次最大列数（防止 TDengine 行超限）
     private static final int MAX_COLUMNS_PER_INSERT = 100;
 
-    // 拆分超级表的列数阈值
-    private static final int SPLIT_STABLE_COLUMN_THRESHOLD = 100;
+    // 默认 VARCHAR 字段长度限制（用于未提供实际长度时的兜底）
+    private static final int DEFAULT_VARCHAR_LENGTH = 16;
 
-    // 默认 VARCHAR 字段长度限制
-    private static final int DEFAULT_VARCHAR_LENGTH = 128;
+    // TDengine 数据保留天数（超过此天数的数据自动删除）
+    private static final int DATA_RETENTION_DAYS = 180;
 
     // 东八区时区偏移（避免重复创建）
     private static final ZoneOffset ZONE_OFFSET_8 = ZoneOffset.of("+8");
@@ -237,7 +238,7 @@ public class TdEngineService {
         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 + controller_id 列（TDengine 要求至少 2 列）
             String stableSql = String.format(
@@ -292,17 +293,18 @@ public class TdEngineService {
     private void splitAndInsertToMultipleStables(String dbName, String stableName, String controllerId,
                                                  List<Map<String, Object>> dataList,
                                                  Map<String, String> columnTypes) throws SQLException {
+
         // 按首字符 UTF-8 值分组
-        Map<Integer, Map<String, String>> groupColumnTypes = new LinkedHashMap<>();
+        Map<Integer, Map<String, String>> groupColumnTypeMap = new LinkedHashMap<>();
         for (Map.Entry<String, String> entry : columnTypes.entrySet()) {
             int groupId = getFirstCharGroupId(entry.getKey());
-            groupColumnTypes.computeIfAbsent(groupId, k -> new LinkedHashMap<>()).put(entry.getKey(), entry.getValue());
+            groupColumnTypeMap.computeIfAbsent(groupId, k -> new LinkedHashMap<>()).put(entry.getKey(), entry.getValue());
         }
 
-        log.info("拆分后超级表数量: {} | 分组: {}", groupColumnTypes.size(), groupColumnTypes.keySet());
+        log.info("拆分后超级表数量: {} | 分组: {}", groupColumnTypeMap.size(), groupColumnTypeMap.keySet());
 
         // 每个分组作为一个超级表，处理其对应的列和数据
-        for (Map.Entry<Integer, Map<String, String>> group : groupColumnTypes.entrySet()) {
+        for (Map.Entry<Integer, Map<String, String>> group : groupColumnTypeMap.entrySet()) {
             int groupId = group.getKey();
             // 超级表名 = superTableName + groupId (例如 charge + 3 = charge3)
             String newStableName = stableName+"_"+groupId;
@@ -330,7 +332,7 @@ public class TdEngineService {
             return 0;
         }
         String firstChar = columnName.substring(0, 1);
-        byte[] bytes = firstChar.getBytes(java.nio.charset.StandardCharsets.UTF_8);
+        byte[] bytes = firstChar.getBytes(StandardCharsets.UTF_8);
         return (bytes[0] & 0xFF) % 10;
     }
 
@@ -340,7 +342,9 @@ public class TdEngineService {
     private List<Map<String, Object>> filterDataByColumnGroup(List<Map<String, Object>> dataList, Set<String> columns) {
         List<Map<String, Object>> filtered = new ArrayList<>();
         for (Map<String, Object> data : dataList) {
-            if (data == null) continue;
+            if (data == null) {
+                continue;
+            }
             Map<String, Object> filteredRow = new LinkedHashMap<>();
             for (String col : columns) {
                 if (data.containsKey(col)) {
@@ -352,11 +356,6 @@ public class TdEngineService {
         return filtered;
     }
 
-    private String extractSuperTableName(String table) {
-        int idx = table.lastIndexOf('_');
-        return idx > 0 ? table.substring(0, idx) : table;
-    }
-
     /**
      * 收集数据中所有动态列及其类型
      */
@@ -379,6 +378,31 @@ 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,
@@ -432,7 +456,7 @@ public class TdEngineService {
         if (!hasData) {
             return null;
         }
-        log.info("生成的 INSERT SQL | 列类型: {} | SQL 前100字符: {}", columnTypes, sql.toString().substring(0, Math.min(100, sql.toString().length())));
+        log.info("生成的 INSERT SQL | 列类型: {} | SQL 前100字符: {}", columnTypes, sql.substring(0, Math.min(100, sql.toString().length())));
         sql.setLength(sql.length() - 1);
         return sql.toString();
     }
@@ -456,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);
@@ -473,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) {
@@ -564,7 +590,7 @@ public class TdEngineService {
     /**
      * 根据 TdEngine 列类型获取创建列的 SQL 类型
      */
-    private String getColumnTypeForDDL(String tdType, String columnName) {
+    private String getColumnTypeForDDL(String tdType, Integer maxLen) {
         switch (tdType) {
             case "BOOL":
             case "BIGINT":
@@ -573,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 + ")";
         }
     }
 
@@ -617,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);
 
@@ -646,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) {
@@ -655,7 +679,7 @@ public class TdEngineService {
                 }
 
                 if (!existingColumns.contains(col)) {
-                    String ddlType = getColumnTypeForDDL(colType, col);
+                    String ddlType = getColumnTypeForDDL(colType, maxLen);
                     String alterSql = String.format(
                             "ALTER STABLE %s.%s ADD COLUMN %s %s",
                             wrapName(dbName),
@@ -752,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

@@ -25,32 +25,14 @@ 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;
+    public VehicleSyncTask(StringRedisTemplate stringRedisTemplate,
+                           SysRealtimeService sysrealtimeService) {
+
         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) {
@@ -83,53 +65,57 @@ public class VehicleSyncTask {
     }
 
     private void doSyncRedisToMySQL() {
-        Set<String> activeKeys = stringRedisTemplate.opsForSet().members("DSB:active:devices");
+        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("DSB:active:devices", redisKey);
+                    stringRedisTemplate.opsForSet().remove("iot:dsb:active:devices", redisKey);
                     continue;
                 }
 
-                String[] parts = redisKey.split(":", 3);
-                if (parts.length != 3 || !"DSB".equals(parts[0])) {
-                    log.warn("跳过非法 key: {}", redisKey);
+                // redisKey 格式: pe_iot_<controllerId>_<deviceId>
+                // 从右往左取，避免 id 本身含下划线的问题
+                String[] parts = redisKey.split("_");
+                if (parts.length < 3) {
+                    log.warn("跳过非法 redis key: {}", redisKey);
                     continue;
                 }
-                String controllerId = parts[1];
+                String controllerId = parts[parts.length - 2];
+                String deviceId = parts[parts.length - 1];
+                String tableName = "pe_iot_" + controllerId.substring(0, 2);
 
                 try {
-                    sysrealtimeService.createRealtime(controllerId);
+                    sysrealtimeService.createRealtime(tableName);
                 } catch (Exception e) {
-                    log.error("创建表失败: {} | {}", controllerId, e.getMessage(), e);
+                    log.error("创建表失败: {} | {}", tableName, 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) {
+
+                if (createTime == null || timestamp == null) {
                     continue;
                 }
 
-                List<String> existingKeys = sysrealtimeService.selectAllKeys(controllerId);
+                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)) {
+                    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(controllerId, createTime, fieldValue, timestamp, fieldKey, deviceId);
+                        sysrealtimeService.updateTables(tableName, createTime, fieldValue, timestamp, fieldKey, controllerId, deviceId);
                     } else {
-                        sysrealtimeService.insertTables(controllerId, createTime, deviceId, timestamp, fieldKey, fieldValue);
+                        sysrealtimeService.insertTables(tableName, createTime, controllerId, deviceId, timestamp, fieldKey, fieldValue);
                     }
                 }
             } catch (RedisConnectionFailureException e) {

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>