mqtt消息记录开发完成

2025-05-23 11:03:42 +08:00
parent abf14bd42f
commit 49f1220310
27 changed files with 1787 additions and 66 deletions
--- a/springboot-init-main/doc/mqtt_communication_log_plan.md
+++ b/springboot-init-main/doc/mqtt_communication_log_plan.md
@@ -0,0 +1,283 @@
+# MQTT通信日志系统开发方案
+
+## 1. 项目概述
+
+### 1.1 目标与意义
+
+开发MQTT通信日志系统，用于完整记录单片机与服务器系统之间的所有MQTT消息通信。该系统具有以下价值：
+
+- **通信可视化**：直观展示单片机与服务器之间的通信细节和消息内容
+- **故障诊断**：为消息处理失败、设备离线等异常场景提供完整的调试信息
+- **性能分析**：通过日志记录消息量、响应时间等指标，评估系统性能
+- **安全审计**：记录所有通信活动，便于安全审计和问题追踪
+- **业务分析**：通过通信数据分析设备使用模式和业务流程
+
+### 1.2 系统定位
+
+该系统为纯日志记录系统，**不干扰**正常的业务流程：
+
+- 不消费消息队列中的数据
+- 不修改消息内容或状态
+- 仅作为"旁路监听者"记录通信过程
+- 以非阻塞方式运行，不影响主业务性能
+
+## 2. 技术架构设计
+
+### 2.1 总体架构
+
+![MQTT日志系统架构](https://placeholder-for-architecture-diagram.com)
+
+系统采用"监听者模式"，不直接参与消息处理流程：
+
+- **MQTT消息监听层**：订阅与主业务相同的Topic，但只读取不处理
+- **日志记录层**：异步将消息内容写入数据库
+- **查询展示层**：提供日志查询、过滤、导出功能的管理界面
+
+### 2.2 数据库结构
+
+已有数据表 `mqtt_communication_log` 包含了足够的字段设计：
+
+```sql
+CREATE TABLE `mqtt_communication_log` (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '日志ID',
+  `message_id` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '消息的唯一标识',
+  `direction` varchar(10) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '消息方向: UPSTREAM (设备->服务器), DOWNSTREAM (服务器->设备)',
+  `client_id` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '相关的客户端ID',
+  `topic` varchar(512) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT 'MQTT 主题',
+  `payload_format` varchar(20) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT 'TEXT' COMMENT 'Payload 格式',
+  `payload` mediumtext CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL COMMENT '消息原文',
+  `qos` tinyint(4) NULL DEFAULT NULL COMMENT '消息QoS级别',
+  `is_retained` tinyint(1) NULL DEFAULT NULL COMMENT '是否为保留消息',
+  `log_timestamp` timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '日志记录时间戳',
+  `backend_processing_status` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '后端处理状态',
+  `backend_processing_info` text CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL COMMENT '后端处理附加信息',
+  `related_session_id` bigint(20) NULL DEFAULT NULL COMMENT '关联的充电会话ID',
+  `related_task_id` bigint(20) NULL DEFAULT NULL COMMENT '关联的机器人任务ID',
+  PRIMARY KEY (`id`) USING BTREE,
+  INDEX `idx_log_timestamp`(`log_timestamp`) USING BTREE,
+  INDEX `idx_topic`(`topic`(255)) USING BTREE,
+  INDEX `idx_client_id`(`client_id`) USING BTREE,
+  INDEX `idx_related_session_id`(`related_session_id`) USING BTREE,
+  INDEX `idx_related_task_id`(`related_task_id`) USING BTREE
+) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = 'MQTT通信日志表'
+```
+
+### 2.3 核心组件设计
+
+#### 2.3.1 MQTT日志监听器
+
+```java
+@Component
+public class MqttCommunicationLogger {
+    // 依赖注入日志服务和MQTT客户端
+}
+```
+
+#### 2.3.2 日志记录服务
+
+```java
+@Service
+public class MqttLogService {
+    // 异步记录消息到数据库
+}
+```
+
+#### 2.3.3 日志查询Controller
+
+```java
+@RestController
+@RequestMapping("/api/admin/mqtt-logs")
+public class MqttLogController {
+    // 提供日志查询API
+}
+```
+
+## 3. 实现步骤
+
+### 3.1 后端实现
+
+1. **创建数据访问层**
+   - 实现 `MqttCommunicationLogMapper` 接口
+   - 实现 `MqttCommunicationLogService` 接口及其实现类
+
+2. **实现消息监听与记录**
+   - 创建独立的 MQTT 客户端用于日志记录
+   - 配置该客户端订阅所有业务相关的主题
+   - 实现监听回调，异步记录收到的消息
+   - 为发送的消息添加拦截记录逻辑
+
+3. **实现关联解析**
+   - 解析消息内容，提取关联的会话ID、任务ID
+   - 根据消息主题和内容推断消息类型和处理状态
+
+4. **实现管理API**
+   - 创建日志查询Controller
+   - 实现分页查询、条件过滤、时间范围查询
+   - 实现日志导出功能
+
+### 3.2 前端实现
+
+1. **日志查询页面**
+   - 创建管理员专用的日志查询页面
+   - 实现条件筛选表单（时间范围、客户端ID、主题等）
+   - 实现分页表格展示日志记录
+
+2. **日志详情展示**
+   - 实现日志详情模态框
+   - 格式化JSON消息内容，便于阅读
+   - 显示相关的业务信息（会话、任务）链接
+
+3. **日志分析功能**
+   - 实现简单的统计分析（消息量、成功率等）
+   - 提供日志导出功能
+
+## 4. 技术实现细节
+
+### 4.1 MQTT日志客户端配置
+
+```java
+@Configuration
+public class MqttLoggerConfig {
+    @Bean(name = "mqttLoggerClient")
+    public MqttClient mqttLoggerClient() {
+        // 配置单独的MQTT客户端用于日志记录
+        // 与主业务使用相同的broker但不同的clientId
+    }
+}
+```
+
+### 4.2 消息记录实现
+
+```java
+@Component
+public class MqttLogSubscriber implements MqttCallback {
+    
+    @Async
+    public void messageArrived(String topic, MqttMessage message) {
+        // 1. 解析消息，提取元数据
+        // 2. 构建日志记录对象
+        // 3. 异步保存到数据库
+        // 注意：完全不干扰消息的正常传递
+    }
+}
+```
+
+### 4.3 发送消息记录
+
+```java
+@Aspect
+@Component
+public class MqttPublishAspect {
+    
+    @Around("execution(* com.yupi.project.mqtt.MqttService.publish*(..)) || execution(* com.yupi.project.mqtt.MqttService.sendCommand*(..))")
+    public Object logMqttPublish(ProceedingJoinPoint joinPoint) throws Throwable {
+        // 前置：记录发送前信息
+        Object result = joinPoint.proceed(); // 执行原方法
+        // 后置：记录发送后信息
+        return result;
+    }
+}
+```
+
+### 4.4 业务关联解析
+
+```java
+@Component
+public class MqttPayloadParser {
+    
+    public MqttLogBusinessInfo parsePayload(String topic, String payload) {
+        // 根据主题和内容解析出业务相关信息
+        // 例如：会话ID、任务ID、指令类型等
+    }
+}
+```
+
+## 5. 部署与配置
+
+### 5.1 配置项
+
+在 `application.yml` 中添加MQTT日志系统的配置：
+
+```yaml
+mqtt:
+  logger:
+    enabled: true
+    client-id-prefix: logger-
+    topics: 
+      - charging/spot/+/status
+      - charging/spot/+/command/#
+      - charging/spot/+/heartbeat
+    async:
+      core-pool-size: 2
+      max-pool-size: 5
+      queue-capacity: 500
+    retention:
+      days: 30  # 日志保留天数
+```
+
+### 5.2 数据库维护
+
+- 设计日志清理策略，避免数据过度膨胀
+- 实现定时任务，清理超过保留期的日志数据
+- 考虑按时间分区表策略，优化大量日志的存储和查询性能
+
+## 6. 测试计划
+
+### 6.1 功能测试
+
+- 验证各类消息的正确记录
+- 验证消息内容完整性
+- 验证不同格式消息的解析效果
+- 验证业务关联信息的正确提取
+
+### 6.2 性能测试
+
+- 高并发消息下的记录性能
+- 验证日志记录对主业务的性能影响（应小于5%）
+- 验证大数据量下的查询性能
+
+### 6.3 稳定性测试
+
+- 长时间运行测试
+- 异常情况下的日志记录可靠性测试
+- 系统重启后的连续性测试
+
+## 7. 注意事项与风险
+
+### 7.1 技术风险
+
+- **性能影响**：确保日志记录不成为系统瓶颈
+  - 解决方案：异步记录，独立线程池，必要时考虑消息缓冲
+  
+- **存储膨胀**：MQTT消息量大可能导致数据库快速增长
+  - 解决方案：设置合理的日志保留策略，定期清理或归档
+
+- **消息完整性**：某些异常情况可能导致日志不完整
+  - 解决方案：增加重试机制，添加监控告警
+
+### 7.2 业务风险
+
+- **敏感信息**：日志可能包含敏感业务数据
+  - 解决方案：实现字段级别的脱敏，严格控制日志访问权限
+
+- **一致性**：日志状态与实际业务状态可能不一致
+  - 解决方案：明确标注日志仅用于参考，不作为业务状态的权威来源
+
+## 8. 后续优化方向
+
+- 实现更复杂的消息分析功能
+- 添加可视化图表，直观展示通信模式和趋势
+- 实现异常模式的自动检测和告警
+- 与系统监控平台集成，提供更全面的系统健康视图
+
+## 9. 项目计划
+
+| 阶段 | 内容 | 时间估计 |
+|------|------|----------|
+| 设计与准备 | 详细设计、环境准备 | 2天 |
+| 后端开发 | 核心记录功能实现 | 3天 |
+| 前端开发 | 日志查询界面实现 | 3天 |
+| 测试与优化 | 功能测试、性能优化 | 2天 |
+| 文档与部署 | 编写文档、系统部署 | 1天 |
+
+**总计划时间**：约11个工作日 
--- a/springboot-init-main/sql/mqtt_power.sql
+++ b/springboot-init-main/sql/mqtt_power.sql
@@ -1,17 +1,17 @@
 /*
- Navicat Premium Data Transfer
+ Navicat Premium Dump SQL

- Source Server         : yuyun
+ Source Server         : yuyun-us1.stormrain.cn
 Source Server Type    : MySQL
- Source Server Version : 50744
+ Source Server Version : 50744 (5.7.44)
 Source Host           : yuyun-us1.stormrain.cn:3306
 Source Schema         : mqtt_power

 Target Server Type    : MySQL
- Target Server Version : 50744
+ Target Server Version : 50744 (5.7.44)
 File Encoding         : 65001

- Date: 18/05/2025 19:54:49
+ Date: 22/05/2025 19:36:58
 */

 SET NAMES utf8mb4;
@@ -89,6 +89,33 @@ CREATE TABLE `charging_session`  (
  PRIMARY KEY (`id`) USING BTREE
 ) ENGINE = InnoDB AUTO_INCREMENT = 8 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_german2_ci COMMENT = '充电记录表' ROW_FORMAT = Dynamic;

+-- ----------------------------
+-- Table structure for mqtt_communication_log
+-- ----------------------------
+DROP TABLE IF EXISTS `mqtt_communication_log`;
+CREATE TABLE `mqtt_communication_log`  (
+  `id` bigint(20) NOT NULL AUTO_INCREMENT COMMENT '日志ID',
+  `message_id` varchar(128) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '消息的唯一标识 (例如 MQTT v5 的 Message ID 或应用生成的UUID)',
+  `direction` varchar(10) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT '消息方向: UPSTREAM (设备->服务器), DOWNSTREAM (服务器->设备)',
+  `client_id` varchar(255) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '相关的客户端ID',
+  `topic` varchar(512) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NOT NULL COMMENT 'MQTT 主题',
+  `payload_format` varchar(20) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT 'TEXT' COMMENT 'Payload 格式 (TEXT, JSON, BINARY)',
+  `payload` mediumtext CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL COMMENT '消息原文 (Payload)',
+  `qos` tinyint(4) NULL DEFAULT NULL COMMENT '消息QoS级别',
+  `is_retained` tinyint(1) NULL DEFAULT NULL COMMENT '是否为保留消息',
+  `log_timestamp` timestamp(3) NOT NULL DEFAULT CURRENT_TIMESTAMP(3) COMMENT '日志记录时间戳 (精确到毫秒)',
+  `backend_processing_status` varchar(50) CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL DEFAULT NULL COMMENT '后端处理状态 (RECEIVED, PROCESSING, SUCCESS, FAILED)',
+  `backend_processing_info` text CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci NULL COMMENT '后端处理附加信息 (如错误信息、关联业务ID)',
+  `related_session_id` bigint(20) NULL DEFAULT NULL COMMENT '关联的充电会话ID (如果适用)',
+  `related_task_id` bigint(20) NULL DEFAULT NULL COMMENT '关联的机器人任务ID (如果适用)',
+  PRIMARY KEY (`id`) USING BTREE,
+  INDEX `idx_log_timestamp`(`log_timestamp`) USING BTREE,
+  INDEX `idx_topic`(`topic`(255)) USING BTREE,
+  INDEX `idx_client_id`(`client_id`) USING BTREE,
+  INDEX `idx_related_session_id`(`related_session_id`) USING BTREE,
+  INDEX `idx_related_task_id`(`related_task_id`) USING BTREE
+) ENGINE = InnoDB AUTO_INCREMENT = 1 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_unicode_ci COMMENT = 'MQTT通信日志表' ROW_FORMAT = Dynamic;
+
 -- ----------------------------
 -- Table structure for parking_spot
 -- ----------------------------
--- a/springboot-init-main/src/main/java/com/yupi/project/aop/MqttPublishLogAspect.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/aop/MqttPublishLogAspect.java
@@ -0,0 +1,105 @@
+package com.yupi.project.aop;
+
+import com.yupi.project.config.properties.MqttLoggerProperties;
+import com.yupi.project.model.entity.RobotTask;
+import com.yupi.project.service.MqttCommunicationLogService;
+import com.yupi.project.service.RobotTaskService;
+import lombok.extern.slf4j.Slf4j;
+import org.aspectj.lang.ProceedingJoinPoint;
+import org.aspectj.lang.annotation.Around;
+import org.aspectj.lang.annotation.Aspect;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+
+import java.nio.charset.StandardCharsets;
+import java.util.UUID;
+
+@Aspect
+@Component
+@Slf4j
+public class MqttPublishLogAspect {
+
+    private final MqttCommunicationLogService logService;
+    private final MqttLoggerProperties loggerProperties;
+    private final RobotTaskService robotTaskService;
+    private final MqttClient mainMqttClient;
+
+    @Autowired
+    public MqttPublishLogAspect(MqttCommunicationLogService logService,
+                                MqttLoggerProperties loggerProperties,
+                                RobotTaskService robotTaskService,
+                                @Qualifier("mqttClientBean") MqttClient mainMqttClient) {
+        this.logService = logService;
+        this.loggerProperties = loggerProperties;
+        this.robotTaskService = robotTaskService;
+        this.mainMqttClient = mainMqttClient;
+    }
+
+    @Around("execution(* com.yupi.project.service.impl.MqttServiceImpl.sendCommand(..)) || execution(* com.yupi.project.service.MqttService.sendCommand(..))")
+    public Object logMqttSendCommand(ProceedingJoinPoint joinPoint) throws Throwable {
+        if (!loggerProperties.isEnabled()) {
+            return joinPoint.proceed();
+        }
+
+        Object[] args = joinPoint.getArgs();
+        String robotIdArg = null;
+        String commandTypeArg = null;
+        String payloadJsonArg = null;
+        Long sessionIdArg = null;
+        Long taskIdForLog = null;
+
+        if (args.length >= 3) {
+            robotIdArg = (String) args[0];
+            commandTypeArg = args[1].toString();
+            payloadJsonArg = (String) args[2];
+            if (args.length >= 4 && args[3] instanceof Long) {
+                 sessionIdArg = (Long) args[3];
+            }
+        }
+        
+        String topic = null;
+        Integer qos = null;
+        Boolean retained = false;
+        String messageIdForLog = UUID.randomUUID().toString();
+
+        Object result = null;
+        try {
+            result = joinPoint.proceed();
+        } catch (Throwable throwable) {
+            logService.asyncLogDownstreamMessage(
+                    "UNKNOWN_TOPIC_PRE_SEND_FAILURE",
+                    payloadJsonArg,
+                    qos,
+                    retained,
+                    mainMqttClient.getClientId(),
+                    messageIdForLog,
+                    sessionIdArg,
+                    null
+            );
+            log.error("Exception during sendCommand, logged with placeholder data. RobotId: {}, Command: {}", robotIdArg, commandTypeArg, throwable);
+            throw throwable;
+        }
+
+        if (robotIdArg != null && sessionIdArg !=null) {
+            RobotTask latestTask = robotTaskService.findLatestTaskByRobotIdAndSessionId(robotIdArg,sessionIdArg);
+            if(latestTask != null){
+                taskIdForLog = latestTask.getId();
+            }
+        }
+
+        logService.asyncLogDownstreamMessage(
+                args.length > 0 ? ("CMD_TO_" + args[0]) : "UNKNOWN_TOPIC_POST_SEND",
+                payloadJsonArg,
+                1,
+                false,
+                mainMqttClient.getClientId(),
+                messageIdForLog, 
+                sessionIdArg,
+                taskIdForLog 
+        );
+
+        return result;
+    }
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/config/MqttLoggerAsyncConfigurer.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/config/MqttLoggerAsyncConfigurer.java
@@ -0,0 +1,39 @@
+package com.yupi.project.config;
+
+import com.yupi.project.config.properties.MqttLoggerProperties;
+import lombok.RequiredArgsConstructor;
+import org.springframework.aop.interceptor.AsyncUncaughtExceptionHandler;
+import org.springframework.aop.interceptor.SimpleAsyncUncaughtExceptionHandler;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.AsyncConfigurer;
+import org.springframework.scheduling.annotation.EnableAsync;
+import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
+
+import java.util.concurrent.Executor;
+
+@Configuration
+@EnableAsync
+@RequiredArgsConstructor
+public class MqttLoggerAsyncConfigurer implements AsyncConfigurer {
+
+    private final MqttLoggerProperties mqttLoggerProperties;
+
+    @Override
+    @Bean(name = "mqttLoggerThreadPoolTaskExecutor")
+    public Executor getAsyncExecutor() {
+        ThreadPoolTaskExecutor executor = new ThreadPoolTaskExecutor();
+        MqttLoggerProperties.Async asyncProps = mqttLoggerProperties.getAsync();
+        executor.setCorePoolSize(asyncProps.getCorePoolSize());
+        executor.setMaxPoolSize(asyncProps.getMaxPoolSize());
+        executor.setQueueCapacity(asyncProps.getQueueCapacity());
+        executor.setThreadNamePrefix(asyncProps.getThreadNamePrefix());
+        executor.initialize();
+        return executor;
+    }
+
+    @Override
+    public AsyncUncaughtExceptionHandler getAsyncUncaughtExceptionHandler() {
+        return new SimpleAsyncUncaughtExceptionHandler();
+    }
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/config/MqttLoggerClientConfig.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/config/MqttLoggerClientConfig.java
@@ -0,0 +1,45 @@
+package com.yupi.project.config;
+
+import com.yupi.project.config.properties.MqttLoggerProperties;
+import com.yupi.project.config.properties.MqttProperties;
+import com.yupi.project.mqtt.MqttLoggerCallbackHandler;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Lazy;
+
+import java.util.UUID;
+
+@Slf4j
+@Configuration
+@RequiredArgsConstructor
+public class MqttLoggerClientConfig {
+
+    private final MqttProperties mainMqttProperties;
+    private final MqttLoggerProperties mqttLoggerProperties;
+    private final MqttConnectOptions mqttConnectOptions;
+
+    @Bean(name = "mqttLoggerClient")
+    @Lazy
+    public MqttClient mqttLoggerClient(@Qualifier("mqttLoggerCallbackHandler") MqttLoggerCallbackHandler loggerCallbackHandler)
+            throws MqttException {
+        if (!mqttLoggerProperties.isEnabled()) {
+            log.info("MQTT Logger Client is disabled via configuration.");
+            return null;
+        }
+
+        String clientId = mqttLoggerProperties.getClientIdPrefix() + UUID.randomUUID().toString().replace("-", "");
+        log.info("Initializing MQTT Logger Client with Broker URL: {} and Client ID: {}", mainMqttProperties.getBrokerUrl(), clientId);
+
+        MqttClient client = new MqttClient(mainMqttProperties.getBrokerUrl(), clientId, new MemoryPersistence());
+        client.setCallback(loggerCallbackHandler);
+        loggerCallbackHandler.setMqttClient(client);
+        return client;
+    }
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/config/properties/MqttLoggerProperties.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/config/properties/MqttLoggerProperties.java
@@ -0,0 +1,32 @@
+package com.yupi.project.config.properties;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+@Data
+@Component
+@ConfigurationProperties(prefix = "mqtt.logger")
+public class MqttLoggerProperties {
+
+    private boolean enabled = false;
+    private String clientIdPrefix;
+    private List<String> topics;
+    private Async async = new Async();
+    private Retention retention = new Retention();
+
+    @Data
+    public static class Async {
+        private int corePoolSize = 1;
+        private int maxPoolSize = 5;
+        private int queueCapacity = 100;
+        private String threadNamePrefix = "mqtt-log-async-";
+    }
+
+    @Data
+    public static class Retention {
+        private int days = 30;
+    }
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/controller/MqttCommunicationLogController.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/controller/MqttCommunicationLogController.java
@@ -0,0 +1,114 @@
+package com.yupi.project.controller;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import com.yupi.project.common.BaseResponse;
+import com.yupi.project.common.ErrorCode;
+import com.yupi.project.common.ResultUtils;
+import com.yupi.project.exception.BusinessException;
+import com.yupi.project.model.dto.mqttlog.MqttLogQueryRequest;
+import com.yupi.project.model.entity.MqttCommunicationLog;
+import com.yupi.project.service.MqttCommunicationLogService;
+import com.yupi.project.annotation.AuthCheck;
+import com.yupi.project.constant.UserConstant;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.web.bind.annotation.*;
+
+import javax.servlet.http.HttpServletRequest;
+import java.util.Date;
+
+@RestController
+@RequestMapping("/admin/mqtt-log")
+@RequiredArgsConstructor
+@Slf4j
+public class MqttCommunicationLogController {
+
+    private final MqttCommunicationLogService mqttCommunicationLogService;
+
+    /**
+     * 分页获取MQTT通信日志列表
+     *
+     * @param mqttLogQueryRequest
+     * @param request
+     * @return
+     */
+    @PostMapping("/list/page")
+    @AuthCheck(mustRole = UserConstant.ADMIN_ROLE)
+    public BaseResponse<Page<MqttCommunicationLog>> listMqttCommunicationLogsByPage(@RequestBody MqttLogQueryRequest mqttLogQueryRequest, HttpServletRequest request) {
+        if (mqttLogQueryRequest == null) {
+            throw new BusinessException(ErrorCode.PARAMS_ERROR);
+        }
+        long current = mqttLogQueryRequest.getCurrent();
+        long size = mqttLogQueryRequest.getPageSize();
+        // 限制爬虫
+        if (size > 50) {
+            throw new BusinessException(ErrorCode.PARAMS_ERROR, "请求数据量过大");
+        }
+
+        QueryWrapper<MqttCommunicationLog> queryWrapper = getQueryWrapper(mqttLogQueryRequest);
+        Page<MqttCommunicationLog> logPage = mqttCommunicationLogService.page(new Page<>(current, size), queryWrapper);
+        return ResultUtils.success(logPage);
+    }
+
+    private QueryWrapper<MqttCommunicationLog> getQueryWrapper(MqttLogQueryRequest queryRequest) {
+        QueryWrapper<MqttCommunicationLog> queryWrapper = new QueryWrapper<>();
+        if (queryRequest == null) {
+            return queryWrapper;
+        }
+
+        String messageId = queryRequest.getMessageId();
+        String direction = queryRequest.getDirection();
+        String clientId = queryRequest.getClientId();
+        String topic = queryRequest.getTopic();
+        String payloadContains = queryRequest.getPayloadContains();
+        Integer qos = queryRequest.getQos();
+        Date startTime = queryRequest.getStartTime();
+        Date endTime = queryRequest.getEndTime();
+        Long relatedSessionId = queryRequest.getRelatedSessionId();
+        Long relatedTaskId = queryRequest.getRelatedTaskId();
+        String sortField = queryRequest.getSortField();
+        String sortOrder = queryRequest.getSortOrder();
+
+        if (StringUtils.isNotBlank(messageId)) {
+            queryWrapper.like("message_id", messageId);
+        }
+        if (StringUtils.isNotBlank(direction)) {
+            queryWrapper.eq("direction", direction);
+        }
+        if (StringUtils.isNotBlank(clientId)) {
+            queryWrapper.like("client_id", clientId);
+        }
+        if (StringUtils.isNotBlank(topic)) {
+            queryWrapper.like("topic", topic);
+        }
+        if (StringUtils.isNotBlank(payloadContains)) {
+            queryWrapper.like("payload", payloadContains);
+        }
+        if (qos != null) {
+            queryWrapper.eq("qos", qos);
+        }
+        if (startTime != null) {
+            queryWrapper.ge("log_timestamp", startTime);
+        }
+        if (endTime != null) {
+            queryWrapper.le("log_timestamp", endTime);
+        }
+        if (relatedSessionId != null) {
+            queryWrapper.eq("related_session_id", relatedSessionId);
+        }
+        if (relatedTaskId != null) {
+            queryWrapper.eq("related_task_id", relatedTaskId);
+        }
+
+        // 默认按日志时间降序排序
+        if (StringUtils.isNotBlank(sortField)) {
+            boolean isAsc = "asc".equalsIgnoreCase(sortOrder);
+            queryWrapper.orderBy(true, isAsc, sortField);
+        } else {
+            queryWrapper.orderByDesc("log_timestamp");
+        }
+        return queryWrapper;
+    }
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/mapper/MqttCommunicationLogMapper.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/mapper/MqttCommunicationLogMapper.java
@@ -0,0 +1,7 @@
+package com.yupi.project.mapper;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import com.yupi.project.model.entity.MqttCommunicationLog;
+
+public interface MqttCommunicationLogMapper extends BaseMapper<MqttCommunicationLog> {
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/model/dto/mqttlog/MqttLogQueryRequest.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/model/dto/mqttlog/MqttLogQueryRequest.java
@@ -0,0 +1,32 @@
+package com.yupi.project.model.dto.mqttlog;
+
+import com.yupi.project.common.PageRequest;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import org.springframework.format.annotation.DateTimeFormat;
+
+import java.io.Serializable;
+import java.util.Date;
+
+@EqualsAndHashCode(callSuper = true)
+@Data
+public class MqttLogQueryRequest extends PageRequest implements Serializable {
+
+    private String messageId;
+    private String direction;
+    private String clientId;
+    private String topic;
+    private String payloadContains;
+    private Integer qos;
+
+    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    private Date startTime;
+
+    @DateTimeFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    private Date endTime;
+
+    private Long relatedSessionId;
+    private Long relatedTaskId;
+
+    private static final long serialVersionUID = 1L;
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/model/entity/MqttCommunicationLog.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/model/entity/MqttCommunicationLog.java
@@ -0,0 +1,46 @@
+package com.yupi.project.model.entity;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+import java.io.Serializable;
+import java.util.Date;
+
+@TableName(value = "mqtt_communication_log")
+@Data
+public class MqttCommunicationLog implements Serializable {
+    @TableId(type = IdType.AUTO)
+    private Long id;
+
+    private String messageId;
+
+    private String direction; // UPSTREAM, DOWNSTREAM
+
+    private String clientId;
+
+    private String topic;
+
+    private String payloadFormat; // TEXT, JSON, BINARY
+
+    private String payload;
+
+    private Integer qos;
+
+    private Boolean isRetained;
+
+    private Date logTimestamp;
+
+    private String backendProcessingStatus; // RECEIVED, PROCESSING, SUCCESS, FAILED
+
+    private String backendProcessingInfo;
+
+    private Long relatedSessionId;
+
+    private Long relatedTaskId;
+
+    @TableField(exist = false)
+    private static final long serialVersionUID = 1L;
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/mqtt/MqttConnectionManager.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/mqtt/MqttConnectionManager.java
@@ -1,24 +1,31 @@
 package com.yupi.project.mqtt;

-import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.context.ApplicationListener;
 import org.springframework.context.event.ContextRefreshedEvent;
 import org.springframework.stereotype.Component;

@Slf4j
@Component
-@RequiredArgsConstructor
 public class MqttConnectionManager implements ApplicationListener<ContextRefreshedEvent>, DisposableBean {

    private final MqttClient mqttClient;
    private final MqttConnectOptions mqttConnectOptions;
    // private final MqttProperties mqttProperties; // Injected if needed for logging brokerUrl, or get from mqttClient.getServerURI()

+    @Autowired
+    public MqttConnectionManager(@Qualifier("mqttClientBean") MqttClient mqttClient, 
+                                 MqttConnectOptions mqttConnectOptions) {
+        this.mqttClient = mqttClient;
+        this.mqttConnectOptions = mqttConnectOptions;
+    }
+
    @Override
    public void onApplicationEvent(ContextRefreshedEvent event) {
        // Ensure this logic runs only once, for the root application context
--- a/springboot-init-main/src/main/java/com/yupi/project/mqtt/MqttLoggerCallbackHandler.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/mqtt/MqttLoggerCallbackHandler.java
@@ -0,0 +1,91 @@
+package com.yupi.project.mqtt;
+
+import com.yupi.project.config.properties.MqttLoggerProperties;
+import com.yupi.project.config.properties.MqttProperties; // 主业务的 properties，获取QoS等
+import com.yupi.project.service.MqttCommunicationLogService;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
+import org.eclipse.paho.client.mqttv3.MqttCallbackExtended;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+import java.util.UUID;
+
+@Slf4j
+@Component("mqttLoggerCallbackHandler") // 指定Bean名称
+public class MqttLoggerCallbackHandler implements MqttCallbackExtended {
+
+    private final MqttCommunicationLogService logService;
+    private final MqttLoggerProperties loggerProperties;
+    private final MqttProperties mainMqttProperties; // 用于获取默认QoS等
+    private MqttClient mqttLoggerClient; // 通过setter注入
+
+    public MqttLoggerCallbackHandler(MqttCommunicationLogService logService,
+                                     MqttLoggerProperties loggerProperties,
+                                     MqttProperties mainMqttProperties) {
+        this.logService = logService;
+        this.loggerProperties = loggerProperties;
+        this.mainMqttProperties = mainMqttProperties;
+    }
+
+    public void setMqttClient(@Qualifier("mqttLoggerClient") MqttClient mqttLoggerClient) {
+        this.mqttLoggerClient = mqttLoggerClient;
+    }
+
+    @Override
+    public void connectComplete(boolean reconnect, String serverURI) {
+        log.info("MQTT Logger Client connection {} to broker: {}", reconnect ? "re-established" : "established", serverURI);
+        if (!loggerProperties.isEnabled() || mqttLoggerClient == null) {
+            log.warn("MQTT Logger is disabled or client not initialized. Cannot subscribe to topics for logging.");
+            return;
+        }
+        try {
+            List<String> topicsToLog = loggerProperties.getTopics();
+            if (topicsToLog != null && !topicsToLog.isEmpty()) {
+                for (String topicFilter : topicsToLog) {
+                    // 使用主配置的QoS等级进行订阅，或者在loggerProperties中单独配置
+                    mqttLoggerClient.subscribe(topicFilter, mainMqttProperties.getDefaultQos());
+                    log.info("MQTT Logger Client subscribed to topic: {} for logging.", topicFilter);
+                }
+            } else {
+                log.warn("No topics configured for MQTT Logger Client to subscribe.");
+            }
+        } catch (Exception e) {
+            log.error("Error subscribing MQTT Logger Client to topics: ", e);
+        }
+    }
+
+    @Override
+    public void connectionLost(Throwable cause) {
+        log.error("MQTT Logger Client connection lost!", cause);
+    }
+
+    @Override
+    public void messageArrived(String topic, MqttMessage message) throws Exception {
+        if (!loggerProperties.isEnabled()) {
+            return; // 如果日志功能被禁用，则不处理
+        }
+        // 异步记录上行消息 (UPSTREAM)
+        // clientId 可以从 mqttLoggerClient.getClientId() 获取，表示是日志客户端收到的
+        // messageIdStr可以尝试从 MQTTv5 属性获取，如果使用的是 MQTTv3，Paho 的 message.getId() 是内部ID，对于 broker 可能无意义
+        // 对于日志记录，如果 message.getId() > 0，可以用它，否则生成一个UUID
+        String messageIdStr = message.getId() > 0 ? String.valueOf(message.getId()) : UUID.randomUUID().toString();
+        logService.asyncLogMessage(topic, message, "UPSTREAM", mqttLoggerClient.getClientId(), messageIdStr, null, null, null, null);
+    }
+
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken token) {
+        // Logger client typically does not publish, so this might not be very relevant
+        // unless it's used for some diagnostic publishing.
+        try {
+            if (token != null && token.isComplete() && token.getMessage() != null) {
+                log.trace("MQTT Logger Client: Delivery complete for message ID: {}", token.getMessageId());
+            }
+        } catch (Exception e) {
+            log.error("Error in MQTT Logger Client deliveryComplete: ", e);
+        }
+    }
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/mqtt/MqttLoggerConnectionManager.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/mqtt/MqttLoggerConnectionManager.java
@@ -0,0 +1,76 @@
+package com.yupi.project.mqtt;
+
+import com.yupi.project.config.properties.MqttLoggerProperties;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.springframework.beans.factory.DisposableBean;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.ContextRefreshedEvent;
+import org.springframework.stereotype.Component;
+
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class MqttLoggerConnectionManager implements ApplicationListener<ContextRefreshedEvent>, DisposableBean {
+
+    @Qualifier("mqttLoggerClient")
+    private final MqttClient mqttLoggerClient; // 注入日志专用客户端
+    private final MqttConnectOptions mqttConnectOptions; // 可复用主连接配置
+    private final MqttLoggerProperties mqttLoggerProperties;
+
+    @Override
+    public void onApplicationEvent(ContextRefreshedEvent event) {
+        if (event.getApplicationContext().getParent() == null) { // Ensure root context
+            if (mqttLoggerProperties.isEnabled() && mqttLoggerClient != null) {
+                connectToMqtt();
+            }
+        }
+    }
+
+    private void connectToMqtt() {
+        try {
+            if (!mqttLoggerClient.isConnected()) {
+                log.info("Attempting to connect MQTT Logger Client: {} to broker: {}", mqttLoggerClient.getClientId(), mqttLoggerClient.getServerURI());
+                mqttLoggerClient.connect(mqttConnectOptions);
+                // Subscription logic is handled by MqttLoggerCallbackHandler.connectComplete
+            } else {
+                log.info("MQTT Logger Client {} is already connected.", mqttLoggerClient.getClientId());
+            }
+        } catch (MqttException e) {
+            log.error("Error connecting MQTT Logger Client: ", e);
+        } catch (Exception e) {
+            log.error("Unexpected error during MQTT Logger Client connection: ", e);
+        }
+    }
+
+    @Override
+    public void destroy() throws Exception {
+        if (mqttLoggerProperties.isEnabled() && mqttLoggerClient != null) {
+            disconnectFromMqtt();
+        }
+    }
+
+    private void disconnectFromMqtt() {
+        try {
+            if (mqttLoggerClient.isConnected()) {
+                log.info("Disconnecting MQTT Logger Client: {} from broker: {}", mqttLoggerClient.getClientId(), mqttLoggerClient.getServerURI());
+                mqttLoggerClient.disconnect();
+                log.info("MQTT Logger Client {} disconnected successfully.", mqttLoggerClient.getClientId());
+            }
+        } catch (MqttException e) {
+            log.error("Error disconnecting MQTT Logger Client: ", e);
+        } finally {
+            try {
+                log.info("Closing MQTT Logger Client: {}", mqttLoggerClient.getClientId());
+                mqttLoggerClient.close();
+                log.info("MQTT Logger Client {} closed successfully.", mqttLoggerClient.getClientId());
+            } catch (MqttException e) {
+                log.error("Error closing MQTT Logger Client: ", e);
+            }
+        }
+    }
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/service/MqttCommunicationLogService.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/service/MqttCommunicationLogService.java
@@ -0,0 +1,31 @@
+package com.yupi.project.service;
+
+import com.baomidou.mybatisplus.extension.service.IService;
+import com.yupi.project.model.entity.MqttCommunicationLog;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+
+public interface MqttCommunicationLogService extends IService<MqttCommunicationLog> {
+
+    /**
+     * 异步记录MQTT消息
+     *
+     * @param topic       主题
+     * @param message     MQTT消息对象
+     * @param direction   方向 (UPSTREAM/DOWNSTREAM)
+     * @param clientId    客户端ID
+     * @param messageIdStr MQTT v5 的 Message ID 或应用生成的UUID
+     * @param backendProcessingStatus 后端处理状态 (可选)
+     * @param backendProcessingInfo   后端处理信息 (可选)
+     * @param relatedSessionId        关联的会话ID (可选)
+     * @param relatedTaskId           关联的任务ID (可选)
+     */
+    void asyncLogMessage(String topic, MqttMessage message, String direction, String clientId,
+                         String messageIdStr, String backendProcessingStatus,
+                         String backendProcessingInfo, Long relatedSessionId, Long relatedTaskId);
+
+    /**
+     * 异步记录出站 (DOWNSTREAM) MQTT消息的简化版本。
+     * 通常在消息发布后调用。
+     */
+    void asyncLogDownstreamMessage(String topic, String payload, Integer qos, Boolean retained, String clientId, String messageIdStr, Long relatedSessionId, Long relatedTaskId);
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/service/RobotTaskService.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/service/RobotTaskService.java
@@ -101,4 +101,13 @@ public interface RobotTaskService extends IService<RobotTask> {
     */
    boolean markTaskAsFailed(Long taskId, String errorMessage, Date failedTime);

+    /**
+     * 根据机器人ID和会话ID查找最新的任务。
+     *
+     * @param robotId 机器人ID
+     * @param sessionId 会话ID
+     * @return 如果找到，则返回任务实体，否则返回 null
+     */
+    RobotTask findLatestTaskByRobotIdAndSessionId(String robotId, Long sessionId);
+
 } 
--- a/springboot-init-main/src/main/java/com/yupi/project/service/impl/MqttCommunicationLogServiceImpl.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/service/impl/MqttCommunicationLogServiceImpl.java
@@ -0,0 +1,105 @@
+package com.yupi.project.service.impl;
+
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import com.yupi.project.mapper.MqttCommunicationLogMapper;
+import com.yupi.project.model.entity.MqttCommunicationLog;
+import com.yupi.project.service.MqttCommunicationLogService;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.scheduling.annotation.Async;
+import org.springframework.stereotype.Service;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Date;
+import java.util.UUID;
+
+@Service
+@Slf4j
+public class MqttCommunicationLogServiceImpl extends ServiceImpl<MqttCommunicationLogMapper, MqttCommunicationLog>
+        implements MqttCommunicationLogService {
+
+    private static final String DIRECTION_UPSTREAM = "UPSTREAM";
+    private static final String DIRECTION_DOWNSTREAM = "DOWNSTREAM";
+    private static final String PAYLOAD_FORMAT_TEXT = "TEXT";
+    private static final String PAYLOAD_FORMAT_JSON = "JSON"; // Assume JSON if parseable
+    private static final String PAYLOAD_FORMAT_BINARY = "BINARY";
+
+    @Override
+    @Async("mqttLoggerThreadPoolTaskExecutor") // Ensure this bean name matches your AsyncConfigurer
+    public void asyncLogMessage(String topic, MqttMessage message, String direction, String clientId,
+                                String messageIdStr, String backendProcessingStatus,
+                                String backendProcessingInfo, Long relatedSessionId, Long relatedTaskId) {
+        try {
+            MqttCommunicationLog logEntry = new MqttCommunicationLog();
+            logEntry.setMessageId(messageIdStr != null ? messageIdStr : (message.getId() > 0 ? String.valueOf(message.getId()) : null) );
+            logEntry.setDirection(direction);
+            logEntry.setClientId(clientId);
+            logEntry.setTopic(topic);
+
+            byte[] payloadBytes = message.getPayload();
+            String payloadString = new String(payloadBytes, StandardCharsets.UTF_8);
+            logEntry.setPayload(payloadString); // Store as string
+
+            // Basic payload format detection (can be enhanced)
+            if (isJson(payloadString)) {
+                logEntry.setPayloadFormat(PAYLOAD_FORMAT_JSON);
+            } else {
+                logEntry.setPayloadFormat(PAYLOAD_FORMAT_TEXT); // Default to TEXT, could be BINARY if not UTF-8 decodable well
+            }
+            // For true binary, would need different handling or indication
+
+            logEntry.setQos(message.getQos());
+            logEntry.setIsRetained(message.isRetained());
+            logEntry.setLogTimestamp(new Date()); // Record time when log entry is created
+
+            logEntry.setBackendProcessingStatus(backendProcessingStatus);
+            logEntry.setBackendProcessingInfo(backendProcessingInfo);
+            logEntry.setRelatedSessionId(relatedSessionId);
+            logEntry.setRelatedTaskId(relatedTaskId);
+
+            this.save(logEntry);
+        } catch (Exception e) {
+            log.error("Failed to asynchronously log MQTT message. Topic: {}, Direction: {}, ClientId: {}", topic, direction, clientId, e);
+        }
+    }
+
+    @Override
+    @Async("mqttLoggerThreadPoolTaskExecutor")
+    public void asyncLogDownstreamMessage(String topic, String payload, Integer qos, Boolean retained, String clientId, String messageIdStr, Long relatedSessionId, Long relatedTaskId) {
+        try {
+            MqttCommunicationLog logEntry = new MqttCommunicationLog();
+            logEntry.setMessageId(messageIdStr != null ? messageIdStr : UUID.randomUUID().toString()); // Generate UUID if not provided
+            logEntry.setDirection(DIRECTION_DOWNSTREAM);
+            logEntry.setClientId(clientId);
+            logEntry.setTopic(topic);
+            logEntry.setPayload(payload);
+
+            if (isJson(payload)) {
+                logEntry.setPayloadFormat(PAYLOAD_FORMAT_JSON);
+            } else {
+                logEntry.setPayloadFormat(PAYLOAD_FORMAT_TEXT);
+            }
+
+            logEntry.setQos(qos);
+            logEntry.setIsRetained(retained != null ? retained : false);
+            logEntry.setLogTimestamp(new Date());
+
+            // For downstream, processing status/info is typically not set at the moment of logging send attempt
+            // It might be updated later if a response is expected and tracked.
+            logEntry.setRelatedSessionId(relatedSessionId);
+            logEntry.setRelatedTaskId(relatedTaskId);
+
+            this.save(logEntry);
+        } catch (Exception e) {
+            log.error("Failed to asynchronously log downstream MQTT message. Topic: {}, ClientId: {}", topic, clientId, e);
+        }
+    }
+
+    private boolean isJson(String str) {
+        if (str == null || str.isEmpty()) {
+            return false;
+        }
+        String trimmedStr = str.trim();
+        return (trimmedStr.startsWith("{") && trimmedStr.endsWith("}")) || (trimmedStr.startsWith("[") && trimmedStr.endsWith("]"));
+    }
+} 
--- a/springboot-init-main/src/main/java/com/yupi/project/service/impl/MqttServiceImpl.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/service/impl/MqttServiceImpl.java
@@ -10,6 +10,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.eclipse.paho.client.mqttv3.MqttException;
 import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;

@@ -18,13 +19,20 @@ import java.util.Date;

@Service
@Slf4j
-@RequiredArgsConstructor
 public class MqttServiceImpl implements MqttService {

    private final MqttClient mqttClient; // Autowired by Spring from MqttConfig
    private final MqttProperties mqttProperties;
    private final RobotTaskService robotTaskService;

+    public MqttServiceImpl(@Qualifier("mqttClientBean") MqttClient mqttClient,
+                          MqttProperties mqttProperties,
+                          RobotTaskService robotTaskService) {
+        this.mqttClient = mqttClient;
+        this.mqttProperties = mqttProperties;
+        this.robotTaskService = robotTaskService;
+    }
+
    @Override
    @Transactional(rollbackFor = Exception.class) // Ensure rollback if MQTT publish fails after task creation
    public boolean sendCommand(String robotId, CommandTypeEnum commandType, String payloadJson, Long sessionId) throws Exception {
--- a/springboot-init-main/src/main/java/com/yupi/project/service/impl/RobotTaskServiceImpl.java
+++ b/springboot-init-main/src/main/java/com/yupi/project/service/impl/RobotTaskServiceImpl.java
@@ -316,4 +316,25 @@ public class RobotTaskServiceImpl extends ServiceImpl<RobotTaskMapper, RobotTask
        }
        return updated;
    }
+
+    @Override
+    public RobotTask findLatestTaskByRobotIdAndSessionId(String robotId, Long sessionId) {
+        if (robotId == null || sessionId == null) {
+            log.warn("Cannot find latest task: robotId or sessionId is null.");
+            return null;
+        }
+        QueryWrapper<RobotTask> queryWrapper = new QueryWrapper<>();
+        queryWrapper.eq("robot_id", robotId)
+                .eq("related_session_id", sessionId)
+                .orderByDesc("create_time")
+                .last("LIMIT 1");
+
+        RobotTask task = this.getOne(queryWrapper);
+        if (task != null) {
+            log.info("Found latest task with ID {} for Robot ID: {} and Session ID: {}.", task.getId(), robotId, sessionId);
+        } else {
+            log.debug("No task found for Robot ID: {} and Session ID: {}.", robotId, sessionId);
+        }
+        return task;
+    }
 } 
--- a/springboot-init-main/src/main/resources/application.yml
+++ b/springboot-init-main/src/main/resources/application.yml
@@ -62,4 +62,20 @@ mqtt:
  status-topic-base: yupi_mqtt_power_project/robot/status   # Prefixed base topic for receiving status
  task: # Task specific configurations
    timeoutSeconds: 300 # Default 300 seconds (5 minutes) for a task to be considered timed out
-    timeoutCheckRateMs: 60000 # Default 60000 ms (1 minute) for how often to check for timed out tasks
+    timeoutCheckRateMs: 60000 # Default 60000 ms (1 minute) for how often to check for timed out tasks
+  logger:
+    enabled: true
+    client-id-prefix: backend-yupi-mqtt-power-logger- # Ensure this is different from the main client-id-prefix
+    # Topics to subscribe to for logging. Use '#' for all sub-topics.
+    # These should cover all topics the main application interacts with for upstream messages.
+    topics:
+      - "yupi_mqtt_power_project/robot/status/#" # Example: status messages from robots
+      # Add other topics for upstream messages (e.g., heartbeats, specific acknowledgements if not covered by status)
+      # - "yupi_mqtt_power_project/robot/heartbeat/#" # If heartbeats are on a separate root
+    async: # Async properties for logging tasks
+      core-pool-size: 2
+      max-pool-size: 5
+      queue-capacity: 1000 # Queue for logging tasks
+      thread-name-prefix: "mqtt-log-async-"
+    retention:
+      days: 30 # Log retention period in days (for future cleanup tasks)