From 759a8112474b541d48b467bcb3c10936b083b514 Mon Sep 17 00:00:00 2001 From: xy <748613696@qq.com> Date: Fri, 17 Jul 2026 13:44:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(consumer):=20=E6=B7=BB=E5=8A=A0=E8=AE=BE?= =?UTF-8?q?=E5=A4=87=E8=BD=AF=E4=BB=B6=E4=BF=A1=E6=81=AF=E6=9B=B4=E6=96=B0?= =?UTF-8?q?=E5=92=8C=E5=8D=87=E7=BA=A7=E5=A4=84=E7=90=86=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在CommonConsumer中新增多个Feign客户端依赖注入，包括设备、软件信息、数据管理等服务 - 实现设备软件版本信息的解析、存储和更新逻辑，支持版本信息的动态维护 - 添加设备升级响应处理机制，根据升级结果更新升级日志状态 - 新增设备重启消息发送功能，支持升级成功后自动重启设备 - 修改DeviceServiceImpl中的upgrade方法返回类型为String并优化升级流程 - 更新控制器接口返回值类型以匹配新的升级处理逻辑 - 移除原有的轮询等待响应机制，改为异步处理方式提升性能 --- .../njcn/access/utils/SendMessageUtil.java | 3 +- .../zlevent/controller/DeviceController.java | 6 +- .../njcn/zlevent/service/IDeviceService.java | 2 +- .../service/impl/DeviceServiceImpl.java | 52 +------ .../njcn/message/consumer/CommonConsumer.java | 128 +++++++++++++++++- 5 files changed, 135 insertions(+), 56 deletions(-) diff --git a/iot-access/access-api/src/main/java/com/njcn/access/utils/SendMessageUtil.java b/iot-access/access-api/src/main/java/com/njcn/access/utils/SendMessageUtil.java index 3296865..f71ed03 100644 --- a/iot-access/access-api/src/main/java/com/njcn/access/utils/SendMessageUtil.java +++ b/iot-access/access-api/src/main/java/com/njcn/access/utils/SendMessageUtil.java @@ -85,7 +85,8 @@ public class SendMessageUtil { return response; } try { - Thread.sleep(100); // 100ms 轮询一次 + // 100ms 轮询一次 + Thread.sleep(100); } catch (InterruptedException e) { Thread.currentThread().interrupt(); break; diff --git a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/controller/DeviceController.java b/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/controller/DeviceController.java index 4fc9120..91db860 100644 --- a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/controller/DeviceController.java +++ b/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/controller/DeviceController.java @@ -54,10 +54,10 @@ public class DeviceController extends BaseController { @ApiImplicitParam(name = "devId", value = "设备ID", required = true), @ApiImplicitParam(name = "edDataId", value = "程序版本Id", required = true) }) - public HttpResult upgrade(@RequestParam("devId") String devId, @RequestParam("edDataId") String edDataId) { + public HttpResult upgrade(@RequestParam("devId") String devId, @RequestParam("edDataId") String edDataId) { String methodDescribe = getMethodDescribe("upgrade"); - boolean res = deviceService.upgrade(devId, edDataId); - return HttpResultUtil.assembleCommonResponseResult(res ? CommonResponseEnum.SUCCESS : CommonResponseEnum.FAIL, res, methodDescribe); + String res = deviceService.upgrade(devId, edDataId); + return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, res, methodDescribe); } @OperateInfo(info = LogEnum.BUSINESS_COMMON) diff --git a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/IDeviceService.java b/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/IDeviceService.java index cf51482..f214953 100644 --- a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/IDeviceService.java +++ b/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/IDeviceService.java @@ -56,7 +56,7 @@ public interface IDeviceService { * @param devId * @param edDataId */ - boolean upgrade(String devId, String edDataId); + String upgrade(String devId, String edDataId); /** * 重启设备 diff --git a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/DeviceServiceImpl.java b/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/DeviceServiceImpl.java index 3677e53..ea16a04 100644 --- a/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/DeviceServiceImpl.java +++ b/iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/DeviceServiceImpl.java @@ -1,7 +1,6 @@ package com.njcn.zlevent.service.impl; import cn.hutool.core.util.IdUtil; -import cn.hutool.core.util.StrUtil; import com.alibaba.fastjson.JSON; import com.github.tocrhz.mqtt.publisher.MqttPublisher; import com.njcn.access.enums.TypeEnum; @@ -12,7 +11,6 @@ import com.njcn.csdevice.api.CsUpgradeLogsFeignClient; import com.njcn.csdevice.api.EquipmentFeignClient; import com.njcn.csdevice.pojo.dto.CsEquipmentDeliveryDTO; import com.njcn.csdevice.pojo.po.CsEdDataPO; -import com.njcn.csdevice.pojo.po.CsSoftInfoPO; import com.njcn.csdevice.pojo.po.CsUpgradeLogs; import com.njcn.middle.rocket.domain.BaseMessage; import com.njcn.oss.utils.FileStorageUtil; @@ -176,10 +174,7 @@ public class DeviceServiceImpl implements IDeviceService { } @Override - public boolean upgrade(String devId, String edDataId) { - // 先获取旧的版本信息 - //DevVersionResponeDTO.VersionInfo oldVersionInfo = this.getDeviceVersion(devId); - + public String upgrade(String devId, String edDataId) { List listHttpResult = equipmentFeignClient.queryDeviceById(Collections.singletonList(devId)).getData(); CsEquipmentDeliveryDTO csEquipmentDeliveryDTO = listHttpResult.get(0); @@ -190,7 +185,6 @@ public class DeviceServiceImpl implements IDeviceService { CsUpgradeLogs csUpgradeLogs = new CsUpgradeLogs(); csUpgradeLogs.setDevId(devId); csUpgradeLogs.setVersionNo(csEdDataPO.getVersionNo()); - csUpgradeLogs.setResult(0); UpgradeRequestDTO requestDTO = new UpgradeRequestDTO(); requestDTO.setDevId(devId); @@ -204,54 +198,16 @@ public class DeviceServiceImpl implements IDeviceService { msg.setName(filePath); detail1.setMsg(msg); requestDTO.setDetail(detail1); - BaseMessage message = new BaseMessage(); message.setMessageBody(JSON.toJSONString(requestDTO)); - // 使用 Redis 存储 guid 用于后续查询 redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_REQUEST + requestDTO.getGuid(), "pending", 120L); - // 发送 commonProducer.send(message, requestDTO.getFrontId()); - - // 轮询 Redis 等待响应 - UpgradeResponeDTO responeDTO = JSON.parseObject(sendMessageUtil.waitForResponse(requestDTO.getGuid(), 10), UpgradeResponeDTO.class); - UpgradeResponeDTO.Detail detail2 = responeDTO.getDetail(); - - if (detail2.getCode() == 200) { - // 修改数据库记录 - String softinfoId = csEquipmentDeliveryDTO.getSoftinfoId(); - - if (StrUtil.isNotBlank(softinfoId)) { - csSoftInfoFeignClient.removeSoftInfo(softinfoId); - } - CsSoftInfoPO softInfoPO = new CsSoftInfoPO(); - softInfoPO.setId(IdUtil.fastSimpleUUID()); - softInfoPO.setAppCheck(csEdDataPO.getCrc()); - softInfoPO.setAppDate(csEdDataPO.getVersionDate()); - softInfoPO.setAppVersion(csEdDataPO.getVersionNo()); - softInfoPO.setOpAttr("r"); - softInfoPO.setOsName("VxWorks"); - softInfoPO.setOsVersion("VxWorks"); - softInfoPO.setSoftUpdate("yes"); - csSoftInfoFeignClient.saveSoftInfo(softInfoPO); - equipmentFeignClient.updateSoftInfo(csEquipmentDeliveryDTO.getNdid(), softInfoPO.getId()); - - // 重新获取升级后的版本信息 - DevVersionResponeDTO.VersionInfo newVersionInfo = this.getDeviceVersion(devId); - if (newVersionInfo.getAppVersion().equals(csEdDataPO.getVersionNo()) && newVersionInfo.getCloudProtocolVer().equals(csEdDataPO.getVersionAgreement())) { - // 修改数据库记录 - equipmentFeignClient.updateSoftInfo(csEquipmentDeliveryDTO.getNdid(), softInfoPO.getId()); - csUpgradeLogs.setResult(1); - - csUpgradeLogsFeignClient.add(csUpgradeLogs); - return true; - } - csUpgradeLogsFeignClient.add(csUpgradeLogs); - return false; - } + // 记录升级日志 + csUpgradeLogs.setResult(3); csUpgradeLogsFeignClient.add(csUpgradeLogs); - return false; + return "指令下发成功"; } @Override diff --git a/iot-message/message-boot/src/main/java/com/njcn/message/consumer/CommonConsumer.java b/iot-message/message-boot/src/main/java/com/njcn/message/consumer/CommonConsumer.java index e778606..d2c8c07 100644 --- a/iot-message/message-boot/src/main/java/com/njcn/message/consumer/CommonConsumer.java +++ b/iot-message/message-boot/src/main/java/com/njcn/message/consumer/CommonConsumer.java @@ -1,22 +1,42 @@ package com.njcn.message.consumer; +import cn.hutool.core.util.IdUtil; import com.alibaba.fastjson.JSON; +import com.njcn.access.enums.TypeEnum; +import com.njcn.csdevice.api.CsEdDataFeignClient; +import com.njcn.csdevice.api.CsSoftInfoFeignClient; +import com.njcn.csdevice.api.CsUpgradeLogsFeignClient; +import com.njcn.csdevice.api.EquipmentFeignClient; +import com.njcn.csdevice.pojo.param.CsEdDataAddParm; +import com.njcn.csdevice.pojo.param.CsEdDataQueryParm; +import com.njcn.csdevice.pojo.po.CsEdDataPO; +import com.njcn.csdevice.pojo.po.CsSoftInfoPO; +import com.njcn.csdevice.pojo.vo.CsEquipmentDeliveryVO; import com.njcn.middle.rocket.constant.EnhanceMessageConstant; import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler; import com.njcn.mq.constant.BusinessTopic; import com.njcn.mq.constant.MessageStatus; +import com.njcn.mq.message.CloudTopicMessage; +import com.njcn.mq.template.CloudTopicMessageTemplate; import com.njcn.redis.pojo.enums.AppRedisKey; import com.njcn.redis.pojo.enums.RedisKeyEnum; import com.njcn.redis.utils.RedisUtil; +import com.njcn.system.api.DictTreeFeignClient; import com.njcn.system.api.RocketMqLogFeignClient; import com.njcn.system.pojo.po.RocketmqMsgErrorLog; +import com.njcn.system.pojo.po.SysDicTreePO; import com.njcn.zlevent.pojo.dto.CommonBaseMessage; +import com.njcn.zlevent.pojo.dto.DevVersionResponeDTO; +import com.njcn.zlevent.pojo.dto.UpgradeResponeDTO; import lombok.extern.slf4j.Slf4j; import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; import org.apache.rocketmq.spring.core.RocketMQListener; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Service; import javax.annotation.Resource; +import java.time.ZoneId; +import java.util.Date; import java.util.Objects; /** @@ -35,18 +55,120 @@ public class CommonConsumer extends EnhanceConsumerMessageHandler