修改问题：1、所有表 timestamp 存时间类型 pe_iot 后缀80去掉

     2、accept、fault、issue 这3个改成redis存，redis判断有变化，每30s备份到mysql， 读直接从redis读（跟pe_iot的逻辑一样）
     3、status表 fleet_id去掉， 逻辑也是redis为主 （跟pe_iot的逻辑一样） id必须有
     4、issue、fault、accept 的 update_time没用，去掉
     5、redis数据变化逻辑，忽略时间timetsamp变化
     6、alert 坐标去掉 加上id

iot-platform/src/main/java/com/iot/platform/domain/SysRealtime.java

@@ -1,12 +1,21 @@
 package com.iot.platform.domain;
 
 public class SysRealtime {
+    private Long id;
     private String createTime;
     private String deviceId;
     private String timestamp;
     private String k;
     private String v;
 
+    public Long getId() {
+        return id;
+    }
+
+    public void setId(Long id) {
+        this.id = id;
+    }
+
     public String getCreateTime() {
         return createTime;
     }

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

@@ -11,7 +11,5 @@ public interface SysAlarmMapper {
                      @Param("createtime")String createtime,
                      @Param("messageType")String messageType,
                      @Param("controllerId")String controllerId,
-                     @Param("deviceId")String deviceId,
-                     @Param("longitude")String longitude,
-                     @Param("latitude")String latitude);
+                     @Param("deviceId")String deviceId);
 }

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

@@ -40,8 +40,19 @@ public interface SysControllerMapper {
                                 @Param("fleetId") String fleetId,
                                 @Param("name") String name,
                                 @Param("path") String path,
-                                @Param("deviceId") String deviceId,
-                                @Param("updateTime") String updateTime);
+                                @Param("deviceId") String deviceId);
+
+    void updateSysControllerCmd(@Param("controllerId") String controllerId,
+                                @Param("timestamp") String timestamp,
+                                @Param("fleetId") String fleetId,
+                                @Param("name") String name,
+                                @Param("path") String path);
+
+    void updateSysControllerFault(@Param("controllerId") String controllerId,
+                                  @Param("timestamp") String timestamp,
+                                  @Param("fleetId") String fleetId,
+                                  @Param("name") String name,
+                                  @Param("path") String path);
 
     SysController selectControllerPath(@Param("path") String path);
 

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

@@ -6,25 +6,23 @@ import org.apache.ibatis.annotations.Param;
 @Mapper
 public interface SysFaultMapper {
 
-    void insertFault(@Param("faultId") String faultId,
+    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,
-                     @Param("longitude") String longitude,
-                     @Param("latitude") String latitude,
-                     @Param("readpeople") String readpeople);
+                     @Param("readpeople") String readpeople,
+                     @Param("timestamp") String timestamp);
 
-    void updateFault(@Param("faultstatus")String faultstatus,
+    void updateAlertData(@Param("faultstatus")String faultstatus,
                      @Param("messageType")String messageType,
-                     @Param("longitude")String longitude,
-                     @Param("latitude")String latitude,
                      @Param("faultdescs")String faultdescs,
                      @Param("controllerId")String controllerId,
                      @Param("deviceId")String deviceId,
-                     @Param("createtime")String createtime);
+                     @Param("createtime")String createtime,
+                     @Param("timestamp")String timestamp);
 
 
     Integer selectFaultCount(@Param("fleetId")String fleetId);

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

@@ -8,6 +8,7 @@ import java.util.List;
 @Mapper
 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);

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

@@ -3,7 +3,6 @@ package com.iot.platform.mqtt;
 import com.alibaba.fastjson2.JSON;
 import com.alibaba.fastjson2.TypeReference;
 import com.iot.platform.config.IotProperties;
-import com.iot.platform.domain.SysDevice;
 import com.iot.platform.domain.SysFault;
 import com.iot.platform.service.*;
 import com.iot.platform.common.utils.NumericIdGenerator;
@@ -14,8 +13,10 @@ import org.springframework.web.client.RestTemplate;
 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.*;
 import java.util.concurrent.ExecutorService;
@@ -38,6 +39,9 @@ public class MqttFaultConsumer extends AbstractMqttConsumer {
     private static final String ALARM_STATUS_TRIGGER = "0";
     private static final String ALARM_STATUS_RECOVERED = "1";
     private static final String DATE_TIME_PATTERN = "yyyy-MM-dd HH:mm:ss";
+    private static final DateTimeFormatter DATE_TIME_FORMATTER = DateTimeFormatter
+            .ofPattern(DATE_TIME_PATTERN)
+            .withZone(ZoneId.systemDefault());
     private static final String COMPANY_ID_PREFIX = "GJ";
     private static final Pattern CONTROLLER_ID_PATTERN = Pattern.compile("^[a-zA-Z0-9_]+$");
 
@@ -174,25 +178,19 @@ public class MqttFaultConsumer extends AbstractMqttConsumer {
             sysFaultService.createMessage(controllerName);
         }
 
-        SysDevice longitude = sysControllerService.selectCoordinates(topics[0], "经度");
-        SysDevice latitude = sysControllerService.selectCoordinates(topics[0], "纬度");
-        if (longitude == null || latitude == null) {
-            log.info("未查询到控制器经纬度信息: {}", topics[0]);
-            return;
-        }
-        String longitudeValue = longitude.getV();
-        String latitudeValue = latitude.getV();
+        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, longitudeValue, latitudeValue);
-            sysFaultService.insertFault(companyId, desc, ALARM_STATUS_TRIGGER, currentTimeStr, ALARM_STATUS_TRIGGER, controllerId, deviceId, longitudeValue, latitudeValue, "");
+//            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);
         } else if ("恢复".equals(type)) {
-            sysAlarmService.insertAlarm(controllerName, companyId, desc, ALARM_STATUS_RECOVERED, currentTimeStr, ALARM_STATUS_TRIGGER, controllerId, deviceId, longitudeValue, latitudeValue);
-            sysFaultService.updateFault(ALARM_STATUS_RECOVERED, ALARM_STATUS_TRIGGER, longitudeValue, latitudeValue, desc, controllerId, deviceId, currentTimeStr);
+//            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);
         }
 
         // controllerId 白名单校验，防止 SSRF
@@ -210,4 +208,16 @@ public class MqttFaultConsumer extends AbstractMqttConsumer {
             log.info("Webhook 调用失败，不影响主流程: {}", e.getMessage());
         }
     }
-}
+
+    private String toDateTime(String timestamp) {
+        long ts;
+        try {
+            ts = Long.parseLong(timestamp);
+        } catch (NumberFormatException e) {
+            log.info("timestamp 格式错误: {}，使用当前时间", timestamp);
+            ts = System.currentTimeMillis();
+        }
+        return DATE_TIME_FORMATTER.format(Instant.ofEpochMilli(ts));
+    }
+
+}

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

@@ -14,8 +14,11 @@ import java.time.Instant;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.TimeUnit;
 
 /**
  * 存储控制器数据
@@ -31,6 +34,16 @@ public class MqttGenericConsumer extends AbstractMqttConsumer {
             .ofPattern("yyyy-MM-dd HH:mm:ss")
             .withZone(ZoneId.systemDefault());
 
+    private String toDateTime(String timestamp) {
+        long ts;
+        try {
+            ts = Long.parseLong(timestamp);
+        } catch (NumberFormatException e) {
+            log.info("timestamp 格式错误: {}，使用当前时间", timestamp);
+            ts = System.currentTimeMillis();
+        }
+        return DATE_TIME_FORMATTER.format(Instant.ofEpochMilli(ts));
+    }
     @Autowired
     public MqttGenericConsumer(@Qualifier("abstractConsumerExecutor") ExecutorService executorService,
                                IotProperties iotProperties,
@@ -96,6 +109,8 @@ public class MqttGenericConsumer extends AbstractMqttConsumer {
 
     private int processTopics(List<Topics> topics, String controllerId, String timestamp, String fleetId) {
         int newControllerCount = 0;
+        String dateTime = toDateTime(timestamp);
+
         for (Topics topicsMap : topics) {
             String path = topicsMap.getPath();
             String name = topicsMap.getName();
@@ -108,27 +123,45 @@ public class MqttGenericConsumer extends AbstractMqttConsumer {
                 log.info("path 中缺少 deviceId，跳过: path={}", path);
                 continue;
             }
-            Integer count = sysControllerService.selectSysControllerCount(path);
-            if (count <= 0) {
-                stringRedisTemplate.opsForHash().put(controllerId + ":" + name, "path", path);
-                sysControllerService.insertSysController(controllerId, timestamp, fleetId, name, path, deviceId);
-                newControllerCount++;
+
+            String redisKey = controllerId + "_" + name;
+            // 读取 Redis 旧数据用于比较
+            Map<Object, Object> oldData = stringRedisTemplate.opsForHash().entries(redisKey);
+            boolean redisEmpty = (oldData == null || oldData.isEmpty());
+
+            // 写入最新数据到 Redis
+            Map<String, String> newData = new HashMap<>();
+            newData.put("path", path);
+            newData.put("timestamp", dateTime);
+            newData.put("name", name);
+            newData.put("deviceId", deviceId);
+            stringRedisTemplate.opsForHash().putAll(redisKey, newData);
+            stringRedisTemplate.expire(redisKey, 24, TimeUnit.HOURS);
+
+            if (redisEmpty) {
+                // 首次写入：查一次 MySQL 判断新增还是更新（Redis 过期但 MySQL 有数据时走更新）
+                Integer count = sysControllerService.selectSysControllerCount(path);
+                if (count <= 0) {
+                    sysControllerService.insertSysController(controllerId, dateTime, fleetId, name, path, deviceId);
+                    newControllerCount++;
+                } else {
+                    sysControllerService.updateControllerAccept(controllerId, dateTime, fleetId, name, path, deviceId);
+                }
             } else {
-                long ts;
-                try {
-                    ts = Long.parseLong(timestamp);
-                } catch (NumberFormatException e) {
-                    log.info("timestamp 格式错误: {}，使用当前时间", timestamp);
-                    ts = System.currentTimeMillis();
+                // Redis 有旧数据：比较关键字段（忽略 timestamp），变化时直接更新 MySQL
+                if (!equals(oldData.get("path"), path)
+                        || !equals(oldData.get("name"), name)
+                        || !equals(oldData.get("deviceId"), deviceId)) {
+                    sysControllerService.updateControllerAccept(controllerId, dateTime, fleetId, name, path, deviceId);
                 }
-                String date = DATE_TIME_FORMATTER.format(Instant.ofEpochMilli(ts));
-                sysControllerService.updateControllerAccept(controllerId, timestamp, fleetId, name, path, deviceId, date);
             }
         }
         return newControllerCount;
     }
 
     private void processCmdTopics(List<Topics> cmdtopics, String controllerId, String timestamp, String fleetId) {
+        String dateTime = toDateTime(timestamp);
+
         for (Topics cmdtopicsMap : cmdtopics) {
             String path = cmdtopicsMap.getPath();
             String name = cmdtopicsMap.getName();
@@ -136,10 +169,30 @@ public class MqttGenericConsumer extends AbstractMqttConsumer {
                 log.info("CmdTopic 数据不完整，跳过: path={}, name={}", path, name);
                 continue;
             }
-            Integer count = sysControllerService.selectSysControllerCmdCount(path);
-            if (count <= 0) {
-                stringRedisTemplate.opsForHash().put(controllerId + "_cmd:" + name, "path", path);
-                sysControllerService.insertSysControllerCmd(controllerId, timestamp, fleetId, name, path);
+
+            String redisKey = controllerId + "_issue_" + name;
+            Map<Object, Object> oldData = stringRedisTemplate.opsForHash().entries(redisKey);
+            boolean redisEmpty = (oldData == null || oldData.isEmpty());
+
+            Map<String, String> newData = new HashMap<>();
+            newData.put("path", path);
+            newData.put("timestamp", dateTime);
+            newData.put("name", name);
+            stringRedisTemplate.opsForHash().putAll(redisKey, newData);
+            stringRedisTemplate.expire(redisKey, 24, TimeUnit.HOURS);
+
+            if (redisEmpty) {
+                Integer count = sysControllerService.selectSysControllerCmdCount(path);
+                if (count <= 0) {
+                    sysControllerService.insertSysControllerCmd(controllerId, dateTime, fleetId, name, path);
+                } else {
+                    sysControllerService.updateSysControllerCmd(controllerId, dateTime, fleetId, name, path);
+                }
+            } else {
+                if (!equals(oldData.get("path"), path)
+                        || !equals(oldData.get("name"), name)) {
+                    sysControllerService.updateSysControllerCmd(controllerId, dateTime, fleetId, name, path);
+                }
             }
         }
     }
@@ -154,10 +207,35 @@ public class MqttGenericConsumer extends AbstractMqttConsumer {
             log.info("FaultProt 数据不完整，跳过: path={}, name={}", path, name);
             return;
         }
-        Integer count = sysControllerService.selectSysControllerFaultCount(path);
-        if (count <= 0) {
-            stringRedisTemplate.opsForHash().put(controllerId + "_fault:" + name, "path", path);
-            sysControllerService.insertSysControllerFault(controllerId, timestamp, fleetId, name, path);
+
+        String dateTime = toDateTime(timestamp);
+        String redisKey = controllerId + "_fault_" + name;
+        Map<Object, Object> oldData = stringRedisTemplate.opsForHash().entries(redisKey);
+        boolean redisEmpty = (oldData == null || oldData.isEmpty());
+
+        Map<String, String> newData = new HashMap<>();
+        newData.put("path", path);
+        newData.put("timestamp", dateTime);
+        newData.put("name", name);
+        stringRedisTemplate.opsForHash().putAll(redisKey, newData);
+        stringRedisTemplate.expire(redisKey, 24, TimeUnit.HOURS);
+
+        if (redisEmpty) {
+            Integer count = sysControllerService.selectSysControllerFaultCount(path);
+            if (count <= 0) {
+                sysControllerService.insertSysControllerFault(controllerId, dateTime, fleetId, name, path);
+            } else {
+                sysControllerService.updateSysControllerFault(controllerId, dateTime, fleetId, name, path);
+            }
+        } else {
+            if (!equals(oldData.get("path"), path)
+                    || !equals(oldData.get("name"), name)) {
+                sysControllerService.updateSysControllerFault(controllerId, dateTime, fleetId, name, path);
+            }
         }
     }
+
+    private static boolean equals(Object a, Object b) {
+        return (a == b) || (a != null && a.equals(b));
+    }
 }

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

@@ -2,30 +2,43 @@ package com.iot.platform.mqtt;
 
 import com.alibaba.fastjson.JSONObject;
 import com.iot.platform.config.IotProperties;
+import com.iot.platform.service.SysStatusService;
 import org.apache.commons.lang3.StringUtils;
 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;
+import java.util.concurrent.TimeUnit;
 
 /**
- * MQTT 状态消息消费者，将控制器状态写入 Redis。
+ * MQTT 状态消息消费者，优先写入 Redis，数据变化时同步 MySQL。
  */
 @Component
 public class MqttStatusConsumer extends AbstractMqttConsumer {
 
     private static final String KEY_CONTROLLER_ID = "controller_id";
-    private static final String KEY_FLEET_ID = "fleet_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;
 
     public MqttStatusConsumer(@Qualifier("abstractConsumerExecutor") ExecutorService executorService,
                               IotProperties iotProperties,
-                              StringRedisTemplate stringRedisTemplate) {
+                              StringRedisTemplate stringRedisTemplate,
+                              SysStatusService sysStatusService) {
         super(executorService, iotProperties);
         this.stringRedisTemplate = stringRedisTemplate;
+        this.sysStatusService = sysStatusService;
     }
 
     @Override
@@ -55,18 +68,50 @@ public class MqttStatusConsumer extends AbstractMqttConsumer {
 
     private void processStatusPayload(JSONObject payload) {
         String controllerId = payload.getString(KEY_CONTROLLER_ID);
-        String fleetId = payload.getString(KEY_FLEET_ID);
         String status = payload.getString(KEY_STATUS);
 
-        if (StringUtils.isBlank(controllerId) || StringUtils.isBlank(fleetId)
-                || StringUtils.isBlank(status)) {
-            log.info("MQTT 状态消息缺少必填字段 | controllerId={} | fleetId={} | status={}",
-                    controllerId, fleetId, status);
+        if (StringUtils.isBlank(controllerId) || StringUtils.isBlank(status)) {
+            log.info("MQTT 状态消息缺少必填字段 | controllerId={} | status={}",
+                    controllerId, status);
             return;
         }
 
-        String redisKey = controllerId + "_" + fleetId;
-        stringRedisTemplate.opsForValue().set(redisKey, status);
-        log.debug("状态已写入 Redis | key={} | status={}", redisKey, status);
+        String dateTime = toDateTime();
+        String redisKey = "status_" + controllerId;
+
+        // 读取 Redis 旧数据用于比较
+        Map<Object, Object> oldData = stringRedisTemplate.opsForHash().entries(redisKey);
+        boolean redisEmpty = (oldData == null || oldData.isEmpty());
+
+        // 写入最新数据到 Redis
+        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);
+
+        if (redisEmpty) {
+            // 首次写入：查一次 MySQL 判断新增还是更新
+            Integer count = sysStatusService.selectStatusCount(controllerId);
+            if (count <= 0) {
+                sysStatusService.insertSysStatus(controllerId, "", status, dateTime);
+            } else {
+                sysStatusService.updateStatus(controllerId, "", status, dateTime);
+            }
+        } else {
+            // Redis 有旧数据：比较关键字段（忽略 createTime），变化时直接更新 MySQL
+            if (!equals(oldData.get("status"), status)) {
+                sysStatusService.updateStatus(controllerId, "", status, dateTime);
+            }
+        }
+    }
+
+    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/SysAlarmService.java

@@ -16,11 +16,11 @@ public class SysAlarmService {
     @Resource
     public SysAlarmMapper sysAlarmMapper;
 
-    public void insertAlarm(String tableName,String faultId,String faultdescs,String faultstatus,String createtime,String messageType,String controllerId,String deviceId,String longitude,String latitude){
+    public void insertAlarm(String tableName,String faultId,String faultdescs,String faultstatus,String createtime,String messageType,String controllerId,String deviceId){
         if (tableName == null || !TABLE_NAME_PATTERN.matcher(tableName).matches()) {
             throw new IllegalArgumentException("非法表名: " + tableName);
         }
-        sysAlarmMapper.insertAlarm(tableName,faultId, faultdescs, faultstatus, createtime, messageType,controllerId,deviceId,longitude,latitude);
+        sysAlarmMapper.insertAlarm(tableName,faultId, faultdescs, faultstatus, createtime, messageType,controllerId,deviceId);
     }
 
 }

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

@@ -45,8 +45,18 @@ public class SysControllerService {
     }
 
     public void updateControllerAccept(String controllerId, String timestamp, String fleetId,
-                                       String name, String path, String deviceId, String updateTime) {
-        sysControllerMapper.updateControllerAccept(controllerId, timestamp, fleetId, name, path, deviceId, updateTime);
+                                       String name, String path, String deviceId) {
+        sysControllerMapper.updateControllerAccept(controllerId, timestamp, fleetId, name, path, deviceId);
+    }
+
+    public void updateSysControllerCmd(String controllerId, String timestamp, String fleetId,
+                                       String name, String path) {
+        sysControllerMapper.updateSysControllerCmd(controllerId, timestamp, fleetId, name, path);
+    }
+
+    public void updateSysControllerFault(String controllerId, String timestamp, String fleetId,
+                                         String name, String path) {
+        sysControllerMapper.updateSysControllerFault(controllerId, timestamp, fleetId, name, path);
     }
 
     public SysController selectControllerPath(String path) {

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

@@ -15,29 +15,27 @@ public class SysFaultService {
     @Resource
     public SysFaultMapper sysFaultMapper;
 
-    public void insertFault(String faultId,
+    public void insertAlertData(String faultId,
                             String faultdescs,
                             String faultstatus,
                             String createtime,
                             String messageType,
                             String controllerId,
                             String deviceId,
-                            String longitude,
-                            String latitude,
-                            String readpeople){
-        sysFaultMapper.insertFault(faultId,faultdescs,faultstatus,createtime,messageType,controllerId,deviceId,longitude,latitude,readpeople);
+                            String readpeople,
+                            String timestamp){
+        sysFaultMapper.insertAlertData(faultId,faultdescs,faultstatus,createtime,messageType,controllerId,deviceId,readpeople,timestamp);
     }
 
-    public void updateFault(
+    public void updateAlertData(
                      String faultstatus,
                      String messageType,
-                     String longitude,
-                     String latitude,
                      String faultdescs,
                      String controllerId,
                      String deviceId,
-                     String createtime){
-        sysFaultMapper.updateFault(faultstatus, messageType, longitude, latitude, faultdescs, controllerId, deviceId, createtime);
+                     String createtime,
+                     String timestamp){
+        sysFaultMapper.updateAlertData(faultstatus, messageType, faultdescs, controllerId, deviceId, createtime, timestamp);
     }
 
     public Integer selectFaultCount(@Param("fleetId")String fleetId){

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

@@ -19,6 +19,11 @@ public class SysRealtimeService {
     @Resource
     public SysRealtimeMapper sysRealtimeMapper;
 
+    public boolean tableExists(String tableName) {
+        validateTableName(tableName);
+        return sysRealtimeMapper.tableExists(tableName) > 0;
+    }
+
     public void createRealtime(String tableName) {
         validateTableName(tableName);
         log.info("准备创建表: [{}]", tableName);

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

@@ -17,6 +17,10 @@ import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestClientException;
 import org.springframework.web.client.RestTemplate;
 
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
 import java.util.*;
 import java.util.concurrent.TimeUnit;
 
@@ -27,7 +31,6 @@ public class VehicleSyncTask {
 
     private final StringRedisTemplate stringRedisTemplate;
     private final SysRealtimeService sysrealtimeService;
-    private final Set<String> createdTables = new HashSet<>();
 
     public VehicleSyncTask(StringRedisTemplate stringRedisTemplate,
                            SysRealtimeService sysrealtimeService) {
@@ -91,13 +94,11 @@ public class VehicleSyncTask {
                 }
                 String controllerId = remainder.substring(0, lastUnderscore);
                 String deviceId = remainder.substring(lastUnderscore + 1);
-                String dbPrefix = controllerId.length() >= 2 ? controllerId.substring(0, 2) : controllerId;
-                String tableName = "pe_iot_" + dbPrefix;
+                String tableName = "pe_iot";
 
-                if (!createdTables.contains(tableName)) {
+                if (!sysrealtimeService.tableExists(tableName)) {
                     try {
                         sysrealtimeService.createRealtime(tableName);
-                        createdTables.add(tableName);
                     } catch (Exception e) {
                         log.error("创建表失败: {} | {}", tableName, e.getMessage(), e);
                         continue;
@@ -111,23 +112,9 @@ public class VehicleSyncTask {
                     continue;
                 }
 
-                List<String> existingKeys = sysrealtimeService.selectAllKeys(tableName, controllerId);
-                Set<String> existingKeySet = existingKeys != null ? new HashSet<>(existingKeys) : Collections.emptySet();
+                timestamp = formatTimestamp(timestamp);
 
-                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);
-                    }
-                }
+                syncToTable(tableName, dataMap, createTime, timestamp, controllerId, deviceId);
             } catch (RedisConnectionFailureException e) {
                 log.error("Redis 连接失败: {} | {}", redisKey, e.getMessage());
             } catch (DataAccessException e) {
@@ -138,8 +125,47 @@ public class VehicleSyncTask {
         }
     }
 
+    private void syncToTable(String tableName, Map<Object, Object> dataMap,
+                               String createTime, String timestamp,
+                               String controllerId, String deviceId) {
+        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);
+            }
+        }
+    }
+
     private String getStringValue(Map<Object, Object> map, String key) {
         Object val = map.get(key);
         return val == null ? null : val.toString().trim();
     }
+
+    private static final DateTimeFormatter TS_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
+
+    private String formatTimestamp(String ts) {
+        try {
+            long tsLong = Long.parseLong(ts);
+            if (ts.length() >= 13) {
+                tsLong = tsLong / 1000;
+            }
+            return LocalDateTime.ofInstant(Instant.ofEpochSecond(tsLong), ZoneId.systemDefault())
+                    .format(TS_FORMATTER);
+        } catch (NumberFormatException e) {
+            log.warn("timestamp 格式无法解析，原样返回: {}", ts);
+            return ts;
+        }
+    }
 }

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

@@ -14,7 +14,7 @@
     </resultMap>
 
     <insert id="insertAlarm">
-        insert into `${tableName}`(faultId,faultdescs,faultstatus,createtime,messageType,controller_id,device_id,longitude,latitude)
-        values(#{faultId},#{faultdescs},#{faultstatus},#{createtime},#{messageType},#{controllerId},#{deviceId},#{longitude},#{latitude})
+        insert into `${tableName}`(faultId,faultdescs,faultstatus,createtime,messageType,controller_id,device_id)
+        values(#{faultId},#{faultdescs},#{faultstatus},#{createtime},#{messageType},#{controllerId},#{deviceId})
     </insert>
 </mapper>

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

@@ -48,8 +48,25 @@
             timestamp=#{timestamp},
             fleet_id=#{fleetId},
             name=#{name},
-            device_id=#{deviceId},
-            update_time=#{updateTime}
+            device_id=#{deviceId}
+        where path = #{path}
+    </update>
+
+    <update id="updateSysControllerCmd">
+        update controller_issue
+        set controller_id=#{controllerId},
+            timestamp=#{timestamp},
+            fleet_id=#{fleetId},
+            name=#{name}
+        where path = #{path}
+    </update>
+
+    <update id="updateSysControllerFault">
+        update controller_fault
+        set controller_id=#{controllerId},
+            timestamp=#{timestamp},
+            fleet_id=#{fleetId},
+            name=#{name}
         where path = #{path}
     </update>
 

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

@@ -5,22 +5,21 @@
 <mapper namespace="com.iot.platform.mapper.SysFaultMapper">
     <!-- 通用查询映射结果 -->
     <resultMap id="BaseResultMap" type="com.iot.platform.domain.SysFault">
-        <result column="deviceId" property="device_id"/>
+        <result column="deviceId" property="deviceId"/>
         <result column="timestamp" property="timestamp"/>
         <result column="type" property="type"/>
         <result column="code" property="code"/>
         <result column="desc" property="desc"/>
     </resultMap>
 
-    <insert id="insertFault">
-        insert into alert_data(faultId,faultdescs,faultstatus,createtime,messageType,controller_id,device_id,longitude,latitude,readpeople)
-        VALUES(#{faultId},#{faultdescs},#{faultstatus},#{createtime},#{messageType},#{controllerId},#{deviceId},#{longitude},#{latitude},#{readpeople})
+    <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>
-    <insert id="updateFault">
+    <insert id="updateAlertData">
         update alert_data set faultstatus=#{faultstatus},
                              messageType=#{messageType},
-                             longitude=#{longitude},
-                             latitude=#{latitude}
+                             timestamp=#{timestamp}
         where faultdescs=#{faultdescs}
           and controller_id=#{controllerId}
           and device_id=#{deviceId}

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

@@ -5,20 +5,28 @@
 <mapper namespace="com.iot.platform.mapper.SysRealtimeMapper">
     <!-- 通用查询映射结果 -->
     <resultMap id="BaseResultMap" type="com.iot.platform.domain.SysRealtime">
+        <id column="id" property="id"/>
         <result column="createTime" property="createTime"/>
         <result column="deviceId" property="deviceId"/>
         <result column="timestamp" property="timestamp"/>
         <result column="k" property="k"/>
         <result column="v" property="v"/>
     </resultMap>
+    <select id="tableExists" resultType="int">
+        SELECT COUNT(*) FROM information_schema.tables
+        WHERE table_schema = DATABASE() AND table_name = #{tableName}
+    </select>
+
     <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 '时间戳',
             k VARCHAR(255) NOT NULL COMMENT 'key',
             v VARCHAR(255) NOT NULL COMMENT '值',
+            PRIMARY KEY (id),
             INDEX idx_controller_id (controller_id),
             INDEX idx_controller_device (controller_id, device_id),
             INDEX idx_controller_k (controller_id, k),

iot-platform/src/test/java/com/iot/platform/mqtt/MqttStatusConsumerTest.java

@@ -2,6 +2,7 @@ package com.iot.platform.mqtt;
 
 import com.alibaba.fastjson.JSONObject;
 import com.iot.platform.config.IotProperties;
+import com.iot.platform.service.SysStatusService;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
@@ -33,13 +34,16 @@ class MqttStatusConsumerTest {
     @Mock
     private IotProperties iotProperties;
 
+    @Mock
+    private SysStatusService sysStatusService;
+
     private MqttStatusConsumer mqttStatusConsumer;
 
     private final ExecutorService executorService = Executors.newSingleThreadExecutor();
 
     @BeforeEach
     void setUp() {
-        mqttStatusConsumer = new MqttStatusConsumer(executorService, iotProperties, stringRedisTemplate);
+        mqttStatusConsumer = new MqttStatusConsumer(executorService, iotProperties, stringRedisTemplate, sysStatusService);
         lenient().when(stringRedisTemplate.opsForValue()).thenReturn(valueOperations);
     }