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

- 在CommonConsumer中新增多个Feign客户端依赖注入，包括设备、软件信息、数据管理等服务
- 实现设备软件版本信息的解析、存储和更新逻辑，支持版本信息的动态维护
- 添加设备升级响应处理机制，根据升级结果更新升级日志状态
- 新增设备重启消息发送功能，支持升级成功后自动重启设备
- 修改DeviceServiceImpl中的upgrade方法返回类型为String并优化升级流程
- 更新控制器接口返回值类型以匹配新的升级处理逻辑
- 移除原有的轮询等待响应机制，改为异步处理方式提升性能

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;

iot-access/access-boot/src/main/java/com/njcn/access/config/MqttAccessSchedulerConfig.java

@@ -49,4 +49,21 @@ public class MqttAccessSchedulerConfig {
         log.info("初始化MQTT接入信号量, 并发上限={}", concurrencyLimit);
         return new Semaphore(concurrencyLimit);
     }
+
+
+    /**
+     * MQTT消息异步处理线程池
+     * 用于将 @MqttSubscribe 回调中的重活(Feign调用、Kafka发送)异步化
+     * 避免阻塞Paho单线程回调，导致心跳消息排队延迟
+     */
+    @Bean(destroyMethod = "shutdownNow")
+    public ExecutorService mqttMessageExecutor() {
+        return new ThreadPoolExecutor(
+                10, 20,
+                60L, TimeUnit.SECONDS,
+                new LinkedBlockingQueue<>(500),
+                r -> new Thread(r, "mqtt-msg-handler"),
+                new ThreadPoolExecutor.CallerRunsPolicy()
+        );
+    }
 }

iot-access/access-boot/src/main/java/com/njcn/access/config/MqttOptionsFixer.java

@@ -0,0 +1,21 @@
+package com.njcn.access.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.config.BeanPostProcessor;
+import org.springframework.stereotype.Component;
+
+@Component
+@Slf4j
+public class MqttOptionsFixer implements BeanPostProcessor {
+
+    @Override
+    public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
+        if (bean instanceof MqttConnectOptions) {
+            MqttConnectOptions options = (MqttConnectOptions) bean;
+            options.setMaxInflight(200);
+        }
+        return bean;
+    }
+}

iot-access/access-boot/src/main/java/com/njcn/access/runner/UpdateDevStatusTimer.java

@@ -18,7 +18,7 @@ public class UpdateDevStatusTimer {
 
     private final ICsHeartService csHeartService;
 
-    @Scheduled(cron = "0 0/30 * * * ?")
+    @Scheduled(cron = "0 0/10 * * * ?")
     public void timer() {
         csHeartService.updateDevStatus();
     }

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsDevModelServiceImpl.java

@@ -43,6 +43,7 @@ import com.njcn.system.pojo.param.CsWaveParam;
 import com.njcn.system.pojo.param.EleEpdPqdParam;
 import com.njcn.system.pojo.param.EleEvtParam;
 import com.njcn.system.pojo.po.EleEpdPqd;
+import com.njcn.system.pojo.po.SysDicTreePO;
 import com.njcn.system.pojo.vo.DictTreeVO;
 import com.njcn.web.utils.RequestUtil;
 import lombok.AllArgsConstructor;
@@ -125,7 +126,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
             //2.录入数据集、详细数据(主设备、模块、监测设备、便携式设备...)
             analysisDataSet(templateDto,csDevModelPo.getId());
             //3.录入监测点模板表(记录当前模板有几个监测点，治理类型的模板目前规定1个监测点，电能质量模板根据逻辑子设备来)
-            addCsLineModel(templateDto,csDevModelPo.getId());
+            addCsLineModel(templateDto,csDevModelPo.getId(),devType);
             //4.如果是云前置的模板录入,重新录入完需要将模板关系信息重新录入
             if (Objects.equals(devType,DicDataEnum.DEV_CLD.getCode())) {
                 List<CsEquipmentDeliveryPO> list = eequipmentFeignClient.getAll().getData();
@@ -1205,7 +1206,7 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
     /**
      * 根据模板文件来录入监测点模板
      */
-    private void addCsLineModel(TemplateDto templateDto,String pId) {
+    private void addCsLineModel(TemplateDto templateDto,String pId,String devType) {
         List<CsLineModel> result = new ArrayList<>();
         //fixme 先用数据类型来区分模板的类型
         if (templateDto.getDataList().contains("Apf") || templateDto.getDataList().contains("Dvr")){
@@ -1217,6 +1218,10 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
             result.add(csLineModel);
         } else {
             List<ClDevDto> list = templateDto.getClDevDtoList();
+            //根据设备型号获取设备类型，根据设备类型生成监测点模板表数据
+            DictTreeVO vo1 = dictTreeFeignClient.queryByCode(devType).getData();
+            SysDicTreePO po1 = dictTreeFeignClient.queryById(vo1.getPid()).getData();
+            if (Objects.equals(po1.getCode(),"Direct_Connected_Device") || Objects.equals(po1.getCode(),"Portable")){
                 list.forEach(item->{
                     CsLineModel csLineModel = new CsLineModel();
                     csLineModel.setLineId(IdUtil.fastSimpleUUID());
@@ -1227,9 +1232,22 @@ public class CsDevModelServiceImpl implements ICsDevModelService {
                     } else if (Objects.equals(item.getClDid(),2) || Objects.equals(item.getLocation(), TypeEnum.LOAD.getCode())){
                         csLineModel.setName("负载侧监测点");
                         csLineModel.setPosition(dicDataFeignClient.getDicDataByCode(DicDataEnum.LOAD_SIDE.getCode()).getData().getId());
+                    } else {
+                        csLineModel.setName("输出侧监测点");
+                        csLineModel.setPosition(dicDataFeignClient.getDicDataByCode(DicDataEnum.OUTPUT_SIDE.getCode()).getData().getId());
                     }
                     result.add(csLineModel);
                 });
+            } else {
+                list.forEach(item->{
+                    CsLineModel csLineModel = new CsLineModel();
+                    csLineModel.setLineId(IdUtil.fastSimpleUUID());
+                    csLineModel.setPid(pId);
+                    csLineModel.setName(item.getClDid() + "#监测点");
+                    csLineModel.setPosition("");
+                    result.add(csLineModel);
+                });
+            }
         }
         if (CollectionUtil.isNotEmpty(result)){
             csLineModelService.addList(result);

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/CsHeartServiceImpl.java

@@ -122,7 +122,8 @@ public class CsHeartServiceImpl implements ICsHeartService {
         csEquipmentDeliveryService.updateRunStatusBynDid(nDid, AccessEnum.OFFLINE.getCode());
         //装置调整为注册状态
         csEquipmentDeliveryService.updateStatusBynDid(nDid,AccessEnum.REGISTERED.getCode(),null,null);
-        sendMessage(nDid);
+        //设备离线通知关闭
+        //sendMessage(nDid);
         //记录装置掉线时间
         PqsCommunicateDto dto = new PqsCommunicateDto();
         dto.setTime(LocalDateTime.now().format(DateTimeFormatter.ofPattern(DatePattern.NORM_DATETIME_PATTERN)));

iot-access/access-boot/src/main/java/com/njcn/access/service/impl/HeartbeatServiceImpl.java

@@ -21,7 +21,7 @@ public class HeartbeatServiceImpl implements IHeartbeatService {
     @Resource
     private RedisUtil redisUtil;
     private static final String HEARTBEAT_REDIS_KEY_PREFIX = "HEARTBEAT:";
-    private static final int DELAY_LEVEL_4MIN = 7;
+    private static final int DELAY_LEVEL_3MIN = 7;
     private static final long HEARTBEAT_EXPIRE_SECONDS = 180;
 
     @Override
@@ -29,14 +29,12 @@ public class HeartbeatServiceImpl implements IHeartbeatService {
         String redisKey = HEARTBEAT_REDIS_KEY_PREFIX + nDid;
         long currentTime = System.currentTimeMillis();
 
-        redisUtil.saveByKey(redisKey, currentTime);
+        redisUtil.saveByKeyWithExpire(redisKey, currentTime, HEARTBEAT_EXPIRE_SECONDS);
-        redisUtil.expire(redisKey, HEARTBEAT_EXPIRE_SECONDS);
 
         HeartbeatTimeoutMessage message = new HeartbeatTimeoutMessage();
-
         message.setNDid(nDid);
         message.setTimestamp(currentTime);
-        message.setDelayLevel(DELAY_LEVEL_4MIN);
+        message.setDelayLevel(DELAY_LEVEL_3MIN);
         heartbeatTimeoutMessageTemplate.sendMember(message);
     }
 

iot-analysis/analysis-stat/stat-boot/src/main/java/com/njcn/stat/service/impl/StatServiceImpl.java

@@ -308,6 +308,9 @@ public class StatServiceImpl implements IStatService {
             tags.put(InfluxDBTableConstant.LINE_ID,lineId);
             tags.put(InfluxDBTableConstant.PHASIC_TYPE,dataArrayList.get(i).getPhase());
             tags.put(InfluxDBTableConstant.VALUE_TYPE,statMethod.toUpperCase());
+            tags.put(InfluxDBTableConstant.CL_DID,clDid.toString());
+            tags.put(InfluxDBTableConstant.PROCESS,process.toString());
+
             if (Objects.isNull(item.getDataTag())) {
                 tags.put(InfluxDBTableConstant.QUALITY_FLAG,"0");
             } else {
@@ -320,8 +323,6 @@ public class StatServiceImpl implements IStatService {
             } else {
                 fields.put(dataArrayList.get(i).getInfluxDbName(),Objects.equals(floats.get(i),3.14159f) ? null:floats.get(i));
             }
-            fields.put(InfluxDBTableConstant.CL_DID,clDid.toString());
-            fields.put(InfluxDBTableConstant.PROCESS,process.toString());
 
             Point point = influxDbUtils.pointBuilder(tableName, adjustedTimeSec, TimeUnit.SECONDS, tags, fields);
             BatchPoints batchPoints = BatchPoints.database(influxDbUtils.getDbName()).retentionPolicy("").consistency(InfluxDB.ConsistencyLevel.ALL).build();
@@ -387,8 +388,8 @@ public class StatServiceImpl implements IStatService {
                 data.forEach(item->{
                     DataArrayLiteDto dataArrayLiteDto = new DataArrayLiteDto();
                     dataArrayLiteDto.setName(item.getName());
-                    dataArrayLiteDto.setInfluxDbName(item.getInfluxDbName());
                     dataArrayLiteDto.setPhase(item.getPhase());
+                    dataArrayLiteDto.setInfluxDbName(item.getInfluxDbName());
                     result.add(dataArrayLiteDto);
                 });
             }

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<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);
+        String res = deviceService.upgrade(devId, edDataId);
-        return HttpResultUtil.assembleCommonResponseResult(res ? CommonResponseEnum.SUCCESS : CommonResponseEnum.FAIL, res, methodDescribe);
+        return HttpResultUtil.assembleCommonResponseResult(CommonResponseEnum.SUCCESS, res, methodDescribe);
     }
 
     @OperateInfo(info = LogEnum.BUSINESS_COMMON)

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);
 
     /**
      * 重启设备

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/CsAlarmServiceImpl.java

@@ -21,7 +21,6 @@ import com.njcn.system.pojo.po.EleEpdPqd;
 import com.njcn.system.pojo.po.SysDicTreePO;
 import com.njcn.zlevent.mapper.CsEventMapper;
 import com.njcn.zlevent.pojo.po.CsEventLogs;
-//import com.njcn.zlevent.service.AppNotificationService;
 import com.njcn.zlevent.service.ICsAlarmService;
 import com.njcn.zlevent.service.ICsEventLogsService;
 import com.njcn.zlevent.service.ICsEventService;
@@ -138,11 +137,9 @@ public class CsAlarmServiceImpl extends ServiceImpl<CsEventMapper, CsEventPO> im
                     if (Objects.isNull(item.getCode())){
                         msgSendParam.setEventName(item.getName());
                         msgSendFeignClient.appMsgSend(msgSendParam);
-//                        appNotificationService.sendAppNotification(2,item.getType(),po.getId(),item.getName(),eventTime,id,po.getNdid(),null,null,null);
                     } else {
                         msgSendParam.setEventName(item.getCode());
                         msgSendFeignClient.appMsgSend(msgSendParam);
-//                        appNotificationService.sendAppNotification(2,item.getType(),po.getId(),item.getCode(),eventTime,id,po.getNdid(),null, null,null);
                         //更新字典信息
                         EleEpdPqd eleEpdPqd = epdFeignClient.findByName(item.getName()).getData();
                         EleEpdPqdParam.EleEpdPqdUpdateParam updateParam = new EleEpdPqdParam.EleEpdPqdUpdateParam();

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) {
+    public String upgrade(String devId, String edDataId) {
-        // 先获取旧的版本信息
-        //DevVersionResponeDTO.VersionInfo oldVersionInfo = this.getDeviceVersion(devId);
-
         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 等待响应
+        csUpgradeLogs.setResult(3);
-        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;
+        return "指令下发成功";
-            }
-            csUpgradeLogsFeignClient.add(csUpgradeLogs);
-            return false;
-        }
-        csUpgradeLogsFeignClient.add(csUpgradeLogs);
-        return false;
     }
 
     @Override

iot-analysis/analysis-zl-event/zl-event-boot/src/main/java/com/njcn/zlevent/service/impl/EventServiceImpl.java

@@ -247,6 +247,7 @@ public class EventServiceImpl implements IEventService {
                             }
                         }
                     }
+                    if (Objects.equals(item.getType(),"2")){
                         MsgSendParam msgSendParam = new MsgSendParam();
                         msgSendParam.setEventType(1);
                         msgSendParam.setType("2");
@@ -259,6 +260,7 @@ public class EventServiceImpl implements IEventService {
                         msgSendParam.setPersistTime(persistTime);
                         msgSendParam.setDropZone(dropZone);
                         msgSendFeignClient.appMsgSend(msgSendParam);
+                    }
                     //如果是暂降事件，则异步发送短信
                     if (Objects.equals(item.getName(), "Evt_Sys_DipStr")) {
                         MsgSendParam msgSendParam2 = new MsgSendParam();
@@ -312,7 +314,9 @@ public class EventServiceImpl implements IEventService {
         rmpEventDetailPo.setDealFlag(0);
         rmpEventDetailPo.setFileFlag(0);
         rmpEventDetailPo.setPhase(item.getPhase());
+        if (!Objects.isNull(item.getSagSource())) {
             rmpEventDetailPo.setSagsource(getSagSource(item.getSagSource()));
+        }
         rmpEventDetailPo.setSeverity(item.getSeverity());
         wlRmpEventDetailMapper.insert(rmpEventDetailPo);
     }
@@ -427,6 +431,7 @@ public class EventServiceImpl implements IEventService {
 
     @Override
     public void getCldEventData(CldLogMessage cldLogMessage) {
+        log.info("数据报文=====：" + cldLogMessage);
         CsEventPO po = new CsEventPO();
         po.setStartTime(LocalDateTime.parse(cldLogMessage.getTime(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss")));
         po.setTag(cldLogMessage.getLog());

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<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;
     }
 
     /***