refactor: migrate MQTT Consumer thread pools to Spring-managed beans

- AbstractMqttConsumer: remove private executorService, use constructor injection
- MqttDynamicConsumer/MqttChargeStationConsumer: remove private core/write executors,
  inject via @Qualifier from ExecutorConfig
- ExecutorConfig: add mqttCoreExecutor, mqttWriteExecutor, abstractConsumerExecutor
  with destroyMethod="shutdown"
- VehicleSyncTask: refine scanKeys() exception handling, add per-record try-catch
  in doUpdateSysCar() loop
- Remove unused jjwt 0.9.1 (CVE-2019-17195) and springfox-boot-starter 3.0.0 deps

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

@@ -39,4 +39,36 @@ public class ExecutorConfig {
                 new ThreadPoolExecutor.CallerRunsPolicy()
         );
     }
+
+    @Bean(destroyMethod = "shutdown")
+    public ScheduledExecutorService mqttCoreExecutor() {
+        return Executors.newSingleThreadScheduledExecutor(r -> {
+            Thread t = new Thread(r, "mqtt-core-scheduler");
+            t.setDaemon(true);
+            return t;
+        });
+    }
+
+    @Bean(destroyMethod = "shutdown")
+    public ExecutorService mqttWriteExecutor() {
+        return new ThreadPoolExecutor(
+                4, 8, 60L, TimeUnit.SECONDS,
+                new LinkedBlockingQueue<>(5000),
+                r -> {
+                    Thread t = new Thread(r, "mqtt-write-" + UUID.randomUUID().toString().substring(0, 4));
+                    t.setDaemon(true);
+                    return t;
+                },
+                new ThreadPoolExecutor.CallerRunsPolicy()
+        );
+    }
+
+    @Bean(destroyMethod = "shutdown")
+    public ExecutorService abstractConsumerExecutor() {
+        return Executors.newSingleThreadExecutor(r -> {
+            Thread t = new Thread(r, "mqtt-abstract-consumer");
+            t.setDaemon(true);
+            return t;
+        });
+    }
 }

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

@@ -12,7 +12,6 @@ import javax.annotation.PreDestroy;
 import java.net.InetSocketAddress;
 import java.net.Socket;
 import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 
 public abstract class AbstractMqttConsumer {
 
@@ -21,6 +20,8 @@ public abstract class AbstractMqttConsumer {
     @Autowired
     protected IotProperties iotProperties;
 
+    protected final ExecutorService executorService;
+
     private String brokerUrl;
     private String brokerHost;
     private int brokerPort;
@@ -35,7 +36,10 @@ public abstract class AbstractMqttConsumer {
     protected MqttClient mqttClient;
     protected MqttConnectOptions connOpts;
     protected volatile boolean isMqttConnected = false;
-    protected final ExecutorService executorService = Executors.newSingleThreadExecutor();
+
+    protected AbstractMqttConsumer(ExecutorService executorService) {
+        this.executorService = executorService;
+    }
 
     protected abstract String getSubscribeTopic();
 
@@ -168,7 +172,6 @@ public abstract class AbstractMqttConsumer {
                 mqttClient.close();
                 log.info("MQTT连接已断开");
             }
-            executorService.shutdown();
             onDestroy();
         } catch (MqttException e) {
             log.error("MQTT断开连接失败：", e);

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

@@ -7,6 +7,7 @@ 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;
@@ -56,27 +57,15 @@ public class MqttChargeStationConsumer {
     private final AtomicBoolean isConnected = new AtomicBoolean(false);
     private final Object lock = new Object();
 
-    // 核心调度线程池
-    private final ScheduledExecutorService coreExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
-        Thread t = new Thread(r);
-        t.setName("mqtt-core-scheduler");
-        t.setDaemon(false); // ← 改这里！
-        t.setPriority(Thread.NORM_PRIORITY - 1);
-        return t;
-    });
-
-    // TDengine 写入线程池
-    private final ExecutorService writeExecutor = new ThreadPoolExecutor(
-            4, 8, 60L, TimeUnit.SECONDS,
-            new LinkedBlockingQueue<>(5000),
-            r -> {
-                Thread t = new Thread(r);
-                t.setName("td-write-worker-" + UUID.randomUUID().toString().substring(0, 4));
-                t.setDaemon(false); // ← 改这里！
-                return t;
-            },
-            new ThreadPoolExecutor.CallerRunsPolicy()
-    );
+    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"})
@@ -459,8 +448,6 @@ public class MqttChargeStationConsumer {
 
     public void disconnect() {
         synchronized (lock) {
-            shutdownExecutor(writeExecutor, "TDengine 写入");
-
             if (mqttClient != null) {
                 try {
                     if (mqttClient.isConnected()) {
@@ -480,27 +467,9 @@ public class MqttChargeStationConsumer {
         }
     }
 
-    private void shutdownExecutor(ExecutorService executor, String name) {
-        if (executor == null || executor.isShutdown()) return;
-        try {
-            executor.shutdown();
-            if (!executor.awaitTermination(20, TimeUnit.SECONDS)) {
-                List<Runnable> remaining = executor.shutdownNow();
-                log.error(name + " 线程池强制关闭，剩余任务数：" + remaining.size());
-            } else {
-                log.info(name + " 线程池已优雅关闭");
-            }
-        } catch (InterruptedException e) {
-            executor.shutdownNow();
-            Thread.currentThread().interrupt();
-            log.error(name + " 线程池关闭被中断");
-        }
-    }
-
     @PreDestroy
     public void destroy() {
         log.info(">>> 服务正在关闭...");
         disconnect();
-        shutdownExecutor(coreExecutor, "MQTT 核心");
     }
 }

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

@@ -9,6 +9,7 @@ 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;
@@ -64,27 +65,15 @@ public class MqttDynamicConsumer {
     private final AtomicBoolean isConnected = new AtomicBoolean(false);
     private final Object lock = new Object();
 
-    // 核心调度线程池
-    private final ScheduledExecutorService coreExecutor = Executors.newSingleThreadScheduledExecutor(r -> {
-        Thread t = new Thread(r);
-        t.setName("mqtt-core-scheduler");
-        t.setDaemon(false); // ← 改这里！
-        t.setPriority(Thread.NORM_PRIORITY - 1);
-        return t;
-    });
-
-    // TDengine 写入线程池
-    private final ExecutorService writeExecutor = new ThreadPoolExecutor(
-            4, 8, 60L, TimeUnit.SECONDS,
-            new LinkedBlockingQueue<>(5000),
-            r -> {
-                Thread t = new Thread(r);
-                t.setName("td-write-worker-" + UUID.randomUUID().toString().substring(0, 4));
-                t.setDaemon(false); // ← 改这里！
-                return t;
-            },
-            new ThreadPoolExecutor.CallerRunsPolicy()
-    );
+    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"})
@@ -522,8 +511,6 @@ public class MqttDynamicConsumer {
 
     public void disconnect() {
         synchronized (lock) {
-            shutdownExecutor(writeExecutor, "TDengine 写入");
-
             if (mqttClient != null) {
                 try {
                     if (mqttClient.isConnected()) {
@@ -564,6 +551,5 @@ public class MqttDynamicConsumer {
     public void destroy() {
         log.info(">>> 服务正在关闭...");
         disconnect();
-        shutdownExecutor(coreExecutor, "MQTT 核心");
     }
 }

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

@@ -6,6 +6,7 @@ import com.iot.platform.domain.SysFault;
 import com.iot.platform.service.*;
 import com.iot.platform.common.utils.NumericIdGenerator;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Component;
 import org.springframework.web.client.RestTemplate;
 
@@ -43,7 +44,9 @@ public class MqttFaultConsumer extends AbstractMqttConsumer {
     private final ExecutorService mqttFaultExecutor;
 
     @Autowired
-    public MqttFaultConsumer(ExecutorService mqttFaultExecutor) {
+    public MqttFaultConsumer(@Qualifier("mqttFaultExecutor") ExecutorService mqttFaultExecutor,
+                             @Qualifier("abstractConsumerExecutor") ExecutorService executorService) {
+        super(executorService);
         this.mqttFaultExecutor = mqttFaultExecutor;
     }
 

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

@@ -5,10 +5,12 @@ import com.iot.platform.domain.ControllerData;
 import com.iot.platform.domain.topics;
 import com.iot.platform.service.SysControllerService;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 import java.util.List;
+import java.util.concurrent.ExecutorService;
 
 /**
  * 存储控制器数据
@@ -25,6 +27,11 @@ public class MqttGenericConsumer extends AbstractMqttConsumer {
 
     private final ObjectMapper objectMapper = new ObjectMapper();
 
+    @Autowired
+    public MqttGenericConsumer(@Qualifier("abstractConsumerExecutor") ExecutorService executorService) {
+        super(executorService);
+    }
+
     @Override
     protected String getSubscribeTopic() {
         return "+/generics";

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

@@ -4,12 +4,14 @@ import com.fasterxml.jackson.databind.ObjectMapper;
 import com.iot.platform.service.SysControllerService;
 import com.iot.platform.service.SysStatusService;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.stereotype.Component;
 
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
 
 /**
  * 存储控制器状态数据
@@ -26,6 +28,11 @@ public class MqttStatusConsumer extends AbstractMqttConsumer {
 
     private final ObjectMapper objectMapper = new ObjectMapper();
 
+    @Autowired
+    public MqttStatusConsumer(@Qualifier("abstractConsumerExecutor") ExecutorService executorService) {
+        super(executorService);
+    }
+
     @Override
     protected String getSubscribeTopic() {
         return "+/status";

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

@@ -79,25 +79,31 @@ public class VehicleSyncTask {
     private void doUpdateSysCar() {
         List<SysCar> sysCarList = sysCarService.selectcontrollerId();
         for (SysCar sysCar : sysCarList) {
-            if (sysCar.getControllerId() == null || sysCar.getControllerId().isEmpty()) {
-                continue;
-            }
-            SysDevice latitude = sysDeviceService.selectsysdevice(sysCar.getControllerId(), "纬度");
-            SysDevice longitude = sysDeviceService.selectsysdevice(sysCar.getControllerId(), "经度");
+            try {
+                if (sysCar.getControllerId() == null || sysCar.getControllerId().isEmpty()) {
+                    continue;
+                }
+                SysDevice latitude = sysDeviceService.selectsysdevice(sysCar.getControllerId(), "纬度");
+                SysDevice longitude = sysDeviceService.selectsysdevice(sysCar.getControllerId(), "经度");
 
-            String redisKeyPattern = "workorder:coordinate:" + sysCar.getControllerId() + ":*";
-            Set<String> keys = scanKeys(redisKeyPattern);
+                String redisKeyPattern = "workorder:coordinate:" + sysCar.getControllerId() + ":*";
+                Set<String> keys = scanKeys(redisKeyPattern);
 
-            if (keys == null || keys.isEmpty()) {
-                updateCarPosition(sysCar, latitude, longitude);
-            } else {
-                for (String key : keys) {
-                    Map<Object, Object> coordinateMap = stringRedisTemplate.opsForHash().entries(key);
-                    if (coordinateMap.get("latitude").equals(latitude.getV()) && coordinateMap.get("longitude").equals(longitude.getV())) {
-                        continue;
-                    }
+                if (keys == null || keys.isEmpty()) {
                     updateCarPosition(sysCar, latitude, longitude);
+                } else {
+                    for (String key : keys) {
+                        Map<Object, Object> coordinateMap = stringRedisTemplate.opsForHash().entries(key);
+                        if (coordinateMap.get("latitude").equals(latitude.getV()) && coordinateMap.get("longitude").equals(longitude.getV())) {
+                            continue;
+                        }
+                        updateCarPosition(sysCar, latitude, longitude);
+                    }
                 }
+            } catch (DataAccessException e) {
+                log.error("更新车辆位置失败 carId={}: {}", sysCar.getCarId(), e.getMessage(), e);
+            } catch (Exception e) {
+                log.error("更新车辆位置异常 carId={}: {}", sysCar.getCarId(), e.getMessage(), e);
             }
         }
     }
@@ -130,8 +136,10 @@ public class VehicleSyncTask {
                 cursor.close();
                 return null;
             });
+        } catch (RedisConnectionFailureException e) {
+            log.warn("Redis SCAN 连接失败 pattern={}: {}", pattern, e.getMessage());
         } catch (Exception e) {
-            log.error("Redis SCAN失败 pattern={}: {}", pattern, e.getMessage());
+            log.error("Redis SCAN 失败 pattern={}: {}", pattern, e.getMessage(), e);
         }
         return keys;
     }

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

@@ -3,21 +3,26 @@ package com.iot.platform.mqtt;
 import org.junit.jupiter.api.DisplayName;
 import org.junit.jupiter.api.Test;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
 import static org.assertj.core.api.Assertions.assertThat;
 
 class MqttGenericConsumerTest {
 
+    private final ExecutorService executorService = Executors.newSingleThreadExecutor();
+
     @Test
     @DisplayName("getSubscribeTopic 应返回 +/generics")
     void getSubscribeTopic_returnsCorrectValue() {
-        MqttGenericConsumer consumer = new MqttGenericConsumer();
+        MqttGenericConsumer consumer = new MqttGenericConsumer(executorService);
         assertThat(consumer.getSubscribeTopic()).isEqualTo("+/generics");
     }
 
     @Test
     @DisplayName("generateClientId 应包含 mqttx 前缀")
     void generateClientId_containsPrefix() {
-        MqttGenericConsumer consumer = new MqttGenericConsumer();
+        MqttGenericConsumer consumer = new MqttGenericConsumer(executorService);
         String clientId = consumer.generateClientId();
         assertThat(clientId).startsWith("mqttx_e216fbf16");
     }

pom.xml

@@ -20,11 +20,9 @@
         <maven-jar-plugin.version>3.1.1</maven-jar-plugin.version>
         <spring-boot.version>2.5.15</spring-boot.version>
         <druid.version>1.2.23</druid.version>
-        <swagger.version>3.0.0</swagger.version>
         <pagehelper.boot.version>1.4.7</pagehelper.boot.version>
         <fastjson.version>2.0.57</fastjson.version>
         <commons.io.version>2.19.0</commons.io.version>
-        <jwt.version>0.9.1</jwt.version>
         <!-- override dependency version -->
         <tomcat.version>9.0.106</tomcat.version>
         <logback.version>1.2.13</logback.version>
@@ -109,19 +107,6 @@
                 <version>${pagehelper.boot.version}</version>
             </dependency>
 
-            <!-- Swagger3依赖 -->
-            <dependency>
-                <groupId>io.springfox</groupId>
-                <artifactId>springfox-boot-starter</artifactId>
-                <version>${swagger.version}</version>
-                <exclusions>
-                    <exclusion>
-                        <groupId>io.swagger</groupId>
-                        <artifactId>swagger-models</artifactId>
-                    </exclusion>
-                </exclusions>
-            </dependency>
-
             <!-- io常用工具类 -->
             <dependency>
                 <groupId>commons-io</groupId>
@@ -129,14 +114,6 @@
                 <version>${commons.io.version}</version>
             </dependency>
 
-            <!-- Token生成与解析-->
-            <dependency>
-                <groupId>io.jsonwebtoken</groupId>
-                <artifactId>jjwt</artifactId>
-                <version>${jwt.version}</version>
-            </dependency>
-
-
         </dependencies>
     </dependencyManagement>