refactor(P0): 提取 AbstractDynamicMqttConsumer 基类消除400+行重复

- 新建 AbstractDynamicMqttConsumer 抽象基类，抽取连接/订阅/重连公共逻辑
- MqttDynamicConsumer extends 基类，仅保留 fetchTopics + processMessage
- MqttChargeStationConsumer extends 基类，仅保留 fetchTopics + processMessage
- 配置化 ChargeStation topic（IotProperties + application.yml）
- 修复 @DependsOn 位置（方法→类级别）
- 从 disconnect() 中移除 tdengineService.close()（Spring单例不应被消费者关闭）
- 基类日志统一使用 {} 占位符，移除 emoji

iot-platform/src/main/java/com/iot/platform/config/IotProperties.java

@@ -56,6 +56,7 @@ public class IotProperties {
         private String brokerUrl = "tcp://47.104.204.180:1883";
         private String username = "";
         private String password = "";
+        private String chargeStationTopic = "station/ChargeStation/device/+/post/json";
 
         public String getBrokerUrl() {
             return brokerUrl;
@@ -80,6 +81,14 @@ public class IotProperties {
         public void setPassword(String password) {
             this.password = password;
         }
+
+        public String getChargeStationTopic() {
+            return chargeStationTopic;
+        }
+
+        public void setChargeStationTopic(String chargeStationTopic) {
+            this.chargeStationTopic = chargeStationTopic;
+        }
     }
 
     /**

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

@@ -0,0 +1,442 @@
+package com.iot.platform.mqtt;
+
+import com.fasterxml.jackson.core.type.TypeReference;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.iot.platform.config.IotProperties;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.net.InetSocketAddress;
+import java.net.Socket;
+import java.util.*;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * 动态 Topic MQTT 消费者基类。
+ * 抽取 MqttDynamicConsumer 与 MqttChargeStationConsumer 的公共连接/订阅管理逻辑，
+ * 子类只需实现 {@link #fetchTopics()} 和 {@link #processMessage(String, String)}。
+ */
+public abstract class AbstractDynamicMqttConsumer {
+
+    protected final Logger log = LoggerFactory.getLogger(getClass());
+
+    protected final IotProperties iotProperties;
+    protected final ScheduledExecutorService coreExecutor;
+    protected final ExecutorService writeExecutor;
+
+    // MQTT 配置
+    protected String brokerUrl;
+    protected String brokerHost;
+    protected int brokerPort;
+    protected String mqttUsername;
+    protected String mqttPassword;
+
+    private static final int QOS = 1;
+    private static final int CONNECT_TIMEOUT = 3000;
+    private static final int RECONNECT_INTERVAL = 5000;
+    private static final int MAX_BATCH_SIZE = 50;
+
+    protected MqttClient mqttClient;
+    protected MqttConnectOptions connOpts;
+    protected final Set<String> currentTopicSet = new CopyOnWriteArraySet<>();
+    protected final AtomicBoolean isConnected = new AtomicBoolean(false);
+    protected final Object lock = new Object();
+
+    protected AbstractDynamicMqttConsumer(IotProperties iotProperties,
+                                          ScheduledExecutorService coreExecutor,
+                                          ExecutorService writeExecutor) {
+        this.iotProperties = iotProperties;
+        this.coreExecutor = coreExecutor;
+        this.writeExecutor = writeExecutor;
+    }
+
+    /**
+     * 子类实现：获取当前需要订阅的 Topic 列表。
+     */
+    protected abstract List<String> fetchTopics();
+
+    /**
+     * 子类实现：处理收到的 MQTT 消息内容。
+     *
+     * @param content 消息原始字符串（JSON）
+     * @param topic   消息所属 Topic
+     */
+    protected abstract void processMessage(String content, String topic) throws Exception;
+
+    @PostConstruct
+    public void initMqttConnection() {
+        this.brokerUrl = iotProperties.getMqtt().getBrokerUrl();
+        String brokerAddr = this.brokerUrl.replace("tcp://", "");
+        int colonIdx = brokerAddr.lastIndexOf(':');
+        this.brokerHost = brokerAddr.substring(0, colonIdx);
+        this.brokerPort = Integer.parseInt(brokerAddr.substring(colonIdx + 1));
+        this.mqttUsername = iotProperties.getMqtt().getUsername();
+        this.mqttPassword = iotProperties.getMqtt().getPassword();
+
+        log.info("开始初始化 MQTT 动态订阅服务...");
+
+        CompletableFuture<Boolean> connectFuture = CompletableFuture.supplyAsync(() -> {
+            int initRetry = 0;
+            while (initRetry < 3 && !isConnected.get()) {
+                try {
+                    if (refreshMqttSubscription()) {
+                        log.info("MQTT 动态订阅初始化成功");
+                        return true;
+                    }
+                } catch (Exception e) {
+                    log.error("MQTT 启动初始化失败（第{}次）", initRetry + 1, e);
+                }
+                initRetry++;
+                try {
+                    Thread.sleep(RECONNECT_INTERVAL * initRetry);
+                } catch (InterruptedException ex) {
+                    Thread.currentThread().interrupt();
+                    break;
+                }
+            }
+            return false;
+        }, coreExecutor);
+
+        try {
+            Boolean connected = connectFuture.get(10, TimeUnit.SECONDS);
+            if (!connected) {
+                log.error("MQTT 启动失败（超时），触发后台重连机制");
+                triggerReconnect();
+            }
+        } catch (TimeoutException e) {
+            log.error("MQTT 初始化超时（10秒），触发后台重连机制");
+            triggerReconnect();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.error("MQTT 初始化被中断");
+            triggerReconnect();
+        } catch (ExecutionException e) {
+            log.error("MQTT 初始化执行异常", e.getCause());
+            triggerReconnect();
+        }
+    }
+
+    private void initMqttConnectOptions() {
+        if (connOpts != null) return;
+        connOpts = new MqttConnectOptions();
+        connOpts.setCleanSession(false);
+        connOpts.setAutomaticReconnect(true);
+        connOpts.setConnectionTimeout(10);
+        connOpts.setKeepAliveInterval(60);
+        connOpts.setMaxInflight(10);
+        connOpts.setMaxReconnectDelay(30);
+        connOpts.setUserName(mqttUsername);
+        connOpts.setPassword(mqttPassword.toCharArray());
+    }
+
+    public boolean refreshMqttSubscription() {
+        synchronized (lock) {
+            try {
+                initMqttConnectOptions();
+                List<String> latestTopicList = fetchTopics();
+                log.info("查询到 Topic 列表: {}", latestTopicList);
+
+                if (latestTopicList == null || latestTopicList.isEmpty()) {
+                    log.error("未查询到 Topic，取消所有订阅");
+                    unsubscribeAll();
+                    currentTopicSet.clear();
+                    return false;
+                }
+
+                Set<String> latestTopicSet = new HashSet<>();
+                for (String topic : latestTopicList) {
+                    if (topic != null && !topic.trim().isEmpty()) {
+                        latestTopicSet.add(topic.trim());
+                    }
+                }
+
+                if (latestTopicSet.equals(currentTopicSet)) {
+                    log.info("Topic 列表无变化，无需刷新订阅关系");
+                    if (!isConnected.get() && !latestTopicSet.isEmpty()) {
+                        return connectAndSubscribe(latestTopicSet);
+                    }
+                    return true;
+                }
+
+                Set<String> topicsToAdd = new HashSet<>(latestTopicSet);
+                topicsToAdd.removeAll(currentTopicSet);
+                Set<String> topicsToRemove = new HashSet<>(currentTopicSet);
+                topicsToRemove.removeAll(latestTopicSet);
+
+                if (!isConnected.get()) {
+                    return connectAndSubscribe(latestTopicSet);
+                } else {
+                    if (!topicsToRemove.isEmpty()) unsubscribeTopics(topicsToRemove);
+                    if (!topicsToAdd.isEmpty()) subscribeTopics(topicsToAdd);
+                }
+
+                currentTopicSet.clear();
+                currentTopicSet.addAll(latestTopicSet);
+                log.info("MQTT 订阅刷新完成，当前数量：{}", currentTopicSet.size());
+                return true;
+
+            } catch (Exception e) {
+                log.error("MQTT 订阅刷新失败", e);
+                return false;
+            }
+        }
+    }
+
+    private boolean connectAndSubscribe(Set<String> topicSet) {
+        synchronized (lock) {
+            if (isConnected.get() && mqttClient != null && mqttClient.isConnected()) {
+                return true;
+            }
+            try {
+                if (!checkServerAvailability()) {
+                    log.error("Broker 不可达，连接失败");
+                    return false;
+                }
+
+                if (mqttClient == null) {
+                    String clientId = generateUniqueClientId();
+                    MemoryPersistence persistence = new MemoryPersistence();
+                    mqttClient = new MqttClient(brokerUrl, clientId, persistence);
+                    setMqttCallback();
+                }
+
+                int connectRetry = 0;
+                while (connectRetry < 3 && !mqttClient.isConnected()) {
+                    try {
+                        mqttClient.connect(connOpts);
+                        if (mqttClient.isConnected()) {
+                            isConnected.set(true);
+                            log.info("MQTT 连接成功，客户端 ID：{}", mqttClient.getClientId());
+                            break;
+                        }
+                    } catch (MqttException e) {
+                        log.error("MQTT 连接失败（第{}次）：{}", connectRetry + 1, e.getMessage());
+                        connectRetry++;
+                        if (connectRetry < 3) Thread.sleep(RECONNECT_INTERVAL * (connectRetry + 1));
+                    }
+                }
+
+                if (!mqttClient.isConnected()) {
+                    isConnected.set(false);
+                    return false;
+                }
+
+                if (!topicSet.isEmpty()) {
+                    List<String> topicList = new ArrayList<>(topicSet);
+                    for (int i = 0; i < topicList.size(); i += MAX_BATCH_SIZE) {
+                        int end = Math.min(i + MAX_BATCH_SIZE, topicList.size());
+                        List<String> batch = topicList.subList(i, end);
+                        batchSubscribeTopics(new HashSet<>(batch));
+                        Thread.sleep(100);
+                    }
+                }
+                return true;
+
+            } catch (MqttException e) {
+                log.error("MQTT 连接 + 订阅失败", e);
+                isConnected.set(false);
+                triggerReconnect();
+                return false;
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.error("MQTT 连接 + 订阅被中断");
+                isConnected.set(false);
+                return false;
+            }
+        }
+    }
+
+    private void batchSubscribeTopics(Set<String> topicSet) {
+        if (topicSet.isEmpty() || !isConnected.get() || mqttClient == null) return;
+        try {
+            List<String> topicList = new ArrayList<>(topicSet);
+            String[] topics = topicList.toArray(new String[0]);
+            int[] qosArr = new int[topics.length];
+            Arrays.fill(qosArr, QOS);
+            mqttClient.subscribe(topics, qosArr);
+            log.info("订阅完成：{} 个 Topic", topicList.size());
+        } catch (MqttException e) {
+            log.error("批量订阅失败：{}", e.getMessage());
+            retrySubscribeBatch(new ArrayList<>(topicSet), 1);
+        }
+    }
+
+    private void retrySubscribeBatch(List<String> batchTopics, int retryTimes) {
+        for (int retry = 1; retry <= retryTimes; retry++) {
+            try {
+                Thread.sleep(1000 * retry);
+                if (!isConnected.get() || mqttClient == null) continue;
+                String[] topics = batchTopics.toArray(new String[0]);
+                int[] qosArr = new int[topics.length];
+                Arrays.fill(qosArr, QOS);
+                mqttClient.subscribe(topics, qosArr);
+                log.info("重试订阅成功（第{}次），数量：{}", retry, batchTopics.size());
+                return;
+            } catch (MqttException e) {
+                log.error("重试订阅失败（第{}次）：{}", retry, e.getMessage());
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                log.error("重试订阅被中断");
+                break;
+            }
+        }
+        log.error("批次订阅最终失败：{}", batchTopics);
+    }
+
+    private void subscribeTopics(Set<String> topicsToAdd) {
+        if (topicsToAdd.isEmpty() || !isConnected.get()) return;
+        batchSubscribeTopics(topicsToAdd);
+    }
+
+    private void unsubscribeTopics(Set<String> topicsToRemove) {
+        if (topicsToRemove.isEmpty() || !isConnected.get() || mqttClient == null) return;
+        try {
+            String[] topics = topicsToRemove.toArray(new String[0]);
+            mqttClient.unsubscribe(topics);
+            log.info("取消订阅完成，数量：{}", topicsToRemove.size());
+        } catch (MqttException e) {
+            log.error("取消订阅失败：{}", e.getMessage());
+        }
+    }
+
+    private void unsubscribeAll() {
+        if (currentTopicSet.isEmpty() || !isConnected.get() || mqttClient == null) return;
+        try {
+            String[] topics = currentTopicSet.toArray(new String[0]);
+            mqttClient.unsubscribe(topics);
+            log.info("取消所有订阅，数量：{}", currentTopicSet.size());
+        } catch (MqttException e) {
+            log.error("取消所有订阅失败：{}", e.getMessage());
+        }
+    }
+
+    private void triggerReconnect() {
+        coreExecutor.schedule(() -> {
+            int maxAttempts = 3;
+            int attempt = 1;
+            while (attempt <= maxAttempts && !isConnected.get()) {
+                try {
+                    log.info("MQTT 重连（第{}次）", attempt);
+                    if (connectAndSubscribe(currentTopicSet)) {
+                        log.info("MQTT 重连成功");
+                        break;
+                    }
+                    attempt++;
+                    if (attempt <= maxAttempts) Thread.sleep(RECONNECT_INTERVAL * 2 * attempt);
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    break;
+                } catch (Exception e) {
+                    log.error("MQTT 重连失败（第{}次）：{}", attempt, e.getMessage());
+                    attempt++;
+                }
+            }
+            if (attempt > maxAttempts) {
+                log.error("MQTT 重连达最大次数，停止尝试");
+                isConnected.set(false);
+            }
+        }, 5, TimeUnit.SECONDS);
+    }
+
+    private boolean checkServerAvailability() {
+        try (Socket socket = new Socket()) {
+            socket.setSoTimeout(CONNECT_TIMEOUT);
+            socket.setTcpNoDelay(true);
+            socket.connect(new InetSocketAddress(brokerHost, brokerPort), CONNECT_TIMEOUT);
+            return true;
+        } catch (Exception e) {
+            log.error("MQTT Broker 不可达：{}", e.getMessage());
+            return false;
+        }
+    }
+
+    private String generateUniqueClientId() {
+        String osPrefix = System.getProperty("os.name").toLowerCase().contains("windows") ? "mqtt_win_" : "mqtt_linux_";
+        return osPrefix + System.currentTimeMillis() + "_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8);
+    }
+
+    private void setMqttCallback() {
+        if (mqttClient == null) return;
+        mqttClient.setCallback(new MqttCallback() {
+            @Override
+            public void connectionLost(Throwable cause) {
+                log.error("MQTT 连接断开：{}", cause.getMessage());
+                isConnected.set(false);
+                coreExecutor.schedule(AbstractDynamicMqttConsumer.this::triggerReconnect, 5, TimeUnit.SECONDS);
+            }
+
+            @Override
+            public void messageArrived(String topic, MqttMessage message) {
+                writeExecutor.submit(() -> {
+                    try {
+                        String content = new String(message.getPayload(), java.nio.charset.StandardCharsets.UTF_8);
+                        if (content == null || content.trim().isEmpty()) return;
+                        processMessage(content, topic);
+                    } catch (Exception e) {
+                        log.error("MQTT 消息处理失败 (Topic: {}): {}", topic, e.getMessage(), e);
+                    }
+                });
+            }
+
+            @Override
+            public void deliveryComplete(IMqttDeliveryToken token) {
+                try {
+                    token.waitForCompletion(1000);
+                } catch (MqttException ignored) {
+                }
+            }
+        });
+    }
+
+    protected Map<String, Object> deepCopyMap(Map<String, Object> original) {
+        if (original == null) return new HashMap<>();
+        try {
+            ObjectMapper mapper = new ObjectMapper();
+            String json = mapper.writeValueAsString(original);
+            return mapper.readValue(json, new TypeReference<Map<String, Object>>() {
+            });
+        } catch (Exception e) {
+            Map<String, Object> copy = new HashMap<>(original.size());
+            original.forEach((k, v) -> {
+                if (v instanceof Map) {
+                    copy.put(k, deepCopyMap((Map<String, Object>) v));
+                } else {
+                    copy.put(k, v);
+                }
+            });
+            return copy;
+        }
+    }
+
+    public void disconnect() {
+        synchronized (lock) {
+            if (mqttClient != null) {
+                try {
+                    if (mqttClient.isConnected()) {
+                        unsubscribeAll();
+                        mqttClient.disconnect(10000);
+                    }
+                    mqttClient.close();
+                    log.info("MQTT 连接已断开");
+                } catch (MqttException e) {
+                    log.error("断开 MQTT 连接失败：{}", e.getMessage());
+                } finally {
+                    isConnected.set(false);
+                    mqttClient = null;
+                }
+            }
+        }
+    }
+
+    @PreDestroy
+    public void destroy() {
+        log.info("服务正在关闭...");
+        disconnect();
+    }
+}

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

@@ -1,475 +1,73 @@
 package com.iot.platform.mqtt;
+
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.iot.platform.domain.SysController;
-import com.iot.platform.service.*;
-import org.eclipse.paho.client.mqttv3.*;
-import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
 import com.iot.platform.config.IotProperties;
+import com.iot.platform.service.TDengineService;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.DependsOn;
-import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-import java.io.IOException;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.nio.charset.StandardCharsets;
+
 import java.time.LocalDate;
-import java.time.LocalDateTime;
-import java.time.format.DateTimeFormatter;
 import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
 
-/**
- * TDegine
- * 添加接驳庄数据到TDeinge
- */
+@SuppressWarnings("unchecked")
 @Component
-public class MqttChargeStationConsumer {
-
-    private static final Logger log = LoggerFactory.getLogger(MqttChargeStationConsumer.class);
+@DependsOn({"tdengineService"})
+public class MqttChargeStationConsumer extends AbstractDynamicMqttConsumer {
 
-    @Autowired
-    private TDengineService tdengineService;
+    private final TDengineService tdengineService;
     private final ObjectMapper objectMapper = new ObjectMapper();
-    @Autowired
-    private IotProperties iotProperties;
-
-    // MQTT 配置
-    private String brokerUrl;
-    private String brokerHost;
-    private int brokerPort;
-    private static final int QOS = 1;
-    private static final int CONNECT_TIMEOUT = 3000;
-    private static final int RECONNECT_INTERVAL = 5000;
-    private static final int MAX_BATCH_SIZE = 50;
-    private String mqttUsername;
-    private String mqttPassword;
-    private MqttClient mqttClient;
-    private MqttConnectOptions connOpts;
-    private final Set<String> currentTopicSet = new CopyOnWriteArraySet<>();
-    private final AtomicBoolean isConnected = new AtomicBoolean(false);
-    private final Object lock = new Object();
-
-    private final ScheduledExecutorService coreExecutor;
-    private final ExecutorService writeExecutor;
 
     @Autowired
-    public MqttChargeStationConsumer(@Qualifier("mqttCoreExecutor") ScheduledExecutorService coreExecutor,
-                                     @Qualifier("mqttWriteExecutor") ExecutorService writeExecutor) {
-        this.coreExecutor = coreExecutor;
-        this.writeExecutor = writeExecutor;
-    }
-
-    @PostConstruct
-    @DependsOn({"sysControllerService", "tdengineService"})
-    public void initMqttConnection() {
-        this.brokerUrl = iotProperties.getMqtt().getBrokerUrl();
-        String brokerAddr = this.brokerUrl.replace("tcp://", "");
-        int colonIdx = brokerAddr.lastIndexOf(':');
-        this.brokerHost = brokerAddr.substring(0, colonIdx);
-        this.brokerPort = Integer.parseInt(brokerAddr.substring(colonIdx + 1));
-        this.mqttUsername = iotProperties.getMqtt().getUsername();
-        this.mqttPassword = iotProperties.getMqtt().getPassword();
-
-        log.info(">>> 开始初始化 MQTT 服务...");
-
-        CompletableFuture<Boolean> connectFuture = CompletableFuture.supplyAsync(() -> {
-            int initRetry = 0;
-            while (initRetry < 3 && !isConnected.get()) {
-                try {
-                    if (refreshMqttSubscription()) {
-                        log.info(">>> MQTT 初始化成功");
-                        return true;
-                    }
-                } catch (Exception e) {
-                    log.error("MQTT 启动初始化失败（第" + (initRetry + 1) + "次）：", e);
-                }
-                initRetry++;
-                try {
-                    Thread.sleep(RECONNECT_INTERVAL * initRetry);
-                } catch (InterruptedException ex) {
-                    Thread.currentThread().interrupt();
-                    break;
-                }
-            }
-            return false;
-        }, coreExecutor);
-
-        try {
-            Boolean connected = connectFuture.get(10, TimeUnit.SECONDS);
-            if (!connected) {
-                log.error("!!! MQTT 启动失败（超时），触发后台重连机制");
-                triggerReconnect();
-            }
-        } catch (TimeoutException e) {
-            log.error("MQTT 初始化超时（10秒），触发后台重连机制");
-            triggerReconnect();
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            log.error("MQTT 初始化被中断");
-            triggerReconnect();
-        } catch (ExecutionException e) {
-            log.error("MQTT 初始化执行异常: ", e.getCause());
-            triggerReconnect();
-        }
-    }
-
-    private void initMqttConnectOptions() {
-        if (connOpts != null) return;
-        try {
-            connOpts = new MqttConnectOptions();
-            connOpts.setCleanSession(false);
-            connOpts.setAutomaticReconnect(true);
-            connOpts.setConnectionTimeout(10);
-            connOpts.setKeepAliveInterval(60);
-            connOpts.setMaxInflight(10);
-            connOpts.setMaxReconnectDelay(30);
-            connOpts.setUserName(mqttUsername);
-            connOpts.setPassword(mqttPassword.toCharArray());
-        } catch (Exception e) {
-            log.error("初始化 MQTT 配置失败：" + e.getMessage());
-            throw new RuntimeException(e);
-        }
-    }
-
-    public boolean refreshMqttSubscription() {
-        synchronized (lock) {
-            try {
-                initMqttConnectOptions();
-                List<String> latestTopicList =new ArrayList<>();
-                latestTopicList.add("station/ChargeStation/device/+/post/json");
-                log.info("🔍 查询到 Topic 列表: " + latestTopicList);
-
-                if (latestTopicList == null || latestTopicList.isEmpty()) {
-                    log.error("未查询到 Topic，取消所有订阅");
-                    unsubscribeAll();
-                    currentTopicSet.clear();
-                    return false;
-                }
-
-                Set<String> latestTopicSet = new HashSet<>();
-                for (String topic : latestTopicList) {
-                    if (topic != null && !topic.trim().isEmpty()) {
-                        latestTopicSet.add(topic.trim());
-                    }
-                }
-
-                if (latestTopicSet.equals(currentTopicSet)) {
-                    log.info("Topic 列表无变化，无需刷新订阅关系");
-                    if (!isConnected.get() && !latestTopicSet.isEmpty()) {
-                        return connectAndSubscribe(latestTopicSet);
-                    }
-                    return true;
-                }
-
-                Set<String> topicsToAdd = new HashSet<>(latestTopicSet);
-                topicsToAdd.removeAll(currentTopicSet);
-                Set<String> topicsToRemove = new HashSet<>(currentTopicSet);
-                topicsToRemove.removeAll(latestTopicSet);
-
-                if (!isConnected.get()) {
-                    return connectAndSubscribe(latestTopicSet);
-                } else {
-                    if (!topicsToRemove.isEmpty()) unsubscribeTopics(topicsToRemove);
-                    if (!topicsToAdd.isEmpty()) subscribeTopics(topicsToAdd);
-                }
-
-                currentTopicSet.clear();
-                currentTopicSet.addAll(latestTopicSet);
-                log.info("MQTT 订阅刷新完成，当前数量：" + currentTopicSet.size());
-                return true;
-
-            } catch (Exception e) {
-                log.error("MQTT 订阅刷新失败：", e);
-                return false;
-            }
-        }
-    }
-
-    private boolean connectAndSubscribe(Set<String> topicSet) {
-        synchronized (lock) {
-            if (isConnected.get() && mqttClient != null && mqttClient.isConnected()) {
-                return true;
-            }
-            try {
-                if (!checkServerAvailability()) {
-                    log.error("Broker 不可达，连接失败");
-                    return false;
-                }
-
-                if (mqttClient == null) {
-                    String clientId = generateUniqueClientId();
-                    MemoryPersistence persistence = new MemoryPersistence();
-                    mqttClient = new MqttClient(brokerUrl, clientId, persistence);
-                    setMqttCallback();
-                }
-
-                int connectRetry = 0;
-                while (connectRetry < 3 && !mqttClient.isConnected()) {
-                    try {
-                        mqttClient.connect(connOpts);
-                        if (mqttClient.isConnected()) {
-                            isConnected.set(true);
-                            log.info("MQTT 连接成功，客户端 ID：" + mqttClient.getClientId());
-                            break;
-                        }
-                    } catch (MqttException e) {
-                        log.error("MQTT 连接失败（第" + (connectRetry + 1) + "次）：" + e.getMessage());
-                        connectRetry++;
-                        if (connectRetry < 3) Thread.sleep(RECONNECT_INTERVAL * (connectRetry + 1));
-                    }
-                }
-
-                if (!mqttClient.isConnected()) {
-                    isConnected.set(false);
-                    return false;
-                }
-
-                if (!topicSet.isEmpty()) {
-                    List<String> topicList = new ArrayList<>(topicSet);
-                    for (int i = 0; i < topicList.size(); i += MAX_BATCH_SIZE) {
-                        int end = Math.min(i + MAX_BATCH_SIZE, topicList.size());
-                        List<String> batch = topicList.subList(i, end);
-                        batchSubscribeTopics(new HashSet<>(batch));
-                        Thread.sleep(100);
-                    }
-                }
-                return true;
-
-            } catch (MqttException e) {
-                log.error("MQTT 连接 + 订阅失败：", e);
-                isConnected.set(false);
-                triggerReconnect();
-                return false;
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                log.error("MQTT 连接 + 订阅被中断");
-                isConnected.set(false);
-                return false;
-            }
-        }
-    }
-
-    private void batchSubscribeTopics(Set<String> topicSet) {
-        if (topicSet.isEmpty() || !isConnected.get() || mqttClient == null) return;
-        try {
-            List<String> topicList = new ArrayList<>(topicSet);
-            String[] topics = topicList.toArray(new String[0]);
-            int[] qosArr = new int[topics.length];
-            Arrays.fill(qosArr, QOS);
-            mqttClient.subscribe(topics, qosArr);
-            log.info("订阅完成：" + topicList.size() + "个 Topic");
-        } catch (MqttException e) {
-            log.error("批量订阅失败：" + e.getMessage());
-            retrySubscribeBatch(new ArrayList<>(topicSet), 1);
-        }
-    }
-
-    private void retrySubscribeBatch(List<String> batchTopics, int retryTimes) {
-        for (int retry = 1; retry <= retryTimes; retry++) {
-            try {
-                Thread.sleep(1000 * retry);
-                if (!isConnected.get() || mqttClient == null) continue;
-                String[] topics = batchTopics.toArray(new String[0]);
-                int[] qosArr = new int[topics.length];
-                Arrays.fill(qosArr, QOS);
-                mqttClient.subscribe(topics, qosArr);
-                log.info("重试订阅成功（第" + retry + "次），数量：" + batchTopics.size());
-                return;
-            } catch (MqttException e) {
-                log.error("重试订阅失败（第" + retry + "次）：" + e.getMessage());
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                log.error("重试订阅被中断");
-                break;
-            }
-        }
-        log.error("批次订阅最终失败：" + batchTopics);
-    }
-
-    private void subscribeTopics(Set<String> topicsToAdd) {
-        if (topicsToAdd.isEmpty() || !isConnected.get()) return;
-        batchSubscribeTopics(topicsToAdd);
-    }
-
-    private void unsubscribeTopics(Set<String> topicsToRemove) {
-        if (topicsToRemove.isEmpty() || !isConnected.get() || mqttClient == null) return;
-        try {
-            String[] topics = topicsToRemove.toArray(new String[0]);
-            mqttClient.unsubscribe(topics);
-            log.info("取消订阅完成，数量：" + topicsToRemove.size());
-        } catch (MqttException e) {
-            log.error("取消订阅失败：" + e.getMessage());
+    public MqttChargeStationConsumer(IotProperties iotProperties,
+                                     @Qualifier("mqttCoreExecutor") ScheduledExecutorService coreExecutor,
+                                     @Qualifier("mqttWriteExecutor") ExecutorService writeExecutor,
+                                     TDengineService tdengineService) {
+        super(iotProperties, coreExecutor, writeExecutor);
+        this.tdengineService = tdengineService;
+    }
+
+    @Override
+    protected List<String> fetchTopics() {
+        String topic = iotProperties.getMqtt().getChargeStationTopic();
+        if (topic == null || topic.trim().isEmpty()) {
+            log.warn("ChargeStation topic 未配置");
+            return Collections.emptyList();
         }
+        return Collections.singletonList(topic.trim());
     }
 
-    private void unsubscribeAll() {
-        if (currentTopicSet.isEmpty() || !isConnected.get() || mqttClient == null) return;
-        try {
-            String[] topics = currentTopicSet.toArray(new String[0]);
-            mqttClient.unsubscribe(topics);
-            log.info("取消所有订阅，数量：" + currentTopicSet.size());
-        } catch (MqttException e) {
-            log.error("取消所有订阅失败：" + e.getMessage());
-        }
-    }
-
-    private void triggerReconnect() {
-        coreExecutor.schedule(() -> {
-            int maxAttempts = 3;
-            int attempt = 1;
-            while (attempt <= maxAttempts && !isConnected.get()) {
-                try {
-                    log.info("MQTT 重连（第" + attempt + "次）");
-                    if (connectAndSubscribe(currentTopicSet)) {
-                        log.info("MQTT 重连成功");
-                        break;
-                    }
-                    attempt++;
-                    if (attempt <= maxAttempts) Thread.sleep(RECONNECT_INTERVAL * 2 * attempt);
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                    break;
-                } catch (Exception e) {
-                    log.error("MQTT 重连失败（第" + attempt + "次）：" + e.getMessage());
-                    attempt++;
-                }
-            }
-            if (attempt > maxAttempts) {
-                log.error("MQTT 重连达最大次数，停止尝试");
-                isConnected.set(false);
-            }
-        }, 5, TimeUnit.SECONDS);
-    }
-
-    private boolean checkServerAvailability() {
-        try (Socket socket = new Socket()) {
-            socket.setSoTimeout(CONNECT_TIMEOUT);
-            socket.setTcpNoDelay(true);
-            socket.connect(new InetSocketAddress(brokerHost, brokerPort), CONNECT_TIMEOUT);
-            return true;
-        } catch (Exception e) {
-            log.error("MQTT Broker 不可达：" + e.getMessage());
-            return false;
-        }
+    @Override
+    protected void processMessage(String content, String topic) throws Exception {
+        List<Map<String, Object>> messageList = objectMapper.readValue(content, new TypeReference<List<Map<String, Object>>>() {});
+        processMessageAndWriteToTDengine(messageList, topic);
     }
 
-    private String generateUniqueClientId() {
-        String osPrefix = System.getProperty("os.name").toLowerCase().contains("windows") ? "mqtt_win_" : "mqtt_linux_";
-        return osPrefix + System.currentTimeMillis() + "_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8);
-    }
-
-    private void setMqttCallback() {
-        if (mqttClient == null) return;
-        mqttClient.setCallback(new MqttCallback() {
-            @Override
-            public void connectionLost(Throwable cause) {
-                log.error("MQTT 连接断开：" + cause.getMessage());
-                isConnected.set(false);
-                coreExecutor.schedule(() -> triggerReconnect(), 5, TimeUnit.SECONDS);
-            }
-
-            @Override
-            public void messageArrived(String topic, MqttMessage message) {
-                // 直接提交到写入线程池处理
-                writeExecutor.submit(() -> {
-                    try {
-                        String content = new String(message.getPayload(), StandardCharsets.UTF_8);
-                        if (content == null || content.trim().isEmpty()) return;
-                        List<Map<String, Object>> messageList = objectMapper.readValue(content, new TypeReference<List<Map<String, Object>>>() {});
-
-                        processMessageAndWriteToTDengine(messageList, topic);
-                    } catch (Exception e) {
-                        log.error("MQTT 消息处理失败（Topic：" + topic + "）：", e);
-                    }
-                });
-            }
-
-            @Override
-            public void deliveryComplete(IMqttDeliveryToken token) {
-                try { token.waitForCompletion(1000); } catch (MqttException ignored) {}
-            }
-        });
-    }
-
-    // 参数改为 List
     private void processMessageAndWriteToTDengine(List<Map<String, Object>> dataList, String topic) throws Exception {
         String[] topicParts = topic.split("/");
         if (topicParts.length < 2) {
             throw new IllegalArgumentException("无效的 Topic 格式：" + topic);
         }
-        // TDengine 批量插入通常需要一个列表
+
         List<Map<String, Object>> batchToInsert = new ArrayList<>();
         for (Map<String, Object> dataMap : dataList) {
-            // 检查数据是否为空
             if (dataMap == null || dataMap.isEmpty()) {
                 continue;
             }
-            // 深拷贝
             Map<String, Object> list = deepCopyMap(dataMap);
             batchToInsert.add(list);
         }
 
-        // 如果有数据，才写入数据库
         if (!batchToInsert.isEmpty()) {
             String dbName = topicParts[1];
             String superTable = topicParts[3];
             LocalDate date = LocalDate.now();
             String tableName = superTable + "_" + date.getYear() + date.getMonthValue();
-            // 调用写入方法（假设 insertBatch 接收 List）
             tdengineService.insertBatch(dbName, tableName, batchToInsert);
         }
     }
-
-    private Map<String, Object> deepCopyMap(Map<String, Object> original) {
-        if (original == null) return new HashMap<>();
-        try {
-            String json = objectMapper.writeValueAsString(original);
-            return objectMapper.readValue(json, new TypeReference<Map<String, Object>>() {});
-        } catch (Exception e) {
-            Map<String, Object> copy = new HashMap<>(original.size());
-            original.forEach((k, v) -> {
-                if (v instanceof Map) {
-                    copy.put(k, deepCopyMap((Map<String, Object>) v));
-                } else {
-                    copy.put(k, v);
-                }
-            });
-            return copy;
-        }
-    }
-
-    public void disconnect() {
-        synchronized (lock) {
-            if (mqttClient != null) {
-                try {
-                    if (mqttClient.isConnected()) {
-                        unsubscribeAll();
-                        mqttClient.disconnect(10000);
-                    }
-                    mqttClient.close();
-                    log.info("MQTT 连接已断开");
-                } catch (MqttException e) {
-                    log.error("断开 MQTT 连接失败：" + e.getMessage());
-                } finally {
-                    isConnected.set(false);
-                    mqttClient = null;
-                }
-            }
-            if (tdengineService != null) tdengineService.close();
-        }
-    }
-
-    @PreDestroy
-    public void destroy() {
-        log.info(">>> 服务正在关闭...");
-        disconnect();
-    }
-}
+}

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

@@ -2,411 +2,58 @@ package com.iot.platform.mqtt;
 
 import com.fasterxml.jackson.core.type.TypeReference;
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.iot.platform.config.IotProperties;
 import com.iot.platform.domain.SysController;
 import com.iot.platform.service.SysControllerService;
 import com.iot.platform.service.TDengineService;
 import org.eclipse.paho.client.mqttv3.*;
-import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
-import com.iot.platform.config.IotProperties;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.annotation.DependsOn;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
-import javax.annotation.PostConstruct;
-import javax.annotation.PreDestroy;
-import java.net.InetSocketAddress;
-import java.net.Socket;
-import java.nio.charset.StandardCharsets;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
-import java.util.concurrent.*;
-import java.util.concurrent.atomic.AtomicBoolean;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 
-@Component
 @SuppressWarnings("unchecked")
-public class MqttDynamicConsumer {
-
-    private static final Logger log = LoggerFactory.getLogger(MqttDynamicConsumer.class);
-
-    @Autowired
-    private SysControllerService sysControllerService;
-
-    @Autowired
-    private TDengineService tdengineService;
+@Component
+@DependsOn({"sysControllerService", "tdengineService"})
+public class MqttDynamicConsumer extends AbstractDynamicMqttConsumer {
 
+    private final SysControllerService sysControllerService;
+    private final TDengineService tdengineService;
+    private final StringRedisTemplate stringRedisTemplate;
     private final ObjectMapper objectMapper = new ObjectMapper();
-    @Autowired
-    private StringRedisTemplate stringRedisTemplate;
 
     @Autowired
-    private IotProperties iotProperties;
-
-    // MQTT 配置
-    private String brokerUrl;
-    private String brokerHost;
-    private int brokerPort;
-    private static final int QOS = 1;
-    private static final int CONNECT_TIMEOUT = 3000;
-    private static final int RECONNECT_INTERVAL = 5000;
-    private static final int MAX_BATCH_SIZE = 50;
-
-    private String mqttUsername;
-    private String mqttPassword;
-
-    private MqttClient mqttClient;
-    private MqttConnectOptions connOpts;
-    private final Set<String> currentTopicSet = new CopyOnWriteArraySet<>();
-    private final AtomicBoolean isConnected = new AtomicBoolean(false);
-    private final Object lock = new Object();
-
-    private final ScheduledExecutorService coreExecutor;
-    private final ExecutorService writeExecutor;
-
-    @Autowired
-    public MqttDynamicConsumer(@Qualifier("mqttCoreExecutor") ScheduledExecutorService coreExecutor,
-                               @Qualifier("mqttWriteExecutor") ExecutorService writeExecutor) {
-        this.coreExecutor = coreExecutor;
-        this.writeExecutor = writeExecutor;
-    }
-
-    @PostConstruct
-    @DependsOn({"sysControllerService", "tdengineService"})
-    public void initMqttConnection() {
-        this.brokerUrl = iotProperties.getMqtt().getBrokerUrl();
-        String brokerAddr = this.brokerUrl.replace("tcp://", "");
-        int colonIdx = brokerAddr.lastIndexOf(':');
-        this.brokerHost = brokerAddr.substring(0, colonIdx);
-        this.brokerPort = Integer.parseInt(brokerAddr.substring(colonIdx + 1));
-        this.mqttUsername = iotProperties.getMqtt().getUsername();
-        this.mqttPassword = iotProperties.getMqtt().getPassword();
-
-        log.info(">>> 开始初始化 MQTT 服务...");
-
-        CompletableFuture<Boolean> connectFuture = CompletableFuture.supplyAsync(() -> {
-            int initRetry = 0;
-            while (initRetry < 3 && !isConnected.get()) {
-                try {
-                    if (refreshMqttSubscription()) {
-                        log.info(">>> MQTT 初始化成功");
-                        return true;
-                    }
-                } catch (Exception e) {
-                    log.error("MQTT 启动初始化失败（第" + (initRetry + 1) + "次）：", e);
-                }
-                initRetry++;
-                try {
-                    Thread.sleep(RECONNECT_INTERVAL * initRetry);
-                } catch (InterruptedException ex) {
-                    Thread.currentThread().interrupt();
-                    break;
-                }
-            }
-            return false;
-        }, coreExecutor);
-
-        try {
-            Boolean connected = connectFuture.get(10, TimeUnit.SECONDS);
-            if (!connected) {
-                log.error("!!! MQTT 启动失败（超时），触发后台重连机制");
-                triggerReconnect();
-            }
-        } catch (TimeoutException e) {
-            log.error("MQTT 初始化超时（10秒），触发后台重连机制");
-            triggerReconnect();
-        } catch (InterruptedException e) {
-            Thread.currentThread().interrupt();
-            log.error("MQTT 初始化被中断");
-            triggerReconnect();
-        } catch (ExecutionException e) {
-            log.error("MQTT 初始化执行异常: ", e.getCause());
-            triggerReconnect();
-        }
+    public MqttDynamicConsumer(IotProperties iotProperties,
+                               @Qualifier("mqttCoreExecutor") ScheduledExecutorService coreExecutor,
+                               @Qualifier("mqttWriteExecutor") ExecutorService writeExecutor,
+                               SysControllerService sysControllerService,
+                               TDengineService tdengineService,
+                               StringRedisTemplate stringRedisTemplate) {
+        super(iotProperties, coreExecutor, writeExecutor);
+        this.sysControllerService = sysControllerService;
+        this.tdengineService = tdengineService;
+        this.stringRedisTemplate = stringRedisTemplate;
     }
 
-    private void initMqttConnectOptions() {
-        if (connOpts != null) return;
-        try {
-            connOpts = new MqttConnectOptions();
-            connOpts.setCleanSession(false);
-            connOpts.setAutomaticReconnect(true);
-            connOpts.setConnectionTimeout(10);
-            connOpts.setKeepAliveInterval(60);
-            connOpts.setMaxInflight(10);
-            connOpts.setMaxReconnectDelay(30);
-            connOpts.setUserName(mqttUsername);
-            connOpts.setPassword(mqttPassword.toCharArray());
-        } catch (Exception e) {
-            log.error("初始化 MQTT 配置失败：" + e.getMessage());
-            throw new RuntimeException(e);
-        }
+    @Override
+    protected List<String> fetchTopics() {
+        return sysControllerService.selectall();
     }
 
-    public boolean refreshMqttSubscription() {
-        synchronized (lock) {
-            try {
-                initMqttConnectOptions();
-                List<String> latestTopicList = sysControllerService.selectall();
-                log.info("🔍 查询到 Topic 列表: " + latestTopicList);
-
-                if (latestTopicList == null || latestTopicList.isEmpty()) {
-                    log.error("未查询到 Topic，取消所有订阅");
-                    unsubscribeAll();
-                    currentTopicSet.clear();
-                    return false;
-                }
-
-                Set<String> latestTopicSet = new HashSet<>();
-                for (String topic : latestTopicList) {
-                    if (topic != null && !topic.trim().isEmpty()) {
-                        latestTopicSet.add(topic.trim());
-                    }
-                }
-
-                if (latestTopicSet.equals(currentTopicSet)) {
-                    log.info("Topic 列表无变化，无需刷新订阅关系");
-                    if (!isConnected.get() && !latestTopicSet.isEmpty()) {
-                        return connectAndSubscribe(latestTopicSet);
-                    }
-                    return true;
-                }
-
-                Set<String> topicsToAdd = new HashSet<>(latestTopicSet);
-                topicsToAdd.removeAll(currentTopicSet);
-                Set<String> topicsToRemove = new HashSet<>(currentTopicSet);
-                topicsToRemove.removeAll(latestTopicSet);
-
-                if (!isConnected.get()) {
-                    return connectAndSubscribe(latestTopicSet);
-                } else {
-                    if (!topicsToRemove.isEmpty()) unsubscribeTopics(topicsToRemove);
-                    if (!topicsToAdd.isEmpty()) subscribeTopics(topicsToAdd);
-                }
-
-                currentTopicSet.clear();
-                currentTopicSet.addAll(latestTopicSet);
-                log.info("MQTT 订阅刷新完成，当前数量：" + currentTopicSet.size());
-                return true;
-
-            } catch (Exception e) {
-                log.error("MQTT 订阅刷新失败：", e);
-                return false;
-            }
-        }
-    }
-
-    private boolean connectAndSubscribe(Set<String> topicSet) {
-        synchronized (lock) {
-            if (isConnected.get() && mqttClient != null && mqttClient.isConnected()) {
-                return true;
-            }
-            try {
-                if (!checkServerAvailability()) {
-                    log.error("Broker 不可达，连接失败");
-                    return false;
-                }
-
-                if (mqttClient == null) {
-                    String clientId = generateUniqueClientId();
-                    MemoryPersistence persistence = new MemoryPersistence();
-                    mqttClient = new MqttClient(brokerUrl, clientId, persistence);
-                    setMqttCallback();
-                }
-
-                int connectRetry = 0;
-                while (connectRetry < 3 && !mqttClient.isConnected()) {
-                    try {
-                        mqttClient.connect(connOpts);
-                        if (mqttClient.isConnected()) {
-                            isConnected.set(true);
-                            log.info("MQTT 连接成功，客户端 ID：" + mqttClient.getClientId());
-                            break;
-                        }
-                    } catch (MqttException e) {
-                        log.error("MQTT 连接失败（第" + (connectRetry + 1) + "次）：" + e.getMessage());
-                        connectRetry++;
-                        if (connectRetry < 3) Thread.sleep(RECONNECT_INTERVAL * (connectRetry + 1));
-                    }
-                }
-
-                if (!mqttClient.isConnected()) {
-                    isConnected.set(false);
-                    return false;
-                }
-
-                if (!topicSet.isEmpty()) {
-                    List<String> topicList = new ArrayList<>(topicSet);
-                    for (int i = 0; i < topicList.size(); i += MAX_BATCH_SIZE) {
-                        int end = Math.min(i + MAX_BATCH_SIZE, topicList.size());
-                        List<String> batch = topicList.subList(i, end);
-                        batchSubscribeTopics(new HashSet<>(batch));
-                        Thread.sleep(100);
-                    }
-                }
-                return true;
-
-            } catch (MqttException e) {
-                log.error("MQTT 连接 + 订阅失败：", e);
-                isConnected.set(false);
-                triggerReconnect();
-                return false;
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                log.error("MQTT 连接 + 订阅被中断");
-                isConnected.set(false);
-                return false;
-            }
-        }
-    }
-
-
-
-
-    private void batchSubscribeTopics(Set<String> topicSet) {
-        if (topicSet.isEmpty() || !isConnected.get() || mqttClient == null) return;
-        try {
-            List<String> topicList = new ArrayList<>(topicSet);
-            String[] topics = topicList.toArray(new String[0]);
-            int[] qosArr = new int[topics.length];
-            Arrays.fill(qosArr, QOS);
-            mqttClient.subscribe(topics, qosArr);
-            log.info("订阅完成：" + topicList.size() + "个 Topic");
-        } catch (MqttException e) {
-            log.error("批量订阅失败：" + e.getMessage());
-            retrySubscribeBatch(new ArrayList<>(topicSet), 1);
-        }
-    }
-
-    private void retrySubscribeBatch(List<String> batchTopics, int retryTimes) {
-        for (int retry = 1; retry <= retryTimes; retry++) {
-            try {
-                Thread.sleep(1000 * retry);
-                if (!isConnected.get() || mqttClient == null) continue;
-                String[] topics = batchTopics.toArray(new String[0]);
-                int[] qosArr = new int[topics.length];
-                Arrays.fill(qosArr, QOS);
-                mqttClient.subscribe(topics, qosArr);
-                log.info("重试订阅成功（第" + retry + "次），数量：" + batchTopics.size());
-                return;
-            } catch (MqttException e) {
-                log.error("重试订阅失败（第" + retry + "次）：" + e.getMessage());
-            } catch (InterruptedException e) {
-                Thread.currentThread().interrupt();
-                log.error("重试订阅被中断");
-                break;
-            }
-        }
-        log.error("批次订阅最终失败：" + batchTopics);
-    }
-
-    private void subscribeTopics(Set<String> topicsToAdd) {
-        if (topicsToAdd.isEmpty() || !isConnected.get()) return;
-        batchSubscribeTopics(topicsToAdd);
-    }
-
-    private void unsubscribeTopics(Set<String> topicsToRemove) {
-        if (topicsToRemove.isEmpty() || !isConnected.get() || mqttClient == null) return;
-        try {
-            String[] topics = topicsToRemove.toArray(new String[0]);
-            mqttClient.unsubscribe(topics);
-            log.info("取消订阅完成，数量：" + topicsToRemove.size());
-        } catch (MqttException e) {
-            log.error("取消订阅失败：" + e.getMessage());
-        }
-    }
-
-    private void unsubscribeAll() {
-        if (currentTopicSet.isEmpty() || !isConnected.get() || mqttClient == null) return;
-        try {
-            String[] topics = currentTopicSet.toArray(new String[0]);
-            mqttClient.unsubscribe(topics);
-            log.info("取消所有订阅，数量：" + currentTopicSet.size());
-        } catch (MqttException e) {
-            log.error("取消所有订阅失败：" + e.getMessage());
-        }
-    }
-
-    private void triggerReconnect() {
-        coreExecutor.schedule(() -> {
-            int maxAttempts = 3;
-            int attempt = 1;
-            while (attempt <= maxAttempts && !isConnected.get()) {
-                try {
-                    log.info("MQTT 重连（第" + attempt + "次）");
-                    if (connectAndSubscribe(currentTopicSet)) {
-                        log.info("MQTT 重连成功");
-                        break;
-                    }
-                    attempt++;
-                    if (attempt <= maxAttempts) Thread.sleep(RECONNECT_INTERVAL * 2 * attempt);
-                } catch (InterruptedException e) {
-                    Thread.currentThread().interrupt();
-                    break;
-                } catch (Exception e) {
-                    log.error("MQTT 重连失败（第" + attempt + "次）：" + e.getMessage());
-                    attempt++;
-                }
-            }
-            if (attempt > maxAttempts) {
-                log.error("MQTT 重连达最大次数，停止尝试");
-                isConnected.set(false);
-            }
-        }, 5, TimeUnit.SECONDS);
-    }
-
-    private boolean checkServerAvailability() {
-        try (Socket socket = new Socket()) {
-            socket.setSoTimeout(CONNECT_TIMEOUT);
-            socket.setTcpNoDelay(true);
-            socket.connect(new InetSocketAddress(brokerHost, brokerPort), CONNECT_TIMEOUT);
-            return true;
-        } catch (Exception e) {
-            log.error("MQTT Broker 不可达：" + e.getMessage());
-            return false;
-        }
-    }
-
-    private String generateUniqueClientId() {
-        String osPrefix = System.getProperty("os.name").toLowerCase().contains("windows") ? "mqtt_win_" : "mqtt_linux_";
-        return osPrefix + System.currentTimeMillis() + "_" + UUID.randomUUID().toString().replace("-", "").substring(0, 8);
-    }
-
-    private void setMqttCallback() {
-        if (mqttClient == null) return;
-        mqttClient.setCallback(new MqttCallback() {
-            @Override
-            public void connectionLost(Throwable cause) {
-                log.error("MQTT 连接断开：" + cause.getMessage());
-                isConnected.set(false);
-                coreExecutor.schedule(() -> triggerReconnect(), 5, TimeUnit.SECONDS);
-            }
-
-            @Override
-            public void messageArrived(String topic, MqttMessage message) {
-                // 直接提交到写入线程池处理
-                writeExecutor.submit(() -> {
-                    try {
-                        String content = new String(message.getPayload(), StandardCharsets.UTF_8);
-                        if (content == null || content.trim().isEmpty()) return;
-
-                        Map<String, Object> messageMap = objectMapper.readValue(content, new TypeReference<Map<String, Object>>() {});
-                        processMessageAndWriteToTDengine(messageMap, topic);
-                    } catch (Exception e) {
-                        log.error("MQTT 消息处理失败（Topic：" + topic + "）：", e);
-                    }
-                });
-            }
-
-            @Override
-            public void deliveryComplete(IMqttDeliveryToken token) {
-                try { token.waitForCompletion(1000); } catch (MqttException ignored) {}
-            }
-        });
+    @Override
+    protected void processMessage(String content, String topic) throws Exception {
+        Map<String, Object> weather = objectMapper.readValue(content, new TypeReference<Map<String, Object>>() {});
+        processMessageAndWriteToTDengine(weather, topic);
+        insertredis(weather, topic);
     }
 
     private void processMessageAndWriteToTDengine(Map<String, Object> weather, String topic) throws Exception {
@@ -433,7 +80,6 @@ public class MqttDynamicConsumer {
         list.put("timestamp", weather.get("timestamp"));
         list.put("device_id", weather.get("device_id"));
 
-        // 构建单条数据并立即写入（可改为批量缓冲）
         String dbName = topicParts[0];
         String superTable = topicParts[1];
         LocalDate date = LocalDate.now();
@@ -441,16 +87,13 @@ public class MqttDynamicConsumer {
 
         List<Map<String, Object>> batch = Collections.singletonList(list);
         tdengineService.insertBatch(dbName, tableName, batch);
-        insertredis(weather,topic);
     }
 
-
     public void insertredis(Map<String, Object> weather, String topic) throws Exception {
         String[] topicParts = topic.split("/", 2);
         if (topicParts.length < 2) return;
 
         String controllerId = topicParts[0];
-        String deviceIdFromTopic = topicParts[1]; // 虽然不用在 key 中，但可记录
 
         SysController sysController = sysControllerService.selectcontrollerpath(topic);
         if (sysController == null || sysController.getName() == null) return;
@@ -461,14 +104,12 @@ public class MqttDynamicConsumer {
         Map<String, Object> metricData = (Map<String, Object>) weather.get(sysController.getName());
         if (metricData == null) return;
 
-        // ✅ 使用标准格式：DSB:controllerId:metricName
         String redisKey = "DSB:" + controllerId + ":" + sysController.getName();
 
-        // 准备写入的数据（包含系统字段）
         Map<String, String> hashData = new HashMap<>();
         hashData.put("createTime", createTime);
         hashData.put("timestamp", getStringFromMap(weather, "timestamp"));
-        hashData.put("device_id", getStringFromMap(weather, "device_id")); // UUID
+        hashData.put("device_id", getStringFromMap(weather, "device_id"));
 
         for (Map.Entry<String, Object> entry : metricData.entrySet()) {
             if (entry.getValue() != null) {
@@ -479,7 +120,6 @@ public class MqttDynamicConsumer {
         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);
     }
@@ -488,51 +128,4 @@ public class MqttDynamicConsumer {
         Object val = map.get(key);
         return val == null ? "" : val.toString().trim();
     }
-
-
-
-    private Map<String, Object> deepCopyMap(Map<String, Object> original) {
-        if (original == null) return new HashMap<>();
-        try {
-            String json = objectMapper.writeValueAsString(original);
-            return objectMapper.readValue(json, new TypeReference<Map<String, Object>>() {});
-        } catch (Exception e) {
-            Map<String, Object> copy = new HashMap<>(original.size());
-            original.forEach((k, v) -> {
-                if (v instanceof Map) {
-                    copy.put(k, deepCopyMap((Map<String, Object>) v));
-                } else {
-                    copy.put(k, v);
-                }
-            });
-            return copy;
-        }
-    }
-
-    public void disconnect() {
-        synchronized (lock) {
-            if (mqttClient != null) {
-                try {
-                    if (mqttClient.isConnected()) {
-                        unsubscribeAll();
-                        mqttClient.disconnect(10000);
-                    }
-                    mqttClient.close();
-                    log.info("MQTT 连接已断开");
-                } catch (MqttException e) {
-                    log.error("断开 MQTT 连接失败：" + e.getMessage());
-                } finally {
-                    isConnected.set(false);
-                    mqttClient = null;
-                }
-            }
-            if (tdengineService != null) tdengineService.close();
-        }
-    }
-
-    @PreDestroy
-    public void destroy() {
-        log.info(">>> 服务正在关闭...");
-        disconnect();
-    }
-}
+}

iot-platform/src/main/resources/application.yml

@@ -68,6 +68,7 @@ iot:
     broker-url: tcp://47.104.204.180:1883
     username: ${MQTT_USERNAME:}
     password: ${MQTT_PASSWORD:}
+    charge-station-topic: ${MQTT_CHARGE_STATION_TOPIC:station/ChargeStation/device/+/post/json}
   tdengine:
     url: jdbc:TAOS://localhost:6030/
     username: ${TDENGINE_USERNAME:}