fix: 物联网平台，amqp数据接入并插入TD数据库相关逻辑编码。

intc-modules/intc-iot/src/main/java/com/intc/iot/config/AliyunIotConfiguration.java

@@ -6,6 +6,7 @@ import com.aliyuncs.profile.DefaultProfile;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -17,6 +18,7 @@ import org.springframework.context.annotation.Configuration;
 @Slf4j
 @Configuration
 @RequiredArgsConstructor
+@EnableConfigurationProperties(AliyunIotProperties.class)
 @ConditionalOnProperty(prefix = "aliyun.living-iot", name = "access-key-id")
 public class AliyunIotConfiguration {
 

intc-modules/intc-iot/src/main/java/com/intc/iot/config/AliyunIotProperties.java

@@ -2,7 +2,6 @@ package com.intc.iot.config;
 
 import lombok.Data;
 import org.springframework.boot.context.properties.ConfigurationProperties;
-import org.springframework.stereotype.Component;
 
 /**
  * 阿里云生活物联网平台（飞燕平台）配置属性
@@ -10,7 +9,6 @@ import org.springframework.stereotype.Component;
  * @author intc
  */
 @Data
-@Component
 @ConfigurationProperties(prefix = "aliyun.living-iot")
 public class AliyunIotProperties {
 
@@ -50,10 +48,15 @@ public class AliyunIotProperties {
     private String categoryKey;
 
     /**
-     * MQTT 配置
+     * MQTT 配置（可选，用于设备直连）
      */
     private MqttConfig mqtt = new MqttConfig();
 
+    /**
+     * AMQP 服务端订阅配置（推荐）
+     */
+    private AmqpConfig amqp = new AmqpConfig();
+
     @Data
     public static class MqttConfig {
         /**
@@ -97,4 +100,62 @@ public class AliyunIotProperties {
         private Boolean cleanSession = true;
     }
 
+    /**
+     * AMQP 服务端订阅配置（使用数据同步的 AppKey + AppSecret）
+     */
+    @Data
+    public static class AmqpConfig {
+        /**
+         * 是否启用 AMQP 订阅
+         */
+        private Boolean enabled = false;
+
+        /**
+         * 数据同步 AppKey（与外层的 appKey 不同！）
+         */
+        private String dataSyncAppKey;
+
+        /**
+         * 数据同步 AppSecret
+         */
+        private String dataSyncAppSecret;
+
+        /**
+         * AMQP 接入点地址（使用 amqps 协议）
+         * 中国内地：amqps://ilop.iot-amqp.cn-shanghai.aliyuncs.com:5671
+         * 新加坡：amqps://ilop.iot-amqp.ap-southeast-1.aliyuncs.com:5671
+         */
+        private String endpoint;
+
+        /**
+         * 消费组 ID
+         */
+        private String consumerGroupId;
+
+        /**
+         * 客户端 ID（建议使用机器 UUID、MAC 地址等唯一标识）
+         */
+        private String clientId;
+
+        /**
+         * 连接超时时间（毫秒）
+         */
+        private Integer connectionTimeout = 80000;
+
+        /**
+         * 自动重连
+         */
+        private Boolean autoReconnect = true;
+
+        /**
+         * 最大重连次数
+         */
+        private Integer maxReconnectAttempts = 10;
+
+        /**
+         * 重连延迟（毫秒）
+         */
+        private Integer reconnectDelay = 30000;
+    }
+
 }

intc-modules/intc-iot/src/main/java/com/intc/iot/config/AmqpConfiguration.java

@@ -0,0 +1,254 @@
+package com.intc.iot.config;
+
+import com.intc.iot.service.AmqpMessageHandler;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.qpid.jms.JmsConnection;
+import org.apache.qpid.jms.JmsConnectionListener;
+import org.apache.qpid.jms.message.JmsInboundMessageDispatch;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import javax.crypto.Mac;
+import javax.crypto.spec.SecretKeySpec;
+import javax.jms.*;
+import javax.jms.Connection;
+import javax.naming.Context;
+import javax.naming.InitialContext;
+import org.apache.commons.codec.binary.Hex;
+import java.net.URI;
+import java.util.Hashtable;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * 阿里云 IoT AMQP 服务端订阅配置
+ * 
+ * 使用说明：
+ * 1. 在阿里云IoT控制台开启设备数据同步
+ * 2. 配置 AMQP 连接参数（endpoint、consumerGroupId 等）
+ * 3. 启用配置：aliyun.living-iot.amqp.enabled=true
+ * 
+ * 注意：
+ * - 阿里云生活物联网平台使用 AMQP 1.0 协议
+ * - 使用 AppKey + AppSecret 认证（不是 AccessKey）
+ * - 连接地址：amqps://ilop.iot-amqp.cn-shanghai.aliyuncs.com:5671
+ *
+ * @author intc
+ */
+@Slf4j
+@Configuration
+@RequiredArgsConstructor
+@EnableConfigurationProperties(AliyunIotProperties.class)
+@ConditionalOnProperty(prefix = "aliyun.living-iot.amqp", name = "enabled", havingValue = "true")
+public class AmqpConfiguration {
+
+    private final AliyunIotProperties iotProperties;
+    private final AmqpMessageHandler messageHandler;
+
+    /**
+     * 创建 AMQP 1.0 连接（使用 JMS）
+     */
+    @Bean
+    public Connection amqpConnection() throws Exception {
+        AliyunIotProperties.AmqpConfig amqp = iotProperties.getAmqp();
+        // 使用数据同步的 AppKey/AppSecret 进行 AMQP 认证
+        String dataSyncAppKey = amqp.getDataSyncAppKey();
+        String dataSyncAppSecret = amqp.getDataSyncAppSecret();
+        
+        log.info("========== AMQP 连接配置 ==========");
+        log.info("Endpoint: {}", amqp.getEndpoint());
+        log.info("数据同步 AppKey: {}", dataSyncAppKey);
+        log.info("数据同步 AppSecret: {}...", dataSyncAppSecret != null && dataSyncAppSecret.length() > 4 
+            ? dataSyncAppSecret.substring(0, 4) + "****" : "NULL");
+        log.info("ConsumerGroupId: {}", amqp.getConsumerGroupId());
+        log.info("ClientId: {}", amqp.getClientId());
+        log.info("ConnectionTimeout: {}ms", amqp.getConnectionTimeout());
+        log.info("====================================");
+        
+        try {
+            // 生成随机数（按照官方示例）
+            long random = ThreadLocalRandom.current().nextLong(Long.MAX_VALUE);
+            
+            // 构建用户名（完全按照官方示例）
+            String userName = amqp.getClientId() + "|authMode=appkey"
+                + ",signMethod=SHA256"
+                + ",random=" + random
+                + ",appKey=" + dataSyncAppKey
+                + ",groupId=" + amqp.getConsumerGroupId() + "|";
+            
+            // 计算密码（完全按照官方示例）
+            String signContent = "random=" + random;
+            String password = doSign(signContent, dataSyncAppSecret, "HmacSHA256");
+            
+            log.info("用户名: {}", userName);
+            log.info("签名内容: {}", signContent);
+            log.info("密码（完整）: {}", password);
+            log.info("密码长度: {}", password.length());
+            log.info("ConsumerGroupId（用于groupId）: {}", amqp.getConsumerGroupId());
+            
+            // 构建连接 URL（带故障转移和 SSL 配置）
+            String connectionUrl = String.format(
+                "failover:(%s?amqp.idleTimeout=%d&transport.verifyHost=false)?failover.maxReconnectAttempts=%d&failover.reconnectDelay=%d",
+                amqp.getEndpoint(),
+                amqp.getConnectionTimeout(),
+                amqp.getMaxReconnectAttempts(),
+                amqp.getReconnectDelay()
+            );
+            
+            log.info("连接 URL: {}", connectionUrl);
+            
+            // 使用 JNDI 初始化连接
+            Hashtable<String, String> hashtable = new Hashtable<>();
+            hashtable.put("connectionfactory.SBCF", connectionUrl);
+            hashtable.put("queue.QUEUE", "default");
+            hashtable.put(Context.INITIAL_CONTEXT_FACTORY, "org.apache.qpid.jms.jndi.JmsInitialContextFactory");
+            
+            Context context = new InitialContext(hashtable);
+            ConnectionFactory cf = (ConnectionFactory) context.lookup("SBCF");
+            Destination queue = (Destination) context.lookup("QUEUE");
+            
+            log.info("正在连接到 AMQP 服务器...");
+            log.info("请耐必等待，连接可能需要几秒...");
+            
+            // 创建连接
+            Connection connection = cf.createConnection(userName, password);
+            
+            // 添加连接监听器
+            if (connection instanceof JmsConnection) {
+                ((JmsConnection) connection).addConnectionListener(new JmsConnectionListener() {
+                    @Override
+                    public void onConnectionEstablished(URI remoteURI) {
+                        log.info("✅ AMQP 连接建立成功: {}", remoteURI);
+                    }
+
+                    @Override
+                    public void onConnectionFailure(Throwable error) {
+                        log.error("❌ AMQP 连接失败回调");
+                        log.error("失败原因: {}", error.getMessage());
+                        log.error("异常类型: {}", error.getClass().getName());
+                        if (error.getCause() != null) {
+                            log.error("根本原因: {}", error.getCause().getMessage());
+                            log.error("根本原因类型: {}", error.getCause().getClass().getName());
+                        }
+                        // 打印完整堆栈
+                        log.error("完整异常信息:", error);
+                    }
+
+                    @Override
+                    public void onConnectionInterrupted(URI remoteURI) {
+                        log.warn("⚠️ AMQP 连接中断: {}", remoteURI);
+                    }
+
+                    @Override
+                    public void onConnectionRestored(URI remoteURI) {
+                        log.info("✅ AMQP 连接恢复: {}", remoteURI);
+                    }
+
+                    @Override
+                    public void onInboundMessage(JmsInboundMessageDispatch envelope) {}
+
+                    @Override
+                    public void onSessionClosed(Session session, Throwable cause) {}
+
+                    @Override
+                    public void onConsumerClosed(MessageConsumer consumer, Throwable cause) {}
+
+                    @Override
+                    public void onProducerClosed(MessageProducer producer, Throwable cause) {}
+                });
+            }
+            
+            // 启动连接
+            connection.start();
+            log.info("✅ AMQP 连接创建成功！");
+            
+            // 启动消息消费
+            startConsuming(connection, queue);
+            
+            return connection;
+        } catch (Exception e) {
+            log.error("❌ AMQP 连接失败", e);
+            log.error("===== 详细错误信息 =====");
+            log.error("异常类型: {}", e.getClass().getName());
+            log.error("异常信息: {}", e.getMessage());
+            
+            // 打印完整异常链
+            Throwable cause = e.getCause();
+            int level = 1;
+            while (cause != null) {
+                log.error("原因 {} - 类型: {}", level, cause.getClass().getName());
+                log.error("原因 {} - 信息: {}", level, cause.getMessage());
+                cause = cause.getCause();
+                level++;
+            }
+            
+            log.error("===== 配置信息 =====");
+            log.error("  - Endpoint: {}", amqp.getEndpoint());
+            log.error("  - 数据同步 AppKey: {}", dataSyncAppKey);
+            log.error("  - 数据同步 AppSecret: {}...", dataSyncAppSecret != null && dataSyncAppSecret.length() > 4 ? dataSyncAppSecret.substring(0, 4) + "****" : "NULL");
+            log.error("  - ConsumerGroupId: {}", amqp.getConsumerGroupId());
+            log.error("  - ClientId: {}", amqp.getClientId());
+            
+            log.error("===== 可能原因 =====");
+            log.error("1. 网络无法连接到服务器: {}", amqp.getEndpoint());
+            log.error("2. 数据同步 AppKey 或 AppSecret 错误");
+            log.error("3. 消费组 ID 不存在或未授权");
+            log.error("4. 未开启设备数据同步");
+            log.error("5. 防火墙阻止了端口 5671");
+            log.error("6. 签名算法或格式错误");
+            
+            throw new RuntimeException("创建 AMQP 连接失败: " + e.getMessage(), e);
+        }
+    }
+
+    /**
+     * 启动消息消费
+     */
+    private void startConsuming(Connection connection, Destination queue) {
+        new Thread(() -> {
+            try {
+                // 创建会话（自动 ACK）
+                Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+                
+                // 创建消费者
+                MessageConsumer consumer = session.createConsumer(queue);
+                
+                log.info("开始消费 AMQP 消息...");
+                
+                // 设置消息监听器
+                consumer.setMessageListener(message -> {
+                    try {
+                        // 阿里云 IoT 发送的是 BytesMessage
+                        byte[] body = message.getBody(byte[].class);
+                        String content = new String(body, "UTF-8");
+                        
+                        // 处理消息（异步）
+                        messageHandler.handleMessage(content);
+                        
+                    } catch (Exception e) {
+                        log.error("处理 AMQP 消息失败", e);
+                    }
+                });
+                
+            } catch (Exception e) {
+                log.error("启动 AMQP 消费失败", e);
+            }
+        }, "AMQP-Consumer").start();
+    }
+
+    /**
+     * 计算签名（完全按照官方示例）
+     */
+    private String doSign(String toSignString, String secret, String signMethod) throws Exception {
+        SecretKeySpec signingKey = new SecretKeySpec(secret.getBytes(), signMethod);
+        Mac mac = Mac.getInstance(signMethod);
+        mac.init(signingKey);
+        byte[] rawHmac = mac.doFinal(toSignString.getBytes());
+        
+        // 使用 Hex 编码（按照官方示例）
+        return Hex.encodeHexString(rawHmac);
+    }
+
+}

intc-modules/intc-iot/src/main/java/com/intc/iot/config/MqttConfiguration.java

@@ -5,6 +5,7 @@ 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.persist.MemoryPersistence;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnClass;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -17,6 +18,7 @@ import org.springframework.context.annotation.Configuration;
 @Slf4j
 @Configuration
 @RequiredArgsConstructor
+@ConditionalOnClass(MqttClient.class)
 @ConditionalOnProperty(prefix = "aliyun.living-iot.mqtt", name = "broker-url")
 public class MqttConfiguration {
 
@@ -24,26 +26,40 @@ public class MqttConfiguration {
 
     /**
      * 创建 MQTT 客户端
+     * 添加容错机制，连接失败时不影响应用启动
      */
     @Bean
-    public MqttClient mqttClient() throws Exception {
+    public MqttClient mqttClient() {
         AliyunIotProperties.MqttConfig mqtt = iotProperties.getMqtt();
         
-        MemoryPersistence persistence = new MemoryPersistence();
-        MqttClient client = new MqttClient(mqtt.getBrokerUrl(), mqtt.getClientId(), persistence);
+        try {
+            MemoryPersistence persistence = new MemoryPersistence();
+            MqttClient client = new MqttClient(mqtt.getBrokerUrl(), mqtt.getClientId(), persistence);
 
-        MqttConnectOptions options = new MqttConnectOptions();
-        options.setUserName(mqtt.getUsername());
-        options.setPassword(mqtt.getPassword().toCharArray());
-        options.setConnectionTimeout(mqtt.getConnectionTimeout());
-        options.setKeepAliveInterval(mqtt.getKeepAliveInterval());
-        options.setAutomaticReconnect(mqtt.getAutoReconnect());
-        options.setCleanSession(mqtt.getCleanSession());
+            MqttConnectOptions options = new MqttConnectOptions();
+            options.setUserName(mqtt.getUsername());
+            options.setPassword(mqtt.getPassword().toCharArray());
+            options.setConnectionTimeout(mqtt.getConnectionTimeout());
+            options.setKeepAliveInterval(mqtt.getKeepAliveInterval());
+            options.setAutomaticReconnect(mqtt.getAutoReconnect());
+            options.setCleanSession(mqtt.getCleanSession());
 
-        client.connect(options);
-        log.info("MQTT 客户端连接成功，Broker: {}", mqtt.getBrokerUrl());
-        
-        return client;
+            client.connect(options);
+            log.info("MQTT 客户端连接成功，Broker: {}", mqtt.getBrokerUrl());
+            
+            return client;
+        } catch (Exception e) {
+            log.error("MQTT 客户端连接失败，将返回未连接的客户端: {}", e.getMessage());
+            
+            try {
+                // 返回未连接的客户端，由 MqttServiceImpl 中的自动重连机制处理
+                MemoryPersistence persistence = new MemoryPersistence();
+                return new MqttClient(mqtt.getBrokerUrl(), mqtt.getClientId(), persistence);
+            } catch (Exception ex) {
+                log.error("MQTT 客户端创建失败，请检查配置: {}", ex.getMessage());
+                throw new RuntimeException("创建 MQTT 客户端失败", ex);
+            }
+        }
     }
 
 }

intc-modules/intc-iot/src/main/java/com/intc/iot/config/VmsMnsAutoStarter.java

@@ -0,0 +1,41 @@
+package com.intc.iot.config;
+
+import com.intc.iot.service.VmsMnsConsumerService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
+import org.springframework.stereotype.Component;
+
+import jakarta.annotation.PreDestroy;
+
+/**
+ * VMS MNS 消费服务自动启动器
+ * 在 Spring 容器启动后自动启动 MNS 消费
+ *
+ * @author intc-iot
+ */
+@Component
+@RequiredArgsConstructor
+@ConditionalOnBean(VmsMnsConsumerService.class)
+@Slf4j
+public class VmsMnsAutoStarter implements ApplicationRunner {
+
+    private final VmsMnsConsumerService vmsMnsConsumerService;
+
+    @Override
+    public void run(ApplicationArguments args) throws Exception {
+        log.info("自动启动 VMS MNS 回执消费服务");
+        vmsMnsConsumerService.start();
+    }
+
+    /**
+     * 容器销毁时停止服务
+     */
+    @PreDestroy
+    public void onDestroy() {
+        log.info("容器销毁，停止 VMS MNS 回执消费服务");
+        vmsMnsConsumerService.stop();
+    }
+}