修改问题：

1.时间改timestamp类型
2.pe_iot 改 pe_ems
3.改hash方法
4.修改tdeing时间格式

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

@@ -9,7 +9,6 @@ public interface SysFaultMapper {
     void insertAlertData(@Param("faultId") String faultId,
                      @Param("faultdescs") String faultdescs,
                      @Param("faultstatus") String faultstatus,
-                     @Param("createtime") String createtime,
                      @Param("messageType") String messageType,
                      @Param("controllerId") String controllerId,
                      @Param("deviceId") String deviceId,
@@ -21,7 +20,6 @@ public interface SysFaultMapper {
                      @Param("faultdescs")String faultdescs,
                      @Param("controllerId")String controllerId,
                      @Param("deviceId")String deviceId,
-                     @Param("createtime")String createtime,
                      @Param("timestamp")String timestamp);
 
 

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

@@ -10,8 +10,8 @@ public interface SysRealtimeMapper {
     void createRealtime(@Param("tableName")String tableName);
     int tableExists(@Param("tableName")String tableName);
     List<String> selectTables();
-    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);
+    void insertTables(@Param("tableName")String tableName, @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("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/mapper/SysStatusMapper.java

@@ -6,8 +6,8 @@ import org.apache.ibatis.annotations.Param;
 @Mapper
 public interface SysStatusMapper {
 
-    void insertSysStatus(@Param("controllerId")String controllerId,@Param("fleetId")String fleetId,@Param("status")String status,@Param("createTime")String createTime);
-    void updateStatus(@Param("controllerId")String controllerId,@Param("fleetId")String fleetId,@Param("status")String status,@Param("createTime")String createTime);
+    void insertSysStatus(@Param("controllerId")String controllerId,@Param("fleetId")String fleetId,@Param("status")String status);
+    void updateStatus(@Param("controllerId")String controllerId,@Param("fleetId")String fleetId,@Param("status")String status);
 
     Integer selectStatusCount(@Param("controllerId")String controllerId);
 }

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

@@ -71,15 +71,10 @@ public class MqttChargeStationConsumer extends AbstractDynamicMqttConsumer {
 
         String dbName = topicParts[1];
         String dbPrefix = dbName.length() >= 2 ? dbName.substring(0, 2) : dbName;
-        String dbNamePrefix = "pe_iot_" + dbPrefix;
+        String dbNamePrefix = "pe_ems_" + dbPrefix;
         //先传输为deviceid，到后面进行优化
         String superTable = topicParts[3];
-//        String tableName = superTable + "_" + date.getYear() + String.format("%02d", date.getMonthValue());
-                           //deviceid_controllerid
         String tableName = dbName;
-//        superTable + "_" +
-//        int i = ((int) "g".getBytes("UTF-8")[0]) % 10;
-
         tdengineService.insertBatch(dbNamePrefix, superTable, tableName, batchToInsert);
     }
 }

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

@@ -106,7 +106,7 @@ public class MqttDynamicConsumer extends AbstractDynamicMqttConsumer {
         List<Map<String, Object>> batch = Collections.singletonList(data);
 
         String controllerPrefix = ctx.controllerId.length() >= 2 ? ctx.controllerId.substring(0, 2) : ctx.controllerId;
-        String dbNamePrefix = "pe_iot_" + controllerPrefix;
+        String dbNamePrefix = "pe_ems_" + controllerPrefix;
         String controllerId = ctx.controllerId;
         tdengineService.insertBatch(dbNamePrefix, ctx.deviceId, controllerId, batch);
     }
@@ -115,7 +115,6 @@ public class MqttDynamicConsumer extends AbstractDynamicMqttConsumer {
         String redisKey = "pe_iot_" + ctx.controllerId + "_" + ctx.deviceId;
 
         Map<String, String> hashData = new HashMap<>();
-        hashData.put("createTime", ctx.currentTime);
         hashData.put("timestamp", ctx.timestamp);
 
         for (Map.Entry<String, Object> entry : ctx.metricData.entrySet()) {

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

@@ -15,7 +15,6 @@ import org.springframework.web.util.UriComponentsBuilder;
 import java.sql.SQLException;
 import java.time.Instant;
 import java.time.LocalDate;
-import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
@@ -181,16 +180,12 @@ public class MqttFaultConsumer extends AbstractMqttConsumer {
         String timestamp = faultData.getTimestamp();
         String dateTime = toDateTime(timestamp);
         String companyId = COMPANY_ID_PREFIX + numericIdGenerator.nextId();
-        LocalDateTime currentTime = LocalDateTime.now();
-        DateTimeFormatter formatter = DateTimeFormatter.ofPattern(DATE_TIME_PATTERN);
-        String currentTimeStr = currentTime.format(formatter);
-
         if ("触发".equals(type)) {
 //            sysAlarmService.insertAlarm(controllerName, companyId, desc, ALARM_STATUS_TRIGGER, currentTimeStr, ALARM_STATUS_TRIGGER, controllerId, deviceId);
-            sysFaultService.insertAlertData(companyId, desc, ALARM_STATUS_TRIGGER, currentTimeStr, ALARM_STATUS_TRIGGER, controllerId, deviceId, "", dateTime);
+            sysFaultService.insertAlertData(companyId, desc, ALARM_STATUS_TRIGGER, ALARM_STATUS_TRIGGER, controllerId, deviceId, "", dateTime);
         } else if ("恢复".equals(type)) {
 //            sysAlarmService.insertAlarm(controllerName, companyId, desc, ALARM_STATUS_RECOVERED, currentTimeStr, ALARM_STATUS_TRIGGER, controllerId, deviceId);
-            sysFaultService.updateAlertData(ALARM_STATUS_RECOVERED, ALARM_STATUS_TRIGGER, desc, controllerId, deviceId, currentTimeStr, dateTime);
+            sysFaultService.updateAlertData(ALARM_STATUS_RECOVERED, ALARM_STATUS_TRIGGER, desc, controllerId, deviceId, dateTime);
         }
 
         // controllerId 白名单校验，防止 SSRF

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

@@ -8,9 +8,6 @@ import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
-import java.time.Instant;
-import java.time.ZoneId;
-import java.time.format.DateTimeFormatter;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.ExecutorService;
@@ -25,10 +22,6 @@ public class MqttStatusConsumer extends AbstractMqttConsumer {
     private static final String KEY_CONTROLLER_ID = "controller_id";
     private static final String KEY_STATUS = "status";
 
-    private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter
-            .ofPattern("yyyy-MM-dd HH:mm:ss")
-            .withZone(ZoneId.systemDefault());
-
     private final StringRedisTemplate stringRedisTemplate;
     private final SysStatusService sysStatusService;
 
@@ -76,7 +69,6 @@ public class MqttStatusConsumer extends AbstractMqttConsumer {
             return;
         }
 
-        String dateTime = toDateTime();
         String redisKey = "status_" + controllerId;
 
         // 读取 Redis 旧数据用于比较
@@ -87,7 +79,6 @@ public class MqttStatusConsumer extends AbstractMqttConsumer {
         Map<String, String> newData = new HashMap<>();
         newData.put("controllerId", controllerId);
         newData.put("status", status);
-        newData.put("createTime", dateTime);
         stringRedisTemplate.opsForHash().putAll(redisKey, newData);
         stringRedisTemplate.expire(redisKey, 24, TimeUnit.HOURS);
 
@@ -95,22 +86,18 @@ public class MqttStatusConsumer extends AbstractMqttConsumer {
             // 首次写入：查一次 MySQL 判断新增还是更新
             Integer count = sysStatusService.selectStatusCount(controllerId);
             if (count <= 0) {
-                sysStatusService.insertSysStatus(controllerId, "", status, dateTime);
+                sysStatusService.insertSysStatus(controllerId, "", status);
             } else {
-                sysStatusService.updateStatus(controllerId, "", status, dateTime);
+                sysStatusService.updateStatus(controllerId, "", status);
             }
         } else {
-            // Redis 有旧数据：比较关键字段（忽略 createTime），变化时直接更新 MySQL
+            // Redis 有旧数据：比较关键字段，变化时直接更新 MySQL
             if (!equals(oldData.get("status"), status)) {
-                sysStatusService.updateStatus(controllerId, "", status, dateTime);
+                sysStatusService.updateStatus(controllerId, "", status);
             }
         }
     }
 
-    private String toDateTime() {
-        return DATE_TIME_FORMATTER.format(Instant.now());
-    }
-
     private static boolean equals(Object a, Object b) {
         return (a == b) || (a != null && a.equals(b));
     }

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

@@ -18,13 +18,12 @@ public class SysFaultService {
     public void insertAlertData(String faultId,
                             String faultdescs,
                             String faultstatus,
-                            String createtime,
                             String messageType,
                             String controllerId,
                             String deviceId,
                             String readpeople,
                             String timestamp){
-        sysFaultMapper.insertAlertData(faultId,faultdescs,faultstatus,createtime,messageType,controllerId,deviceId,readpeople,timestamp);
+        sysFaultMapper.insertAlertData(faultId,faultdescs,faultstatus,messageType,controllerId,deviceId,readpeople,timestamp);
     }
 
     public void updateAlertData(
@@ -33,9 +32,8 @@ public class SysFaultService {
                      String faultdescs,
                      String controllerId,
                      String deviceId,
-                     String createtime,
                      String timestamp){
-        sysFaultMapper.updateAlertData(faultstatus, messageType, faultdescs, controllerId, deviceId, createtime, timestamp);
+        sysFaultMapper.updateAlertData(faultstatus, messageType, faultdescs, controllerId, deviceId, timestamp);
     }
 
     public Integer selectFaultCount(@Param("fleetId")String fleetId){

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

@@ -35,14 +35,14 @@ public class SysRealtimeService {
         return sysRealtimeMapper.selectTables();
     }
 
-    public void insertTables(String tableName, String createTime, String controllerId, String deviceId, String timestamp, String k, Object v) {
+    public void insertTables(String tableName, String controllerId, String deviceId, String timestamp, String k, Object v) {
         validateTableName(tableName);
-        sysRealtimeMapper.insertTables(tableName, createTime, controllerId, deviceId, timestamp, k, v);
+        sysRealtimeMapper.insertTables(tableName, controllerId, deviceId, timestamp, k, v);
     }
 
-    public void updateTables(String tableName, String createTime, Object v, String timestamp, String k, String controllerId, String deviceId) {
+    public void updateTables(String tableName, Object v, String timestamp, String k, String controllerId, String deviceId) {
         validateTableName(tableName);
-        sysRealtimeMapper.updateTables(tableName, createTime, v, timestamp, k, controllerId, deviceId);
+        sysRealtimeMapper.updateTables(tableName, v, timestamp, k, controllerId, deviceId);
     }
 
     public Integer selectKey(String tableName, String k, String controllerId) {

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

@@ -10,11 +10,11 @@ public class SysStatusService {
     @Resource
     public SysStatusMapper sysStatusMapper;
 
-    public void insertSysStatus(String controllerId, String fleetId, String status,String createTime){
-        sysStatusMapper.insertSysStatus(controllerId, fleetId, status,createTime);
+    public void insertSysStatus(String controllerId, String fleetId, String status){
+        sysStatusMapper.insertSysStatus(controllerId, fleetId, status);
     }
-    public void updateStatus(String controllerId,String fleetId, String status, String createTime){
-        sysStatusMapper.updateStatus(controllerId,fleetId, status, createTime);
+    public void updateStatus(String controllerId,String fleetId, String status){
+        sysStatusMapper.updateStatus(controllerId,fleetId, status);
     }
     public Integer selectStatusCount(String controllerId){
         return sysStatusMapper.selectStatusCount(controllerId);

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

@@ -325,15 +325,13 @@ public class TdEngineService {
     }
 
     /**
-     * 获取列名首字符的 UTF-8 值模 10，作为分组 ID
+     * 使用列名完整 hashCode 模 10 作为分组 ID，分布更均匀
      */
     private int getFirstCharGroupId(String columnName) {
         if (columnName == null || columnName.isEmpty()) {
             return 0;
         }
-        String firstChar = columnName.substring(0, 1);
-        byte[] bytes = firstChar.getBytes(StandardCharsets.UTF_8);
-        return (bytes[0] & 0xFF) % 10;
+        return Math.abs(columnName.hashCode()) % 10;
     }
 
     /**
@@ -368,7 +366,7 @@ public class TdEngineService {
             for (Map.Entry<String, Object> entry : data.entrySet()) {
                 String key = entry.getKey();
                 if (isValidFieldName(key) && !isReservedColumn(key) && !columnTypes.containsKey(key)) {
-                    columnTypes.put(key, getValueType(entry.getValue()));
+                    columnTypes.put(key, getValueType(key, entry.getValue()));
                 }
             }
         }
@@ -566,7 +564,11 @@ public class TdEngineService {
     /**
      * 判断值类型，返回 TdEngine 对应的数据类型
      */
-    private String getValueType(Object value) {
+    private String getValueType(String columnName, Object value) {
+        // timestamp 和 createTime 字段始终使用 TIMESTAMP 类型
+        if ("timestamp".equalsIgnoreCase(columnName) || "createTime".equalsIgnoreCase(columnName)) {
+            return "TIMESTAMP";
+        }
         if (value == null) {
             return "VARCHAR";
         }
@@ -642,6 +644,43 @@ public class TdEngineService {
             if (value instanceof java.time.Instant) {
                 return String.valueOf(((java.time.Instant) value).toEpochMilli());
             }
+            // 字符串值：尝试解析为 epoch 毫秒或格式化日期
+            String str = value.toString();
+            if (str == null || str.isEmpty()) {
+                return "NULL";
+            }
+            // 尝试解析为纯数字（epoch 毫秒/秒）
+            try {
+                long num = Long.parseLong(str);
+                if (num > 0) {
+                    // 如果是秒级时间戳（10位），转为毫秒
+                    if (str.length() == 10) {
+                        num = num * 1000;
+                    }
+                    return String.valueOf(num);
+                }
+                return "NULL";
+            } catch (NumberFormatException ignored) {
+                // 非数字，尝试按日期格式解析
+            }
+            // 尝试常见日期格式
+            java.time.format.DateTimeFormatter[] formatters = {
+                java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"),
+                java.time.format.DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss"),
+                java.time.format.DateTimeFormatter.ofPattern("yyyy/MM/dd HH:mm:ss"),
+                java.time.format.DateTimeFormatter.ISO_LOCAL_DATE_TIME
+            };
+            for (java.time.format.DateTimeFormatter fmt : formatters) {
+                try {
+                    java.time.LocalDateTime ldt = java.time.LocalDateTime.parse(str, fmt);
+                    return String.valueOf(ldt.toInstant(ZONE_OFFSET_8).toEpochMilli());
+                } catch (Exception ignored) {
+                    // 继续尝试下一个格式
+                }
+            }
+            // 所有格式都失败，返回当前时间
+            log.warn("无法解析 TIMESTAMP 列 {} 的值 '{}'，使用当前时间", columnName, str);
+            return String.valueOf(System.currentTimeMillis());
         }
 
         // 字符串类型：使用实际值长度 + 5 冗余，不截断

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

@@ -105,16 +105,15 @@ public class VehicleSyncTask {
                     }
                 }
 
-                String createTime = getStringValue(dataMap, "createTime");
                 String timestamp = getStringValue(dataMap, "timestamp");
 
-                if (createTime == null || timestamp == null) {
+                if (timestamp == null) {
                     continue;
                 }
 
                 timestamp = formatTimestamp(timestamp);
 
-                syncToTable(tableName, dataMap, createTime, timestamp, controllerId, deviceId);
+                syncToTable(tableName, dataMap, timestamp, controllerId, deviceId);
             } catch (RedisConnectionFailureException e) {
                 log.error("Redis 连接失败: {} | {}", redisKey, e.getMessage());
             } catch (DataAccessException e) {
@@ -126,7 +125,7 @@ public class VehicleSyncTask {
     }
 
     private void syncToTable(String tableName, Map<Object, Object> dataMap,
-                               String createTime, String timestamp,
+                               String timestamp,
                                String controllerId, String deviceId) {
         List<String> existingKeys = sysrealtimeService.selectAllKeys(tableName, controllerId);
         Set<String> existingKeySet = existingKeys != null ? new HashSet<>(existingKeys) : Collections.emptySet();
@@ -141,9 +140,9 @@ public class VehicleSyncTask {
             if (fieldValue == null) continue;
 
             if (existingKeySet.contains(fieldKey)) {
-                sysrealtimeService.updateTables(tableName, createTime, fieldValue, timestamp, fieldKey, controllerId, deviceId);
+                sysrealtimeService.updateTables(tableName, fieldValue, timestamp, fieldKey, controllerId, deviceId);
             } else {
-                sysrealtimeService.insertTables(tableName, createTime, controllerId, deviceId, timestamp, fieldKey, fieldValue);
+                sysrealtimeService.insertTables(tableName, controllerId, deviceId, timestamp, fieldKey, fieldValue);
             }
         }
     }

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

@@ -13,8 +13,8 @@
     </resultMap>
 
     <insert id="insertAlertData">
-        insert into alert_data(faultId,faultdescs,faultstatus,createtime,messageType,controller_id,device_id,readpeople,timestamp)
-        VALUES(#{faultId},#{faultdescs},#{faultstatus},#{createtime},#{messageType},#{controllerId},#{deviceId},#{readpeople},#{timestamp})
+        insert into alert_data(faultId,faultdescs,faultstatus,messageType,controller_id,device_id,readpeople,timestamp)
+        VALUES(#{faultId},#{faultdescs},#{faultstatus},#{messageType},#{controllerId},#{deviceId},#{readpeople},#{timestamp})
     </insert>
     <insert id="updateAlertData">
         update alert_data set faultstatus=#{faultstatus},
@@ -23,9 +23,6 @@
         where faultdescs=#{faultdescs}
           and controller_id=#{controllerId}
           and device_id=#{deviceId}
-          AND createtime &lt; #{createtime}  -- 1. 必须是当前传输时间之前的
-        ORDER BY createtime DESC         -- 2. 按时间倒序，取最近的一条
-        LIMIT 1;
     </insert>
 
     <select id="selectFaultCount" resultType="Integer">
@@ -36,7 +33,6 @@
         `faultId` varchar(255) DEFAULT NULL COMMENT '消息id',
         `faultdescs` varchar(255) DEFAULT NULL COMMENT '故障描述',
         `faultstatus` varchar(255) DEFAULT NULL COMMENT '故障是否恢复（0：触发，1：恢复）',
-        `createtime` varchar(255) DEFAULT NULL COMMENT '故障发生事件',
         `messageType` varchar(255) DEFAULT NULL COMMENT '消息状态（0未读，1已读）',
         `controller_id` varchar(255) DEFAULT NULL COMMENT '控制器id',
         `device_id` varchar(255) DEFAULT NULL COMMENT '设备id',

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

@@ -20,7 +20,6 @@
     <update id="createRealtime">
         CREATE TABLE IF NOT EXISTS `${tableName}` (
             id BIGINT NOT NULL AUTO_INCREMENT COMMENT '主键',
-            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 '时间戳',
@@ -31,7 +30,6 @@
             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)
         ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COMMENT='实时数据信息';
     </update>
@@ -40,10 +38,10 @@
         SHOW TABLES;
     </select>
     <insert id="insertTables">
-        insert into `${tableName}`(create_time,controller_id,device_id,timestamp,k,v) values(#{createTime},#{controllerId},#{deviceId},#{timestamp},#{k},#{v})
+        insert into `${tableName}`(controller_id,device_id,timestamp,k,v) values(#{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} and controller_id=#{controllerId}
+        update `${tableName}` set v=#{v},timestamp=#{timestamp} where k=#{k} and device_id=#{deviceId} and controller_id=#{controllerId}
     </update>
 
     <select id="selectKey" resultType="Integer">

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

@@ -10,10 +10,10 @@
         <result column="status" property="status"/>
     </resultMap>
     <insert id="insertSysStatus">
-        Insert into controller_status(controller_id,fleet_id,status,create_time) values(#{controllerId},#{fleetId},#{status},#{createTime})
+        Insert into controller_status(controller_id,fleet_id,status) values(#{controllerId},#{fleetId},#{status})
     </insert>
     <update id="updateStatus">
-        update controller_status set fleet_id=#{fleetId},status=#{status},create_time=#{createTime} where controller_id=#{controllerId}
+        update controller_status set fleet_id=#{fleetId},status=#{status} where controller_id=#{controllerId}
     </update>
     <select id="selectStatusCount" resultType="Integer">
         select COUNT(*) from controller_status where controller_id=#{controllerId}