MQTT通信修改 使用服务器数据库

docs/mqtt-embedded-communication.md

@@ -2,6 +2,8 @@
 
 本文档说明后端服务与嵌入式设备之间的 MQTT Topic、通信方向、Payload 类型和 JSON 数据格式。
 
+> **Topic 标识变更（v5.6 起）**：除首次注册回复 `registerDeviceNo` 仍按 **MAC 地址** 下发外，其余 MQTT 下行命令 Topic 第一段统一使用**设备编号 deviceNo**。后端上行处理通过 `DeviceIdentityResolver` 兼容 MAC / 设备编号两种入参；下行命令请按 `deviceNo` 订阅 `subscriber/cmd` 或 `subscriber/schedule`。业务字段（payload 中的 `deviceNo`、数据库主键、API 入参）仍是设备编号，未变更。
+
 当前配置位置：
 
 - 后端订阅配置：`water-admin/src/main/resources/application.yml`
@@ -14,8 +16,8 @@
 | --- | --- |
 | Broker | `mqtt.broker-url` |
 | QoS | `mqtt.qos`，当前默认 `1` |
-| 设备编号 | Topic 第一段 `{deviceNo}` |
+| 设备标识 | 上行 Topic 第一段可为设备 **MAC** 或 `deviceNo`；下行命令 Topic 第一段统一使用 `deviceNo`，仅 `registerDeviceNo` 例外仍使用 MAC |
-| Topic 变量 | 文档中的 `{deviceNo}` 替换为真实设备编号，例如 `/01/publish/register` |
+| Topic 变量 | 上行文档中的 `{deviceNo}` 表示设备标识占位符；下行文档中的 `{deviceNo}` 即设备编号 |
 | JSON 编码 | UTF-8 |
 | 时间格式 | `yyyy-MM-dd HH:mm:ss` |
 
@@ -52,22 +54,19 @@
 | Topic | `/{deviceNo}/publish/register` |
 | Payload 类型 | JSON |
 
-后端以 Topic 中的 `{deviceNo}` 作为设备编号，payload 中的 `deviceNo` 即使传入也会被 Topic 覆盖。
+后端以 Topic 中的设备标识作为解析入口，payload 中的 `deviceNo` 即使传入也会被 Topic 覆盖。
 
 示例：
 
 ```json
 {
   "deviceName": "一号浇水设备",
-  "status": "1",
-  "workStatus": "0",
   "powerLevel": "86",
-  "wifiName": "office-wifi",
   "deviceEm": "WATER-EM-01",
   "deviceSn": "SN202606100001",
   "fwVer": "1.0.0",
-  "macAddress": "AA:BB:CC:DD:EE:FF",
+  "macAddress": "AA:BB:CC:DD:EE:FF"
-  "bindToken": "123456"
+ 
 }
 ```
 
@@ -76,15 +75,11 @@
 | 字段 | 类型 | 必填 | 说明 |
 | --- | --- | --- | --- |
 | deviceName | string | 否 | 设备名称 |
-| status | string | 否 | 设备状态：`1` 在线，`0` 离线，`2` 到期，`3` 故障；为空时后端默认 `2` |
-| workStatus | string | 否 | 工作状态：`0` 休息，`1` 工作 |
 | powerLevel | string | 否 | 电量 |
-| wifiName | string | 否 | WiFi 名称 |
 | deviceEm | string | 否 | 设备型号 |
 | deviceSn | string | 否 | 设备序列号 |
 | fwVer | string | 否 | 固件版本 |
 | macAddress | string | 否 | MAC 地址 |
-| bindToken | string | 否 | 设备绑定令牌 |
 
 ## 电量上报
 
@@ -99,7 +94,9 @@
 
 ```json
 {
-  "powerLevel": "86"
+  "deviceName":	"Waterer_01",
+  "powerLevel": "86",
+  "charging":	1
 }
 ```
 
@@ -193,6 +190,17 @@ receive
 }
 ```
 
+## 设备编号下发
+
+| 项目 | 内容 |
+| --- | --- |
+| 功能名称 | 设备编号下发 |
+| 通信方向 | 后端发布，设备订阅 |
+| Topic | `/{deviceMac}/subscriber/cmd` |
+| Payload 类型 | JSON |
+
+说明：这是首次注册后的唯一例外，后端按设备 MAC 回传设备编号；其余下行命令都使用 `deviceNo` 作为 Topic 第一段。
+
 ## 排程任务完成上报
 
 | 项目 | 内容 |
@@ -300,7 +308,7 @@ JSON 示例：
 
 1. 设备启动后发布注册消息到 `/{deviceNo}/publish/register`。
 2. 设备定时或电量变化时发布电量到 `/{deviceNo}/publish/power`。
-3. 设备订阅自己的命令 Topic：`/{deviceNo}/subscriber/cmd`。
+3. 设备订阅自己的命令 Topic：`/{deviceNo}/subscriber/cmd` 和 `/{deviceNo}/subscriber/schedule`。
 4. 设备收到命令后立即返回 ACK 到 `/{deviceNo}/publish/ack`，推荐返回 JSON 并携带 `commandId`。
 5. 排程或按键浇水完成后，分别发布到 `/{deviceNo}/publish/finish/schedule` 或 `/{deviceNo}/publish/finish/key`。
 6. 硬件异常时发布到 `/{deviceNo}/publish/error`。

water-admin/src/main/resources/application-dev.yml

@@ -47,9 +47,9 @@ spring:
           driverClassName: com.mysql.cj.jdbc.Driver
           # jdbc 所有参数配置参考 https://lionli.blog.csdn.net/article/details/122018562
           # rewriteBatchedStatements=true 批处理优化 大幅提升批量插入更新删除性能(对数据库有性能损耗 使用批量操作应考虑性能问题)
-          url: jdbc:mysql://localhost:3306/water?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&autoReconnect=true&rewriteBatchedStatements=true&allowPublicKeyRetrieval=true&nullCatalogMeansCurrent=true
+          url: jdbc:mysql://47.97.217.123:3306/water?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&autoReconnect=true&rewriteBatchedStatements=true&allowPublicKeyRetrieval=true&nullCatalogMeansCurrent=true
           username: root
-          password: 123456
+          password: 267a976b31c7a649
 #        # 从库数据源
 #        slave:
 #          lazy: true

water-admin/src/main/resources/application-prod.yml

@@ -50,9 +50,9 @@ spring:
           driverClassName: com.mysql.cj.jdbc.Driver
           # jdbc 所有参数配置参考 https://lionli.blog.csdn.net/article/details/122018562
           # rewriteBatchedStatements=true 批处理优化 大幅提升批量插入更新删除性能(对数据库有性能损耗 使用批量操作应考虑性能问题)
-          url: jdbc:mysql://localhost:3306/ry-vue?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&autoReconnect=true&rewriteBatchedStatements=true&allowPublicKeyRetrieval=true&nullCatalogMeansCurrent=true
+          url: jdbc:mysql://localhost:3306/water?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8&autoReconnect=true&rewriteBatchedStatements=true&allowPublicKeyRetrieval=true&nullCatalogMeansCurrent=true
           username: root
-          password: root
+          password: 267a976b31c7a649
 #        # 从库数据源
 #        slave:
 #          lazy: true
@@ -105,7 +105,7 @@ spring.data:
     # 数据库索引
     database: 0
     # redis 密码必须配置
-    password: water123
+    password: waterx123
     # 连接超时时间
     timeout: 10s
     # 是否开启ssl
@@ -136,17 +136,17 @@ redisson:
 
 --- # mail 邮件发送
 mail:
-  enabled: false
+  enabled: true
   host: smtp.163.com
   port: 465
   # 是否需要用户名密码验证
   auth: true
   # 发送方，遵循RFC-822标准
-  from: xxx@163.com
+  from: watexdevice@163.com
   # 用户名（注意：如果使用foxmail邮箱，此处user为qq号）
-  user: xxx@163.com
+  user: watexdevice@163.com
   # 密码（注意，某些邮箱需要为SMTP服务单独设置密码，详情查看相关帮助）
-  pass: xxxxxxxxxx
+  pass: XZVDXjRdVrjiX8N3
   # 使用 STARTTLS安全连接，STARTTLS是对纯文本通信协议的扩展。
   starttlsEnable: true
   # 使用SSL安全连接
@@ -175,10 +175,10 @@ sms:
       # 框架定义的厂商名称标识，标定此配置是哪个厂商，详细请看厂商标识介绍部分
       supplier: alibaba
       # 有些称为accessKey有些称之为apiKey，也有称为sdkKey或者appId。
-      access-key-id: 您的accessKey
+      access-key-id: LTAI5t7xHqxFKG9DwMgurvrn
       # 称为accessSecret有些称之为apiSecret
-      access-key-secret: 您的accessKeySecret
+      access-key-secret: nNFjR8YwNJUcMrrtdc4pkpPsHmYtfp
-      signature: 您的短信签名
+      signature: 山东春雨润集雨节灌设备
       sdk-app-id: 您的sdkAppId
     config2:
       # 厂商标识，标定此配置是哪个厂商，详细请看厂商标识介绍部分

water-admin/src/main/resources/application.yml

@@ -59,6 +59,7 @@ mqtt:
     enabled: true
     interval-ms: 90000
   topics:
+    # 订阅/下发主题第一段为设备 MAC（小写、去分隔符），上行业务侧由 DeviceIdentityResolver 解析为设备编号
     subscribe:
 
       - /+/publish/finish/schedule   #排程任务完成上报

water-common/water-common-json/src/main/java/org/dromara/common/json/config/JacksonConfig.java

@@ -33,9 +33,9 @@ public class JacksonConfig {
     public Module registerJavaTimeModule() {
         // 全局配置序列化返回 JSON 处理
         JavaTimeModule javaTimeModule = new JavaTimeModule();
-        javaTimeModule.addSerializer(Long.class, BigNumberSerializer.INSTANCE);
+        javaTimeModule.addSerializer(Long.class, ToStringSerializer.instance);
         javaTimeModule.addSerializer(Long.TYPE, BigNumberSerializer.INSTANCE);
-        javaTimeModule.addSerializer(BigInteger.class, BigNumberSerializer.INSTANCE);
+        javaTimeModule.addSerializer(BigInteger.class, ToStringSerializer.instance);
         javaTimeModule.addSerializer(BigDecimal.class, ToStringSerializer.instance);
         DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
         javaTimeModule.addSerializer(LocalDateTime.class, new LocalDateTimeSerializer(formatter));

water-common/water-common-mqtt/pom.xml

@@ -43,5 +43,10 @@
             <groupId>org.dromara</groupId>
             <artifactId>water-app</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 </project>

water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/DeviceMqttCommandPublisher.java

@@ -3,6 +3,7 @@ package org.dromara.mqtt;
 import lombok.RequiredArgsConstructor;
 import org.dromara.app.domain.mqtt.DeviceCommand;
 import org.dromara.app.service.IDeviceCommandPublisher;
+import org.dromara.common.core.exception.ServiceException;
 import org.dromara.common.core.utils.StringUtils;
 import org.dromara.common.json.utils.JsonUtils;
 import org.dromara.mqtt.config.properties.MqttProperties;
@@ -44,14 +45,18 @@ public class DeviceMqttCommandPublisher implements IDeviceCommandPublisher {
     }
 
     private String buildCommandTopic(String deviceNo) {
+        if (StringUtils.isBlank(deviceNo)) {
+            throw new ServiceException("命令下发缺少设备编号，无法拼接主题");
+        }
+        String device = deviceNo.trim().toLowerCase();
         String prefix = mqttProperties.getTopics().getPublishPrefix();
         if (StringUtils.isBlank(prefix)) {
-            return "/" + deviceNo + "/subscriber/cmd";
+            return "/" + device + "/subscriber/cmd";
         }
         String normalizedPrefix = prefix.startsWith("/") ? prefix : "/" + prefix;
         if (normalizedPrefix.endsWith("/")) {
             normalizedPrefix = normalizedPrefix.substring(0, normalizedPrefix.length() - 1);
         }
-        return normalizedPrefix + "/" + deviceNo + "/subscriber/cmd";
+        return normalizedPrefix + "/" + device + "/subscriber/cmd";
     }
 }

water-modules/water-app/pom.xml

@@ -123,6 +123,11 @@
             <groupId>org.dromara</groupId>
             <artifactId>water-system</artifactId>
         </dependency>
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+            <scope>test</scope>
+        </dependency>
 
     </dependencies>
 

water-modules/water-app/src/main/java/org/dromara/app/controller/AppController.java

@@ -14,7 +14,6 @@ import org.dromara.app.domain.bo.AppDeviceBo;
 import org.dromara.app.domain.bo.AppScheduleBo;
 import org.dromara.app.domain.bo.AppSchedulingDeviceBo;
 import org.dromara.app.domain.bo.AppWateringLogBo;
-import org.dromara.app.domain.mqtt.DeviceCommand;
 import org.dromara.app.domain.vo.*;
 import org.dromara.app.service.*;
 import org.dromara.app.service.impl.AppScheduleServiceImpl;
@@ -37,11 +36,16 @@ import org.dromara.common.mybatis.helper.DataPermissionHelper;
 import org.dromara.common.satoken.utils.LoginHelper;
 import org.dromara.common.web.core.BaseController;
 import org.dromara.system.domain.bo.SysUserBo;
+import org.dromara.system.domain.vo.SysOssUploadVo;
+import org.dromara.system.domain.vo.SysOssVo;
 import org.dromara.system.domain.vo.SysUserVo;
 import org.dromara.system.domain.vo.UserInfoVo;
+import org.dromara.system.service.ISysOssService;
 import org.dromara.system.service.ISysUserService;
+import org.springframework.http.MediaType;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
+import org.springframework.web.multipart.MultipartFile;
 
 import java.util.*;
 
@@ -61,6 +65,7 @@ public class AppController extends BaseController {
     private final ISysUserService userService;
     private final IAppWateringLogService appWateringLogService;
     private final IDeviceCommandService deviceCommandService;
+    private final ISysOssService ossService;
 
     /**
      *查询设备
@@ -243,10 +248,15 @@ public class AppController extends BaseController {
                     continue;
                 }
                 assertDeviceOwned(deviceNo);
+                if (appSchedulingDeviceService.queryByScheduleIdAndDeviceNo(scheduleId, deviceNo) != null) {
+                    continue;
+                }
                 AppSchedulingDeviceBo schedulingDeviceBo = new AppSchedulingDeviceBo();
                 schedulingDeviceBo.setDeviceNo(deviceNo);
                 schedulingDeviceBo.setScheduleId(scheduleId);
-                appSchedulingDeviceService.insertByBo(schedulingDeviceBo);
+                if (appSchedulingDeviceService.insertByBo(schedulingDeviceBo)) {
+                    deviceCommandService.sendScheduleBindCommand(deviceNo, buildScheduleBindPayload(scheduleId, deviceNo));
+                }
             }
             return R.ok();
         } catch (Exception e) {
@@ -277,6 +287,47 @@ public class AppController extends BaseController {
             .toList();
     }
 
+    private Map<String, Object> buildScheduleBindPayload(Long scheduleId, String deviceNo) {
+     //   AppScheduleVo schedule = getOwnedSchedule(scheduleId);
+        List<AppScheduleDetailVo> details = appScheduleDetailService.queryByScheduleIdByStatus(scheduleId);
+
+//        Map<String, Object> schedulePayload = new HashMap<>();
+//        schedulePayload.put("id", schedule.getId());
+//        schedulePayload.put("name", schedule.getName());
+//        schedulePayload.put("status", schedule.getStatus());
+
+        Map<String, Object> payload = new HashMap<>();
+       // payload.put("cmd", -1);
+        payload.put("deviceNo", deviceNo);
+    //    payload.put("schedule", schedulePayload);
+        payload.put("details", toScheduleDetailPayload(details));
+        return payload;
+       // return toScheduleDetailPayload(details);
+    }
+
+    private List<Map<String, Object>> toScheduleDetailPayload(List<AppScheduleDetailVo> details) {
+        List<Map<String, Object>> detailPayload = new ArrayList<>();
+        if (details == null) {
+            return detailPayload;
+        }
+        for (AppScheduleDetailVo detail : details) {
+            List<Object> timeSlots = new ArrayList<>();
+            JSONArray timeArray = JSONUtil.parseArray(detail.getTimeData());
+            for (Object timeSlot : timeArray) {
+                timeSlots.add(timeSlot);
+            }
+
+            Map<String, Object> item = new HashMap<>();
+            item.put("id", detail.getId());
+            item.put("weekday", detail.getWeekday());
+            item.put("timeData", timeSlots);
+            item.put("triggerType", detail.getTriggerType());
+            item.put("status", detail.getStatus());
+            detailPayload.add(item);
+        }
+        return detailPayload;
+    }
+
     /**
      * 删除排程与设备绑定
      * @param scheduleId,deviceNo
@@ -297,7 +348,17 @@ public class AppController extends BaseController {
             Long scheduleIdValue = Long.valueOf(scheduleId);
             assertScheduleOwned(scheduleIdValue);
             assertDeviceOwned(deviceNo);
-            return toAjax(appSchedulingDeviceService.deleteWithValidByScheduleIdAndDeviceNo(scheduleIdValue, deviceNo));
+            Boolean deleted = appSchedulingDeviceService.deleteWithValidByScheduleIdAndDeviceNo(scheduleIdValue, deviceNo);
+            if (deleted) {
+                // 通知设备解绑排程
+                try {
+                    deviceCommandService.sendScheduleUnbindCommand(deviceNo, scheduleIdValue);
+                } catch (Exception mqttEx) {
+                    log.warn("[排程通知] 删除排程设备绑定后下发解绑命令失败 排程ID={} 设备编号={} 原因={}",
+                        scheduleIdValue, deviceNo, mqttEx.getMessage());
+                }
+            }
+            return toAjax(deleted);
         } catch (Exception e) {
             return fail(e);
         }
@@ -406,7 +467,12 @@ public class AppController extends BaseController {
         try {
             assertScheduleOwned(bo.getId());
             bo.setUserId(LoginHelper.getUserId());
-            return toAjax(appScheduleService.updateStatusByBo(bo));
+            Boolean updated = appScheduleService.updateStatusByBo(bo);
+            if (updated) {
+                // 向所有绑定设备下发排程状态变更
+                notifyBoundDevicesScheduleUpdate(bo.getId());
+            }
+            return toAjax(updated);
         } catch (Exception e) {
             return fail(e);
         }
@@ -424,6 +490,8 @@ public class AppController extends BaseController {
             bo.setUserId(LoginHelper.getUserId());
             Map<String, Object> map = appScheduleService.updateByBo(bo);
             if (Boolean.valueOf(map.get("flag").toString())){
+                // 向所有绑定设备下发排程更新
+                notifyBoundDevicesScheduleUpdate(bo.getId());
                 return R.ok(map.get("appSchedule"));
             }else {
                 return R.fail();
@@ -448,7 +516,17 @@ public class AppController extends BaseController {
             for (Long id : ids) {
                 assertScheduleOwned(id);
             }
-            return toAjax(appScheduleService.deleteWithValidByIds(List.of(ids), true));
+            Map<Long, List<String>> boundDeviceNosBySchedule = new HashMap<>();
+            for (Long id : ids) {
+                boundDeviceNosBySchedule.put(id, findBoundDeviceNos(id));
+            }
+            Boolean deleted = appScheduleService.deleteWithValidByIds(List.of(ids), true);
+            if (deleted) {
+                for (Map.Entry<Long, List<String>> entry : boundDeviceNosBySchedule.entrySet()) {
+                    notifyScheduleUnbind(entry.getKey(), entry.getValue());
+                }
+            }
+            return toAjax(deleted);
         } catch (Exception e) {
             return fail(e);
         }
@@ -475,7 +553,7 @@ public class AppController extends BaseController {
                     vo.setDeviceNo(appDeviceVo.getDeviceNo());
                 }
             }
-            vo.setTriggerType(vo.getTriggerType().equals("0")?"排程":"手动");
+          //  vo.setTriggerType(vo.getTriggerType().equals("0")?"排程":"手动");
             l.add(vo);
         }
         appWateringLogVoTableDataInfo.setRows(l);
@@ -563,6 +641,31 @@ public class AppController extends BaseController {
         }
     }
 
+    /**
+     * 上传图片到OSS对象存储
+     */
+    @PostMapping(value = "/uploadImage", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
+    public R<SysOssUploadVo> uploadImage(@RequestPart("file") MultipartFile file) {
+        try {
+            if (file == null || file.isEmpty()) {
+                throw new ServiceException("上传图片不能为空");
+            }
+            String contentType = file.getContentType();
+            if (StringUtils.isBlank(contentType) || !contentType.toLowerCase(Locale.ROOT).startsWith("image/")) {
+                throw new ServiceException("只能上传图片文件");
+            }
+
+            SysOssVo oss = ossService.upload(file);
+            SysOssUploadVo uploadVo = new SysOssUploadVo();
+            uploadVo.setUrl(oss.getUrl());
+            uploadVo.setFileName(oss.getOriginalName());
+            uploadVo.setOssId(oss.getOssId() == null ? null : oss.getOssId().toString());
+            return R.ok(uploadVo);
+        } catch (Exception e) {
+            return fail(e);
+        }
+    }
+
     /**
      * 修改用户
      */
@@ -638,9 +741,6 @@ public class AppController extends BaseController {
     @GetMapping("/test")
     public  R<Void>  test (){
         try {
-
-            DeviceCommand deviceCommand = new DeviceCommand();
-
             String bindDevice = deviceCommandService.sendBindDeviceCommand("01");
 
             // String s = deviceCommandService.sendSwitchCommand("01", "1", "2026-06-10 14:35:31", 20);
@@ -706,4 +806,48 @@ public class AppController extends BaseController {
         }
     }
 
+    /**
+     * 查找排程关联的所有设备编号
+     */
+    private List<String> findBoundDeviceNos(Long scheduleId) {
+        AppSchedulingDeviceBo bo = new AppSchedulingDeviceBo();
+        bo.setScheduleId(scheduleId);
+        List<AppSchedulingDeviceVo> bindings = appSchedulingDeviceService.queryList(bo);
+        return bindings.stream()
+            .map(AppSchedulingDeviceVo::getDeviceNo)
+            .filter(StringUtils::isNotBlank)
+            .toList();
+    }
+
+    /**
+     * 向排程关联的所有设备重新下发排程信息（用于排程内容/状态变更）
+     */
+    private void notifyBoundDevicesScheduleUpdate(Long scheduleId) {
+        List<String> deviceNos = findBoundDeviceNos(scheduleId);
+        for (String deviceNo : deviceNos) {
+            try {
+                deviceCommandService.sendScheduleBindCommand(deviceNo, buildScheduleBindPayload(scheduleId, deviceNo));
+            } catch (Exception e) {
+                log.warn("[排程通知] 下发排程更新失败 排程ID={} 设备编号={} 原因={}", scheduleId, deviceNo, e.getMessage());
+            }
+        }
+    }
+
+    /**
+     * 向排程关联的所有设备下发排程解绑通知
+     */
+    private void notifyBoundDevicesScheduleUnbind(Long scheduleId) {
+        notifyScheduleUnbind(scheduleId, findBoundDeviceNos(scheduleId));
+    }
+
+    private void notifyScheduleUnbind(Long scheduleId, List<String> deviceNos) {
+        for (String deviceNo : deviceNos) {
+            try {
+                deviceCommandService.sendScheduleUnbindCommand(deviceNo, scheduleId);
+            } catch (Exception e) {
+                log.warn("[排程通知] 下发排程解绑失败 排程ID={} 设备编号={} 原因={}", scheduleId, deviceNo, e.getMessage());
+            }
+        }
+    }
+
 }

water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceCommand.java

@@ -15,6 +15,7 @@ public class DeviceCommand implements Serializable {
 
     private String commandId;
     private String deviceNo;
+    private String deviceMac;
     private String commandType;
     private String topic;
     private Map<String, Object> payload = new HashMap<>();

water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceDataHandler.java

@@ -81,14 +81,16 @@ public class DeviceDataHandler implements MqttTopicHandler {
 
     /**
      * 刷新设备在线状态到 Redis（与 MqttCommandAckService 逻辑一致）
-     * 写入 Redis 后 TTL 到期自动过期 = 离线
+     * 写入幂等，未拿到锁则跳过由后续上报兜底
      */
     private void refreshDeviceOnline(String deviceNo) {
         RLock lock = RedisUtils.getClient().getLock(DEVICE_STATUS_LOCK_PREFIX + deviceNo);
         boolean locked = false;
         try {
-            lock.lock(10, TimeUnit.SECONDS);
+            locked = lock.tryLock(0, 10, TimeUnit.SECONDS);
-            locked = true;
+            if (!locked) {
+                return;
+            }
             Map<String, Object> statusCache = new HashMap<>();
             statusCache.put("deviceNo", deviceNo);
             statusCache.put("status", "1");
@@ -103,6 +105,9 @@ public class DeviceDataHandler implements MqttTopicHandler {
                     .set(AppDevice::getStatus, "1")
                     .eq(AppDevice::getDeviceNo, deviceNo)
             );
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.warn("[MQTT] 刷新设备在线状态被中断 设备编号={}", deviceNo);
         } finally {
             if (locked && lock.isHeldByCurrentThread()) {
                 lock.unlock();

water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceRegisterHandler.java

@@ -47,8 +47,6 @@ DeviceRegisterHandler implements MqttTopicHandler {
     @Value("${mqtt.command-ack.device-status-cache-ttl-seconds}")
     private int deviceStatusCacheTtlSeconds;
     private final DeviceIdentityResolver deviceIdentityResolver;
-    @Value("${mqtt.topics.publish-prefix:}")
-    private String mqttPublishPrefix;
 
     @Override
     public Pattern topicPattern() {
@@ -77,8 +75,7 @@ DeviceRegisterHandler implements MqttTopicHandler {
             AppDeviceBo device = buildRegisterDevice(deviceNo, normalizedDeviceMac, dto);
             appDeviceService.registerByMqtt(device);
             refreshDeviceOnline(deviceNo);
-            String commandTarget = normalizedDeviceMac == null ? deviceIdentity : normalizedDeviceMac;
+            sendDeviceNoToDevice(deviceNo, normalizedDeviceMac);
-            sendDeviceNoToDevice(commandTarget, deviceNo);
             log.info("[MQTT] 设备注册 时间={} MAC={} 设备编号={} 消息体={}",
                 HandlerLogTime.now(), normalizedDeviceMac, deviceNo, payload);
         } catch (Exception e) {
@@ -115,19 +112,25 @@ DeviceRegisterHandler implements MqttTopicHandler {
         return text.isBlank() ? null : text.toLowerCase(Locale.ROOT);
     }
 
-    private void sendDeviceNoToDevice(String deviceMac, String deviceNo) {
+    private void sendDeviceNoToDevice(String deviceNo, String deviceMac) {
         IDeviceCommandPublisher commandPublisher = commandPublisherProvider.getIfAvailable();
         if (commandPublisher == null) {
             log.warn("[MQTT] 设备编号下发失败，未找到命令发布器 时间={} MAC={} 设备编号={}",
                 HandlerLogTime.now(), deviceMac, deviceNo);
             return;
         }
+        if (deviceMac == null) {
+            log.warn("[MQTT] 设备编号下发失败，缺少设备 MAC 时间={} 设备编号={}",
+                HandlerLogTime.now(), deviceNo);
+            return;
+        }
 
         try {
             DeviceCommand command = new DeviceCommand();
             command.setDeviceNo(deviceNo);
+            command.setDeviceMac(deviceMac);
             command.setCommandType("registerDeviceNo");
-            command.setTopic(buildCommandTopic(deviceMac));
+            command.setTopic("/" + deviceMac.trim().toLowerCase(Locale.ROOT) + "/subscriber/cmd");
             command.getPayload().put("deviceNo", deviceNo);
             command.getPayload().put("deviceMac", deviceMac);
 
@@ -140,17 +143,6 @@ DeviceRegisterHandler implements MqttTopicHandler {
         }
     }
 
-    private String buildCommandTopic(String deviceMac) {
-        if (mqttPublishPrefix == null || mqttPublishPrefix.isBlank()) {
-            return "/" + deviceMac + "/subscriber/cmd";
-        }
-        String normalizedPrefix = mqttPublishPrefix.startsWith("/") ? mqttPublishPrefix : "/" + mqttPublishPrefix;
-        if (normalizedPrefix.endsWith("/")) {
-            normalizedPrefix = normalizedPrefix.substring(0, normalizedPrefix.length() - 1);
-        }
-        return normalizedPrefix + "/" + deviceMac + "/subscriber/cmd";
-    }
-
     private String firstNotBlank(Object first, Object second) {
         String firstText = valueAsString(first);
         if (firstText != null && !firstText.isBlank()) {
@@ -165,14 +157,16 @@ DeviceRegisterHandler implements MqttTopicHandler {
     }
 
     /**
-     * 刷新设备在线状态到 Redis
+     * 刷新设备在线状态到 Redis；状态写入幂等，未拿到锁则跳过由后续上报兜底
      */
     private void refreshDeviceOnline(String deviceNo) {
         RLock lock = RedisUtils.getClient().getLock(DEVICE_STATUS_LOCK_PREFIX + deviceNo);
         boolean locked = false;
         try {
-            lock.lock(10, TimeUnit.SECONDS);
+            locked = lock.tryLock(0, 10, TimeUnit.SECONDS);
-            locked = true;
+            if (!locked) {
+                return;
+            }
             Map<String, Object> statusCache = new HashMap<>();
             statusCache.put("deviceNo", deviceNo);
             statusCache.put("status", "1");
@@ -187,6 +181,9 @@ DeviceRegisterHandler implements MqttTopicHandler {
                     .set(AppDevice::getStatus, "1")
                     .eq(AppDevice::getDeviceNo, deviceNo)
             );
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            log.warn("[MQTT] 刷新设备在线状态被中断 设备编号={}", deviceNo);
         } finally {
             if (locked && lock.isHeldByCurrentThread()) {
                 lock.unlock();

water-modules/water-app/src/main/java/org/dromara/app/service/IDeviceCommandService.java

@@ -41,8 +41,26 @@ public interface IDeviceCommandService {
      * @return commandId
      */
     String sendCustomCommand(String deviceNo, String commandType, Map<String, Object> extra);
+
+    /**
+     * 下发排程绑定命令。
+     *
+     * @param deviceNo 设备编号
+     * @param payload  排程信息
+     * @return commandId
+     */
+    String sendScheduleBindCommand(String deviceNo, Map<String, Object> payload);
     String sendBindDeviceCommand(String deviceNo);
 
+    /**
+     * 下发排程解绑命令，通知设备清除指定排程。
+     *
+     * @param deviceNo   设备编号
+     * @param scheduleId 排程ID
+     * @return commandId
+     */
+    String sendScheduleUnbindCommand(String deviceNo, Long scheduleId);
+
     /**
      * 下发设备初始化指令。用于解绑设备，允许设备未绑定用户时下发。
      *

water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceServiceImpl.java

@@ -142,6 +142,7 @@ public class AppDeviceServiceImpl implements IAppDeviceService {
             try {
                 return baseMapper.insert(device) > 0;
             } catch (org.springframework.dao.DuplicateKeyException ignored) {
+                device.setDeviceName(null);
                 return baseMapper.updateById(device) > 0;
             }
         }
@@ -219,8 +220,7 @@ public class AppDeviceServiceImpl implements IAppDeviceService {
             throw new ServiceException("设备绑定失败，请重试");
         }
         //下发设备已绑定状态
-        //deviceCommandService.sendBindDeviceCommand(exists.getDeviceNo());
+        deviceCommandService.sendBindDeviceCommand(exists.getDeviceNo());
-        deviceCommandService.sendBindDeviceCommand(exists.getMacAddress());
         return baseMapper.selectVoById(exists.getDeviceNo());
     }
 
@@ -247,8 +247,7 @@ public class AppDeviceServiceImpl implements IAppDeviceService {
         AppDeviceVo device = baseMapper.selectVoById(deviceNo);
         if (ObjectUtil.isNotNull(device) && StringUtils.isNotBlank(device.getDeviceNo())) {
             // 通过业务层统一下发命令（含浇水日志记录）
-            //String commandId = deviceCommandService.sendSwitchCommand(deviceNo, workStatus, startTime, durationMin);
+            String commandId = deviceCommandService.sendSwitchCommand(deviceNo, workStatus, startTime, durationMin);
-            String commandId = deviceCommandService.sendSwitchCommand(device.getMacAddress(), workStatus, startTime, durationMin);
             log.info("[设备] 开关命令已下发 设备编号={} 命令编号={}", deviceNo, commandId);
         }
 

water-modules/water-app/src/main/java/org/dromara/app/service/impl/DeviceCommandServiceImpl.java

@@ -19,10 +19,7 @@ import org.springframework.stereotype.Service;
 
 import java.text.ParseException;
 import java.text.SimpleDateFormat;
-import java.util.Calendar;
+import java.util.*;
-import java.util.Date;
-import java.util.HashMap;
-import java.util.Map;
 
 /**
  * 设备命令下发业务实现。
@@ -49,6 +46,9 @@ public class DeviceCommandServiceImpl implements IDeviceCommandService {
     @Override
     public String sendCommand(DeviceCommand command) {
         validate(command);
+        if (StringUtils.isBlank(command.getDeviceMac())) {
+            command.setDeviceMac(requireDeviceMac(command.getDeviceNo()));
+        }
         String commandId = commandPublisher.send(command);
         log.info("[命令] 命令已下发 设备编号={} 命令类型={} 命令编号={}",
             command.getDeviceNo(), command.getCommandType(), commandId);
@@ -67,6 +67,7 @@ public class DeviceCommandServiceImpl implements IDeviceCommandService {
 
         DeviceCommand command = new DeviceCommand();
         command.setDeviceNo(deviceNo);
+        command.setDeviceMac(requireDeviceMac(deviceNo));
         command.setCommandType("switchDevice");
         command.getPayload().put("deviceNo", deviceNo);
         command.getPayload().put("cmd", workStatus);
@@ -100,6 +101,7 @@ public class DeviceCommandServiceImpl implements IDeviceCommandService {
 
         DeviceCommand command = new DeviceCommand();
         command.setDeviceNo(deviceNo);
+        command.setDeviceMac(requireDeviceMac(deviceNo));
         command.setCommandType(commandType);
         if (extra != null) {
             command.getPayload().putAll(extra);
@@ -107,15 +109,37 @@ public class DeviceCommandServiceImpl implements IDeviceCommandService {
         return sendCommand(command);
     }
 
+    @Override
+    public String sendScheduleBindCommand(String deviceNo, Map<String, Object> payload) {
+        AppDeviceVo device = requireCommandDevice(deviceNo);
+        if (StringUtils.isBlank(device.getMacAddress())) {
+            throw new ServiceException("设备未登记 MAC，无法下发排程：" + deviceNo);
+        }
+
+        DeviceCommand command = new DeviceCommand();
+        command.setDeviceNo(deviceNo);
+        command.setDeviceMac(device.getMacAddress());
+        command.setCommandType("bindSchedule");
+        command.setTopic(buildScheduleTopic(deviceNo));
+        if (payload != null) {
+            command.getPayload().putAll(payload);
+        }
+        return sendCommand(command);
+    }
+
     @Override
     public String sendBindDeviceCommand(String deviceNo) {
         AppDeviceVo device = deviceService.queryById(deviceNo);
         if (ObjectUtil.isNull(device)) {
             throw new ServiceException("设备不存在：" + deviceNo);
         }
+        if (StringUtils.isBlank(device.getMacAddress())) {
+            throw new ServiceException("设备未登记 MAC，无法下发绑定命令：" + deviceNo);
+        }
 
         DeviceCommand command = new DeviceCommand();
         command.setDeviceNo(deviceNo);
+        command.setDeviceMac(device.getMacAddress());
         command.setCommandType("bindDevice");
         Map<String, Object> objectObjectHashMap = new HashMap<>();
         objectObjectHashMap.put("bindStatus", true);
@@ -125,15 +149,38 @@ public class DeviceCommandServiceImpl implements IDeviceCommandService {
         return sendCommand(command);
     }
 
+    @Override
+    public String sendScheduleUnbindCommand(String deviceNo, Long scheduleId) {
+        AppDeviceVo device = requireCommandDevice(deviceNo);
+        if (StringUtils.isBlank(device.getMacAddress())) {
+            throw new ServiceException("设备未登记 MAC，无法下发排程解绑命令：" + deviceNo);
+        }
+
+        DeviceCommand command = new DeviceCommand();
+        command.setDeviceNo(deviceNo);
+        command.setDeviceMac(device.getMacAddress());
+        command.setCommandType("unbindSchedule");
+        command.setTopic(buildScheduleTopic(deviceNo));
+        command.getPayload().put("cmd", -1);
+        command.getPayload().put("deviceNo", deviceNo);
+        command.getPayload().put("scheduleId", scheduleId);
+        command.getPayload().put("unbind", true);
+        return sendCommand(command);
+    }
+
     @Override
     public String sendInitDeviceCommand(String deviceNo) {
         AppDeviceVo device = deviceService.queryById(deviceNo);
         if (ObjectUtil.isNull(device)) {
             throw new ServiceException("设备不存在：" + deviceNo);
         }
+        if (StringUtils.isBlank(device.getMacAddress())) {
+            throw new ServiceException("设备未登记 MAC，无法下发初始化命令：" + deviceNo);
+        }
 
         DeviceCommand command = new DeviceCommand();
         command.setDeviceNo(deviceNo);
+        command.setDeviceMac(device.getMacAddress());
         command.setCommandType("initDevice");
         Map<String, Object> payload = new HashMap<>();
         payload.put("deviceNo", deviceNo);
@@ -167,6 +214,29 @@ public class DeviceCommandServiceImpl implements IDeviceCommandService {
         }
     }
 
+    private String requireDeviceMac(String deviceNo) {
+        AppDeviceVo device = deviceService.queryById(deviceNo);
+        if (device == null || StringUtils.isBlank(device.getMacAddress())) {
+            throw new ServiceException("设备未登记 MAC，无法下发命令：" + deviceNo);
+        }
+        return device.getMacAddress();
+    }
+
+    private AppDeviceVo requireCommandDevice(String deviceNo) {
+        AppDeviceVo device = deviceService.queryById(deviceNo);
+        if (ObjectUtil.isNull(device)) {
+            throw new ServiceException("设备不存在：" + deviceNo);
+        }
+        if (device.getUserId() == null) {
+            throw new ServiceException("设备未绑定用户，无法下发命令：" + deviceNo);
+        }
+        return device;
+    }
+
+    private String buildScheduleTopic(String deviceNo) {
+        return "/" + deviceNo.trim().toLowerCase(Locale.ROOT) + "/subscriber/schedule";
+    }
+
     private void assertSwitchStatus(String workStatus) {
         if (!"0".equals(workStatus) && !"1".equals(workStatus)) {
             throw new ServiceException("设备工作状态只能为0或1");