#update

maibu-common/src/main/java/com/maibu/constant/Constant.java

@@ -22,4 +22,6 @@ public class Constant {
     public static final AttributeKey<String> ATT_SITE_ID = AttributeKey.valueOf("siteId");
 
     public static final String  WheelGainCoefficients =  "WheelGainCoefficients";
+
+    public static final String  mqttClientPrefix =  "common-device-client-";
 }

maibu-common/src/main/java/com/maibu/constant/Constants.java

@@ -129,16 +129,6 @@ public class Constants
      */
     public static final String LOOKUP_LDAPS = "ldaps:";
 
-    /**
-     * 定时任务白名单配置（仅允许访问的包名，如其他需要可以自行添加）
-     */
-    public static final String[] JOB_WHITELIST_STR = { "com.fastbee" };
-
-    /**
-     * 定时任务违规的字符
-     */
-    public static final String[] JOB_ERROR_STR = { "java.net.URL", "javax.naming.InitialContext", "org.yaml.snakeyaml",
-            "org.springframework", "org.apache", "com.fastbee.common.utils.file", "com.fastbee.common.config" };
 
     /**
      * 语言类型

maibu-common/src/main/java/com/maibu/core/business/Device.java

@@ -233,4 +233,7 @@ public class Device extends BaseDO {
     @TableField(exist = false)
     private DeviceRunParam runParam;
 
+
+    private boolean hasRegistered = false;
+
 }

maibu-common/src/main/java/com/maibu/core/business/DeviceModel.java

@@ -8,6 +8,7 @@ import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.maibu.core.domain.BaseDO;
+import com.maibu.core.domain.OrgNoUserDO;
 import com.maibu.core.enums.ConnectType;
 
 import io.swagger.annotations.ApiModel;
@@ -20,7 +21,7 @@ import lombok.EqualsAndHashCode;
 @JsonInclude(JsonInclude.Include.NON_NULL)
 @Data
 @TableName("iot_device_model")
-public class DeviceModel extends BaseDO {
+public class DeviceModel extends OrgNoUserDO {
     private static final long serialVersionUID = 1L;
 
     @TableId(type = IdType.AUTO)

maibu-common/src/main/java/com/maibu/core/business/IoTCommonDevice.java

@@ -6,6 +6,7 @@ import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.fasterxml.jackson.annotation.JsonFormat;
 import com.maibu.core.domain.BaseDO;
+import com.maibu.core.domain.OrgNoUserDO;
 import com.maibu.core.enums.ConnectType;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
@@ -21,7 +22,7 @@ import java.util.Map;
 @EqualsAndHashCode(callSuper = true)
 @Data
 @TableName(value = "iot_device_common", autoResultMap = true)
-public class IoTCommonDevice extends BaseDO {
+public class IoTCommonDevice extends OrgNoUserDO {
     private static final long serialVersionUID = 1L;
 
     @TableId(type = IdType.AUTO)
@@ -165,4 +166,16 @@ public class IoTCommonDevice extends BaseDO {
      */
     @TableField(exist = false)
     private List<DeviceModelProperty> propertyList;
+
+    /**
+     * 属性列表（非数据库字段）
+     */
+    @TableField(exist = false)
+    private DeviceModel deviceModel;
+
+    /**
+     * 属性列表（非数据库字段）
+     */
+    @TableField(exist = false)
+    private Product deviceProduct;
 }

maibu-common/src/main/java/com/maibu/core/business/Product.java

@@ -7,6 +7,7 @@ import com.baomidou.mybatisplus.annotation.TableName;
 import com.baomidou.mybatisplus.extension.handlers.JacksonTypeHandler;
 import com.fasterxml.jackson.annotation.JsonInclude;
 import com.maibu.core.domain.BaseDO;
+import com.maibu.core.domain.OrgNoUserDO;
 import io.swagger.annotations.ApiModel;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
@@ -19,7 +20,7 @@ import java.util.Map;
 @JsonInclude(JsonInclude.Include.NON_NULL)
 @Data
 @TableName(value = "device_product", autoResultMap = true)
-public class Product extends BaseDO {
+public class Product extends OrgNoUserDO {
     private static final long serialVersionUID = 1L;
 
     @TableId(type = IdType.AUTO)
@@ -43,6 +44,8 @@ public class Product extends BaseDO {
 
     private Integer deviceType;
 
+    private Integer integrationMode; //对接方式
+
     @TableField(typeHandler = JacksonTypeHandler.class)
     private Map<String, Object> thingsModels;
 

maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java

@@ -196,6 +196,7 @@ public class NettyDevice extends Connector {
         if (this.getChannel() != null && this.getChannel().isActive()) {
             this.getChannel().writeAndFlush(Unpooled.wrappedBuffer(bytes));
         }
+        log.debug("下发读取配置的报文完成：{}", toHexString(bytes));
     }
 
     public void writeConfigCommand(DeviceRunParam deviceRunParam) {
@@ -205,6 +206,7 @@ public class NettyDevice extends Connector {
         if (this.getChannel() != null && this.getChannel().isActive()) {
             this.getChannel().writeAndFlush(Unpooled.wrappedBuffer(bytes));
         }
+        log.debug("下发修改配置的报文完成：{}", toHexString(bytes));
     }
 
     // todo 优化

maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java

@@ -1,22 +1,19 @@
 package com.maibu.memory;
 
+import com.maibu.core.business.Device;
+import com.maibu.core.business.IoTCommonDevice;
+import com.maibu.influxdb.util.InfluxDBUtil;
+import com.maibu.mqtt.MqttClientUtil;
+import lombok.Data;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 
-import org.springframework.stereotype.Component;
-import org.springframework.util.CollectionUtils;
-
-import com.maibu.core.business.Device;
-import com.maibu.core.business.IoTCommonDevice;
-import com.maibu.core.business.IoTCommonProduct;
-import com.maibu.influxdb.util.InfluxDBUtil;
-import com.maibu.mqtt.MqttClientUtil;
-
-import lombok.Data;
-
 @Component
 @Data
 public class GlobalMemory {
@@ -29,8 +26,6 @@ public class GlobalMemory {
     // key：orgId - siteId
     public static ConcurrentHashMap<Long, Map<Long, SiteMemory>> orgAttrs = new ConcurrentHashMap<>();
 
-    public static ConcurrentHashMap<Long, IoTCommonProduct> productMap = new ConcurrentHashMap<>();
-
     public static InfluxDBUtil influxDBUtil;
 
     public static MqttClientUtil mqttClientUtil;

maibu-common/src/main/java/com/maibu/memory/SiteMemory.java

@@ -6,18 +6,13 @@ import java.util.Map;
 import java.util.Objects;
 import java.util.concurrent.ConcurrentHashMap;
 
+import com.maibu.core.business.*;
+import com.maibu.mqtt.CustomMqttDeviceMonitor;
 import org.apache.commons.lang3.StringUtils;
+import org.eclipse.paho.client.mqttv3.MqttClient;
 import org.springframework.stereotype.Component;
 import org.springframework.util.CollectionUtils;
 
-import com.maibu.core.business.AlarmOrderConfig;
-import com.maibu.core.business.Device;
-import com.maibu.core.business.DevicePlan;
-import com.maibu.core.business.DevicePlanTask;
-import com.maibu.core.business.DeviceRunParam;
-import com.maibu.core.business.ErrorIdentificationStandard;
-import com.maibu.core.business.IoTCommonDevice;
-import com.maibu.core.business.OrderModel;
 import com.maibu.core.domain.entity.SysSite;
 import com.maibu.core.enums.ErrorSource;
 
@@ -35,6 +30,10 @@ public class SiteMemory {
 
     public SysSite sysSite;
 
+    public CustomMqttDeviceMonitor CustomMqttDeviceMonitor;
+
+    public static final ConcurrentHashMap<String, MqttClient> mqttClientMap = new ConcurrentHashMap<>();
+
     public static final ConcurrentHashMap<Long, OrderModel> orderModelMap = new ConcurrentHashMap<>();
 
     public static final ConcurrentHashMap<String, AlarmOrderConfig> orderConfigMap = new ConcurrentHashMap<>();
@@ -45,6 +44,10 @@ public class SiteMemory {
 
     private static final ConcurrentHashMap<String, IoTCommonDevice> commonDeviceMap = new ConcurrentHashMap<>();
 
+    private static final ConcurrentHashMap<Long, Product> productMap = new ConcurrentHashMap<>();
+
+    private static final ConcurrentHashMap<Long, DeviceModel> deviceModelMap = new ConcurrentHashMap<>();
+
     // key deviceId
     public static ConcurrentHashMap<String, List<DevicePlan>> devicePlanMap = new ConcurrentHashMap<>();
     // key deviceId prepare
@@ -86,6 +89,55 @@ public class SiteMemory {
         }
     }
 
+    public void saveProduct(Product product) {
+        productMap.put(product.getId(), product);
+    }
+
+    public Product getProduct(Long id) {
+        if (id != null) {
+            return productMap.get(id);
+        }
+        return null;
+    }
+
+    public List<Product> getAllProduct() {
+        if (!CollectionUtils.isEmpty(productMap)) {
+            return (List<Product>) productMap.values();
+        }
+        return null;
+    }
+
+    public void removeProduct(Long id) {
+        if (id != null) {
+            productMap.remove(id);
+        }
+    }
+
+    public void saveDeviceModel(DeviceModel deviceModel) {
+        deviceModelMap.put(deviceModel.getId(), deviceModel);
+    }
+
+    public DeviceModel getDeviceModel(Long id) {
+        if (id != null) {
+            return deviceModelMap.get(id);
+        }
+        return null;
+    }
+
+    public List<DeviceModel> getAllDeviceModel() {
+        if (!CollectionUtils.isEmpty(deviceModelMap)) {
+            return (List<DeviceModel>) deviceModelMap.values();
+        }
+        return null;
+    }
+
+    public void removeDeviceModel(Long id) {
+        if (id != null) {
+            deviceModelMap.remove(id);
+        }
+    }
+
+
     public void saveDeviceRunParam(DeviceRunParam deviceRunParam) {
         deviceRunParamMap.put(deviceRunParam.getDeviceId(), deviceRunParam);
     }

maibu-common/src/main/java/com/maibu/mqtt/ActionMessageHandler.java

@@ -0,0 +1,83 @@
+package com.maibu.mqtt;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+/**
+ * 示例：属性数据消息处理器
+ * 演示如何处理订阅到的设备属性数据
+ */
+@Slf4j
+@Component
+public class ActionMessageHandler implements CustomMqttMessageHandler {
+
+    private final ObjectMapper objectMapper = new ObjectMapper();
+
+    @Override
+    public String getHandlerType() {
+        return "Action";
+    }
+
+    @Override
+    public void handleMessage(String deviceKey, String topic, String payload, CustomMqttDeviceConfig config) {
+        try {
+            // 方式1: 原始字符串处理
+            log.info("【属性消息】deviceKey={}, topic={}, payload={}", deviceKey, topic, payload);
+
+            // 方式2: JSON 解析示例
+            // JsonNode jsonNode = objectMapper.readTree(payload);
+            // String dataType = jsonNode.path("type").asText();
+            // switch (dataType) {
+            //     case "temperature":
+            //         handleTemperature(deviceKey, jsonNode);
+            //         break;
+            //     case "humidity":
+            //         handleHumidity(deviceKey, jsonNode);
+            //         break;
+            //     default:
+            //         log.warn("未知的数据类型: {}", dataType);
+            // }
+
+            // 方式3: 按 Topic 路由分发
+            if (topic.contains("status")) {
+                log.info("【状态变更】设备 {} 状态更新", deviceKey);
+            } else if (topic.contains("alarm")) {
+                log.warn("【告警消息】设备 {} 触发告警: {}", deviceKey, payload);
+            } else if (topic.contains("data")) {
+                log.debug("【数据上报】设备 {} 数据: {}", deviceKey, payload);
+            }
+
+            // 在这里添加你的业务逻辑
+            // 例如: 存入数据库、推送到 WebSocket、触发告警等
+
+        } catch (Exception e) {
+            log.error("属性消息处理异常: deviceKey={}, topic={}, error={}",
+                    deviceKey, topic, e.getMessage());
+        }
+    }
+
+    @Override
+    public void onConnectionLost(String deviceKey, Throwable cause) {
+        log.warn("设备 {} 连接断开: {}", deviceKey, cause.getMessage());
+        // 可以在这里做一些清理工作，比如更新设备状态为离线
+    }
+
+    @Override
+    public void onSubscribed(String deviceKey, String topic) {
+        log.info("设备 {} 订阅主题: {}", deviceKey, topic);
+    }
+
+    // ==================== 私有处理方法示例 ====================
+
+    // private void handleTemperature(String deviceKey, JsonNode data) {
+    //     double temp = data.path("value").asDouble();
+    //     String unit = data.path("unit").asText("°C");
+    //     log.info("设备 {} 温度: {}{}", deviceKey, temp, unit);
+    // }
+
+    // private void handleHumidity(String deviceKey, JsonNode data) {
+    //     double humidity = data.path("value").asDouble();
+    //     log.info("设备 {} 湿度: {}%", deviceKey, humidity);
+    // }
+}

maibu-common/src/main/java/com/maibu/mqtt/CustomMqttDeviceConfig.java

@@ -0,0 +1,118 @@
+package com.maibu.mqtt;
+
+import lombok.Data;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 自定义外部设备 MQTT 连接配置
+ * 
+ * <p>Topic 支持三种格式：</p>
+ * <ul>
+ *   <li><b>固定主题</b>：{@link #subscribeTopics}，如 {@code mower/device-001/event/status/post}</li>
+ *   <li><b>模板主题</b>：{@link #topicTemplates}，如 {@code mower/%s/event/status/post}，{@code %s} 会自动替换为 deviceKey</li>
+ *   <li><b>MQTT 原生通配符</b>：{@link #topicTemplates} 或 {@link #subscribeTopics}，支持 {@code +}（单层）和 {@code #}（多层），如 {@code mower/+/event/status/post}</li>
+ * </ul>
+ * 
+ * <p>示例：</p>
+ * <pre>
+ * CustomMqttDeviceConfig config = new CustomMqttDeviceConfig();
+ * config.setDeviceKey("device-001");
+ * 
+ * // 方式1: 固定主题
+ * config.setSubscribeTopics(Arrays.asList("mower/device-001/event/status/post"));
+ * 
+ * // 方式2: 模板主题（%s 自动替换为 deviceKey）
+ * config.setTopicTemplates(Arrays.asList("mower/%s/event/status/post", "mower/%s/property/realtime/post"));
+ * 
+ * // 方式3: MQTT 原生通配符
+ * config.setSubscribeTopics(Arrays.asList("mower/+/event/status/post", "mower/+/property/#"));
+ * </pre>
+ */
+@Data
+public class CustomMqttDeviceConfig {
+
+    /**
+     * 设备唯一标识（如 SN 或自定义 ID）
+     */
+    private String deviceKey;
+
+    /**
+     * MQTT Broker 地址，格式: tcp://ip:port 或 ssl://ip:port
+     */
+    private String brokerUrl;
+
+    /**
+     * MQTT 客户端 ID
+     */
+    private String clientId;
+
+    /**
+     * 登录用户名
+     */
+    private String username;
+
+    /**
+     * 登录密码
+     */
+    private String password;
+
+    /**
+     * 固定订阅主题列表（支持 MQTT 原生通配符 + 和 #）
+     * 
+     * <p>示例：</p>
+     * <ul>
+     *   <li>{@code mower/device-001/event/status/post} - 精确匹配</li>
+     *   <li>{@code mower/+/event/status/post} - 单层通配</li>
+     *   <li>{@code mower/+/property/#} - 多层通配</li>
+     * </ul>
+     */
+    private List<String> subscribeTopics;
+
+
+    /**
+     * 消息处理器类型标识，用于分发到对应的 Handler
+     * 例如: "property" / "alert" / "status"
+     */
+    private String handlerType;
+
+    /**
+     * 连接超时时间（秒），默认 10
+     */
+    private Integer connectionTimeout = 10;
+
+    /**
+     * 心跳间隔（秒），默认 60
+     */
+    private Integer keepAliveInterval = 60;
+
+    /**
+     * 是否清除会话（true=离线消息不保留），默认 false
+     */
+    private Boolean cleanSession = false;
+
+    /**
+     * 是否自动重连，默认 true
+     */
+    private Boolean automaticReconnect = true;
+
+    /**
+     * 将 topicTemplates 中的 %s 替换为 deviceKey，合并到 subscribeTopics
+     * 
+     * @return 最终生效的订阅主题列表
+     */
+    public List<String> resolveSubscribeTopics() {
+        List<String> result = new ArrayList<>();
+        if (subscribeTopics != null && deviceKey != null) {
+            for (String template : subscribeTopics) {
+                if (template != null && template.contains("%s")) {
+                    result.add(String.format(template, deviceKey));
+                } else if (template != null) {
+                    result.add(template);
+                }
+            }
+        }
+        return result;
+    }
+}

maibu-common/src/main/java/com/maibu/mqtt/CustomMqttDeviceMonitor.java

@@ -0,0 +1,399 @@
+package com.maibu.mqtt;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PreDestroy;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 自定义外部设备 MQTT 订阅管理器
+ *
+ * <p>负责：</p>
+ * <ul>
+ *   <li>动态注册/注销外部设备 MQTT 连接</li>
+ *   <li>自动重连和心跳保活</li>
+ *   <li>按 handlerType 分发消息到对应处理器</li>
+ *   <li>定时巡检连接状态，断线自动恢复</li>
+ * </ul>
+ *
+ * <p>使用示例：</p>
+ * <pre>
+ * // 1. 注册消息处理器
+ * customMqttDeviceMonitor.registerHandler(new PropertyMessageHandler());
+ *
+ * // 2. 注册设备配置
+ * CustomMqttDeviceConfig config = new CustomMqttDeviceConfig();
+ * config.setDeviceKey("sensor-001");
+ * config.setBrokerUrl("tcp://192.168.1.100:1883");
+ * config.setClientId("client-001");
+ * config.setSubscribeTopics(Arrays.asList("/sensor/data", "/sensor/status"));
+ * config.setHandlerType("property");
+ * customMqttDeviceMonitor.registerDevice(config);
+ *
+ * // 3. 启动监控
+ * customMqttDeviceMonitor.start();
+ * </pre>
+ */
+@Slf4j
+@Component
+public class CustomMqttDeviceMonitor {
+
+    /**
+     * 设备配置缓存：deviceKey -> 配置
+     */
+    private final ConcurrentHashMap<String, CustomMqttDeviceConfig> deviceConfigMap = new ConcurrentHashMap<>();
+
+    /**
+     * 设备连接缓存：deviceKey -> MqttClient
+     */
+    private final ConcurrentHashMap<String, MqttClient> clientMap = new ConcurrentHashMap<>();
+
+    /**
+     * 消息处理器缓存：handlerType -> Handler
+     */
+    private final ConcurrentHashMap<String, CustomMqttMessageHandler> handlerMap = new ConcurrentHashMap<>();
+
+    /**
+     * 心跳/重连定时任务
+     */
+    private ScheduledExecutorService scheduler;
+
+    /**
+     * 巡检间隔（秒）
+     */
+    private static final long MONITOR_INTERVAL_SECONDS = 10;
+
+    // ==================== 消息处理器管理 ====================
+
+
+    public CustomMqttMessageHandler getMessageHandler(String key){
+        return handlerMap.get(key);
+    }
+
+    /**
+     * 注册消息处理器
+     */
+    public void registerHandler(CustomMqttMessageHandler handler) {
+        handlerMap.put(handler.getHandlerType(), handler);
+        log.info("注册消息处理器: type={}", handler.getHandlerType());
+    }
+
+    /**
+     * 注销消息处理器
+     */
+    public void unregisterHandler(String handlerType) {
+        handlerMap.remove(handlerType);
+        log.info("注销消息处理器: type={}", handlerType);
+    }
+
+    // ==================== 设备配置管理 ====================
+
+    /**
+     * 注册设备（添加配置并立即尝试连接）
+     */
+    public void registerDevice(CustomMqttDeviceConfig config) {
+        if (config == null || config.getDeviceKey() == null) {
+            log.warn("注册设备失败: 配置或 deviceKey 为空");
+            return;
+        }
+        deviceConfigMap.put(config.getDeviceKey(), config);
+        log.info("注册设备配置: deviceKey={}, broker={}, topics={}",
+                config.getDeviceKey(), config.getBrokerUrl(), config.resolveSubscribeTopics());
+        connectDevice(config);
+    }
+
+    /**
+     * 批量注册设备
+     */
+    public void registerDevices(List<CustomMqttDeviceConfig> configs) {
+        if (configs != null) {
+            configs.forEach(this::registerDevice);
+        }
+    }
+
+    /**
+     * 注销设备（断开连接并移除配置）
+     */
+    public void unregisterDevice(String deviceKey) {
+        CustomMqttDeviceConfig config = deviceConfigMap.remove(deviceKey);
+        if (config != null) {
+            disconnectDevice(deviceKey);
+            log.info("注销设备: deviceKey={}", deviceKey);
+        }
+    }
+
+    /**
+     * 获取已注册设备配置
+     */
+    public CustomMqttDeviceConfig getDeviceConfig(String deviceKey) {
+        return deviceConfigMap.get(deviceKey);
+    }
+
+    /**
+     * 获取所有已注册设备的配置
+     */
+    public Map<String, CustomMqttDeviceConfig> getAllDeviceConfigs() {
+        return deviceConfigMap;
+    }
+
+    /**
+     * 检查设备是否在线
+     */
+    public boolean isDeviceOnline(String deviceKey) {
+        MqttClient client = clientMap.get(deviceKey);
+        return client != null && client.isConnected();
+    }
+
+    // ==================== 连接管理 ====================
+
+    /**
+     * 启动监控（定时巡检）
+     */
+    public void start() {
+        if (scheduler != null && !scheduler.isShutdown()) {
+            return;
+        }
+        scheduler = Executors.newSingleThreadScheduledExecutor();
+        scheduler.scheduleWithFixedDelay(() -> {
+            try {
+                monitorConnections();
+            } catch (Exception e) {
+                log.error("MQTT 连接巡检异常: {}", e.getMessage());
+            }
+        }, 0, MONITOR_INTERVAL_SECONDS, TimeUnit.SECONDS);
+        log.info("CustomMqttDeviceMonitor 已启动，巡检间隔={}s", MONITOR_INTERVAL_SECONDS);
+    }
+
+    /**
+     * 停止监控并关闭所有连接
+     */
+    @PreDestroy
+    public void stop() {
+        if (scheduler != null) {
+            scheduler.shutdown();
+        }
+        clientMap.keySet().forEach(this::disconnectDevice);
+        log.info("CustomMqttDeviceMonitor 已停止");
+    }
+
+    /**
+     * 建立单个设备的 MQTT 连接
+     */
+    private void connectDevice(CustomMqttDeviceConfig config) {
+        String deviceKey = config.getDeviceKey();
+        try {
+            // 如果已存在连接，先断开
+            disconnectDevice(deviceKey);
+
+            MqttClient client = new MqttClient(
+                    config.getBrokerUrl(),
+                    config.getClientId(),
+                    new MemoryPersistence());
+
+            MqttConnectOptions options = new MqttConnectOptions();
+            options.setCleanSession(config.getCleanSession() != null ? config.getCleanSession() : false);
+            options.setAutomaticReconnect(config.getAutomaticReconnect() != null ? config.getAutomaticReconnect() : true);
+            options.setConnectionTimeout(config.getConnectionTimeout() != null ? config.getConnectionTimeout() : 10);
+            options.setKeepAliveInterval(config.getKeepAliveInterval() != null ? config.getKeepAliveInterval() : 60);
+
+            if (config.getUsername() != null && !config.getUsername().isEmpty()) {
+                options.setUserName(config.getUsername());
+            }
+            if (config.getPassword() != null && !config.getPassword().isEmpty()) {
+                options.setPassword(config.getPassword().toCharArray());
+            }
+
+            // 设置回调：消息处理 + 连接状态
+            client.setCallback(new MqttCallback() {
+                @Override
+                public void connectionLost(Throwable cause) {
+                    log.warn("设备连接断开: deviceKey={}, reason={}", deviceKey, cause.getMessage());
+                    CustomMqttMessageHandler handler = findHandler(config.getHandlerType());
+                    if (handler != null) {
+                        handler.onConnectionLost(deviceKey, cause);
+                    }
+                }
+
+                @Override
+                public void messageArrived(String topic, MqttMessage message) {
+                    String payload = new String(message.getPayload());
+                    log.debug("收到消息: deviceKey={}, topic={}, payload={}", deviceKey, topic, payload);
+                    dispatchMessage(deviceKey, topic, payload, config);
+                }
+
+                @Override
+                public void deliveryComplete(IMqttDeliveryToken token) {
+                    // 发布完成回调，通常无需处理
+                }
+            });
+
+            client.connect(options);
+
+            // 保存连接
+            clientMap.put(deviceKey, client);
+            log.info("设备连接成功: deviceKey={}, broker={}", deviceKey, config.getBrokerUrl());
+
+            // 订阅主题
+            subscribeTopics(client, config);
+
+        } catch (MqttException e) {
+            log.error("设备连接失败: deviceKey={}, error={}", deviceKey, e.getMessage());
+            clientMap.remove(deviceKey);
+        }
+    }
+
+    /**
+     * 断开单个设备的 MQTT 连接
+     */
+    private void disconnectDevice(String deviceKey) {
+        MqttClient client = clientMap.remove(deviceKey);
+        if (client != null) {
+            try {
+                if (client.isConnected()) {
+                    client.disconnect();
+                }
+                client.close();
+                log.info("设备连接已断开: deviceKey={}", deviceKey);
+            } catch (MqttException e) {
+                log.warn("断开设备连接异常: deviceKey={}, error={}", deviceKey, e.getMessage());
+            }
+        }
+    }
+
+    /**
+     * 为设备订阅配置中的所有主题
+     * 自动将 topicTemplates 中的 %s 替换为 deviceKey
+     */
+    private void subscribeTopics(MqttClient client, CustomMqttDeviceConfig config) {
+        List<String> topics = config.resolveSubscribeTopics();
+        if (topics == null || topics.isEmpty()) {
+            log.warn("设备无订阅主题: deviceKey={}", config.getDeviceKey());
+            return;
+        }
+
+        String[] topicArray = topics.toArray(new String[0]);
+        int[] qosArray = new int[topics.size()];
+        for (int i = 0; i < qosArray.length; i++) {
+            qosArray[i] = 1;
+        }
+
+        try {
+            client.subscribe(topicArray, qosArray);
+            log.info("设备订阅成功: deviceKey={}, topics={}", config.getDeviceKey(), topics);
+
+            CustomMqttMessageHandler handler = findHandler(config.getHandlerType());
+            if (handler != null) {
+                for (String topic : topics) {
+                    handler.onSubscribed(config.getDeviceKey(), topic);
+                }
+            }
+        } catch (MqttException e) {
+            log.error("设备订阅失败: deviceKey={}, topics={}, error={}",
+                    config.getDeviceKey(), topics, e.getMessage());
+        }
+    }
+
+    /**
+     * 消息分发：根据 handlerType 找到对应处理器
+     *
+     * <p>匹配策略：</p>
+     * <ol>
+     *   <li>优先使用 supports() 匹配（支持通用处理器处理多种 handlerType）</li>
+     *   <li>回退到精确匹配（与 getHandlerType() 相同的 handler）</li>
+     * </ol>
+     */
+    private void dispatchMessage(String deviceKey, String topic, String payload, CustomMqttDeviceConfig config) {
+        String handlerType = config.getHandlerType();
+        if (handlerType == null || handlerType.isEmpty()) {
+            log.warn("设备未配置 handlerType，消息无法分发: deviceKey={}, topic={}", deviceKey, topic);
+            return;
+        }
+
+        CustomMqttMessageHandler handler = findHandler(handlerType);
+        if (handler == null) {
+            log.warn("未找到对应的消息处理器: handlerType={}, deviceKey={}, topic={}",
+                    handlerType, deviceKey, topic);
+            return;
+        }
+
+        try {
+            handler.handleMessage(deviceKey, topic, payload, config);
+        } catch (Exception e) {
+            log.error("消息处理异常: handlerType={}, deviceKey={}, topic={}, error={}",
+                    handlerType, deviceKey, topic, e.getMessage());
+        }
+    }
+
+    /**
+     * 查找处理器：先通过 supports() 匹配，再回退到精确匹配
+     */
+    private CustomMqttMessageHandler findHandler(String handlerType) {
+        // 优先：supports() 匹配（支持通用处理器）
+        for (CustomMqttMessageHandler handler : handlerMap.values()) {
+            if (handler.supports(handlerType)) {
+                return handler;
+            }
+        }
+        // 回退：精确匹配
+        return handlerMap.get(handlerType);
+    }
+
+    /**
+     * 定时巡检：检查所有连接状态，断线自动重连
+     */
+    private void monitorConnections() {
+        for (Map.Entry<String, CustomMqttDeviceConfig> entry : deviceConfigMap.entrySet()) {
+            String deviceKey = entry.getKey();
+            CustomMqttDeviceConfig config = entry.getValue();
+            MqttClient client = clientMap.get(deviceKey);
+
+            if (client == null || !client.isConnected()) {
+                log.info("检测到设备连接异常，尝试重连: deviceKey={}", deviceKey);
+                connectDevice(config);
+            }
+        }
+    }
+
+    // ==================== 工具方法 ====================
+
+    /**
+     * 向指定设备发布消息
+     */
+    public void publish(String deviceKey, String topic, String payload) throws MqttException {
+        publish(deviceKey, topic, payload, 1);
+    }
+
+    /**
+     * 向指定设备发布消息（指定 QoS）
+     */
+    public void publish(String deviceKey, String topic, String payload, int qos) throws MqttException {
+        MqttClient client = clientMap.get(deviceKey);
+        if (client == null || !client.isConnected()) {
+            log.error("设备未连接: {}", deviceKey);
+            return;
+        }
+        MqttMessage message = new MqttMessage(payload.getBytes());
+        message.setQos(qos);
+        client.publish(topic, message);
+        log.debug("发布消息: deviceKey={}, topic={}", deviceKey, topic);
+    }
+
+    /**
+     * 重新订阅指定设备的所有主题
+     */
+    public void resubscribe(String deviceKey) {
+        CustomMqttDeviceConfig config = deviceConfigMap.get(deviceKey);
+        MqttClient client = clientMap.get(deviceKey);
+        if (config != null && client != null && client.isConnected()) {
+            subscribeTopics(client, config);
+        }
+    }
+}

maibu-common/src/main/java/com/maibu/mqtt/CustomMqttMessageHandler.java

@@ -0,0 +1,56 @@
+package com.maibu.mqtt;
+
+/**
+ * 自定义 MQTT 消息处理器接口
+ * 外部设备订阅消息的处理逻辑实现此接口
+ */
+public interface CustomMqttMessageHandler {
+
+    /**
+     * 获取处理器类型标识
+     * 与 {@link CustomMqttDeviceConfig#getHandlerType()} 对应
+     */
+    String getHandlerType();
+
+    /**
+     * 判断当前处理器是否支持指定的 handlerType
+     *
+     * <p>默认实现为精确匹配（与 getHandlerType() 返回值相同）。</p>
+     * <p>通用处理器（如 GenericPropertyMessageHandler）可覆盖此方法，
+     * 实现一个处理器服务多种 handlerType 的需求。</p>
+     *
+     * @param handlerType 设备配置中的 handlerType
+     * @return 是否支持处理该类型的消息
+     */
+    default boolean supports(String handlerType) {
+        return getHandlerType().equals(handlerType);
+    }
+
+    /**
+     * 处理接收到的 MQTT 消息
+     *
+     * @param deviceKey  设备唯一标识
+     * @param topic      消息主题
+     * @param payload    消息内容
+     * @param config     设备配置
+     */
+    void handleMessage(String deviceKey, String topic, String payload, CustomMqttDeviceConfig config);
+
+    /**
+     * 连接断开回调
+     *
+     * @param deviceKey 设备唯一标识
+     * @param cause     断开原因
+     */
+    default void onConnectionLost(String deviceKey, Throwable cause) {
+    }
+
+    /**
+     * 订阅成功回调
+     *
+     * @param deviceKey 设备唯一标识
+     * @param topic     订阅主题
+     */
+    default void onSubscribed(String deviceKey, String topic) {
+    }
+}

maibu-common/src/main/java/com/maibu/mqtt/DefaultPropertyParser.java

@@ -0,0 +1,281 @@
+package com.maibu.mqtt;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Iterator;
+import java.util.Map;
+
+/**
+ * 默认属性数据解析器
+ *
+ * <p>支持三种 JSON 结构的解析：FLAT（扁平）、NESTED（嵌套）、KEY_VALUE_ARRAY（键值数组）</p>
+ *
+ * <p>使用 Jackson 进行 JSON 解析，支持点号路径访问嵌套字段。</p>
+ */
+@Slf4j
+public class DefaultPropertyParser implements PropertyParser {
+
+    private final ObjectMapper objectMapper = new ObjectMapper();
+
+    @Override
+    public PropertyData parse(String payload, PropertyMappingConfig config, String deviceKey, String topic) {
+        PropertyData data = new PropertyData();
+        data.setDeviceKey(deviceKey);
+        data.setRawPayload(payload);
+        data.setRawTopic(topic);
+        data.setTimestamp(System.currentTimeMillis());
+
+        try {
+            JsonNode root = objectMapper.readTree(payload);
+
+            // 提取 deviceKey（如果配置了路径）
+            if (config.getDeviceKeyPath() != null && !config.getDeviceKeyPath().isEmpty()) {
+                String extracted = getStringByPath(root, config.getDeviceKeyPath());
+                if (extracted != null) {
+                    data.setDeviceKey(extracted);
+                }
+            }
+
+            // 提取 timestamp（如果配置了路径）
+            if (config.getTimestampPath() != null && !config.getTimestampPath().isEmpty()) {
+                Long ts = getLongByPath(root, config.getTimestampPath());
+                if (ts != null) {
+                    data.setTimestamp(ts);
+                }
+            }
+
+            // 根据解析器类型执行不同的解析逻辑
+            switch (config.getParserType()) {
+                case FLAT:
+                case NESTED:
+                    parseByMappings(root, config, data);
+                    break;
+                case KEY_VALUE_ARRAY:
+                    parseKeyValueArray(root, config, data);
+                    break;
+                default:
+                    log.warn("未知的解析器类型: {}", config.getParserType());
+            }
+
+        } catch (Exception e) {
+            log.error("JSON 解析异常: payload={}, error={}", payload, e.getMessage());
+        }
+
+        return data;
+    }
+
+    /**
+     * 按字段映射列表解析（FLAT 和 NESTED 共用）
+     */
+    private void parseByMappings(JsonNode root, PropertyMappingConfig config, PropertyData data) {
+        if (config.getMappings() == null || config.getMappings().isEmpty()) {
+            return;
+        }
+        for (PropertyMappingConfig.FieldMapping mapping : config.getMappings()) {
+            try {
+                JsonNode node = getNodeByPath(root, mapping.getSource());
+                Object value;
+
+                if (node == null || node.isNull() || node.isMissingNode()) {
+                    value = mapping.getDefaultValue();
+                } else {
+                    value = convertValue(node, mapping.getValueType());
+                }
+
+                // 应用值转换
+                if (value != null && mapping.getValueTransform() != null) {
+                    value = applyTransform(value, mapping.getValueTransform());
+                }
+
+                if (value != null) {
+                    data.addProperty(mapping.getTarget(), value);
+                }
+
+            } catch (Exception e) {
+                log.warn("字段映射解析失败: source={}, target={}, error={}",
+                        mapping.getSource(), mapping.getTarget(), e.getMessage());
+            }
+        }
+    }
+
+    /**
+     * 解析键值数组结构
+     */
+    private void parseKeyValueArray(JsonNode root, PropertyMappingConfig config, PropertyData data) {
+        JsonNode arrayNode = getNodeByPath(root, config.getArrayPath());
+        if (arrayNode == null || !arrayNode.isArray()) {
+            log.warn("键值数组路径不存在或不是数组: path={}", config.getArrayPath());
+            return;
+        }
+
+        Iterator<JsonNode> items = arrayNode.elements();
+        while (items.hasNext()) {
+            JsonNode item = items.next();
+            String key = getStringField(item, config.getArrayKeyField());
+            Object value = getFieldValue(item, config.getArrayValueField());
+
+            if (key == null || value == null) {
+                continue;
+            }
+
+            // 应用 key 映射
+            if (config.getArrayKeyMapping() != null && config.getArrayKeyMapping().containsKey(key)) {
+                key = config.getArrayKeyMapping().get(key);
+            }
+
+            data.addProperty(key, value);
+        }
+    }
+
+    /**
+     * 通过点号路径获取 JSON 节点
+     * 示例: "data.temp.value" 对应 json.data.temp.value
+     */
+    private JsonNode getNodeByPath(JsonNode root, String path) {
+        if (path == null || path.isEmpty()) {
+            return null;
+        }
+        String[] parts = path.split("\\.");
+        JsonNode current = root;
+        for (String part : parts) {
+            if (current == null || current.isMissingNode()) {
+                return null;
+            }
+            current = current.get(part);
+        }
+        return current;
+    }
+
+    /**
+     * 获取字符串字段
+     */
+    private String getStringByPath(JsonNode root, String path) {
+        JsonNode node = getNodeByPath(root, path);
+        if (node != null && !node.isNull()) {
+            return node.asText();
+        }
+        return null;
+    }
+
+    /**
+     * 获取长整型字段
+     */
+    private Long getLongByPath(JsonNode root, String path) {
+        JsonNode node = getNodeByPath(root, path);
+        if (node != null && !node.isNull()) {
+            if (node.isNumber()) {
+                return node.asLong();
+            }
+            try {
+                return Long.parseLong(node.asText());
+            } catch (NumberFormatException e) {
+                return null;
+            }
+        }
+        return null;
+    }
+
+    /**
+     * 获取数组元素中的字符串字段
+     */
+    private String getStringField(JsonNode item, String fieldName) {
+        JsonNode node = item.get(fieldName);
+        if (node != null && !node.isNull()) {
+            return node.asText();
+        }
+        return null;
+    }
+
+    /**
+     * 获取数组元素中的值字段
+     */
+    private Object getFieldValue(JsonNode item, String fieldName) {
+        JsonNode node = item.get(fieldName);
+        if (node == null || node.isNull()) {
+            return null;
+        }
+        if (node.isNumber()) {
+            return node.numberValue();
+        }
+        if (node.isBoolean()) {
+            return node.booleanValue();
+        }
+        if (node.isTextual()) {
+            return node.asText();
+        }
+        return node.toString();
+    }
+
+    /**
+     * 根据 ValueType 转换 JSON 值
+     */
+    private Object convertValue(JsonNode node, PropertyMappingConfig.ValueType valueType) {
+        if (node == null || node.isNull()) {
+            return null;
+        }
+
+        switch (valueType) {
+            case STRING:
+                return node.asText();
+            case NUMBER:
+                if (node.isNumber()) {
+                    return node.numberValue();
+                }
+                try {
+                    String text = node.asText();
+                    if (text.contains(".") || text.contains("e") || text.contains("E")) {
+                        return Double.parseDouble(text);
+                    }
+                    return Long.parseLong(text);
+                } catch (NumberFormatException e) {
+                    return node.asText();
+                }
+            case BOOLEAN:
+                if (node.isBoolean()) {
+                    return node.booleanValue();
+                }
+                if (node.isNumber()) {
+                    return node.intValue() != 0;
+                }
+                return Boolean.parseBoolean(node.asText());
+            case AUTO:
+            default:
+                if (node.isNumber()) {
+                    return node.numberValue();
+                }
+                if (node.isBoolean()) {
+                    return node.booleanValue();
+                }
+                if (node.isTextual()) {
+                    return node.asText();
+                }
+                if (node.isArray() || node.isObject()) {
+                    return node.toString();
+                }
+                return node.asText();
+        }
+    }
+
+    /**
+     * 应用值转换映射
+     */
+    private Object applyTransform(Object value, Map<Object, Object> transform) {
+        if (value == null || transform == null || transform.isEmpty()) {
+            return value;
+        }
+        // 先尝试直接匹配
+        Object mapped = transform.get(value);
+        if (mapped != null) {
+            return mapped;
+        }
+        // 尝试字符串匹配（处理类型不一致）
+        for (Map.Entry<Object, Object> entry : transform.entrySet()) {
+            if (entry.getKey().toString().equals(value.toString())) {
+                return entry.getValue();
+            }
+        }
+        return value;
+    }
+}

maibu-common/src/main/java/com/maibu/mqtt/GenericPropertyMessageHandler.java

@@ -0,0 +1,241 @@
+package com.maibu.mqtt;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * 通用属性数据消息处理器
+ *
+ * <p>通过配置驱动的方式处理不同客户的 JSON 格式数据，实现业务逻辑与数据格式的解耦。</p>
+ *
+ * <h3>使用方式：</h3>
+ * <pre>
+ * // 1. 创建通用处理器
+ * GenericPropertyMessageHandler handler = new GenericPropertyMessageHandler();
+ *
+ * // 2. 为不同客户注册映射配置
+ * handler.registerMappingConfig("vendor-a", PropertyMappingConfig.flat(
+ *     Arrays.asList(
+ *         FieldMapping.of("temperature", "temperature"),
+ *         FieldMapping.of("humidity", "humidity"),
+ *         FieldMapping.of("state", "status")
+ *     )
+ * ));
+ *
+ * handler.registerMappingConfig("vendor-b", PropertyMappingConfig.nested(
+ *     Arrays.asList(
+ *         FieldMapping.of("data.temp", "temperature"),
+ *         FieldMapping.of("data.hum", "humidity"),
+ *         FieldMapping.ofTransform("data.code", "status", Map.of(1, "normal", 0, "error"))
+ *     )
+ * ));
+ *
+ * // 3. 注册到 MQTT 监控器
+ * customMqttDeviceMonitor.registerHandler(handler);
+ *
+ * // 4. 为不同设备配置对应的 handlerType
+ * configA.setHandlerType("vendor-a");
+ * configB.setHandlerType("vendor-b");
+ * </pre>
+ *
+ * <h3>执行流程：</h3>
+ * <pre>
+ * MQTT 消息到达
+ *   → CustomMqttDeviceMonitor.dispatchMessage()
+ *   → GenericPropertyMessageHandler.handleMessage()
+ *   → 根据 handlerType 查找映射配置
+ *   → DefaultPropertyParser.parse() 解析 JSON
+ *   → beforeProcess() 预处理（可扩展）
+ *   → afterProcess() 后处理（可扩展）
+ * </pre>
+ */
+@Slf4j
+public class GenericPropertyMessageHandler implements CustomMqttMessageHandler {
+
+    /**
+     * 处理器类型标识，与 CustomMqttDeviceConfig.handlerType 对应
+     */
+    private final String handlerType;
+
+    /**
+     * 映射配置缓存：handlerType → 映射配置
+     * 支持运行时动态注册和修改映射规则
+     */
+    private final ConcurrentHashMap<String, PropertyMappingConfig> mappingConfigMap = new ConcurrentHashMap<>();
+
+    /**
+     * 自定义解析器（可选），默认使用 DefaultPropertyParser
+     */
+    private PropertyParser parser;
+
+    /**
+     * 创建通用处理器，默认 handlerType 为 "property"
+     */
+    public GenericPropertyMessageHandler() {
+        this("property");
+    }
+
+    /**
+     * 创建通用处理器，指定 handlerType
+     */
+    public GenericPropertyMessageHandler(String handlerType) {
+        this.handlerType = handlerType;
+        this.parser = new DefaultPropertyParser();
+    }
+
+    /**
+     * 设置自定义解析器
+     */
+    public GenericPropertyMessageHandler setParser(PropertyParser parser) {
+        this.parser = parser;
+        return this;
+    }
+
+    // ==================== 映射配置管理 ====================
+
+    /**
+     * 注册映射配置
+     *
+     * @param handlerType 与设备配置中的 handlerType 对应
+     * @param config      字段映射配置
+     */
+    public GenericPropertyMessageHandler registerMappingConfig(String handlerType, PropertyMappingConfig config) {
+        if (handlerType == null || config == null) {
+            log.warn("注册映射配置失败: handlerType 或 config 为空");
+            return this;
+        }
+        mappingConfigMap.put(handlerType, config);
+        log.info("注册映射配置: handlerType={}, parserType={}, mappings={}",
+                handlerType, config.getParserType(), config.getMappings().size());
+        return this;
+    }
+
+    /**
+     * 注销映射配置
+     */
+    public GenericPropertyMessageHandler unregisterMappingConfig(String handlerType) {
+        mappingConfigMap.remove(handlerType);
+        log.info("注销映射配置: handlerType={}", handlerType);
+        return this;
+    }
+
+    /**
+     * 获取映射配置
+     */
+    public PropertyMappingConfig getMappingConfig(String handlerType) {
+        return mappingConfigMap.get(handlerType);
+    }
+
+    // ==================== CustomMqttMessageHandler 实现 ====================
+
+    @Override
+    public String getHandlerType() {
+        return handlerType;
+    }
+
+    /**
+     * 判断是否支持指定的 handlerType
+     *
+     * <p>支持条件：</p>
+     * <ol>
+     *   <li>精确匹配（handlerType 相同）</li>
+     *   <li>已注册映射配置（mappingConfigMap 中包含该 handlerType）</li>
+     * </ol>
+     */
+    @Override
+    public boolean supports(String handlerType) {
+        if (handlerType == null) {
+            return false;
+        }
+        return handlerType.equals(this.handlerType) || mappingConfigMap.containsKey(handlerType);
+    }
+
+    @Override
+    public void handleMessage(String deviceKey, String topic, String payload, CustomMqttDeviceConfig config) {
+        try {
+            PropertyMappingConfig mappingConfig = mappingConfigMap.get(config.getHandlerType());
+
+            // 如果没有映射配置，使用默认配置（直接按扁平结构解析）
+            if (mappingConfig == null) {
+                mappingConfig = createDefaultConfig(payload);
+            }
+
+            // 解析 JSON
+            PropertyData data = parser.parse(payload, mappingConfig, deviceKey, topic);
+
+            log.debug("属性解析完成: deviceKey={}, topic={}, properties={}",
+                    data.getDeviceKey(), data.getRawTopic(), data.getProperties());
+
+            // 预处理钩子
+            beforeProcess(data, config);
+            //todo
+            // 核心处理
+            process(data, config);
+
+            // 后处理钩子
+            afterProcess(data, config);
+
+        } catch (Exception e) {
+            log.error("属性消息处理异常: deviceKey={}, topic={}, error={}",
+                    deviceKey, topic, e.getMessage());
+        }
+    }
+
+    @Override
+    public void onConnectionLost(String deviceKey, Throwable cause) {
+        log.warn("设备连接断开: deviceKey={}, reason={}", deviceKey, cause.getMessage());
+    }
+
+    @Override
+    public void onSubscribed(String deviceKey, String topic) {
+        log.info("设备订阅主题: deviceKey={}, topic={}", deviceKey, topic);
+    }
+
+    // ==================== 可扩展的处理钩子 ====================
+
+    /**
+     * 核心处理方法：解析后的统一数据处理
+     *
+     * <p>默认实现为日志输出，子类或外部注册可覆盖此方法实现自定义业务逻辑。</p>
+     *
+     * <p>典型业务逻辑示例：</p>
+     * <ul>
+     *   <li>存入数据库</li>
+     *   <li>推送到 WebSocket</li>
+     *   <li>触发告警</li>
+     *   <li>转发到其他系统</li>
+     * </ul>
+     */
+    protected void process(PropertyData data, CustomMqttDeviceConfig config) {
+        log.info("【属性数据】deviceKey={}, properties={}", data.getDeviceKey(), data.getProperties());
+    }
+
+    /**
+     * 预处理钩子：在解析完成、核心处理之前执行
+     * 可用于数据校验、格式转换等
+     */
+    protected void beforeProcess(PropertyData data, CustomMqttDeviceConfig config) {
+    }
+
+    /**
+     * 后处理钩子：在核心处理之后执行
+     * 可用于日志记录、触发后续动作等
+     */
+    protected void afterProcess(PropertyData data, CustomMqttDeviceConfig config) {
+    }
+
+    // ==================== 私有辅助方法 ====================
+
+    /**
+     * 创建默认映射配置（用于未配置映射的情况）
+     * 会自动识别 JSON 结构并尝试提取所有字段
+     */
+    private PropertyMappingConfig createDefaultConfig(String payload) {
+        PropertyMappingConfig config = new PropertyMappingConfig();
+        config.setParserType(PropertyMappingConfig.ParserType.FLAT);
+        log.warn("未找到映射配置，使用默认解析: payload={}", payload);
+        return config;
+    }
+}

maibu-common/src/main/java/com/maibu/mqtt/PropertyData.java

@@ -0,0 +1,99 @@
+package com.maibu.mqtt;
+
+import lombok.Data;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * 属性数据统一内部结构
+ *
+ * <p>所有外部客户的 JSON 消息，经过解析后都转换成此结构，业务层只需要面对统一的数据格式。</p>
+ *
+ * <p>不同客户的 JSON 格式示例：</p>
+ * <pre>
+ * 客户A（扁平结构）:
+ * { "temperature": 25.5, "humidity": 60, "status": "normal" }
+ *
+ * 客户B（嵌套结构）:
+ * { "data": { "temp": 25.5, "hum": 60, "state": 1 } }
+ *
+ * 客户C（键值数组）:
+ * { "metrics": [ { "key": "temperature", "val": 25.5 }, { "key": "humidity", "val": 60 } ] }
+ *
+ * 经过统一转换后:
+ * PropertyData {
+ *   deviceKey = "device-001"
+ *   properties = { "temperature": 25.5, "humidity": 60, "status": "normal" }
+ * }
+ * </pre>
+ */
+@Data
+public class PropertyData {
+
+    /**
+     * 设备唯一标识
+     */
+    private String deviceKey;
+
+    /**
+     * 时间戳（毫秒），消息产生的时间
+     */
+    private Long timestamp;
+
+    /**
+     * 标准属性键值对
+     * key = 统一后的属性名（如 temperature, humidity, status）
+     * value = 属性值，支持任意类型
+     */
+    private Map<String, Object> properties = new HashMap<>();
+
+    /**
+     * 原始 JSON 报文，用于日志追溯
+     */
+    private String rawPayload;
+
+    /**
+     * 消息来源的原始 Topic
+     */
+    private String rawTopic;
+
+    /**
+     * 添加一个属性
+     */
+    public PropertyData addProperty(String name, Object value) {
+        this.properties.put(name, value);
+        return this;
+    }
+
+    /**
+     * 获取字符串属性
+     */
+    public String getString(String name) {
+        Object v = properties.get(name);
+        return v != null ? v.toString() : null;
+    }
+
+    /**
+     * 获取数值属性
+     */
+    public Number getNumber(String name) {
+        Object v = properties.get(name);
+        if (v instanceof Number) return (Number) v;
+        if (v instanceof String) {
+            try { return Double.parseDouble((String) v); } catch (Exception e) { return null; }
+        }
+        return null;
+    }
+
+    /**
+     * 获取布尔属性
+     */
+    public Boolean getBoolean(String name) {
+        Object v = properties.get(name);
+        if (v instanceof Boolean) return (Boolean) v;
+        if (v instanceof Number) return ((Number) v).intValue() != 0;
+        if (v instanceof String) return Boolean.parseBoolean((String) v);
+        return null;
+    }
+}

maibu-common/src/main/java/com/maibu/mqtt/PropertyMappingConfig.java

@@ -0,0 +1,245 @@
+package com.maibu.mqtt;
+
+import lombok.Data;
+
+import java.util.*;
+
+/**
+ * 属性字段映射配置
+ *
+ * <p>通过配置方式将外部客户的 JSON 字段映射到统一的内部属性名，实现不同客户数据格式的标准化。</p>
+ *
+ * <h3>支持三种 JSON 结构：</h3>
+ *
+ * <h4>1. 扁平结构（FLAT）</h4>
+ * <pre>
+ * JSON: { "temperature": 25.5, "humidity": 60, "state": "ok" }
+ * Config: parserType=FLAT
+ *   mappings: [
+ *     { source="temperature", target="temperature" },
+ *     { source="humidity", target="humidity" },
+ *     { source="state", target="status" }
+ *   ]
+ * Result: { temperature=25.5, humidity=60, status="ok" }
+ * </pre>
+ *
+ * <h4>2. 嵌套结构（NESTED）</h4>
+ * <pre>
+ * JSON: { "data": { "temp": 25.5, "hum": 60, "code": 1 } }
+ * Config: parserType=NESTED
+ *   mappings: [
+ *     { source="data.temp", target="temperature" },
+ *     { source="data.hum", target="humidity" },
+ *     { source="data.code", target="status", valueTransform={1:"normal", 0:"error"} }
+ *   ]
+ * Result: { temperature=25.5, humidity=60, status="normal" }
+ * </pre>
+ *
+ * <h4>3. 键值数组结构（KEY_VALUE_ARRAY）</h4>
+ * <pre>
+ * JSON: { "metrics": [ { "key": "temperature", "val": 25.5 }, { "key": "humidity", "val": 60 } ] }
+ * Config: parserType=KEY_VALUE_ARRAY
+ *   arrayPath: "metrics"
+ *   arrayKeyField: "key"
+ *   arrayValueField: "val"
+ * Result: { temperature=25.5, humidity=60 }
+ * </pre>
+ */
+@Data
+public class PropertyMappingConfig {
+
+    /**
+     * 解析器类型
+     */
+    public enum ParserType {
+        /**
+         * 扁平结构：字段直接在 JSON 根节点
+         * 示例: { "temperature": 25.5, "humidity": 60 }
+         */
+        FLAT,
+
+        /**
+         * 嵌套结构：字段在 JSON 的嵌套路径下
+         * 示例: { "data": { "temp": 25.5, "hum": 60 } }
+         * 使用点号路径，如 "data.temp"
+         */
+        NESTED,
+
+        /**
+         * 键值数组结构：字段以键值对数组形式存在
+         * 示例: { "metrics": [ { "key": "temperature", "val": 25.5 } ] }
+         * 通过 arrayPath 指定数组路径，arrayKeyField 和 arrayValueField 指定键值字段名
+         */
+        KEY_VALUE_ARRAY
+    }
+
+    /**
+     * 值类型，用于自动类型转换
+     */
+    public enum ValueType {
+        STRING,
+        NUMBER,
+        BOOLEAN,
+        /** 自动识别（默认），按 JSON 原始类型处理 */
+        AUTO
+    }
+
+    /**
+     * 字段映射定义
+     */
+    @Data
+    public static class FieldMapping {
+        /**
+         * 源字段路径（外部 JSON 中的字段名或路径）
+         * <ul>
+         *   <li>FLAT: 简单字段名，如 {@code temperature}</li>
+         *   <li>NESTED: 点号分隔路径，如 {@code data.temp.value}</li>
+         *   <li>KEY_VALUE_ARRAY: 不使用此字段</li>
+         * </ul>
+         */
+        private String source;
+
+        /**
+         * 目标属性名（统一后的内部属性名）
+         * 如 source 为 {@code data.temp}，target 为 {@code temperature}
+         */
+        private String target;
+
+        /**
+         * 值类型，用于自动类型转换，默认 AUTO
+         */
+        private ValueType valueType = ValueType.AUTO;
+
+        /**
+         * 值转换映射表，用于将原始值映射为标准值
+         * <p>示例: {@code {1: "normal", 0: "error"}}</p>
+         */
+        private Map<Object, Object> valueTransform;
+
+        /**
+         * 默认值，当源字段不存在时使用
+         */
+        private Object defaultValue;
+
+        /**
+         * 创建字段映射
+         */
+        public static FieldMapping of(String source, String target) {
+            FieldMapping m = new FieldMapping();
+            m.source = source;
+            m.target = target;
+            return m;
+        }
+
+        /**
+         * 创建字段映射（带类型转换）
+         */
+        public static FieldMapping of(String source, String target, ValueType valueType) {
+            FieldMapping m = new FieldMapping();
+            m.source = source;
+            m.target = target;
+            m.valueType = valueType;
+            return m;
+        }
+
+        /**
+         * 创建字段映射（带值转换）
+         */
+        public static FieldMapping ofTransform(String source, String target, Map<Object, Object> transform) {
+            FieldMapping m = new FieldMapping();
+            m.source = source;
+            m.target = target;
+            m.valueTransform = transform;
+            return m;
+        }
+    }
+
+    // ==================== 配置字段 ====================
+
+    /**
+     * 解析器类型，默认 FLAT
+     */
+    private ParserType parserType = ParserType.FLAT;
+
+    /**
+     * 字段映射列表（FLAT 和 NESTED 类型使用）
+     */
+    private List<FieldMapping> mappings = new ArrayList<>();
+
+    // ==================== 以下为公共字段提取配置 ====================
+
+    /**
+     * 设备ID提取路径（可选）
+     * 如果设置，会从 JSON 中提取设备ID覆盖配置中的 deviceKey
+     * <p>示例: "sn" 或 "data.deviceId"</p>
+     */
+    private String deviceKeyPath;
+
+    /**
+     * 时间戳提取路径（可选）
+     * 如果设置，会从 JSON 中提取时间戳
+     * <p>示例: "timestamp" 或 "data.time"</p>
+     */
+    private String timestampPath;
+
+    // ==================== KEY_VALUE_ARRAY 专用配置 ====================
+
+    /**
+     * 键值数组的路径（仅 KEY_VALUE_ARRAY 类型使用）
+     * <p>示例: JSON 为 {@code { "metrics": [...] }}，则 arrayPath = "metrics"</p>
+     */
+    private String arrayPath;
+
+    /**
+     * 键值数组中表示属性名的字段（仅 KEY_VALUE_ARRAY 类型使用）
+     * <p>默认 "key"</p>
+     */
+    private String arrayKeyField = "key";
+
+    /**
+     * 键值数组中表示属性值的字段（仅 KEY_VALUE_ARRAY 类型使用）
+     * <p>默认 "value"</p>
+     */
+    private String arrayValueField = "value";
+
+    /**
+     * 键值数组映射过滤表（仅 KEY_VALUE_ARRAY 类型使用，可选）
+     * <p>用于将数组中的 key 映射为统一的属性名</p>
+     * <p>示例: {@code {"temp": "temperature", "hum": "humidity"}}</p>
+     */
+    private Map<String, String> arrayKeyMapping;
+
+    // ==================== 便捷创建方法 ====================
+
+    /**
+     * 创建 FLAT 类型配置
+     */
+    public static PropertyMappingConfig flat(List<FieldMapping> mappings) {
+        PropertyMappingConfig config = new PropertyMappingConfig();
+        config.parserType = ParserType.FLAT;
+        config.mappings = mappings;
+        return config;
+    }
+
+    /**
+     * 创建 NESTED 类型配置
+     */
+    public static PropertyMappingConfig nested(List<FieldMapping> mappings) {
+        PropertyMappingConfig config = new PropertyMappingConfig();
+        config.parserType = ParserType.NESTED;
+        config.mappings = mappings;
+        return config;
+    }
+
+    /**
+     * 创建 KEY_VALUE_ARRAY 类型配置
+     */
+    public static PropertyMappingConfig keyValueArray(String arrayPath, String keyField, String valueField) {
+        PropertyMappingConfig config = new PropertyMappingConfig();
+        config.parserType = ParserType.KEY_VALUE_ARRAY;
+        config.arrayPath = arrayPath;
+        config.arrayKeyField = keyField;
+        config.arrayValueField = valueField;
+        return config;
+    }
+}

maibu-common/src/main/java/com/maibu/mqtt/PropertyParser.java

@@ -0,0 +1,25 @@
+package com.maibu.mqtt;
+
+/**
+ * 属性数据解析器接口
+ *
+ * <p>负责将外部客户的原始 JSON 报文解析为统一的 {@link PropertyData} 结构。</p>
+ *
+ * <p>不同客户的 JSON 格式不同，通过 {@link PropertyMappingConfig} 配置映射规则，
+ * 解析器按配置执行字段提取、类型转换和值映射。</p>
+ *
+ * <p>如需支持新的 JSON 结构，可实现此接口扩展自定义解析器。</p>
+ */
+public interface PropertyParser {
+
+    /**
+     * 解析 JSON 报文
+     *
+     * @param payload 原始 JSON 报文
+     * @param config  映射配置
+     * @param deviceKey 设备ID（如果 JSON 中未配置 deviceKeyPath，则使用此值）
+     * @param topic   原始 MQTT Topic
+     * @return 解析后的统一属性数据
+     */
+    PropertyData parse(String payload, PropertyMappingConfig config, String deviceKey, String topic);
+}