feat:添加消息发送队列 处理消息积压问题

device-service-server/src/main/java/com/hfln/device/server/controller/DeviceMonitorController.java → device-service-application/src/main/java/com/hfln/device/application/controller/web/DeviceMonitorController.java

@@ -1,6 +1,5 @@
-package com.hfln.device.server.controller;
+package com.hfln.device.application.controller.web;
 
-import com.hfln.device.domain.entity.BehaviorPattern;
 import com.hfln.device.domain.service.DeviceMonitorService;
 import com.hfln.device.domain.service.DeviceService;
 import lombok.extern.slf4j.Slf4j;

device-service-infrastructure/src/main/java/com/hfln/device/infrastructure/config/MqttConfig.java

@@ -63,6 +63,15 @@ public class MqttConfig {
     
     @Value("${mqtt.clean.session:true}")
     private boolean cleanSession;
+    
+    @Value("${mqtt.max.inflight:1000}")
+    private int maxInflight;
+    
+    @Value("${mqtt.connection.pool.size:5}")
+    private int connectionPoolSize;
+    
+    @Value("${mqtt.message.queue.size:10000}")
+    private int messageQueueSize;
 
     // ===========================================
     // 基础配置
@@ -100,6 +109,11 @@ public class MqttConfig {
         options.setKeepAliveInterval(keepAliveInterval);
         options.setCleanSession(cleanSession);
         
+        // 添加重要的MQTT连接参数来缓解"Too many publishes in progress"错误
+        options.setMaxInflight(maxInflight);  // 增加最大并发发布数，默认是10
+        options.setAutomaticReconnect(true);  // 启用自动重连
+        options.setMaxReconnectDelay(30000);  // 最大重连延迟30秒
+        
         factory.setConnectionOptions(options);
         
         return factory;
@@ -313,6 +327,15 @@ public class MqttConfig {
         messageHandler.setAsync(true);
         messageHandler.setDefaultQos(1);
         
+        // 添加重要的配置参数来缓解"Too many publishes in progress"错误
+        messageHandler.setCompletionTimeout(10000);  // 增加完成超时时间到10秒
+        messageHandler.setDefaultRetained(false);    // 默认不保留消息
+        messageHandler.setDefaultTopic("");          // 设置默认主题
+        
+        // 设置异步发送参数
+        messageHandler.setAsyncEvents(true);         // 启用异步事件
+        messageHandler.setAsync(true);               // 确保异步模式
+        
         return messageHandler;
     }
     

device-service-infrastructure/src/main/java/com/hfln/device/infrastructure/config/MqttConnectionPoolManager.java

@@ -0,0 +1,237 @@
+package com.hfln.device.infrastructure.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
+import org.springframework.integration.mqtt.core.MqttPahoClientFactory;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicInteger;
+
+/**
+ * MQTT连接池管理器
+ * 管理多个MQTT连接，提高并发处理能力
+ * 特别针对实时点位数据的高频发送场景
+ */
+@Component
+@Slf4j
+public class MqttConnectionPoolManager {
+
+    @Value("${mqtt.connection.pool.size:5}")
+    private int poolSize;
+    
+    @Value("${mqtt.connection.pool.max.wait:5000}")
+    private long maxWaitTime;
+    
+    @Autowired
+    private MqttPahoClientFactory mqttClientFactory;
+    
+    // 连接池
+    private final BlockingQueue<MqttPahoMessageHandler> connectionPool = new LinkedBlockingQueue<>();
+    
+    // 活跃连接数
+    private final AtomicInteger activeConnections = new AtomicInteger(0);
+    
+    // 连接池状态
+    private volatile boolean initialized = false;
+    
+    @PostConstruct
+    public void init() {
+        initializeConnectionPool();
+    }
+    
+    @PreDestroy
+    public void shutdown() {
+        closeAllConnections();
+    }
+    
+    /**
+     * 初始化连接池
+     */
+    private void initializeConnectionPool() {
+        try {
+            log.info("Initializing MQTT connection pool with {} connections", poolSize);
+            
+            for (int i = 0; i < poolSize; i++) {
+                MqttPahoMessageHandler handler = createNewConnection("pool-" + i);
+                if (handler != null) {
+                    connectionPool.offer(handler);
+                    activeConnections.incrementAndGet();
+                }
+            }
+            
+            initialized = true;
+            log.info("MQTT connection pool initialized successfully. Active connections: {}", activeConnections.get());
+            
+        } catch (Exception e) {
+            log.error("Failed to initialize MQTT connection pool", e);
+        }
+    }
+    
+    /**
+     * 创建新的MQTT连接
+     */
+    private MqttPahoMessageHandler createNewConnection(String connectionId) {
+        try {
+            String clientId = "hfln-device-service-" + connectionId + "-" + System.currentTimeMillis();
+            MqttPahoMessageHandler handler = new MqttPahoMessageHandler(clientId, mqttClientFactory);
+            
+            // 配置连接参数
+            handler.setAsync(true);
+            handler.setDefaultQos(1);
+            handler.setCompletionTimeout(10000);
+            handler.setDefaultRetained(false);
+            handler.setAsyncEvents(true);
+            
+            log.debug("Created new MQTT connection: {}", clientId);
+            return handler;
+            
+        } catch (Exception e) {
+            log.error("Failed to create MQTT connection: {}", connectionId, e);
+            return null;
+        }
+    }
+    
+    /**
+     * 获取可用的连接
+     */
+    public MqttPahoMessageHandler getConnection() {
+        if (!initialized) {
+            log.warn("Connection pool not initialized yet");
+            return null;
+        }
+        
+        try {
+            // 尝试从池中获取连接
+            MqttPahoMessageHandler handler = connectionPool.poll(maxWaitTime, TimeUnit.MILLISECONDS);
+            
+            if (handler != null) {
+                // 检查连接是否仍然有效
+                if (isConnectionHealthy(handler)) {
+                    return handler;
+                } else {
+                    // 连接无效，创建新连接替换
+                    log.warn("Invalid connection detected, creating replacement");
+                    handler = createNewConnection("replacement-" + System.currentTimeMillis());
+                    if (handler != null) {
+                        return handler;
+                    }
+                }
+            }
+            
+            // 如果池中没有可用连接，尝试创建新连接
+            if (activeConnections.get() < poolSize * 2) { // 允许临时超出池大小
+                log.info("Pool exhausted, creating temporary connection. Active: {}", activeConnections.get());
+                MqttPahoMessageHandler newHandler = createNewConnection("temp-" + System.currentTimeMillis());
+                if (newHandler != null) {
+                    activeConnections.incrementAndGet();
+                    return newHandler;
+                }
+            }
+            
+            log.warn("No available MQTT connections. Pool size: {}, Active: {}", poolSize, activeConnections.get());
+            return null;
+            
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.error("Interrupted while waiting for MQTT connection", e);
+            return null;
+        }
+    }
+    
+    /**
+     * 归还连接到池中
+     */
+    public void returnConnection(MqttPahoMessageHandler handler) {
+        if (handler == null) {
+            return;
+        }
+        
+        try {
+            // 检查连接是否仍然有效
+            if (isConnectionHealthy(handler)) {
+                // 如果池未满，归还连接
+                if (connectionPool.size() < poolSize) {
+                    connectionPool.offer(handler);
+                } else {
+                    // 池已满，关闭多余连接
+                    closeConnection(handler);
+                    activeConnections.decrementAndGet();
+                }
+            } else {
+                // 连接无效，关闭并减少计数
+                closeConnection(handler);
+                activeConnections.decrementAndGet();
+                
+                // 尝试创建新连接补充池
+                MqttPahoMessageHandler newHandler = createNewConnection("replacement-" + System.currentTimeMillis());
+                if (newHandler != null) {
+                    connectionPool.offer(newHandler);
+                }
+            }
+        } catch (Exception e) {
+            log.error("Error returning MQTT connection to pool", e);
+            closeConnection(handler);
+            activeConnections.decrementAndGet();
+        }
+    }
+    
+    /**
+     * 检查连接是否健康
+     */
+    private boolean isConnectionHealthy(MqttPahoMessageHandler handler) {
+        try {
+            return handler.isRunning();
+        } catch (Exception e) {
+            log.debug("Connection health check failed", e);
+            return false;
+        }
+    }
+    
+    /**
+     * 关闭连接
+     */
+    private void closeConnection(MqttPahoMessageHandler handler) {
+        try {
+            if (handler != null) {
+                handler.stop();
+            }
+        } catch (Exception e) {
+            log.debug("Error closing MQTT connection", e);
+        }
+    }
+    
+    /**
+     * 关闭所有连接
+     */
+    private void closeAllConnections() {
+        log.info("Closing all MQTT connections. Pool size: {}", connectionPool.size());
+        
+        MqttPahoMessageHandler handler;
+        while ((handler = connectionPool.poll()) != null) {
+            closeConnection(handler);
+            activeConnections.decrementAndGet();
+        }
+        
+        log.info("All MQTT connections closed. Active connections: {}", activeConnections.get());
+    }
+    
+    /**
+     * 获取连接池状态
+     */
+    public String getPoolStatus() {
+        return String.format("MQTT Connection Pool - Pool Size: %d, Available: %d, Active: %d, Initialized: %s",
+                poolSize, connectionPool.size(), activeConnections.get(), initialized);
+    }
+    
+    /**
+     * 健康检查
+     */
+    public boolean isHealthy() {
+        return initialized && activeConnections.get() > 0;
+    }
+} 

device-service-infrastructure/src/main/java/com/hfln/device/infrastructure/config/ObjectInjectConfig.java

@@ -6,7 +6,8 @@ import org.springframework.security.crypto.bcrypt.BCryptPasswordEncoder;
 import org.springframework.web.client.RestTemplate;
 
 @Configuration
-public class ObjectInjectConfig {
+public class
+ObjectInjectConfig {
 
     @Bean
     public RestTemplate restTemplate() {

device-service-infrastructure/src/main/java/com/hfln/device/infrastructure/config/RedisConfig.java

@@ -1,22 +1,89 @@
 package com.hfln.device.infrastructure.config;
 
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Primary;
 import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
+import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.data.redis.serializer.StringRedisSerializer;
+import com.hfln.device.infrastructure.config.FastJson2JsonRedisSerializer;
 
 /**
  * Redis 配置类
- * 手动注册 RedisTemplate，确保与自研 Redis Starter 兼容
+ * 手动注册 RedisTemplate 和 RedisConnectionFactory，确保与自研 Redis Starter 兼容
  */
 @Configuration
 public class RedisConfig {
 
+    @Value("${spring.redis.host:localhost}")
+    private String host;
+    
+    @Value("${spring.redis.port:6379}")
+    private int port;
+    
+    @Value("${spring.redis.database:0}")
+    private int database;
+    
+    @Value("${spring.redis.password:}")
+    private String password;
+    
+    @Value("${spring.redis.timeout:2000}")
+    private String timeout;
+
+    /**
+     * 创建Redis连接工厂
+     */
     @Bean
-    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory connectionFactory) {
+    @Primary
+    @ConditionalOnMissingBean(name = "redisConnectionFactory")
+    public RedisConnectionFactory redisConnectionFactory() {
+        RedisStandaloneConfiguration config = new RedisStandaloneConfiguration();
+        config.setHostName(host);
+        config.setPort(port);
+        config.setDatabase(database);
+        
+        if (password != null && !password.isEmpty()) {
+            config.setPassword(password);
+        }
+        
+        LettuceConnectionFactory factory = new LettuceConnectionFactory(config);
+        long timeoutMs = parseTimeout(timeout);
+        factory.setTimeout(timeoutMs);
+        factory.afterPropertiesSet();
+        
+        return factory;
+    }
+
+    private long parseTimeout(String timeoutStr) {
+        if (timeoutStr == null || timeoutStr.isEmpty()) {
+            return 2000L; // 默认2秒
+        }
+
+        try {
+            if (timeoutStr.endsWith("s")) {
+                return Long.parseLong(timeoutStr.substring(0, timeoutStr.length() - 1)) * 1000;
+            } else if (timeoutStr.endsWith("ms")) {
+                return Long.parseLong(timeoutStr.substring(0, timeoutStr.length() - 2));
+            } else {
+                return Long.parseLong(timeoutStr);
+            }
+        } catch (NumberFormatException e) {
+            return 2000L; // 解析失败时使用默认值
+        }
+    }
+    /**
+     * 创建RedisTemplate
+     */
+    @Bean("redisTemplate")
+    @Primary
+    @ConditionalOnMissingBean(name = "redisTemplate")
+    public RedisTemplate<String, Object> redisTemplate() {
         RedisTemplate<String, Object> template = new RedisTemplate<>();
-        template.setConnectionFactory(connectionFactory);
+        template.setConnectionFactory(redisConnectionFactory());
 
         // 使用 StringRedisSerializer 进行序列化
         StringRedisSerializer stringSerializer = new StringRedisSerializer();

device-service-infrastructure/src/main/java/com/hfln/device/infrastructure/controller/MqttMonitorController.java

@@ -0,0 +1,145 @@
+package com.hfln.device.infrastructure.controller;
+
+import com.hfln.device.infrastructure.config.MqttConnectionPoolManager;
+import com.hfln.device.infrastructure.service.MqttHealthService;
+import com.hfln.device.infrastructure.service.MqttMessageBufferService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * MQTT监控控制器
+ * 提供MQTT系统状态监控接口
+ */
+@RestController
+@RequestMapping("/api/mqtt/monitor")
+@Slf4j
+public class MqttMonitorController {
+
+    @Autowired
+    private MqttHealthService mqttHealthService;
+    
+    @Autowired
+    private MqttMessageBufferService mqttBufferService;
+    
+    @Autowired
+    private MqttConnectionPoolManager connectionPoolManager;
+
+    /**
+     * 获取MQTT系统整体状态
+     */
+    @GetMapping("/status")
+    public Map<String, Object> getMqttStatus() {
+        Map<String, Object> status = new HashMap<>();
+        
+        try {
+            // 连接状态
+            status.put("connectionHealthy", mqttHealthService.isConnected());
+            status.put("connectionPoolStatus", connectionPoolManager.getPoolStatus());
+            status.put("connectionPoolHealthy", connectionPoolManager.isHealthy());
+            
+            // 缓冲队列状态
+            status.put("bufferQueueStatus", mqttBufferService.getQueueStatus());
+            
+            // 性能统计
+            status.put("performanceStats", mqttHealthService.getPerformanceStats());
+            
+            // 时间戳
+            status.put("timestamp", System.currentTimeMillis());
+            
+        } catch (Exception e) {
+            log.error("Error getting MQTT status", e);
+            status.put("error", e.getMessage());
+        }
+        
+        return status;
+    }
+
+    /**
+     * 获取连接池状态
+     */
+    @GetMapping("/connection-pool")
+    public Map<String, Object> getConnectionPoolStatus() {
+        Map<String, Object> status = new HashMap<>();
+        
+        try {
+            status.put("poolStatus", connectionPoolManager.getPoolStatus());
+            status.put("healthy", connectionPoolManager.isHealthy());
+            status.put("timestamp", System.currentTimeMillis());
+            
+        } catch (Exception e) {
+            log.error("Error getting connection pool status", e);
+            status.put("error", e.getMessage());
+        }
+        
+        return status;
+    }
+
+    /**
+     * 获取缓冲队列状态
+     */
+    @GetMapping("/buffer-queue")
+    public Map<String, Object> getBufferQueueStatus() {
+        Map<String, Object> status = new HashMap<>();
+        
+        try {
+            status.put("queueStatus", mqttBufferService.getQueueStatus());
+            status.put("timestamp", System.currentTimeMillis());
+            
+        } catch (Exception e) {
+            log.error("Error getting buffer queue status", e);
+            status.put("error", e.getMessage());
+        }
+        
+        return status;
+    }
+
+    /**
+     * 获取性能统计
+     */
+    @GetMapping("/performance")
+    public Map<String, Object> getPerformanceStats() {
+        Map<String, Object> status = new HashMap<>();
+        
+        try {
+            status.put("performanceStats", mqttHealthService.getPerformanceStats());
+            status.put("timestamp", System.currentTimeMillis());
+            
+        } catch (Exception e) {
+            log.error("Error getting performance stats", e);
+            status.put("error", e.getMessage());
+        }
+        
+        return status;
+    }
+
+    /**
+     * 健康检查接口
+     */
+    @GetMapping("/health")
+    public Map<String, Object> healthCheck() {
+        Map<String, Object> health = new HashMap<>();
+        
+        try {
+            boolean connectionHealthy = mqttHealthService.isConnected();
+            boolean poolHealthy = connectionPoolManager.isHealthy();
+            
+            health.put("status", connectionHealthy && poolHealthy ? "UP" : "DOWN");
+            health.put("mqttConnection", connectionHealthy ? "UP" : "DOWN");
+            health.put("connectionPool", poolHealthy ? "UP" : "DOWN");
+            health.put("timestamp", System.currentTimeMillis());
+            
+        } catch (Exception e) {
+            log.error("Health check failed", e);
+            health.put("status", "DOWN");
+            health.put("error", e.getMessage());
+        }
+        
+        return health;
+    }
+} 

device-service-infrastructure/src/main/java/com/hfln/device/infrastructure/gateway/MqttGatewayImpl.java

@@ -15,15 +15,20 @@ import org.springframework.integration.mqtt.support.MqttHeaders;
 import org.springframework.messaging.Message;
 import org.springframework.messaging.support.MessageBuilder;
 import org.springframework.stereotype.Component;
+import com.hfln.device.infrastructure.service.MqttHealthService;
+import com.hfln.device.infrastructure.service.MqttMessageBufferService;
 
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
 
 /**
  * MQTT网关实现类 - 基于Spring Integration MQTT
  * 合并了原MqttGatewayDefaultImpl的优秀特性
+ * 添加了流量控制机制来缓解"Too many publishes in progress"错误
  */
 @Component
 @Primary
@@ -35,6 +40,17 @@ public class MqttGatewayImpl implements MqttGateway {
 
     @Autowired
     private ObjectMapper objectMapper;
+    
+    @Autowired
+    private MqttHealthService mqttHealthService;
+    
+    @Autowired
+    private MqttMessageBufferService mqttBufferService;
+    
+    // 移除信号量控制，改为使用缓冲服务
+    // 重试配置
+    private static final int MAX_RETRY_ATTEMPTS = 3;
+    private static final long RETRY_DELAY_MS = 100;
 
     @Override
     public void initialize() {
@@ -697,6 +713,7 @@ public class MqttGatewayImpl implements MqttGateway {
     /**
      * 发送消息到MQTT服务器 - 增强版
      * 支持动态QoS和消息保留设置
+     * 使用缓冲服务确保数据不丢失，特别是实时点位数据
      *
      * @param topic   主题
      * @param payload 负载
@@ -705,18 +722,61 @@ public class MqttGatewayImpl implements MqttGateway {
      */
     private void sendToMqtt(String topic, String payload, int qos, boolean retain) {
         try {
-            Message<String> message = MessageBuilder
-                    .withPayload(payload)
-                    .setHeader(MqttHeaders.TOPIC, topic)
-                    .setHeader(MqttHeaders.QOS, qos)
-                    .setHeader(MqttHeaders.RETAINED, retain)
-                    .build();
-
-            mqttOutbound.handleMessage(message);
-            log.trace("MQTT message sent to topic: {}", topic);
+            // 判断是否为高优先级消息（实时点位数据）
+            boolean isHighPriority = isHighPriorityTopic(topic);
+            
+            if (isHighPriority) {
+                // 高优先级消息使用阻塞队列，确保不丢失
+                mqttBufferService.sendHighPriorityMessage(topic, payload, qos, retain);
+                mqttHealthService.recordMessageSent();
+                log.trace("High priority MQTT message queued: topic={}", topic);
+            } else {
+                // 普通消息使用非阻塞队列
+                boolean queued = mqttBufferService.sendNormalPriorityMessage(topic, payload, qos, retain);
+                if (queued) {
+                    mqttHealthService.recordMessageSent();
+                    log.trace("Normal priority MQTT message queued: topic={}", topic);
+                } else {
+                    mqttHealthService.recordMessageDropped();
+                    log.warn("Normal priority message queue full, dropped: topic={}", topic);
+                }
+            }
         } catch (Exception e) {
-            log.error("Failed to send MQTT message to topic: {}, error: {}", topic, e.getMessage());
+            log.error("Failed to queue MQTT message to topic: {}, error: {}", topic, e.getMessage());
+            mqttHealthService.recordMessageFailed();
+        }
+    }
+    
+    /**
+     * 判断是否为高优先级主题
+     * 实时点位数据等关键数据使用高优先级
+     */
+    private boolean isHighPriorityTopic(String topic) {
+        if (topic == null) {
+            return false;
+        }
+        
+        // 实时点位数据 - 最高优先级
+        if (topic.equals("/das/realtime_pos") || 
+            topic.equals(MqttTopics.DAS_REALTIME_POS)) {
+            return true;
+        }
+        
+        // 告警事件 - 高优先级
+        if (topic.contains("/alarm_event") || 
+            topic.contains("/event") ||
+            topic.contains("/fall")) {
+            return true;
+        }
+        
+        // 设备状态 - 中等优先级
+        if (topic.contains("/status") || 
+            topic.contains("/keepalive")) {
+            return false;
         }
+        
+        // 其他消息 - 普通优先级
+        return false;
     }
 
     /**

device-service-infrastructure/src/main/java/com/hfln/device/infrastructure/service/MqttHealthService.java

@@ -0,0 +1,101 @@
+package com.hfln.device.infrastructure.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+/**
+ * MQTT健康检查服务
+ * 监控MQTT连接状态和性能指标，帮助诊断"Too many publishes in progress"问题
+ */
+@Service
+@Slf4j
+public class MqttHealthService {
+
+    @Autowired
+    private MqttPahoMessageHandler mqttOutbound;
+
+    // 性能统计
+    private final AtomicLong totalMessagesSent = new AtomicLong(0);
+    private final AtomicLong failedMessages = new AtomicLong(0);
+    private final AtomicLong droppedMessages = new AtomicLong(0);
+
+    /**
+     * 记录消息发送成功
+     */
+    public void recordMessageSent() {
+        totalMessagesSent.incrementAndGet();
+    }
+
+    /**
+     * 记录消息发送失败
+     */
+    public void recordMessageFailed() {
+        failedMessages.incrementAndGet();
+    }
+
+    /**
+     * 记录消息被丢弃
+     */
+    public void recordMessageDropped() {
+        droppedMessages.incrementAndGet();
+    }
+
+    /**
+     * 获取性能统计信息
+     */
+    public String getPerformanceStats() {
+        long total = totalMessagesSent.get();
+        long failed = failedMessages.get();
+        long dropped = droppedMessages.get();
+        long success = total - failed - dropped;
+        
+        return String.format("MQTT Performance Stats - Total: %d, Success: %d, Failed: %d, Dropped: %d", 
+                total, success, failed, dropped);
+    }
+
+    /**
+     * 定时健康检查 - 每5分钟执行一次
+     */
+    @Scheduled(fixedRate = 300000) // 5分钟
+    public void healthCheck() {
+        try {
+            // 检查MQTT连接状态
+            boolean isConnected = mqttOutbound.isRunning();
+            
+            if (isConnected) {
+                log.info("MQTT Health Check: Connection is healthy");
+                log.info(getPerformanceStats());
+            } else {
+                log.warn("MQTT Health Check: Connection is not healthy");
+            }
+            
+            // 重置计数器（每天重置一次）
+            if (System.currentTimeMillis() % 86400000 < 300000) { // 每天重置
+                totalMessagesSent.set(0);
+                failedMessages.set(0);
+                droppedMessages.set(0);
+                log.info("MQTT performance counters reset");
+            }
+            
+        } catch (Exception e) {
+            log.error("MQTT Health Check failed: {}", e.getMessage(), e);
+        }
+    }
+
+    /**
+     * 获取当前MQTT连接状态
+     */
+    public boolean isConnected() {
+        try {
+            return mqttOutbound.isRunning();
+        } catch (Exception e) {
+            log.error("Failed to check MQTT connection status: {}", e.getMessage());
+            return false;
+        }
+    }
+} 

device-service-infrastructure/src/main/java/com/hfln/device/infrastructure/service/MqttMessageBufferService.java

@@ -0,0 +1,301 @@
+package com.hfln.device.infrastructure.service;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.integration.mqtt.outbound.MqttPahoMessageHandler;
+import org.springframework.integration.mqtt.support.MqttHeaders;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.support.MessageBuilder;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.concurrent.*;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * MQTT消息缓冲服务
+ * 专门处理实时点位数据，确保数据不丢失
+ * 使用多级缓冲策略和智能重试机制
+ */
+@Service
+@Slf4j
+public class MqttMessageBufferService {
+
+    @Autowired
+    private MqttPahoMessageHandler mqttOutbound;
+
+    // 高优先级队列 - 用于实时点位数据
+    private final BlockingQueue<MqttMessage> highPriorityQueue = new LinkedBlockingQueue<>(10000);
+    
+    // 普通优先级队列 - 用于其他消息
+    private final BlockingQueue<MqttMessage> normalPriorityQueue = new LinkedBlockingQueue<>(5000);
+    
+    // 死信队列 - 存储发送失败的消息，用于后续重试
+    private final BlockingQueue<MqttMessage> deadLetterQueue = new LinkedBlockingQueue<>(1000);
+    
+    // 线程池 - 专门处理MQTT消息发送
+    private final ExecutorService mqttExecutor = Executors.newFixedThreadPool(3, r -> {
+        Thread t = new Thread(r, "mqtt-sender-" + System.currentTimeMillis());
+        t.setDaemon(true);
+        return t;
+    });
+    
+    // 控制标志
+    private final AtomicBoolean running = new AtomicBoolean(true);
+    
+    // 统计信息
+    private final AtomicLong totalMessages = new AtomicLong(0);
+    private final AtomicLong sentMessages = new AtomicLong(0);
+    private final AtomicLong failedMessages = new AtomicLong(0);
+    private final AtomicLong bufferedMessages = new AtomicLong(0);
+
+    @PostConstruct
+    public void init() {
+        // 启动消息处理线程
+        startMessageProcessors();
+        log.info("MQTT Message Buffer Service initialized");
+    }
+
+    @PreDestroy
+    public void shutdown() {
+        running.set(false);
+        mqttExecutor.shutdown();
+        try {
+            if (!mqttExecutor.awaitTermination(30, TimeUnit.SECONDS)) {
+                mqttExecutor.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            mqttExecutor.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+        log.info("MQTT Message Buffer Service shutdown completed");
+    }
+
+    /**
+     * 发送高优先级消息（如实时点位数据）
+     * 如果队列满，会阻塞等待，确保数据不丢失
+     */
+    public void sendHighPriorityMessage(String topic, String payload, int qos, boolean retain) {
+        MqttMessage message = new MqttMessage(topic, payload, qos, retain, System.currentTimeMillis(), true);
+        
+        try {
+            // 对于高优先级消息，使用阻塞方式，确保不丢失
+            highPriorityQueue.put(message);
+            totalMessages.incrementAndGet();
+            bufferedMessages.incrementAndGet();
+            
+            if (log.isTraceEnabled()) {
+                log.trace("High priority message queued: topic={}, queueSize={}", topic, highPriorityQueue.size());
+            }
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.error("Failed to queue high priority message: {}", topic, e);
+        }
+    }
+
+    /**
+     * 发送普通优先级消息
+     * 如果队列满，会尝试非阻塞方式，避免阻塞高优先级消息
+     */
+    public boolean sendNormalPriorityMessage(String topic, String payload, int qos, boolean retain) {
+        MqttMessage message = new MqttMessage(topic, payload, qos, retain, System.currentTimeMillis(), false);
+        
+        // 对于普通消息，使用非阻塞方式
+        boolean offered = normalPriorityQueue.offer(message);
+        if (offered) {
+            totalMessages.incrementAndGet();
+            bufferedMessages.incrementAndGet();
+            
+            if (log.isTraceEnabled()) {
+                log.trace("Normal priority message queued: topic={}, queueSize={}", topic, normalPriorityQueue.size());
+            }
+        } else {
+            log.warn("Normal priority queue is full, message dropped: topic={}", topic);
+        }
+        
+        return offered;
+    }
+
+    /**
+     * 启动消息处理器
+     */
+    private void startMessageProcessors() {
+        // 高优先级消息处理器
+        mqttExecutor.submit(() -> processHighPriorityMessages());
+        
+        // 普通优先级消息处理器
+        mqttExecutor.submit(() -> processNormalPriorityMessages());
+        
+        // 死信队列重试处理器
+        mqttExecutor.submit(() -> processDeadLetterQueue());
+    }
+
+    /**
+     * 处理高优先级消息
+     */
+    private void processHighPriorityMessages() {
+        while (running.get()) {
+            try {
+                MqttMessage message = highPriorityQueue.take();
+                if (sendMessageWithRetry(message)) {
+                    sentMessages.incrementAndGet();
+                    bufferedMessages.decrementAndGet();
+                } else {
+                    failedMessages.incrementAndGet();
+                    // 高优先级消息失败后放入死信队列重试
+                    deadLetterQueue.offer(message);
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            } catch (Exception e) {
+                log.error("Error processing high priority message", e);
+            }
+        }
+    }
+
+    /**
+     * 处理普通优先级消息
+     */
+    private void processNormalPriorityMessages() {
+        while (running.get()) {
+            try {
+                MqttMessage message = normalPriorityQueue.poll(100, TimeUnit.MILLISECONDS);
+                if (message != null) {
+                    if (sendMessageWithRetry(message)) {
+                        sentMessages.incrementAndGet();
+                        bufferedMessages.decrementAndGet();
+                    } else {
+                        failedMessages.incrementAndGet();
+                        // 普通消息失败后也放入死信队列重试
+                        deadLetterQueue.offer(message);
+                    }
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            } catch (Exception e) {
+                log.error("Error processing normal priority message", e);
+            }
+        }
+    }
+
+    /**
+     * 处理死信队列
+     */
+    private void processDeadLetterQueue() {
+        while (running.get()) {
+            try {
+                MqttMessage message = deadLetterQueue.poll(5000, TimeUnit.MILLISECONDS);
+                if (message != null) {
+                    // 延迟重试，避免立即重试
+                    if (System.currentTimeMillis() - message.getTimestamp() > 10000) { // 10秒后重试
+                        if (sendMessageWithRetry(message)) {
+                            sentMessages.incrementAndGet();
+                            log.info("Dead letter message retry successful: topic={}", message.getTopic());
+                        } else {
+                            // 重试失败，重新放入死信队列
+                            deadLetterQueue.offer(message);
+                        }
+                    } else {
+                        // 时间未到，重新放入队列
+                        deadLetterQueue.offer(message);
+                    }
+                }
+            } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+                break;
+            } catch (Exception e) {
+                log.error("Error processing dead letter queue", e);
+            }
+        }
+    }
+
+    /**
+     * 发送消息并重试
+     */
+    private boolean sendMessageWithRetry(MqttMessage message) {
+        int maxRetries = message.isHighPriority() ? 5 : 3; // 高优先级消息重试更多次
+        
+        for (int attempt = 1; attempt <= maxRetries; attempt++) {
+            try {
+                Message<String> springMessage = MessageBuilder
+                        .withPayload(message.getPayload())
+                        .setHeader(MqttHeaders.TOPIC, message.getTopic())
+                        .setHeader(MqttHeaders.QOS, message.getQos())
+                        .setHeader(MqttHeaders.RETAINED, message.isRetain())
+                        .build();
+
+                mqttOutbound.handleMessage(springMessage);
+                return true;
+                
+            } catch (Exception e) {
+                if (attempt < maxRetries) {
+                    try {
+                        // 指数退避重试
+                        long delay = Math.min(1000L * (1L << (attempt - 1)), 10000L);
+                        Thread.sleep(delay);
+                    } catch (InterruptedException ie) {
+                        Thread.currentThread().interrupt();
+                        break;
+                    }
+                } else {
+                    log.error("Failed to send message after {} attempts: topic={}, error={}", 
+                            maxRetries, message.getTopic(), e.getMessage());
+                }
+            }
+        }
+        return false;
+    }
+
+    /**
+     * 获取队列状态信息
+     */
+    public String getQueueStatus() {
+        return String.format("Queue Status - HighPriority: %d, Normal: %d, DeadLetter: %d, Total: %d, Sent: %d, Failed: %d, Buffered: %d",
+                highPriorityQueue.size(), normalPriorityQueue.size(), deadLetterQueue.size(),
+                totalMessages.get(), sentMessages.get(), failedMessages.get(), bufferedMessages.get());
+    }
+
+    /**
+     * 定时输出队列状态
+     */
+    @Scheduled(fixedRate = 30000) // 每30秒输出一次状态
+    public void logQueueStatus() {
+        if (log.isInfoEnabled()) {
+            log.info(getQueueStatus());
+        }
+    }
+
+    /**
+     * MQTT消息内部类
+     */
+    private static class MqttMessage {
+        private final String topic;
+        private final String payload;
+        private final int qos;
+        private final boolean retain;
+        private final long timestamp;
+        private final boolean highPriority;
+
+        public MqttMessage(String topic, String payload, int qos, boolean retain, long timestamp, boolean highPriority) {
+            this.topic = topic;
+            this.payload = payload;
+            this.qos = qos;
+            this.retain = retain;
+            this.timestamp = timestamp;
+            this.highPriority = highPriority;
+        }
+
+        // Getters
+        public String getTopic() { return topic; }
+        public String getPayload() { return payload; }
+        public int getQos() { return qos; }
+        public boolean isRetain() { return retain; }
+        public long getTimestamp() { return timestamp; }
+        public boolean isHighPriority() { return highPriority; }
+    }
+} 

device-service-server/src/main/java/com/hfln/device/server/DeviceApplication.java

@@ -3,6 +3,7 @@ package com.hfln.device.server;
 import org.mybatis.spring.annotation.MapperScan;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
 import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
 import org.springframework.cloud.openfeign.EnableFeignClients;
 import org.springframework.context.annotation.ComponentScan;
@@ -16,7 +17,8 @@ import org.springframework.scheduling.annotation.EnableScheduling;
 @MapperScan("com.hfln.device.infrastructure.mapper")
 @EnableFeignClients(basePackages = "com.hfln.device.client.api")
 @SpringBootApplication(
-    scanBasePackages = {"com.hfln.device.**"}
+    scanBasePackages = {"com.hfln.device.**"},
+    exclude = {RedisAutoConfiguration.class}
 )
 @ComponentScan(
     basePackages = {"com.hfln.device.**"}