From 319ebd19adc182611096ef8ce45924e0d4c6cd04 Mon Sep 17 00:00:00 2001 From: yuhaiming <40624989@qq.com> Date: Tue, 30 Jun 2026 08:12:35 +0800 Subject: [PATCH] =?UTF-8?q?MQTT=E9=80=9A=E4=BF=A1=E4=BF=AE=E6=94=B9=20=20?= =?UTF-8?q?=E4=BD=BF=E7=94=A8=E6=9C=8D=E5=8A=A1=E5=99=A8=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=BA=93?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- docs/mqtt-embedded-communication.md | 36 ++-- .../src/main/resources/application-dev.yml | 4 +- .../src/main/resources/application-prod.yml | 20 +-- .../src/main/resources/application.yml | 1 + .../common/json/config/JacksonConfig.java | 4 +- water-common/water-common-mqtt/pom.xml | 5 + .../mqtt/DeviceMqttCommandPublisher.java | 9 +- water-modules/water-app/pom.xml | 5 + .../dromara/app/controller/AppController.java | 162 +++++++++++++++++- .../app/domain/mqtt/DeviceCommand.java | 1 + .../app/handler/DeviceDataHandler.java | 11 +- .../app/handler/DeviceRegisterHandler.java | 37 ++-- .../app/service/IDeviceCommandService.java | 18 ++ .../service/impl/AppDeviceServiceImpl.java | 7 +- .../impl/DeviceCommandServiceImpl.java | 78 ++++++++- 15 files changed, 328 insertions(+), 70 deletions(-) diff --git a/docs/mqtt-embedded-communication.md b/docs/mqtt-embedded-communication.md index 7771ac3..d62ac57 100644 --- a/docs/mqtt-embedded-communication.md +++ b/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 变量 | 文档中的 `{deviceNo}` 替换为真实设备编号，例如 `/01/publish/register` | +| 设备标识 | 上行 Topic 第一段可为设备 **MAC** 或 `deviceNo`；下行命令 Topic 第一段统一使用 `deviceNo`，仅 `registerDeviceNo` 例外仍使用 MAC | +| 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", - "bindToken": "123456" + "macAddress": "AA:BB:CC:DD:EE:FF" + } ``` @@ -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`。 diff --git a/water-admin/src/main/resources/application-dev.yml b/water-admin/src/main/resources/application-dev.yml index 204a118..32a2d5e 100644 --- a/water-admin/src/main/resources/application-dev.yml +++ b/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 diff --git a/water-admin/src/main/resources/application-prod.yml b/water-admin/src/main/resources/application-prod.yml index 0521879..0b56e5c 100644 --- a/water-admin/src/main/resources/application-prod.yml +++ b/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 - signature: 您的短信签名 + access-key-secret: nNFjR8YwNJUcMrrtdc4pkpPsHmYtfp + signature: 山东春雨润集雨节灌设备 sdk-app-id: 您的sdkAppId config2: # 厂商标识，标定此配置是哪个厂商，详细请看厂商标识介绍部分 diff --git a/water-admin/src/main/resources/application.yml b/water-admin/src/main/resources/application.yml index d1354ca..b5c5881 100644 --- a/water-admin/src/main/resources/application.yml +++ b/water-admin/src/main/resources/application.yml @@ -59,6 +59,7 @@ mqtt: enabled: true interval-ms: 90000 topics: + # 订阅/下发主题第一段为设备 MAC（小写、去分隔符），上行业务侧由 DeviceIdentityResolver 解析为设备编号 subscribe: - /+/publish/finish/schedule #排程任务完成上报 diff --git a/water-common/water-common-json/src/main/java/org/dromara/common/json/config/JacksonConfig.java b/water-common/water-common-json/src/main/java/org/dromara/common/json/config/JacksonConfig.java index 10eb7a1..832e224 100644 --- a/water-common/water-common-json/src/main/java/org/dromara/common/json/config/JacksonConfig.java +++ b/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)); diff --git a/water-common/water-common-mqtt/pom.xml b/water-common/water-common-mqtt/pom.xml index d4809f2..2f98a29 100644 --- a/water-common/water-common-mqtt/pom.xml +++ b/water-common/water-common-mqtt/pom.xml @@ -43,5 +43,10 @@ org.dromara water-app + + org.springframework.boot + spring-boot-starter-test + test + diff --git a/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/DeviceMqttCommandPublisher.java b/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/DeviceMqttCommandPublisher.java index cd61323..a62dd74 100644 --- a/water-common/water-common-mqtt/src/main/java/org/dromara/mqtt/DeviceMqttCommandPublisher.java +++ b/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"; } } diff --git a/water-modules/water-app/pom.xml b/water-modules/water-app/pom.xml index 92da839..e44f8cb 100644 --- a/water-modules/water-app/pom.xml +++ b/water-modules/water-app/pom.xml @@ -123,6 +123,11 @@ org.dromara water-system + + org.springframework.boot + spring-boot-starter-test + test + diff --git a/water-modules/water-app/src/main/java/org/dromara/app/controller/AppController.java b/water-modules/water-app/src/main/java/org/dromara/app/controller/AppController.java index eef0c96..9b84ad5 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/controller/AppController.java +++ b/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 buildScheduleBindPayload(Long scheduleId, String deviceNo) { + // AppScheduleVo schedule = getOwnedSchedule(scheduleId); + List details = appScheduleDetailService.queryByScheduleIdByStatus(scheduleId); + +// Map schedulePayload = new HashMap<>(); +// schedulePayload.put("id", schedule.getId()); +// schedulePayload.put("name", schedule.getName()); +// schedulePayload.put("status", schedule.getStatus()); + + Map 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> toScheduleDetailPayload(List details) { + List> detailPayload = new ArrayList<>(); + if (details == null) { + return detailPayload; + } + for (AppScheduleDetailVo detail : details) { + List timeSlots = new ArrayList<>(); + JSONArray timeArray = JSONUtil.parseArray(detail.getTimeData()); + for (Object timeSlot : timeArray) { + timeSlots.add(timeSlot); + } + + Map 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 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> 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> 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 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 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 findBoundDeviceNos(Long scheduleId) { + AppSchedulingDeviceBo bo = new AppSchedulingDeviceBo(); + bo.setScheduleId(scheduleId); + List bindings = appSchedulingDeviceService.queryList(bo); + return bindings.stream() + .map(AppSchedulingDeviceVo::getDeviceNo) + .filter(StringUtils::isNotBlank) + .toList(); + } + + /** + * 向排程关联的所有设备重新下发排程信息（用于排程内容/状态变更） + */ + private void notifyBoundDevicesScheduleUpdate(Long scheduleId) { + List 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 deviceNos) { + for (String deviceNo : deviceNos) { + try { + deviceCommandService.sendScheduleUnbindCommand(deviceNo, scheduleId); + } catch (Exception e) { + log.warn("[排程通知] 下发排程解绑失败 排程ID={} 设备编号={} 原因={}", scheduleId, deviceNo, e.getMessage()); + } + } + } + } diff --git a/water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceCommand.java b/water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceCommand.java index 0d99429..2f2f3e6 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/domain/mqtt/DeviceCommand.java +++ b/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 payload = new HashMap<>(); diff --git a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceDataHandler.java b/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceDataHandler.java index 078c2fe..ba253a4 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceDataHandler.java +++ b/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 = true; + locked = lock.tryLock(0, 10, TimeUnit.SECONDS); + if (!locked) { + return; + } Map 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(); diff --git a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceRegisterHandler.java b/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceRegisterHandler.java index 323de32..0794fbd 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/handler/DeviceRegisterHandler.java +++ b/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(commandTarget, deviceNo); + sendDeviceNoToDevice(deviceNo, normalizedDeviceMac); 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 = true; + locked = lock.tryLock(0, 10, TimeUnit.SECONDS); + if (!locked) { + return; + } Map 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(); diff --git a/water-modules/water-app/src/main/java/org/dromara/app/service/IDeviceCommandService.java b/water-modules/water-app/src/main/java/org/dromara/app/service/IDeviceCommandService.java index e8d0425..9180432 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/service/IDeviceCommandService.java +++ b/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 extra); + + /** + * 下发排程绑定命令。 + * + * @param deviceNo 设备编号 + * @param payload 排程信息 + * @return commandId + */ + String sendScheduleBindCommand(String deviceNo, Map payload); String sendBindDeviceCommand(String deviceNo); + /** + * 下发排程解绑命令，通知设备清除指定排程。 + * + * @param deviceNo 设备编号 + * @param scheduleId 排程ID + * @return commandId + */ + String sendScheduleUnbindCommand(String deviceNo, Long scheduleId); + /** * 下发设备初始化指令。用于解绑设备，允许设备未绑定用户时下发。 * diff --git a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceServiceImpl.java b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceServiceImpl.java index fe014ea..1e5d904 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/AppDeviceServiceImpl.java +++ b/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.getMacAddress()); + deviceCommandService.sendBindDeviceCommand(exists.getDeviceNo()); 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(device.getMacAddress(), workStatus, startTime, durationMin); + String commandId = deviceCommandService.sendSwitchCommand(deviceNo, workStatus, startTime, durationMin); log.info("[设备] 开关命令已下发 设备编号={} 命令编号={}", deviceNo, commandId); } diff --git a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/DeviceCommandServiceImpl.java b/water-modules/water-app/src/main/java/org/dromara/app/service/impl/DeviceCommandServiceImpl.java index c65bd18..d64f3e6 100644 --- a/water-modules/water-app/src/main/java/org/dromara/app/service/impl/DeviceCommandServiceImpl.java +++ b/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.Date; -import java.util.HashMap; -import java.util.Map; +import java.util.*; /** * 设备命令下发业务实现。 @@ -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 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 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 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");