feat(综合搜索)：检验模块检验信息更新后同步更新es

amos-boot-system-tzs/amos-boot-module-common/amos-boot-module-common-api/src/main/java/com/yeejoin/amos/boot/module/common/api/constant/TZSCommonConstant.java

@@ -8,4 +8,9 @@ public class TZSCommonConstant {
      * 陕西省行政区划编码
      */
     public final static String SHAN_XI_REGION_CODE = "610000";
+
+    /**
+     * 数据刷新主题-前缀,第一个为：dataType, 第二个为：操作类型：增删改
+     */
+    public final static String DATA_REFRESH_TOPIC = "data/refresh/%s/%s";
 }

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/refresh/adapter/DataRefreshEventBuilderService.java

+package com.yeejoin.amos.boot.module.jg.biz.refresh.adapter;
+
+import com.yeejoin.amos.boot.module.common.biz.event.CommonPublisher;
+import com.yeejoin.amos.boot.module.common.biz.refresh.DataRefreshEvent;
+import lombok.RequiredArgsConstructor;
+import org.eclipse.paho.client.mqttv3.MqttTopic;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+
+import java.util.List;
+
+@Component
+@RequiredArgsConstructor
+public class DataRefreshEventBuilderService {
+
+    private final CommonPublisher publisher;
+
+    @Transactional(rollbackFor = Exception.class)
+    public void processMessage(String topic, List<String> records) {
+        String[] topicCxt = topic.split(MqttTopic.TOPIC_LEVEL_SEPARATOR);
+        if (topicCxt.length == 4) {
+            String dataType = topicCxt[2];
+            String operation = topicCxt[3];
+            DataRefreshEvent dataRefreshEvent = new DataRefreshEvent(this, records, dataType, DataRefreshEvent.Operation.valueOf(operation));
+            publisher.publish(dataRefreshEvent);
+        }
+    }
+}

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/refresh/adapter/Emq2DataRefreshEvent.java

+package com.yeejoin.amos.boot.module.jg.biz.refresh.adapter;
+
+import com.alibaba.fastjson.JSONObject;
+import lombok.RequiredArgsConstructor;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+import org.typroject.tyboot.component.emq.EmqKeeper;
+import org.typroject.tyboot.component.emq.EmqxListener;
+
+import javax.annotation.PostConstruct;
+import java.nio.charset.StandardCharsets;
+import java.util.List;
+
+@Component
+@RequiredArgsConstructor
+public class Emq2DataRefreshEvent extends EmqxListener {
+
+    private final EmqKeeper emqKeeper;
+
+    @Value("${data.refresh.tp:data/refresh/+/+}")
+    private String dataRefreshTp;
+
+    @Value("${spring.application.name}")
+    private String applicationName;
+
+    private final DataRefreshEventBuilderService dataRefreshEventBuilderService;
+
+    @Override
+    public void processMessage(String topic, MqttMessage message) {
+        byte[] payload = message.getPayload();
+        String str = new String(payload, StandardCharsets.UTF_8);
+        List<String> records = JSONObject.parseArray(str, String.class);
+        dataRefreshEventBuilderService.processMessage(topic, records);
+    }
+
+    @PostConstruct
+    void init() throws Exception {
+        emqKeeper.subscript(this.buildShareTopic(dataRefreshTp), 2, this);
+    }
+
+    private String buildShareTopic(String topic) {
+        return "$share/" + applicationName + "/" + topic;
+    }
+}

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/service/impl/IdxBizJgProjectContraptionServiceImplService.java

@@ -570,10 +570,11 @@ public class IdxBizJgProjectContraptionServiceImplService extends BaseEntityServ
     @Transactional(rollbackFor = Exception.class)
     public void saveOrUpdateDetectionInfoBatch(JSONObject jsonObject) {
         if(Objects.nonNull(jsonObject)){
-            JSONArray records = jsonObject.getJSONArray("records");
-            if(CollectionUtil.isNotEmpty(records)){
+            JSONArray items = jsonObject.getJSONArray("records");
+            Set<String> records = new HashSet<>();
+            if(CollectionUtil.isNotEmpty(items)){
                 List<IdxBizJgInspectionDetectionInfo> jgInspectionDetectionInfos= new ArrayList<>();
-                for (Object record : records) {
+                for (Object record : items) {
                     IdxBizJgInspectionDetectionInfo detectionInfo = JSON.parseObject(JSONObject.toJSONString(record), IdxBizJgInspectionDetectionInfo.class);
                     if(Objects.nonNull(detectionInfo)){
                         if (record instanceof LinkedHashMap){
@@ -581,6 +582,7 @@ public class IdxBizJgProjectContraptionServiceImplService extends BaseEntityServ
                           detectionInfo.setRecDate(new Date());
                           detectionInfo.setSequenceNbr((String) map.get("detectionInfoSequenceNbr"));
                           detectionInfo.setInspectConclusion((String)map.get("inspectConclusionCode"));
+                          records.add(detectionInfo.getRecord());
                           jgInspectionDetectionInfos.add(detectionInfo);
                         }
                     }
@@ -588,6 +590,7 @@ public class IdxBizJgProjectContraptionServiceImplService extends BaseEntityServ
                 if(CollUtil.isNotEmpty(jgInspectionDetectionInfos)){
                     detectionInfoService.saveOrUpdateBatch(jgInspectionDetectionInfos);
                 }
+                eventPublisher.publish(new DataRefreshEvent(this, new ArrayList<>(records), DataRefreshEvent.DataType.equipment.name(), DataRefreshEvent.Operation.UPDATE));
             }
         }
 

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/service/impl/JgInstallationNoticeServiceImpl.java

@@ -76,6 +76,7 @@ import org.redisson.api.RedissonClient;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.http.HttpStatus;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
@@ -197,7 +198,9 @@ public class JgInstallationNoticeServiceImpl extends BaseService<JgInstallationN
     EventPublisher eventPublisher;
     @Autowired
     private JgUseRegistrationServiceImpl jgUseRegistrationService;
+
     @Autowired
+    @Lazy
     private JgReformNoticeServiceImpl jgReformNoticeService;
     @Autowired
     private JgTransferNoticeServiceImpl jgTransferNoticeService;

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/service/impl/JgReformNoticeServiceImpl.java

@@ -58,6 +58,8 @@ import lombok.extern.slf4j.Slf4j;
 import org.redisson.api.RLock;
 import org.redisson.api.RedissonClient;
 import org.springframework.beans.BeanUtils;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Lazy;
 import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.transaction.support.TransactionSynchronization;
@@ -121,7 +123,6 @@ public class JgReformNoticeServiceImpl extends BaseService<JgReformNoticeDto, Jg
     private final RedisUtils redisUtils;
     private final SnowflakeIdUtil sequence;
     private final JgRegistrationHistoryServiceImpl jgRegistrationHistoryService;
-    private final JgInstallationNoticeServiceImpl jgInstallationNoticeService;
     private final JgReformNoticeMapper jgReformNoticeMapper;
     private final CommonServiceImpl commonService;
     private final CmWorkflowServiceImpl cmWorkflowService;
@@ -141,7 +142,10 @@ public class JgReformNoticeServiceImpl extends BaseService<JgReformNoticeDto, Jg
     private final IdxBizJgInspectionDetectionInfoServiceImpl idxBizJgInspectionDetectionInfoService;
     private final IIdxBizJgTechParamsPipelineService iIdxBizJgTechParamsPipelineService;
     private final JgUseRegistrationServiceImpl useRegistrationService;
-    private final UseInfoMapper useInfoMapper;
+
+    @Autowired
+    @Lazy
+    private JgInstallationNoticeServiceImpl jgInstallationNoticeService;
 
 
     @ResultFieldMapping({
@@ -896,7 +900,7 @@ public class JgReformNoticeServiceImpl extends BaseService<JgReformNoticeDto, Jg
     /**
      * 通过和驳回
      *
-     * @param dto 改造信息
+     * @param map 改造信息
      * @param op  通过或驳回
      */
     @GlobalTransactional(rollbackFor = Exception.class)

amos-boot-system-tzs/amos-boot-module-jyjc/amos-boot-module-jyjc-biz/src/main/java/com/yeejoin/amos/boot/module/jyjc/biz/controller/JyjcInspectionResultController.java

@@ -222,6 +222,7 @@ public class JyjcInspectionResultController extends BaseController {
     public ResponseModel<Boolean> useUnitEntryWithSave(@RequestBody Map<String, Object> entryData) {
         return ResponseHelper.buildResponse(jyjcInspectionResultServiceImpl.useUnitEntryWithSave(entryData,getSelectedOrgInfo()));
     }
+
     @TycloudOperation(ApiLevel = UserType.AGENCY)
     @PostMapping(value = "/useUnitEntryWithUpdate")
     @ApiOperation(httpMethod = "POST", value = "使用单位录入检验结果-更新", notes = "使用单位录入检验结果-更新")

amos-boot-system-tzs/amos-boot-module-jyjc/amos-boot-module-jyjc-biz/src/main/java/com/yeejoin/amos/boot/module/jyjc/biz/event/listener/InspectionDetectionSaveToDbEventListener.java

@@ -5,11 +5,13 @@ import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.yeejoin.amos.boot.biz.common.utils.SnowflakeIdUtil;
 import com.yeejoin.amos.boot.module.jyjc.api.entity.JyjcInspectionResult;
 import com.yeejoin.amos.boot.module.jyjc.biz.event.InspectionDetectionSaveToDbEvent;
+import com.yeejoin.amos.boot.module.jyjc.biz.event.listener.service.InspectionDetectionInfoUpdateService;
 import com.yeejoin.amos.boot.module.jyjc.biz.event.publisher.BizEmqPublisher;
 import com.yeejoin.amos.boot.module.jyjc.biz.service.impl.CommonServiceImpl;
 import com.yeejoin.amos.boot.module.ymt.api.entity.InspectionDetectionInfo;
 import com.yeejoin.amos.boot.module.ymt.api.entity.TzsUserInfo;
 import com.yeejoin.amos.boot.module.ymt.api.mapper.InspectionDetectionInfoMapper;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
@@ -31,26 +33,15 @@ import java.util.stream.Collectors;
  */
 @Component
 @Slf4j
+@RequiredArgsConstructor
 public class InspectionDetectionSaveToDbEventListener implements ApplicationListener<InspectionDetectionSaveToDbEvent> {
 
     @Value("${inspect.info.save.thread.number:3}")
     private int threadNumber;
 
-    @Autowired
-    CommonServiceImpl commonService;
-
-    @Autowired
-    private SnowflakeIdUtil sequence;
-
-    @Autowired
-    InspectionDetectionInfoMapper inspectionDetectionInfoMapper;
-
-    @Autowired
-    BizEmqPublisher bizEmqPublisher;
-
-
-    private BlockingQueue<JyjcInspectionResult> blockingQueue = new LinkedBlockingQueue<>();
+    private final BlockingQueue<JyjcInspectionResult> blockingQueue = new LinkedBlockingQueue<>();
 
+    private final InspectionDetectionInfoUpdateService  updateService;
 
 
     @Override
@@ -67,27 +58,12 @@ public class InspectionDetectionSaveToDbEventListener implements ApplicationList
                 while (true) {
                     try {
                         JyjcInspectionResult jyjcInspectionResult = blockingQueue.take();
-                        InspectionDetectionInfo info = new InspectionDetectionInfo();
-                        QueryWrapper<InspectionDetectionInfo> wrapper = new QueryWrapper<>();
-                        wrapper.lambda().eq(InspectionDetectionInfo::getInspectReportNo, jyjcInspectionResult.getResultNo());
-                        List<InspectionDetectionInfo> list = inspectionDetectionInfoMapper.selectList(wrapper);
-                        if (CollectionUtils.isEmpty(list)) {
-                            fillInspectionFields(jyjcInspectionResult, info);
-                            info.setSequenceNbr(sequence.nextId() + "");
-                            inspectionDetectionInfoMapper.insert(info);
-                            bizEmqPublisher.sendInspectionMsgAfterSave(info, "insert");
-                        } else {
-                            info = list.get(0);
-                            fillInspectionFields(jyjcInspectionResult, info);
-                            inspectionDetectionInfoMapper.updateById(info);
-                            bizEmqPublisher.sendInspectionMsgAfterSave(info, "update");
-                        }
+                        updateService.update(jyjcInspectionResult);
                     } catch (Exception e) {
                         log.error(e.getMessage(), e);
                     }
                 }
             });
-
         }
     }
 
@@ -95,25 +71,6 @@ public class InspectionDetectionSaveToDbEventListener implements ApplicationList
 
 
 
-    private void fillInspectionFields(JyjcInspectionResult jyjcInspectionResult, InspectionDetectionInfo info) {
-        info.setInspectOrgName(jyjcInspectionResult.getInspectionTypeName());
-        info.setInspectType(jyjcInspectionResult.getInspectionType());
-        info.setRecord(jyjcInspectionResult.getEquipUnicode());
-        info.setInspectDate(jyjcInspectionResult.getInspectionDate());
-        info.setInspectStaff(getInspectUserName(jyjcInspectionResult));
-        info.setInspectStaffCode(jyjcInspectionResult.getInspector());
-        info.setInspectConclusion(jyjcInspectionResult.getInspectionConclusion());
-        info.setProblemRemark(jyjcInspectionResult.getNonConformance());
-        info.setNextInspectDate(jyjcInspectionResult.getNextInspectionDate());
-        info.setSequenceCode(jyjcInspectionResult.getEquipUnicode());
-        info.setInspectOrgCode(jyjcInspectionResult.getInspectionUnitCode());
-        info.setInspectReportNo(jyjcInspectionResult.getResultNo());
-        info.setRecDate(new Date());
-    }
 
-    private String getInspectUserName(JyjcInspectionResult model) {
-        List<TzsUserInfo> userInfos = commonService.getUserInfosByUnitCode(model.getInspectionUnitCode());
-        return userInfos.stream().filter(u -> model.getInspector().contains(u.getSequenceNbr() + "")).map(TzsUserInfo::getName).collect(Collectors.joining(","));
-    }
 
 }

amos-boot-system-tzs/amos-boot-module-jyjc/amos-boot-module-jyjc-biz/src/main/java/com/yeejoin/amos/boot/module/jyjc/biz/event/listener/service/InspectionDetectionInfoUpdateService.java

+package com.yeejoin.amos.boot.module.jyjc.biz.event.listener.service;
+
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.yeejoin.amos.boot.biz.common.utils.SnowflakeIdUtil;
+import com.yeejoin.amos.boot.module.common.api.constant.TZSCommonConstant;
+import com.yeejoin.amos.boot.module.common.biz.refresh.DataRefreshEvent;
+import com.yeejoin.amos.boot.module.jyjc.api.entity.JyjcInspectionResult;
+import com.yeejoin.amos.boot.module.jyjc.biz.event.publisher.BizEmqPublisher;
+import com.yeejoin.amos.boot.module.jyjc.biz.service.impl.CommonServiceImpl;
+import com.yeejoin.amos.boot.module.ymt.api.entity.InspectionDetectionInfo;
+import com.yeejoin.amos.boot.module.ymt.api.entity.TzsUserInfo;
+import com.yeejoin.amos.boot.module.ymt.api.mapper.InspectionDetectionInfoMapper;
+import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.util.CollectionUtils;
+import org.typroject.tyboot.component.emq.EmqKeeper;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Collections;
+import java.util.Date;
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Component
+@Slf4j
+public class InspectionDetectionInfoUpdateService {
+
+    @Autowired
+    CommonServiceImpl commonService;
+
+    @Autowired
+    private SnowflakeIdUtil sequence;
+
+    @Autowired
+    InspectionDetectionInfoMapper inspectionDetectionInfoMapper;
+
+    @Autowired
+    BizEmqPublisher bizEmqPublisher;
+
+    @Autowired
+    private EmqKeeper emqKeeper;
+
+    @Transactional(rollbackFor = Exception.class)
+    public void update(JyjcInspectionResult jyjcInspectionResult) {
+        InspectionDetectionInfo info = new InspectionDetectionInfo();
+        QueryWrapper<InspectionDetectionInfo> wrapper = new QueryWrapper<>();
+        wrapper.lambda().eq(InspectionDetectionInfo::getInspectReportNo, jyjcInspectionResult.getResultNo());
+        List<InspectionDetectionInfo> list = inspectionDetectionInfoMapper.selectList(wrapper);
+        String record;
+        if (CollectionUtils.isEmpty(list)) {
+            fillInspectionFields(jyjcInspectionResult, info);
+            info.setSequenceNbr(sequence.nextId() + "");
+            inspectionDetectionInfoMapper.insert(info);
+            record = info.getRecord();
+            bizEmqPublisher.sendInspectionMsgAfterSave(info, "insert");
+        } else {
+            info = list.get(0);
+            fillInspectionFields(jyjcInspectionResult, info);
+            inspectionDetectionInfoMapper.updateById(info);
+            record = info.getRecord();
+            bizEmqPublisher.sendInspectionMsgAfterSave(info, "update");
+        }
+        try {
+            emqKeeper.getMqttClient().publish(String.format(TZSCommonConstant.DATA_REFRESH_TOPIC, DataRefreshEvent.DataType.equipment.name(), DataRefreshEvent.Operation.UPDATE), JSONObject.toJSONString(Collections.singletonList(record)).getBytes(StandardCharsets.UTF_8), 2, false);
+        } catch (MqttException e) {
+            log.error("发送数据变更emq消息失败：{}", e.getMessage(), e);
+        }
+    }
+
+    private void fillInspectionFields(JyjcInspectionResult jyjcInspectionResult, InspectionDetectionInfo info) {
+        info.setInspectOrgName(jyjcInspectionResult.getInspectionTypeName());
+        info.setInspectType(jyjcInspectionResult.getInspectionType());
+        info.setRecord(jyjcInspectionResult.getEquipUnicode());
+        info.setInspectDate(jyjcInspectionResult.getInspectionDate());
+        info.setInspectStaff(getInspectUserName(jyjcInspectionResult));
+        info.setInspectStaffCode(jyjcInspectionResult.getInspector());
+        info.setInspectConclusion(jyjcInspectionResult.getInspectionConclusion());
+        info.setProblemRemark(jyjcInspectionResult.getNonConformance());
+        info.setNextInspectDate(jyjcInspectionResult.getNextInspectionDate());
+        info.setSequenceCode(jyjcInspectionResult.getEquipUnicode());
+        info.setInspectOrgCode(jyjcInspectionResult.getInspectionUnitCode());
+        info.setInspectReportNo(jyjcInspectionResult.getResultNo());
+        info.setRecDate(new Date());
+    }
+
+    private String getInspectUserName(JyjcInspectionResult model) {
+        List<TzsUserInfo> userInfos = commonService.getUserInfosByUnitCode(model.getInspectionUnitCode());
+        return userInfos.stream().filter(u -> model.getInspector().contains(u.getSequenceNbr() + "")).map(TzsUserInfo::getName).collect(Collectors.joining(","));
+    }
+}

amos-boot-system-tzs/amos-boot-module-jyjc/amos-boot-module-jyjc-biz/src/main/java/com/yeejoin/amos/boot/module/jyjc/biz/service/impl/JyjcInspectionResultServiceImpl.java

@@ -19,8 +19,10 @@ import com.yeejoin.amos.boot.biz.common.service.impl.DataDictionaryServiceImpl;
 import com.yeejoin.amos.boot.biz.common.utils.RedisKey;
 import com.yeejoin.amos.boot.biz.common.utils.RedisUtils;
 import com.yeejoin.amos.boot.biz.common.utils.SnowflakeIdUtil;
+import com.yeejoin.amos.boot.module.common.api.constant.TZSCommonConstant;
 import com.yeejoin.amos.boot.module.common.api.dao.ESEquipmentCategory;
 import com.yeejoin.amos.boot.module.common.api.dto.ESEquipmentCategoryDto;
+import com.yeejoin.amos.boot.module.common.biz.refresh.DataRefreshEvent;
 import com.yeejoin.amos.boot.module.jg.api.dto.DynamicColumnDto;
 import com.yeejoin.amos.boot.module.jg.api.mapper.CommonMapper;
 import com.yeejoin.amos.boot.module.jg.api.vo.SortVo;
@@ -48,6 +50,7 @@ import com.yeejoin.amos.boot.module.ymt.api.mapper.*;
 import com.yeejoin.amos.feign.systemctl.model.DictionarieModel;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
+import org.eclipse.paho.client.mqttv3.MqttException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.core.io.Resource;
@@ -56,6 +59,7 @@ import org.springframework.stereotype.Service;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.CollectionUtils;
 import org.springframework.util.ObjectUtils;
+import org.typroject.tyboot.component.emq.EmqKeeper;
 import org.typroject.tyboot.core.foundation.context.RequestContext;
 import org.typroject.tyboot.core.foundation.utils.ValidationUtil;
 import org.typroject.tyboot.core.rdbms.orm.entity.BaseEntity;
@@ -63,6 +67,7 @@ import org.typroject.tyboot.core.rdbms.service.BaseService;
 import org.typroject.tyboot.core.restful.exception.instance.BadRequest;
 
 import javax.servlet.http.HttpServletResponse;
+import java.nio.charset.StandardCharsets;
 import java.text.SimpleDateFormat;
 import java.util.*;
 import java.util.function.Function;
@@ -153,6 +158,9 @@ public class JyjcInspectionResultServiceImpl extends BaseService<JyjcInspectionR
     @Autowired
     private IdxBizJgInspectionDetectionInfoMapper idxBizJgInspectionDetectionInfoMapper;
 
+    @Autowired
+    private EmqKeeper emqKeeper;
+
     /**
      * 检验检测单位分页查询
      */
@@ -313,14 +321,24 @@ public class JyjcInspectionResultServiceImpl extends BaseService<JyjcInspectionR
         }
         JyjcInspectionResultModel dbResultModel = this.queryBySeq(model.getSequenceNbr());
         // 更新使用信息、检验信息、技术参数
-        this.updateTechParamAndInspectInfo(dbResultModel, jybgFile);
+        Set<String> records = this.updateTechParamAndInspectInfo(dbResultModel, jybgFile);
         // 更新单据信息的json（公共）
         this.updateHisDataAfterResultUpdate(dbResultModel);
+        sendDataRefreshMsg(records);
         return dbResultModel;
     }
 
+    private void sendDataRefreshMsg(Set<String> records) {
+        try {
+            emqKeeper.getMqttClient().publish(String.format(TZSCommonConstant.DATA_REFRESH_TOPIC, DataRefreshEvent.DataType.equipment.name(), DataRefreshEvent.Operation.UPDATE), JSONObject.toJSONString(new ArrayList<>(records)).getBytes(StandardCharsets.UTF_8), 2, false);
+        } catch (MqttException e) {
+            log.error("发送数据变更emq消息失败：{}", e.getMessage(), e);
+        }
+    }
+
 
-    private void updateTechParamAndInspectInfo(JyjcInspectionResultModel model, JyjcInspectionResultAttachment jybgFile){
+    private Set<String> updateTechParamAndInspectInfo(JyjcInspectionResultModel model, JyjcInspectionResultAttachment jybgFile){
+        Set<String> records = new HashSet<>();
         if(model.getEquList().equals(EquipmentClassifityEnum.YLGD.getCode())){
             // 管道逻辑： 循环更新技术参数、检验信息更新或者插入、使用信息更新(数据库及es)
             JyjcInspectionResultParam param = iJyjcInspectionResultParamService.getOneParamByResultSeq(model.getSequenceNbr());
@@ -334,6 +352,7 @@ public class JyjcInspectionResultServiceImpl extends BaseService<JyjcInspectionR
                     );
                     equips.forEach(e->{
                         String record = String.valueOf(e.get("record"));
+                        records.add(record);
                         // 1.更新管道的技术参数
                         EquipTechParamPipeline techParamPipeline = new EquipTechParamPipeline();
                         BeanUtil.copyProperties(e, techParamPipeline,true);
@@ -421,7 +440,9 @@ public class JyjcInspectionResultServiceImpl extends BaseService<JyjcInspectionR
             useInfoMapper.updateByRecord(model.getEquipUnicode(), model.getNextInspectionDate(), model.getInspectionType(), model.getApplicationNo());
             // 4.es更新下次检验日期
             this.updateEquipNextInspectDate(model, model.getEquipUnicode());
+            records.add(model.getEquipUnicode());
         }
+        return records;
     }
 
     private String getTableName(String paramType) {
@@ -736,6 +757,7 @@ public class JyjcInspectionResultServiceImpl extends BaseService<JyjcInspectionR
      */
     @Transactional(rollbackFor = Exception.class)
     public Boolean useUnitEntryWithSave(Map<String, Object> entryData, ReginParams reginParams) {
+        Set<String> records = new HashSet<>();
         boolean isYLGD = "8000".equals(Objects.toString(entryData.get("equList")));
         JSONArray equList = JSON.parseArray(JSON.toJSONString(entryData.get(EQU_LIST)));
         List<String> applicationNos = tzsServiceFeignClient.applicationFormCode(ApplicationFormTypeEnum.JY.getCode(), equList.size()).getResult();
@@ -810,11 +832,17 @@ public class JyjcInspectionResultServiceImpl extends BaseService<JyjcInspectionR
                 inspectionHistory.setRecUserId(RequestContext.getExeUserId());
                 inspectionHistory.setRecDate(new Date());
                 jyjcInspectionHistoryService.save(inspectionHistory);
+                records.add(inspectionDetectionInfo.getRecord());
             } catch (Exception e) {
                 log.warn(e.getMessage());
                 throw new BadRequest("数据异常，请联系管理员！");
             }
         }
+        try {
+            emqKeeper.getMqttClient().publish(String.format(TZSCommonConstant.DATA_REFRESH_TOPIC, DataRefreshEvent.DataType.equipment.name(), DataRefreshEvent.Operation.UPDATE), JSONObject.toJSONString(records).getBytes(StandardCharsets.UTF_8), 2, false);
+        } catch (MqttException e) {
+            log.error("发送数据变更emq消息失败：{}", e.getMessage(), e);
+        }
         return Boolean.TRUE;
     }
 
@@ -937,7 +965,10 @@ public class JyjcInspectionResultServiceImpl extends BaseService<JyjcInspectionR
             inspectionHistory.setRecUserId(RequestContext.getExeUserId());
             jyjcInspectionHistoryService.update(inspectionHistory,
                     new LambdaUpdateWrapper<JyjcInspectionHistory>().eq(JyjcInspectionHistory::getSSeq, Long.valueOf(Objects.toString(sequenceNbr))));
+            // 发送数据变更消息
+            emqKeeper.getMqttClient().publish(String.format(TZSCommonConstant.DATA_REFRESH_TOPIC, DataRefreshEvent.DataType.equipment.name(), DataRefreshEvent.Operation.UPDATE), JSONObject.toJSONString(Collections.singletonList(inspectionDetectionInfo.getRecord())).getBytes(StandardCharsets.UTF_8), 2, false);
         } catch (Exception e) {
+            log.error(e.getMessage(), e);
             throw new BadRequest("数据异常，请联系管理员！");
         }
         return Boolean.TRUE;
@@ -963,12 +994,15 @@ public class JyjcInspectionResultServiceImpl extends BaseService<JyjcInspectionR
     public Boolean delUseUnitEntry(Long resultSeq) {
         // 删除结果表
         this.baseMapper.deleteById(resultSeq);
+        Set<String> records = idxBizJgInspectionDetectionInfoMapper.selectList(new LambdaQueryWrapper<IdxBizJgInspectionDetectionInfo>().eq(IdxBizJgInspectionDetectionInfo::getResultSeq, resultSeq).select(IdxBizJgInspectionDetectionInfo::getRecord)).stream().map(IdxBizJgInspectionDetectionInfo::getRecord).collect(Collectors.toSet());
         // 删除检验信息表
         idxBizJgInspectionDetectionInfoMapper.delete(new LambdaQueryWrapper<IdxBizJgInspectionDetectionInfo>().eq(IdxBizJgInspectionDetectionInfo::getResultSeq, resultSeq.toString()));
         // 删除附件表
         attachmentService.getBaseMapper().delete(new LambdaQueryWrapper<JyjcInspectionResultAttachment>().eq(JyjcInspectionResultAttachment::getResultSeq,resultSeq.toString()));
         // 删除历史表
         jyjcInspectionHistoryService.getBaseMapper().delete(new LambdaQueryWrapper<JyjcInspectionHistory>().eq(JyjcInspectionHistory::getSSeq, resultSeq));
+        // 发送数据变更消息
+        sendDataRefreshMsg(records);
         return Boolean.TRUE;
     }
 }
\ No newline at end of file