From 25a05ae9d884bf773c07d7e13b475a5b99c6717b Mon Sep 17 00:00:00 2001 From: rqian <1206436827@qq.com> Date: Wed, 23 Sep 2026 14:38:12 +0800 Subject: [PATCH] =?UTF-8?q?#update=20=E8=AE=BE=E5=A4=87=E5=9C=A8=E7=BA=BF?= =?UTF-8?q?=E5=8D=87=E7=BA=A7=E7=9A=84=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../core/business/device/NettyDevice.java | 27 ++++-- .../core/ota/DeviceOtaOperateRecord.java | 2 +- .../java/com/maibu/memory/GlobalMemory.java | 3 +- .../com/maibu/mqtt/HostMessageHandler.java | 25 +++++- .../main/java/com/maibu/mqtt/MqttTopic.java | 15 ++-- .../maibu/controller/DeviceOtaController.java | 7 +- .../maibu/service/impl/DeviceOtaService.java | 83 ++++++++++++++++++- 7 files changed, 140 insertions(+), 22 deletions(-) diff --git a/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java b/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java index 662a8e5..3a0dd1a 100644 --- a/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java +++ b/maibu-common/src/main/java/com/maibu/core/business/device/NettyDevice.java @@ -9,6 +9,7 @@ import com.maibu.core.enums.DeviceTaskStaus; import com.maibu.core.host.HostWorkRecordCreateDTO; import com.maibu.core.host.MutiPointEntity; import com.maibu.core.host.PointEntity; +import com.maibu.core.ota.DeviceOtaFirmware; import com.maibu.mapper.DevicePlanTaskMapper; import com.maibu.mapper.WorkRecordMapper; import com.maibu.memory.GlobalMemory; @@ -26,10 +27,7 @@ import org.eclipse.paho.client.mqttv3.MqttException; import java.nio.ByteBuffer; import java.nio.ByteOrder; import java.time.LocalDateTime; -import java.util.ArrayList; -import java.util.LinkedList; -import java.util.List; -import java.util.Queue; +import java.util.*; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledFuture; import java.util.concurrent.TimeUnit; @@ -182,7 +180,7 @@ public class NettyDevice extends Connector { return true; } - public synchronized boolean failedTask() { + public synchronized boolean failedTask() { if (task != null) { task.setTaskStaus(DeviceTaskStaus.FAILED); task.setUpdateTime(LocalDateTime.now()); @@ -216,7 +214,7 @@ public class NettyDevice extends Connector { // 配置左右轮增益系数 public void sendWheelGainCoefficientsCommand(double leftForwardGain, double leftBackwardGain, - double rightForwardGain, double rightBackwardGain) { + double rightForwardGain, double rightBackwardGain) { ByteBuffer buffer = ByteBuffer.allocate(39); buffer.order(ByteOrder.LITTLE_ENDIAN); @@ -403,4 +401,21 @@ public class NettyDevice extends Connector { } } + public void upgrade(String deviceId, DeviceOtaFirmware firmware) throws MqttException { + Map payload = new HashMap<>(); + payload.put("action", "ota_upgrade"); + payload.put("version", firmware.getVersion()); + payload.put("file_url", firmware.getFileUrl()); + payload.put("firmware_id", firmware.getId()); + payload.put("sn", deviceId); + payload.put("timestamp", System.currentTimeMillis()); + payload.put("message_id", firmware.getId() + "_" + System.currentTimeMillis()); + + GlobalMemory.customMqttDeviceMonitor.publish( + mqttClientId, + String.format(MqttTopic.MOWER_OTA_UPGRADE_SET, deviceId), + JsonUtils.toJsonString(payload) + ); + } + } diff --git a/maibu-common/src/main/java/com/maibu/core/ota/DeviceOtaOperateRecord.java b/maibu-common/src/main/java/com/maibu/core/ota/DeviceOtaOperateRecord.java index b3ac4b2..703f74f 100644 --- a/maibu-common/src/main/java/com/maibu/core/ota/DeviceOtaOperateRecord.java +++ b/maibu-common/src/main/java/com/maibu/core/ota/DeviceOtaOperateRecord.java @@ -27,7 +27,7 @@ public class DeviceOtaOperateRecord extends BaseDO { private OtaOperate operateType; //操作类型 @TableField(typeHandler = JacksonTypeHandler.class) - private List deviceIds; //文件名 + private List deviceModels; //文件名 private String fileName; //名称 diff --git a/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java b/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java index 038ec51..dc66dc1 100644 --- a/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java +++ b/maibu-common/src/main/java/com/maibu/memory/GlobalMemory.java @@ -111,7 +111,8 @@ public class GlobalMemory { MqttTopic.MOWER_WILDCARD_TASK_STATUS,//任务状态 MqttTopic.MOWER_WILDCARD_ERROR,//错误 MqttTopic.MOWER_WILDCARD_REALTIME, //状态消息 - MqttTopic.MOWER_WILDCARD_TASK_TARGET_REPLY //路径规划下发回复 + MqttTopic.MOWER_WILDCARD_TASK_TARGET_REPLY, //路径规划下发回复 + MqttTopic.MOWER_OTA_RESULT_POST )); deviceConfig.setHandlerType(Constant.mqttHandleKey); //唯一的 用于处理上位机mqtt的 customMqttDeviceMonitor.registerDevice(deviceConfig); diff --git a/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java b/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java index 15054f6..006eda9 100644 --- a/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java +++ b/maibu-common/src/main/java/com/maibu/mqtt/HostMessageHandler.java @@ -114,6 +114,9 @@ public class HostMessageHandler implements CustomMqttMessageHandler { case "route_reply": handleRouteReply(deviceId, topic, payload); break; + case "upgrade_reply": + handleUpgradeReply(deviceId, topic, payload); + break; default: log.debug("割草机消息: deviceId={}, topic={}, action={}", deviceId, topic, action); break; @@ -162,7 +165,7 @@ public class HostMessageHandler implements CustomMqttMessageHandler { } public WebStatusMessageDTO createWebDeviceLocationMessage(HostLocationRelTimeDTO locationRelTimeDTO, - String deviceId) { + String deviceId) { WebStatusMessageDTO webStatusMessageDTO = new WebStatusMessageDTO(); List transferData = new ArrayList<>(); @@ -447,6 +450,23 @@ public class HostMessageHandler implements CustomMqttMessageHandler { } } + /** + * 处理控制指令响应 + */ + private void handleUpgradeReply(String deviceId, String topic, String payload) { +// log.debug("【路径任务下发响应】deviceId={}, payload={}", deviceId, payload); +// +// // todo 更新任务接收状态 +// HostNavigationReplyDTO replyDTO = JsonUtils.parseObject(payload, HostNavigationReplyDTO.class); +// NettyDevice device = deviceSessionManager.getDevice(deviceId); +// if (replyDTO.isAccepted()) { +// if (device != null && device.getTask() != null +// && String.valueOf(device.getTask().getId()).equals(replyDTO.getTask_id())) { +// device.getTask().setTaskAccepted(true); +// } +// } + } + /** * 从 Topic 中解析 deviceId * Topic 格式：mower/{deviceId}/... @@ -497,6 +517,9 @@ public class HostMessageHandler implements CustomMqttMessageHandler { return "control_reply"; if (remaining.equals("task/target/reply")) return "route_reply"; + if (remaining.equals("/ota/result/post")) + return "upgrade_reply"; + return null; } diff --git a/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java b/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java index 503c8cc..dbb1dfc 100644 --- a/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java +++ b/maibu-common/src/main/java/com/maibu/mqtt/MqttTopic.java @@ -64,15 +64,12 @@ public class MqttTopic { public static final String MOWER_ERROR_POST = "mower/%s/event/error/post"; // ==================== OTA 固件升级 ==================== - // /** 下发 OTA 升级指令（平台->设备） */ - // public static final String MOWER_OTA_UPGRADE_SET = - // "mower/%s/ota/upgrade/set"; - // /** OTA 升级进度上报（设备->平台） */ - // public static final String MOWER_OTA_PROGRESS_POST = - // "mower/%s/ota/progress/post"; - // /** OTA 升级结果上报（设备->平台） */ - // public static final String MOWER_OTA_RESULT_POST = - // "mower/%s/ota/result/post"; + /** 下发 OTA 升级指令（平台->设备） */ + public static final String MOWER_OTA_UPGRADE_SET = "mower/%s/ota/upgrade/set"; + /** OTA 升级进度上报（设备->平台） */ + public static final String MOWER_OTA_PROGRESS_POST = "mower/+/ota/progress/post"; + /** OTA 升级结果上报（设备->平台） */ + public static final String MOWER_OTA_RESULT_POST = "mower/+/ota/result/post"; // ==================== 设备信息 ==================== // /** 查询设备信息（平台->设备） */ diff --git a/maibu-service/maibu-iot-service/src/main/java/com/maibu/controller/DeviceOtaController.java b/maibu-service/maibu-iot-service/src/main/java/com/maibu/controller/DeviceOtaController.java index 30c28fa..6c5646c 100644 --- a/maibu-service/maibu-iot-service/src/main/java/com/maibu/controller/DeviceOtaController.java +++ b/maibu-service/maibu-iot-service/src/main/java/com/maibu/controller/DeviceOtaController.java @@ -86,7 +86,7 @@ public class DeviceOtaController extends BaseController { @ApiOperation("更新固件") @PostMapping("/firmware/upgrade") public AjaxResult upgrade(@RequestBody DeviceOtaOperateDTO dto) { - return AjaxResult.success(service.upgrade(dto)); + return service.upgrade(dto); } @ApiOperation("批量删除固件") @@ -115,6 +115,11 @@ public class DeviceOtaController extends BaseController { public AjaxResult selectOperateRecordById(@RequestParam Long id) { return AjaxResult.success(service.selectOperateRecordById(id)); } + + @GetMapping("/record/selectListByDeviceId") + public AjaxResult selectListByDeviceId(@RequestParam String deviceId) { + return AjaxResult.success(service.selectListByDeviceId(deviceId)); + } @ApiOperation("批量删除操作记录") @PostMapping("/record/delete") diff --git a/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/DeviceOtaService.java b/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/DeviceOtaService.java index 0852b1c..82f97da 100644 --- a/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/DeviceOtaService.java +++ b/maibu-service/maibu-iot-service/src/main/java/com/maibu/service/impl/DeviceOtaService.java @@ -1,19 +1,27 @@ package com.maibu.service.impl; import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper; +import com.maibu.core.business.device.NettyDevice; +import com.maibu.core.domain.AjaxResult; +import com.maibu.core.enums.OtaOperate; +import com.maibu.core.ota.DeviceOtaFirmware; +import com.maibu.core.ota.DeviceOtaOperateRecord; import com.maibu.dto.BaseQueryDTO; import com.maibu.dto.DeviceOtaOperateDTO; import com.maibu.mapper.DeviceOtaFirmwareMapper; import com.maibu.mapper.DeviceOtaOperateRecordMapper; -import com.maibu.core.ota.DeviceOtaFirmware; -import com.maibu.core.ota.DeviceOtaOperateRecord; +import com.maibu.memory.DeviceSessionManager; import com.maibu.utils.MinioUtil; +import com.maibu.utils.SecurityUtils; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import org.springframework.util.CollectionUtils; +import java.time.LocalDateTime; +import java.util.ArrayList; import java.util.List; @Service @@ -29,6 +37,11 @@ public class DeviceOtaService { @Autowired private MinioUtil minioUtil; + @Autowired + private DeviceSessionManager deviceSessionManager; + + private static final int BATCH_SIZE = 5; + public boolean saveFirmware(DeviceOtaFirmware firmware) { return deviceOtaFirmwareMapper.saveOrUpdate(firmware); } @@ -44,8 +57,63 @@ public class DeviceOtaService { } - public void upgrade(DeviceOtaOperateDTO dto) { + public AjaxResult upgrade(DeviceOtaOperateDTO dto) { + if (dto.getFirmwareId() == null) { + return AjaxResult.error("固件ID不能为空"); + } + if (CollectionUtils.isEmpty(dto.getDeviceIds())) { + return AjaxResult.error("请选择要升级的设备"); + } + DeviceOtaFirmware firmware = deviceOtaFirmwareMapper.selectById(dto.getFirmwareId()); + if (firmware == null) { + return AjaxResult.error("固件不存在"); + } + + DeviceOtaOperateRecord record = new DeviceOtaOperateRecord(); + record.setOperateType(OtaOperate.upgrade); + record.setFirmwareId(firmware.getId()); + record.setFileName(firmware.getFileName()); + record.setDeviceType(firmware.getDeviceType()); + record.setDeviceModels(new ArrayList<>()); + record.setCreateTime(LocalDateTime.now()); + record.setCreateBy(SecurityUtils.getUsername()); + + int totalDevices = dto.getDeviceIds().size(); + int successCount = 0; + int offlineCount = 0; + + for (int i = 0; i < totalDevices; i += BATCH_SIZE) { + List batch = dto.getDeviceIds().subList(i, Math.min(i + BATCH_SIZE, totalDevices)); + log.info("OTA 灰度升级 第 {} 批, 本批设备数: {}", (i / BATCH_SIZE) + 1, batch.size()); + + for (String deviceId : batch) { + NettyDevice device = deviceSessionManager.getDevice(deviceId); + if (device == null || device.getOnlineStatus() == 0) { + log.warn("设备 {} 离线，跳过升级", deviceId); + offlineCount++; + continue; + } + + try { + device.upgrade(deviceId, firmware); + record.getDeviceModels().add(deviceId); + successCount++; + log.info("OTA 升级指令已下发: deviceId={}, version={}", deviceId, firmware.getVersion()); + } catch (Exception e) { + log.error("OTA 升级指令下发失败: deviceId={}", deviceId, e); + } + } + } + + if (!record.getDeviceModels().isEmpty()) { + deviceOtaOperateRecordMapper.insert(record); + } + + String msg = String.format("升级指令下发完成: 成功=%d, 离线=%d, 总计=%d", + successCount, offlineCount, totalDevices); + log.info(msg); + return AjaxResult.success(msg); } @@ -85,6 +153,15 @@ public class DeviceOtaService { return deviceOtaOperateRecordMapper.selectById(id); } + public List selectListByDeviceId(String deviceId) { + LambdaQueryWrapper wrapper = new LambdaQueryWrapper<>(); + wrapper.apply( + "JSON_CONTAINS(device_models, {0})", + "\"" + deviceId + "\"" + ); + return deviceOtaOperateRecordMapper.selectList(wrapper); + } + public int deleteOperateRecord(List ids) { if (CollectionUtils.isEmpty(ids)) { return 0;