Microservice/iot
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 1 Wiki Activity

feat(consumer): 添加设备软件信息更新和升级处理功能

Browse Source 
- 在CommonConsumer中新增多个Feign客户端依赖注入，包括设备、软件信息、数据管理等服务
- 实现设备软件版本信息的解析、存储和更新逻辑，支持版本信息的动态维护
- 添加设备升级响应处理机制，根据升级结果更新升级日志状态
- 新增设备重启消息发送功能，支持升级成功后自动重启设备
- 修改DeviceServiceImpl中的upgrade方法返回类型为String并优化升级流程
- 更新控制器接口返回值类型以匹配新的升级处理逻辑
- 移除原有的轮询等待响应机制，改为异步处理方式提升性能
This commit is contained in:
xy
2026-07-17 13:44:58 +08:00
parent 3c369d0244
commit 759a811247
5 changed files with 135 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
iot-access/access-api/src/main/java/com/njcn/access/utils/SendMessageUtil.java
View File

@@ -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;

6
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/controller/DeviceController.java
View File

@@ -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<Boolean> upgrade(@RequestParam("devId") String devId, @RequestParam("edDataId") String edDataId) {
public HttpResult<String> 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)

2
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/IDeviceService.java
View File

@@ -56,7 +56,7 @@ public interface IDeviceService {
* @param devId
* @param edDataId
*/
boolean upgrade(String devId, String edDataId);
String upgrade(String devId, String edDataId);
/**
* 重启设备

52
iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/DeviceServiceImpl.java
View File

@@ -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<CsEquipmentDeliveryDTO> 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

128
iot-message/message-boot/src/main/java/com/njcn/message/consumer/CommonConsumer.java
View File

@@ -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<CommonBaseMess
@Resource
private RedisUtil redisUtil;
@Resource
private RocketMqLogFeignClient rocketMqLogFeignClient;
@Resource
private EquipmentFeignClient equipmentFeignClient;
@Resource
private CsSoftInfoFeignClient csSoftInfoFeignClient;
@Resource
private CsEdDataFeignClient csEdDataFeignClient;
@Autowired
private DictTreeFeignClient dictTreeFeignClient;
@Resource
private CsUpgradeLogsFeignClient csUpgradeLogsFeignClient;
@Resource
private CloudTopicMessageTemplate cloudTopicMessageTemplate;
@Override
protected void handleMessage(CommonBaseMessage message) throws Exception {
protected void handleMessage(CommonBaseMessage message) {
log.info("@@@@@处理Common信息");
System.out.println(JSON.toJSON(message));
String guid = message.getGuid();
// 将响应结果存入 Redis
redisUtil.saveByKeyWithExpire(AppRedisKey.COMMON_RESOPNSE + guid, JSON.toJSONString(message), 120L);
//fixme 这边有部分开发好的功能，如果修改的话，改动量较大，相当于重做了，这边就先临时处理，后续再完善
//更新设备的软件信息报文
DevVersionResponeDTO responseDTO = JSON.parseObject(JSON.toJSONString(message), DevVersionResponeDTO.class);
if (!Objects.isNull(responseDTO.getDetail())) {
DevVersionResponeDTO.Detail detail2 = responseDTO.getDetail();
if (detail2.getType() == 1112) {
CsEquipmentDeliveryVO vo = equipmentFeignClient.queryEquipmentByndid(responseDTO.getDevMac()).getData();
if (!Objects.isNull(vo)) {
String softInfo = vo.getSoftinfoId();
if (!Objects.isNull(softInfo)) {
csSoftInfoFeignClient.removeSoftInfo(softInfo);
}
DevVersionResponeDTO.VersionInfo versionInfo = detail2.getMsg().getVersionInfo();
CsSoftInfoPO po = new CsSoftInfoPO();
String id = IdUtil.fastSimpleUUID();
po.setId(id);
po.setOpAttr("r");
po.setAppVersion(versionInfo.getAppVersion());
po.setAppDate(versionInfo.getAppDate().atStartOfDay());
po.setAppCheck(versionInfo.getAppChecksum());
po.setSoftUpdate("yes");
csSoftInfoFeignClient.saveSoftInfo(po);
//修改软件信息
equipmentFeignClient.updateSoftInfo(vo.getNdid(), id);
//新增版本(先查看是否已经存在，不存在再录入)
CsEdDataQueryParm param = new CsEdDataQueryParm();
param.setVersionAgreement(versionInfo.getCloudProtocolVer());
param.setVersionNo(versionInfo.getAppVersion());
param.setVersionDate(versionInfo.getAppDate());
param.setCrc(versionInfo.getAppChecksum());
param.setDevType(vo.getDevModel());
CsEdDataPO csEdDataPo = csEdDataFeignClient.findByCondition(param).getData();
if (!Objects.isNull(csEdDataPo)) {
CsEdDataAddParm addParam = new CsEdDataAddParm();
addParam.setDevType(vo.getDevType());
SysDicTreePO dicTreePO = dictTreeFeignClient.queryById(vo.getDevModel()).getData();
if (dicTreePO != null) {
addParam.setDevTypeName(dicTreePO.getName());
}
addParam.setVersionNo(versionInfo.getAppVersion());
addParam.setVersionAgreement(versionInfo.getCloudProtocolVer());
addParam.setVersionDate(
Date.from(versionInfo.getAppDate().atStartOfDay(ZoneId.systemDefault()).toInstant())
);
addParam.setVersionType("通用");
addParam.setCrc(versionInfo.getAppChecksum());
addParam.setType(vo.getDevType());
csEdDataFeignClient.addEdDataFirst(addParam);
}
}
}
}
//升级报文
UpgradeResponeDTO upgradeResponeDTO = JSON.parseObject(JSON.toJSONString(message), UpgradeResponeDTO.class);
if (!Objects.isNull(upgradeResponeDTO.getDetail())) {
UpgradeResponeDTO.Detail detail = upgradeResponeDTO.getDetail();
if (detail.getType() == 1115) {
CsEquipmentDeliveryVO vo = equipmentFeignClient.queryEquipmentByndid(responseDTO.getDevMac()).getData();
String devId = vo.getId();
if (detail.getCode() == 200) {
//更新升级日志
String code = "200";
Integer resultCode = 1;
csUpgradeLogsFeignClient.update(devId,code,resultCode);
//重启装置
CloudTopicMessage message1 = rebootDev(devId, vo.getNodeProcess(), vo.getNodeId(), TypeEnum.DEVICE_REBOOT.getCode());
cloudTopicMessageTemplate.sendMember(message1,vo.getNodeId());
} else if (detail.getCode() == 202){
String code = "202";
Integer resultCode = 2;
csUpgradeLogsFeignClient.update(devId,code,resultCode);
} else {
String code = String.valueOf(detail.getCode());
Integer resultCode = 0;
csUpgradeLogsFeignClient.update(devId,code,resultCode);
}
}
}
}
public CloudTopicMessage rebootDev(String devId, Integer node, String frontId, String code) {
CloudTopicMessage message = new CloudTopicMessage();
message.setGuid(IdUtil.simpleUUID());
message.setDevId(devId);
message.setNode(node);
message.setFrontId(frontId);
CloudTopicMessage.Detail detail = new CloudTopicMessage.Detail();
detail.setType(Integer.valueOf(code));
detail.setMsg(null);
message.setDetail(detail);
return message;
}
/***