feat(wm-iot): #28 MQTT协议适配器 + 设备注册/发现API

- Device实体：设备SN/名称/类型/区域/位置/几何状态
- DeviceMapper：MyBatis-Plus完整CRUD操作
- DeviceService：设备生命周期管理+上线/下线
- DeviceController：REST API + 批量指令下发
- MqttConfig：配置属性注入
- MqttClientService：连接管理/订阅/消息处理
- MqttAdapter：协议适配器增强+指令下发
- Flyway迁移：iot_device表结构更新
- application.yml：MQTT配置属性添加

实现功能：
✅ MQTT协议连接/订阅/消息解析
✅ 设备注册/发现 REST API (CRUD)
✅ 统一设备模型实体(device_sn/type/area/position/geom)
✅ 设备状态管理 + 指令下发
✅ 批量指令下发接口

Issue #28

db/postgresql/V28__iot_device_migration.sql

@@ -0,0 +1,63 @@
+-- =============================================
+-- 智慧水务管理系统 - Issue #28: MQTT协议适配器 + 设备注册/发现API
+-- 版本: V28
+-- =============================================
+
+-- 更新 iot_device 表结构以匹配新的 Device 实体
+-- 添加 JSON 位置字段和 WKT 几何字段，保留原有字段用于兼容性
+
+-- 添加新的字段来统一设备模型
+ALTER TABLE iot_device 
+ADD COLUMN IF NOT EXISTS position JSONB,
+ADD COLUMN IF NOT EXISTS geom GEOMETRY(Point, 4326),
+ADD COLUMN IF NOT EXISTS remark TEXT;
+
+-- 更新现有字段的备注
+COMMENT ON COLUMN iot_device.position IS '设备位置信息 (JSON格式: {lng, lat, address})';
+COMMENT ON COLUMN iot_device.geom IS '设备几何位置 (WKT格式)';
+COMMENT ON COLUMN iot_device.remark IS '设备备注信息';
+
+-- 如果 position 和 geom 为空，从原有字段迁移数据
+-- 注意：这是一个可选的迁移步骤，根据实际需求决定是否执行
+-- UPDATE iot_device 
+-- SET position = json_build_object('lng', loc_lng, 'lat', loc_lat, 'address', address)
+-- WHERE loc_lng IS NOT NULL OR loc_lat IS NOT NULL;
+
+-- 添加索引
+CREATE INDEX IF NOT EXISTS idx_iot_device_position ON iot_device USING GIST(geom);
+CREATE INDEX IF NOT EXISTS idx_iot_device_area ON iot_device(area);
+CREATE INDEX IF NOT EXISTS idx_iot_device_type ON iot_device(device_type);
+CREATE INDEX IF NOT EXISTS idx_iot_device_status ON iot_device(status);
+
+-- 创建设备状态变更触发器函数
+CREATE OR REPLACE FUNCTION iot_device_update_status()
+RETURNS TRIGGER AS $$
+BEGIN
+    IF NEW.status <> OLD.status THEN
+        -- 记录设备状态变更事件
+        INSERT INTO iot_device_event (device_id, device_sn, event_type, event_data)
+        VALUES (NEW.id, NEW.device_sn, NEW.status, json_build_object('old_status', OLD.status));
+    END IF;
+    
+    RETURN NEW;
+END;
+$$ LANGUAGE plpgsql;
+
+-- 创建设备状态变更触发器
+CREATE TRIGGER tr_iot_device_status_change
+BEFORE UPDATE ON iot_device
+FOR EACH ROW
+WHEN (NEW.status <> OLD.status)
+EXECUTE FUNCTION iot_device_update_status();
+
+-- 插入示例数据
+INSERT INTO iot_device (device_sn, device_name, device_type, area, position, geom, status, remark) VALUES
+('SN001', '流量计-1号', 'flow_meter', '东部片区', 
+ '{"lng": 116.1234, "lat": 39.5678, "address": "水厂主入口"}',
+ ST_GeomFromText('POINT(116.1234 39.5678)', 4326), 
+ 'online', '主要流量监测设备'),
+('SN002', '压力传感器-1号', 'pressure_sensor', '西部片区',
+ '{"lng": 116.5678, "lat": 39.8765, "address": "调压站入口"}',
+ ST_GeomFromText('POINT(116.5678 39.8765)', 4326),
+ 'offline', '备用压力监测设备')
+ON CONFLICT (device_sn) DO NOTHING;

wm-iot/pom.xml

@@ -11,6 +11,12 @@
         <dependency><groupId>com.alibaba.cloud</groupId><artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId></dependency>
         <dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency>
         <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-data-redis</artifactId></dependency>
+        <dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-websocket</artifactId></dependency>
+        <dependency>
+            <groupId>org.eclipse.paho</groupId>
+            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
+            <version>1.2.5</version>
+        </dependency>
         <dependency><groupId>org.postgresql</groupId><artifactId>postgresql</artifactId></dependency>
         <dependency><groupId>net.postgis</groupId><artifactId>postgis-jdbc</artifactId></dependency>
         <dependency>

wm-iot/src/main/java/com/water/iot/adapter/MqttAdapter.java

@@ -1,6 +1,8 @@
 package com.water.iot.adapter;
 
 import com.fasterxml.jackson.databind.ObjectMapper;
+import com.water.iot.service.MqttClientService;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Component;
 
@@ -8,9 +10,11 @@ import java.util.*;
 
 @Slf4j
 @Component
+@RequiredArgsConstructor
 public class MqttAdapter implements ProtocolAdapter {
 
-    private final ObjectMapper mapper = new ObjectMapper();
+    private final ObjectMapper mapper;
+    private final MqttClientService mqttClientService;
 
     @Override
     public String protocol() { return "MQTT"; }
@@ -35,12 +39,63 @@ public class MqttAdapter implements ProtocolAdapter {
     @Override
     public byte[] encodeCommand(Map<String, Object> command) {
         try { return mapper.writeValueAsBytes(command); }
-        catch (Exception e) { return null; }
+        catch (Exception e) { 
+            log.error("MQTT encode command error: {}", e.getMessage());
+            return null; 
+        }
     }
 
     @Override
     public boolean authenticate(String deviceSn, String credential) {
-        // TODO: 从设备表查询校验
-        return true;
+        try {
+            // TODO: 可以扩展为从数据库查询设备并校验凭证
+            log.info("设备鉴权: deviceSn={}, credential={}", deviceSn, credential);
+            return true;
+        } catch (Exception e) {
+            log.error("设备鉴权失败: {}", e.getMessage());
+            return false;
+        }
+    }
+
+    /**
+     * 发送指令到设备
+     */
+    public boolean sendCommandToDevice(String deviceSn, Map<String, Object> command) {
+        try {
+            byte[] commandBytes = encodeCommand(command);
+            if (commandBytes == null) {
+                return false;
+            }
+            
+            String commandJson = new String(commandBytes);
+            boolean success = mqttClientService.sendCommandToDevice(deviceSn, commandJson);
+            
+            if (success) {
+                log.info("指令发送成功: deviceSn={}, command={}", deviceSn, commandJson);
+            } else {
+                log.error("指令发送失败: deviceSn={}, command={}", deviceSn, commandJson);
+            }
+            
+            return success;
+        } catch (Exception e) {
+            log.error("发送指令异常: {}", e.getMessage());
+            return false;
+        }
+    }
+
+    /**
+     * 发布设备状态
+     */
+    public boolean publishDeviceStatus(String deviceSn, String status) {
+        try {
+            String topic = "water/iot/status/" + deviceSn;
+            String payload = String.format("{\"deviceSn\": \"%s\", \"status\": \"%s\", \"timestamp\": %d}", 
+                deviceSn, status, System.currentTimeMillis());
+            
+            return mqttClientService.publish(topic, payload);
+        } catch (Exception e) {
+            log.error("发布设备状态异常: {}", e.getMessage());
+            return false;
+        }
     }
 }

wm-iot/src/main/java/com/water/iot/config/MqttConfig.java

@@ -0,0 +1,50 @@
+package com.water.iot.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+/**
+ * MQTT 配置
+ */
+@Data
+@Configuration
+@ConfigurationProperties(prefix = "mqtt")
+public class MqttConfig {
+
+    /** MQTT 服务器地址 */
+    private String brokerUrl = "tcp://localhost:1883";
+
+    /** 客户端ID */
+    private String clientId = "water-iot-client";
+
+    /** 用户名 */
+    private String username = "admin";
+
+    /** 密码 */
+    private String password = "password";
+
+    /** 连接超时时间(秒) */
+    private int connectionTimeout = 10;
+
+    /** 保持连接时间(秒) */
+    private int keepAliveInterval = 20;
+
+    /** 自动重连 */
+    private boolean automaticReconnect = true;
+
+    /** 清理会话 */
+    private boolean cleanSession = true;
+
+    /** 发布主题前缀 */
+    private String publishTopicPrefix = "water/iot/";
+
+    /** 订阅主题前缀 */
+    private String subscribeTopicPrefix = "water/iot/";
+
+    /** 设备状态主题 */
+    private String deviceStatusTopic = "device/status/+";
+
+    /** 设备遥测数据主题 */
+    private String deviceTelemetryTopic = "device/telemetry/+";
+}

wm-iot/src/main/java/com/water/iot/controller/DeviceController.java

@@ -1,14 +1,17 @@
 package com.water.iot.controller;
 
 import com.water.common.core.result.R;
+import com.water.iot.adapter.MqttAdapter;
+import com.water.iot.entity.Device;
+import com.water.iot.service.DeviceService;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.tags.Tag;
 import lombok.RequiredArgsConstructor;
-import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.web.bind.annotation.*;
 
 import java.util.List;
 import java.util.Map;
+import java.util.HashMap;
 
 @Tag(name = "设备管理")
 @RestController
@@ -16,37 +19,199 @@ import java.util.Map;
 @RequiredArgsConstructor
 public class DeviceController {
 
-    private final JdbcTemplate jdbcTemplate;
+    private final DeviceService deviceService;
+    private final MqttAdapter mqttAdapter;
+
+    // 批量指令请求内部类
+    public static class BatchCommandRequest {
+        private List<String> deviceSns;
+        private Map<String, Object> command;
+        
+        public List<String> getDeviceSns() {
+            return deviceSns;
+        }
+        
+        public void setDeviceSns(List<String> deviceSns) {
+            this.deviceSns = deviceSns;
+        }
+        
+        public Map<String, Object> getCommand() {
+            return command;
+        }
+        
+        public void setCommand(Map<String, Object> command) {
+            this.command = command;
+        }
+    }
 
     @Operation(summary = "设备列表")
     @GetMapping("/list")
-    public R<List<Map<String, Object>>> list(@RequestParam(defaultValue = "1") int page,
-                                               @RequestParam(defaultValue = "10") int size) {
-        int offset = (page - 1) * size;
-        String sql = "SELECT id, device_sn, device_name, device_type, area, status, last_report_time FROM iot_device ORDER BY id LIMIT ? OFFSET ?";
-        return R.ok(jdbcTemplate.queryForList(sql, size, offset));
+    public R<List<Device>> list(@RequestParam(defaultValue = "1") int page,
+                                @RequestParam(defaultValue = "10") int size) {
+        List<Device> devices = deviceService.getDevicesPage(page, size);
+        return R.ok(devices);
     }
 
     @Operation(summary = "设备详情")
     @GetMapping("/{id}")
-    public R<Map<String, Object>> getById(@PathVariable Long id) {
-        return R.ok(jdbcTemplate.queryForMap("SELECT * FROM iot_device WHERE id = ?", id));
+    public R<Device> getById(@PathVariable Long id) {
+        Device device = deviceService.getDeviceById(id);
+        if (device == null) {
+            return R.failed("设备不存在");
+        }
+        return R.ok(device);
     }
 
     @Operation(summary = "注册设备")
     @PostMapping
-    public R<String> register(@RequestBody Map<String, Object> body) {
-        jdbcTemplate.update(
-            "INSERT INTO iot_device (device_sn, device_name, device_type, area, loc_lng, loc_lat) VALUES (?,?,?,?,?,?)",
-            body.get("deviceSn"), body.get("deviceName"), body.get("deviceType"), body.get("area"),
-            body.get("lng"), body.get("lat"));
-        return R.ok("注册成功");
+    public R<Device> register(@RequestBody Device device) {
+        try {
+            Device registeredDevice = deviceService.registerDevice(device);
+            return R.ok(registeredDevice, "设备注册成功");
+        } catch (Exception e) {
+            return R.failed("设备注册失败: " + e.getMessage());
+        }
+    }
+
+    @Operation(summary = "设备详情")
+    @GetMapping("/sn/{deviceSn}")
+    public R<Device> getByDeviceSn(@PathVariable String deviceSn) {
+        Device device = deviceService.getDeviceByDeviceSn(deviceSn);
+        if (device == null) {
+            return R.failed("设备不存在");
+        }
+        return R.ok(device);
+    }
+
+    @Operation(summary = "更新设备")
+    @PutMapping("/{id}")
+    public R<Device> update(@PathVariable Long id, @RequestBody Device device) {
+        device.setId(id);
+        try {
+            Device updatedDevice = deviceService.updateDevice(device);
+            return R.ok(updatedDevice, "设备更新成功");
+        } catch (Exception e) {
+            return R.failed("设备更新失败: " + e.getMessage());
+        }
+    }
+
+    @Operation(summary = "删除设备")
+    @DeleteMapping("/{id}")
+    public R<String> delete(@PathVariable Long id) {
+        try {
+            boolean success = deviceService.deleteDevice(id);
+            if (success) {
+                return R.ok("设备删除成功");
+            } else {
+                return R.failed("设备删除失败");
+            }
+        } catch (Exception e) {
+            return R.failed("设备删除失败: " + e.getMessage());
+        }
+    }
+
+    @Operation(summary = "设备上线")
+    @PostMapping("/{deviceSn}/online")
+    public R<String> online(@PathVariable String deviceSn) {
+        try {
+            deviceService.deviceOnline(deviceSn);
+            return R.ok("设备上线成功");
+        } catch (Exception e) {
+            return R.failed("设备上线失败: " + e.getMessage());
+        }
+    }
+
+    @Operation(summary = "设备下线")
+    @PostMapping("/{deviceSn}/offline")
+    public R<String> offline(@PathVariable String deviceSn) {
+        try {
+            deviceService.deviceOffline(deviceSn);
+            return R.ok("设备下线成功");
+        } catch (Exception e) {
+            return R.failed("设备下线失败: " + e.getMessage());
+        }
+    }
+
+    @Operation(summary = "根据区域查询设备")
+    @GetMapping("/area/{area}")
+    public R<List<Device>> getByArea(@PathVariable String area) {
+        List<Device> devices = deviceService.getDevicesByArea(area);
+        return R.ok(devices);
+    }
+
+    @Operation(summary = "根据设备类型查询设备")
+    @GetMapping("/type/{deviceType}")
+    public R<List<Device>> getByDeviceType(@PathVariable String deviceType) {
+        List<Device> devices = deviceService.getDevicesByDeviceType(deviceType);
+        return R.ok(devices);
+    }
+
+    @Operation(summary = "查询设备总数")
+    @GetMapping("/count")
+    public R<Long> count() {
+        long count = deviceService.countDevices();
+        return R.ok(count);
+    }
+
+    @Operation(summary = "设备发现（所有在线设备）")
+    @GetMapping("/discover")
+    public R<List<Device>> discover() {
+        // TODO: 可以扩展为更复杂的发现逻辑，如根据心跳时间判断
+        List<Device> allDevices = deviceService.getAllDevices();
+        // 过滤出在线设备
+        List<Device> onlineDevices = allDevices.stream()
+                .filter(device -> "online".equals(device.getStatus()))
+                .toList();
+        return R.ok(onlineDevices);
     }
 
     @Operation(summary = "下发指令")
     @PostMapping("/{id}/command")
     public R<String> sendCommand(@PathVariable Long id, @RequestBody Map<String, Object> cmd) {
-        // TODO: 实际指令通过 Kafka -> EMQX -> MQTT -> 设备
-        return R.ok("指令已下发");
+        try {
+            Device device = deviceService.getDeviceById(id);
+            if (device == null) {
+                return R.failed("设备不存在");
+            }
+            
+            // 添加设备SN到指令
+            cmd.put("deviceSn", device.getDeviceSn());
+            cmd.put("timestamp", System.currentTimeMillis());
+            
+            boolean success = mqttAdapter.sendCommandToDevice(device.getDeviceSn(), cmd);
+            if (success) {
+                return R.ok("指令已下发");
+            } else {
+                return R.failed("指令下发失败");
+            }
+        } catch (Exception e) {
+            return R.failed("指令下发失败: " + e.getMessage());
+        }
+    }
+
+    @Operation(summary = "批量下发指令")
+    @PostMapping("/command/batch")
+    public R<String> sendBatchCommand(@RequestBody BatchCommandRequest request) {
+        try {
+            int successCount = 0;
+            int failCount = 0;
+            
+            for (String deviceSn : request.getDeviceSns()) {
+                Map<String, Object> cmd = new HashMap<>(request.getCommand());
+                cmd.put("deviceSn", deviceSn);
+                cmd.put("timestamp", System.currentTimeMillis());
+                
+                boolean success = mqttAdapter.sendCommandToDevice(deviceSn, cmd);
+                if (success) {
+                    successCount++;
+                } else {
+                    failCount++;
+                }
+            }
+            
+            return R.ok(String.format("批量指令下发完成，成功: %d，失败: %d", successCount, failCount));
+        } catch (Exception e) {
+            return R.failed("批量指令下发失败: " + e.getMessage());
+        }
     }
 }

wm-iot/src/main/java/com/water/iot/entity/Device.java

@@ -0,0 +1,47 @@
+package com.water.iot.entity;
+
+import lombok.Data;
+import java.time.LocalDateTime;
+
+/**
+ * 统一设备模型实体
+ */
+@Data
+public class Device {
+
+    /** 主键ID */
+    private Long id;
+
+    /** 设备SN(唯一标识) */
+    private String deviceSn;
+
+    /** 设备名称 */
+    private String deviceName;
+
+    /** 设备类型：传感器/执行器/网关等 */
+    private String deviceType;
+
+    /** 所属区域 */
+    private String area;
+
+    /** 位置信息JSON：{"lng": 经度, "lat": 纬度, "address": 地址} */
+    private String position;
+
+    /** 地理几何信息(GIS)：WKT格式 */
+    private String geom;
+
+    /** 设备状态：在线/离线/故障 */
+    private String status;
+
+    /** 最后上报时间 */
+    private LocalDateTime lastReportTime;
+
+    /** 创建时间 */
+    private LocalDateTime createTime;
+
+    /** 更新时间 */
+    private LocalDateTime updateTime;
+
+    /** 备注 */
+    private String remark;
+}

wm-iot/src/main/java/com/water/iot/mapper/DeviceMapper.java

@@ -0,0 +1,101 @@
+package com.water.iot.mapper;
+
+import com.water.iot.entity.Device;
+import org.apache.ibatis.annotations.*;
+import java.util.List;
+
+/**
+ * 设备数据访问层
+ */
+@Mapper
+public interface DeviceMapper {
+
+    /**
+     * 插入设备
+     */
+    @Insert("INSERT INTO iot_device (device_sn, device_name, device_type, area, position, geom, status, create_time, update_time, remark) " +
+            "VALUES (#{deviceSn}, #{deviceName}, #{deviceType}, #{area}, #{position}, #{geom}, #{status}, #{createTime}, #{updateTime}, #{remark})")
+    @Options(useGeneratedKeys = true, keyProperty = "id")
+    int insert(Device device);
+
+    /**
+     * 根据ID更新设备
+     */
+    @Update("UPDATE iot_device SET device_sn = #{deviceSn}, device_name = #{deviceName}, device_type = #{deviceType}, " +
+            "area = #{area}, position = #{position}, geom = #{geom}, status = #{status}, " +
+            "last_report_time = #{lastReportTime}, update_time = #{updateTime}, remark = #{remark} " +
+            "WHERE id = #{id}")
+    int updateById(Device device);
+
+    /**
+     * 根据设备SN更新设备
+     */
+    @Update("UPDATE iot_device SET device_name = #{deviceName}, device_type = #{deviceType}, area = #{area}, " +
+            "position = #{position}, geom = #{geom}, status = #{status}, last_report_time = #{lastReportTime}, " +
+            "update_time = #{updateTime}, remark = #{remark} " +
+            "WHERE device_sn = #{deviceSn}")
+    int updateByDeviceSn(Device device);
+
+    /**
+     * 根据ID删除设备
+     */
+    @Delete("DELETE FROM iot_device WHERE id = #{id}")
+    int deleteById(@Param("id") Long id);
+
+    /**
+     * 根据设备SN删除设备
+     */
+    @Delete("DELETE FROM iot_device WHERE device_sn = #{deviceSn}")
+    int deleteByDeviceSn(@Param("deviceSn") String deviceSn);
+
+    /**
+     * 根据ID查询设备
+     */
+    @Select("SELECT id, device_sn, device_name, device_type, area, position, geom, status, " +
+            "last_report_time, create_time, update_time, remark FROM iot_device WHERE id = #{id}")
+    Device selectById(@Param("id") Long id);
+
+    /**
+     * 根据设备SN查询设备
+     */
+    @Select("SELECT id, device_sn, device_name, device_type, area, position, geom, status, " +
+            "last_report_time, create_time, update_time, remark FROM iot_device WHERE device_sn = #{deviceSn}")
+    Device selectByDeviceSn(@Param("deviceSn") String deviceSn);
+
+    /**
+     * 查询所有设备
+     */
+    @Select("SELECT id, device_sn, device_name, device_type, area, position, geom, status, " +
+            "last_report_time, create_time, update_time, remark FROM iot_device ORDER BY id")
+    List<Device> selectAll();
+
+    /**
+     * 分页查询设备列表
+     */
+    @Select("SELECT id, device_sn, device_name, device_type, area, position, geom, status, " +
+            "last_report_time, create_time, update_time, remark FROM iot_device " +
+            "ORDER BY id LIMIT #{limit} OFFSET #{offset}")
+    List<Device> selectPage(@Param("offset") int offset, @Param("limit") int limit);
+
+    /**
+     * 根据区域查询设备
+     */
+    @Select("SELECT id, device_sn, device_name, device_type, area, position, geom, status, " +
+            "last_report_time, create_time, update_time, remark FROM iot_device WHERE area = #{area} " +
+            "ORDER BY id")
+    List<Device> selectByArea(@Param("area") String area);
+
+    /**
+     * 根据设备类型查询设备
+     */
+    @Select("SELECT id, device_sn, device_name, device_type, area, position, geom, status, " +
+            "last_report_time, create_time, update_time, remark FROM iot_device WHERE device_type = #{deviceType} " +
+            "ORDER BY id")
+    List<Device> selectByDeviceType(@Param("deviceType") String deviceType);
+
+    /**
+     * 查询设备总数
+     */
+    @Select("SELECT COUNT(*) FROM iot_device")
+    int countTotal();
+}

wm-iot/src/main/java/com/water/iot/service/DeviceService.java

@@ -0,0 +1,181 @@
+package com.water.iot.service;
+
+import com.water.iot.entity.Device;
+import com.water.iot.mapper.DeviceMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.time.LocalDateTime;
+import java.util.List;
+
+/**
+ * 设备管理服务
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class DeviceService {
+
+    private final DeviceMapper deviceMapper;
+
+    /**
+     * 注册设备
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public Device registerDevice(Device device) {
+        // 设置创建时间和更新时间
+        LocalDateTime now = LocalDateTime.now();
+        device.setCreateTime(now);
+        device.setUpdateTime(now);
+        
+        // 设置默认状态
+        if (device.getStatus() == null) {
+            device.setStatus("offline");
+        }
+        
+        // 插入设备
+        int result = deviceMapper.insert(device);
+        if (result <= 0) {
+            throw new RuntimeException("设备注册失败");
+        }
+        
+        log.info("设备注册成功: deviceSn={}, deviceName={}", device.getDeviceSn(), device.getDeviceName());
+        return device;
+    }
+
+    /**
+     * 更新设备信息
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public Device updateDevice(Device device) {
+        device.setUpdateTime(LocalDateTime.now());
+        int result = deviceMapper.updateById(device);
+        if (result <= 0) {
+            throw new RuntimeException("设备更新失败");
+        }
+        return device;
+    }
+
+    /**
+     * 根据设备SN更新设备
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public Device updateDeviceByDeviceSn(Device device) {
+        device.setUpdateTime(LocalDateTime.now());
+        int result = deviceMapper.updateByDeviceSn(device);
+        if (result <= 0) {
+            throw new RuntimeException("设备更新失败");
+        }
+        return device;
+    }
+
+    /**
+     * 删除设备
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public boolean deleteDevice(Long id) {
+        int result = deviceMapper.deleteById(id);
+        return result > 0;
+    }
+
+    /**
+     * 根据设备SN删除设备
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public boolean deleteDeviceByDeviceSn(String deviceSn) {
+        int result = deviceMapper.deleteByDeviceSn(deviceSn);
+        return result > 0;
+    }
+
+    /**
+     * 根据ID查询设备
+     */
+    public Device getDeviceById(Long id) {
+        return deviceMapper.selectById(id);
+    }
+
+    /**
+     * 根据设备SN查询设备
+     */
+    public Device getDeviceByDeviceSn(String deviceSn) {
+        return deviceMapper.selectByDeviceSn(deviceSn);
+    }
+
+    /**
+     * 查询所有设备
+     */
+    public List<Device> getAllDevices() {
+        return deviceMapper.selectAll();
+    }
+
+    /**
+     * 分页查询设备列表
+     */
+    public List<Device> getDevicesPage(int page, int size) {
+        int offset = (page - 1) * size;
+        return deviceMapper.selectPage(offset, size);
+    }
+
+    /**
+     * 根据区域查询设备
+     */
+    public List<Device> getDevicesByArea(String area) {
+        return deviceMapper.selectByArea(area);
+    }
+
+    /**
+     * 根据设备类型查询设备
+     */
+    public List<Device> getDevicesByDeviceType(String deviceType) {
+        return deviceMapper.selectByDeviceType(deviceType);
+    }
+
+    /**
+     * 查询设备总数
+     */
+    public int countDevices() {
+        return deviceMapper.countTotal();
+    }
+
+    /**
+     * 设备上线
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public void deviceOnline(String deviceSn) {
+        Device device = new Device();
+        device.setDeviceSn(deviceSn);
+        device.setStatus("online");
+        device.setUpdateTime(LocalDateTime.now());
+        deviceMapper.updateByDeviceSn(device);
+        
+        log.info("设备上线: deviceSn={}", deviceSn);
+    }
+
+    /**
+     * 设备下线
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public void deviceOffline(String deviceSn) {
+        Device device = new Device();
+        device.setDeviceSn(deviceSn);
+        device.setStatus("offline");
+        device.setUpdateTime(LocalDateTime.now());
+        deviceMapper.updateByDeviceSn(device);
+        
+        log.info("设备下线: deviceSn={}", deviceSn);
+    }
+
+    /**
+     * 更新设备最后上报时间
+     */
+    @Transactional(rollbackFor = Exception.class)
+    public void updateLastReportTime(String deviceSn) {
+        Device device = new Device();
+        device.setDeviceSn(deviceSn);
+        device.setLastReportTime(LocalDateTime.now());
+        device.setUpdateTime(LocalDateTime.now());
+        deviceMapper.updateByDeviceSn(device);
+    }
+}

wm-iot/src/main/java/com/water/iot/service/MqttClientService.java

@@ -0,0 +1,211 @@
+package com.water.iot.service;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.water.iot.config.MqttConfig;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.*;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
+import java.util.UUID;
+
+/**
+ * MQTT 客户端服务
+ */
+@Slf4j
+@Service
+@RequiredArgsConstructor
+public class MqttClientService implements MqttCallback {
+
+    private final MqttConfig mqttConfig;
+    private final DeviceService deviceService;
+    private final ObjectMapper objectMapper;
+
+    private MqttClient mqttClient;
+    private volatile boolean connected = false;
+
+    /**
+     * 初始化 MQTT 客户端
+     */
+    @PostConstruct
+    public void init() {
+        try {
+            String clientId = mqttConfig.getClientId() + "-" + UUID.randomUUID().toString();
+            mqttClient = new MqttClient(
+                mqttConfig.getBrokerUrl(),
+                clientId,
+                new MemoryPersistence()
+            );
+
+            // 设置回调
+            mqttClient.setCallback(this);
+
+            // 连接选项
+            MqttConnectOptions options = new MqttConnectOptions();
+            options.setCleanSession(mqttConfig.isCleanSession());
+            options.setConnectionTimeout(mqttConfig.getConnectionTimeout());
+            options.setKeepAliveInterval(mqttConfig.getKeepAliveInterval());
+            options.setAutomaticReconnect(mqttConfig.isAutomaticReconnect());
+            options.setUserName(mqttConfig.getUsername());
+            options.setPassword(mqttConfig.getPassword().toCharArray());
+
+            // 连接
+            log.info("正在连接 MQTT 服务器: {}", mqttConfig.getBrokerUrl());
+            mqttClient.connect(options);
+            connected = true;
+
+            // 订阅主题
+            subscribeTopics();
+
+            log.info("MQTT 客户端连接成功");
+        } catch (Exception e) {
+            log.error("MQTT 客户端连接失败", e);
+            connected = false;
+        }
+    }
+
+    /**
+     * 订阅主题
+     */
+    private void subscribeTopics() {
+        try {
+            // 订阅设备状态主题
+            mqttClient.subscribe(mqttConfig.getDeviceStatusTopic(), this::handleDeviceStatus);
+            
+            // 订阅设备遥测数据主题
+            mqttClient.subscribe(mqttConfig.getDeviceTelemetryTopic(), this::handleDeviceTelemetry);
+
+            log.info("已订阅主题: {}, {}", mqttConfig.getDeviceStatusTopic(), mqttConfig.getDeviceTelemetryTopic());
+        } catch (Exception e) {
+            log.error("订阅主题失败", e);
+        }
+    }
+
+    /**
+     * 处理设备状态消息
+     */
+    private void handleDeviceStatus(String topic, MqttMessage message) {
+        try {
+            // topic 格式: device/status/{deviceSn}
+            String deviceSn = topic.split("/")[2];
+            String payload = new String(message.getPayload());
+            
+            log.info("收到设备状态消息: deviceSn={}, status={}", deviceSn, payload);
+            
+            if ("online".equals(payload)) {
+                deviceService.deviceOnline(deviceSn);
+            } else if ("offline".equals(payload)) {
+                deviceService.deviceOffline(deviceSn);
+            }
+        } catch (Exception e) {
+            log.error("处理设备状态消息失败", e);
+        }
+    }
+
+    /**
+     * 处理设备遥测数据
+     */
+    private void handleDeviceTelemetry(String topic, MqttMessage message) {
+        try {
+            // topic 格式: device/telemetry/{deviceSn}
+            String deviceSn = topic.split("/")[2];
+            String payload = new String(message.getPayload());
+            
+            log.info("收到设备遥测数据: deviceSn={}, payload={}", deviceSn, payload);
+            
+            // 解析遥测数据
+            @SuppressWarnings("unchecked")
+            Map<String, Object> telemetryData = objectMapper.readValue(payload, Map.class);
+            
+            // 更新设备最后上报时间
+            deviceService.updateLastReportTime(deviceSn);
+            
+            // TODO: 可以进一步处理遥测数据，如存储到时序数据库或触发告警
+            
+        } catch (Exception e) {
+            log.error("处理设备遥测数据失败", e);
+        }
+    }
+
+    /**
+     * 发布消息
+     */
+    public boolean publish(String topic, String payload) {
+        try {
+            if (!connected || !mqttClient.isConnected()) {
+                log.warn("MQTT 客户端未连接，无法发布消息");
+                return false;
+            }
+
+            MqttMessage message = new MqttMessage(payload.getBytes());
+            message.setQos(1);
+            message.setRetained(false);
+            
+            mqttClient.publish(topic, message);
+            log.info("发布消息成功: topic={}, payload={}", topic, payload);
+            return true;
+        } catch (Exception e) {
+            log.error("发布消息失败", e);
+            return false;
+        }
+    }
+
+    /**
+     * 发布指令到设备
+     */
+    public boolean sendCommandToDevice(String deviceSn, String command) {
+        String topic = mqttConfig.getPublishTopicPrefix() + "command/" + deviceSn;
+        return publish(topic, command);
+    }
+
+    /**
+     * 连接丢失
+     */
+    @Override
+    public void connectionLost(Throwable cause) {
+        connected = false;
+        log.error("MQTT 连接丢失", cause);
+    }
+
+    /**
+     * 消息送达
+     */
+    @Override
+    public void deliveryComplete(IMqttDeliveryToken token) {
+        // 消息送达回调
+    }
+
+    /**
+     * 消息到达
+     */
+    @Override
+    public void messageArrived(String topic, MqttMessage message) throws Exception {
+        // 已在上面处理
+    }
+
+    /**
+     * 销毁
+     */
+    @PreDestroy
+    public void destroy() {
+        try {
+            if (mqttClient != null && mqttClient.isConnected()) {
+                mqttClient.disconnect();
+                mqttClient.close();
+                log.info("MQTT 客户端已断开连接");
+            }
+        } catch (Exception e) {
+            log.error("关闭 MQTT 客户端失败", e);
+        }
+    }
+
+    /**
+     * 获取连接状态
+     */
+    public boolean isConnected() {
+        return connected && mqttClient != null && mqttClient.isConnected();
+    }
+}

wm-iot/src/main/resources/application.yml

@@ -20,3 +20,18 @@ spring:
     nacos:
       discovery:
         server-addr: ${NACOS_HOST:127.0.0.1}:8848
+
+# MQTT 配置
+mqtt:
+  broker-url: tcp://${MQTT_HOST:127.0.0.1}:1883
+  client-id: water-iot-client
+  username: ${MQTT_USERNAME:admin}
+  password: ${MQTT_PASSWORD:password}
+  connection-timeout: 10
+  keep-alive-interval: 20
+  automatic-reconnect: true
+  clean-session: true
+  publish-topic-prefix: water/iot/
+  subscribe-topic-prefix: water/iot/
+  device-status-topic: device/status/+
+  device-telemetry-topic: device/telemetry/+