fix(jyjc):bug处理

amos-boot-system-tzs/amos-boot-module-jyjc/amos-boot-module-jyjc-api/src/main/java/com/yeejoin/amos/boot/module/jyjc/api/common/BizCommonConstant.java

@@ -40,6 +40,11 @@ public interface BizCommonConstant {
     String PUSH_STATUS_SUCCESS = "2";
 
     /**
+     * 推送状态失败
+     */
+    String PUSH_STATUS_FAILED = "99";
+
+    /**
      * 待分配业务负责人
      */
     String JS_DFP = "66190";

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/InspectionApplicationPushEventListener.java

@@ -28,15 +28,13 @@ import com.yeejoin.amos.boot.module.jyjc.biz.service.impl.CommonServiceImpl;
 import com.yeejoin.amos.boot.module.jyjc.biz.service.impl.JyjcInspectionApplicationEquipServiceImpl;
 import com.yeejoin.amos.boot.module.jyjc.biz.service.impl.JyjcInspectionApplicationPushLogServiceImpl;
 import com.yeejoin.amos.boot.module.jyjc.biz.service.impl.JyjcInspectionApplicationServiceImpl;
-import com.yeejoin.amos.boot.module.jyjc.biz.util.JsonUtils;
 import com.yeejoin.amos.boot.module.jyjc.biz.util.JyjcConstant;
 import com.yeejoin.amos.boot.module.ymt.api.entity.*;
 import com.yeejoin.amos.boot.module.ymt.api.mapper.*;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Value;
-import org.springframework.core.io.Resource;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.event.TransactionalEventListener;
 
@@ -55,90 +53,85 @@ import java.util.stream.Collectors;
  */
 @Component
 @Slf4j
+@RequiredArgsConstructor
 public class InspectionApplicationPushEventListener {
 
-    @Value("classpath:/json/companyCodeRegName.json")
-    private Resource regNameJson;
-
-    private Map<String, String> companyCodeRegNameMap;
-
     private final BlockingQueue<JyjcInspectionApplicationModel> queue = new LinkedBlockingQueue<>();
 
     @Value("${inspection.push.max.deal.thread.num:2}")
     private int threadNum;
 
-    @Autowired
-    JyjcInspectionApplicationPushLogServiceImpl pushLogService;
 
-    @Autowired
-    JyjcInspectionApplicationEquipServiceImpl applicationEquipService;
+    private final JyjcInspectionApplicationPushLogServiceImpl pushLogService;
+
+
+    private final JyjcInspectionApplicationEquipServiceImpl applicationEquipService;
+
+
+    private final IdxBizJgConstructionInfoMapper constructionInfoMapper;
+
+
+    private final IdxBizJgUseInfoMapper useInfoMapper;
+
+
+    private final IdxBizJgRegisterInfoMapper idxBizJgRegisterInfoMapper;
+
+
+    private final TzBaseEnterpriseInfoMapper baseEnterpriseInfoMapper;
+
+
+    private final EquipTechParamBoilerMapper equipTechParamBoilerMapper;
+
+
+    private final EquipTechParamVesselMapper equipTechParamVesselMapper;
 
-    @Autowired
-    IdxBizJgConstructionInfoMapper constructionInfoMapper;
 
-    @Autowired
-    IdxBizJgUseInfoMapper useInfoMapper;
+    private final EquipTechParamElevatorMapper equipTechParamElevatorMapper;
 
-    @Autowired
-    IdxBizJgRegisterInfoMapper idxBizJgRegisterInfoMapper;
 
-    @Autowired
-    TzBaseEnterpriseInfoMapper baseEnterpriseInfoMapper;
+    private final EquipTechParamLiftingMapper equipTechParamLiftingMapper;
 
-    @Autowired
-    private EquipTechParamBoilerMapper equipTechParamBoilerMapper;
 
-    @Autowired
-    private EquipTechParamVesselMapper equipTechParamVesselMapper;
+    private final EquipTechParamVehicleMapper equipTechParamVehicleMapper;
 
-    @Autowired
-    private EquipTechParamElevatorMapper equipTechParamElevatorMapper;
 
-    @Autowired
-    private EquipTechParamLiftingMapper equipTechParamLiftingMapper;
+    private final EquipTechParamRidesMapper equipTechParamRidesMapper;
 
-    @Autowired
-    private EquipTechParamVehicleMapper equipTechParamVehicleMapper;
 
-    @Autowired
-    private EquipTechParamRidesMapper equipTechParamRidesMapper;
+    private final EquipTechParamPipelineMapper equipTechParamPipelineMapper;
 
-    @Autowired
-    private EquipTechParamPipelineMapper equipTechParamPipelineMapper;
 
-    @Autowired
-    private EquipTechParamRopewayMapper equipTechParamRopewayMapper;
+    private final EquipTechParamRopewayMapper equipTechParamRopewayMapper;
 
-    @Autowired
-    KafkaProducer kafkaProducer;
 
-    @Autowired
-    SnowflakeIdUtil sequence;
+    private final KafkaProducer kafkaProducer;
 
-    @Autowired
-    IdxBizJgFactoryInfoMapper factoryInfoMapper;
 
-    @Autowired
-    IdxBizJgMaintenanceRecordInfoMapper maintenanceInfoMapper;
+    private final SnowflakeIdUtil sequence;
 
-    @Autowired
-    JgUseRegistrationManageMapper jgUseRegistrationManageMapper;
 
-    @Autowired
-    IdxBizJgDesignInfoMapper designInfoMapper;
+    private final IdxBizJgFactoryInfoMapper factoryInfoMapper;
 
-    @javax.annotation.Resource
-    IdxBizJgProjectContraptionMapper projectContraptionMapper;
 
+    private final IdxBizJgMaintenanceRecordInfoMapper maintenanceInfoMapper;
 
-    @javax.annotation.Resource
-    private JyjcInspectionApplicationServiceImpl inspectionApplicationService;
 
-    @javax.annotation.Resource
-    private TzBaseEnterpriseInfoMapper enterpriseInfoMapper;
+    private final JgUseRegistrationManageMapper jgUseRegistrationManageMapper;
 
-    @javax.annotation.Resource
-    private CommonServiceImpl commonService;
+
+    private final IdxBizJgDesignInfoMapper designInfoMapper;
+
+
+    private final IdxBizJgProjectContraptionMapper projectContraptionMapper;
+
+
+    private final JyjcInspectionApplicationServiceImpl inspectionApplicationService;
+
+
+    private final TzBaseEnterpriseInfoMapper enterpriseInfoMapper;
+
+
+    private final CommonServiceImpl commonService;
 
 
     /**
@@ -151,15 +144,10 @@ public class InspectionApplicationPushEventListener {
      */
     private final static List<String> JDJY_ARRAY = Arrays.asList("AZJDJY", "GZJDJY", "WXJDJY");
 
-    /**
-     * 定首检-检验类型
-     */
-    private final static List<String> DSJ_ARRAY = Arrays.asList("DQJY", "SCJY");
 
     private static final BigDecimal ZERO = new BigDecimal("0.000");
-    @Autowired
-    private IdxBizJgSupervisionInfoMapper idxBizJgSupervisionInfoMapper;
 
+    private final IdxBizJgSupervisionInfoMapper idxBizJgSupervisionInfoMapper;
 
     @TransactionalEventListener(value = InspectionApplicationPushEvent.class)
     public void onApplicationEvent(InspectionApplicationPushEvent event) {
@@ -170,7 +158,6 @@ public class InspectionApplicationPushEventListener {
     @PostConstruct
     public void init() {
         ExecutorService executorService = Executors.newFixedThreadPool(threadNum);
-        companyCodeRegNameMap = JsonUtils.getResourceJson(regNameJson);
         for (int i = 0; i < threadNum; i++) {
             executorService.execute(() -> {
                 while (true) {
@@ -188,12 +175,15 @@ public class InspectionApplicationPushEventListener {
         }
     }
 
+    /**
+     * 发送kafka指定的主题
+     *
+     * @param dockingUnitCode 对接单位appId
+     * @param pushLog         本地日志表
+     */
     private void pushData2Kafka(String dockingUnitCode, JyjcInspectionApplicationPushLog pushLog) {
-        // 发送kafka指定的主题
         String topic = String.format(INSPECTION_APPLICATION_PUSH_TOPIC, dockingUnitCode);
-        kafkaProducer.sendMessage(topic, pushLog.getPushData());
-        pushLog.setPushStatus(BizCommonConstant.PUSH_STATUS_SUCCESS);
-        pushLogService.updateById(pushLog);
+        kafkaProducer.sendMessageWithLog(topic, pushLog.getPushData(), pushLog);
     }
 
     private JyjcInspectionApplicationPushLog createPushData(JyjcInspectionApplicationModel applicationModel) {
@@ -210,12 +200,6 @@ public class InspectionApplicationPushEventListener {
         return pushLog;
     }
 
-    private List<String> getCanDealInspectionType() {
-        List<String> canDealInspectionTypes = new ArrayList<>(JDJY_ARRAY);
-        canDealInspectionTypes.addAll(DSJ_ARRAY);
-        return canDealInspectionTypes;
-    }
-
     private String calTotalLength(List<PipelinePushItemDto> pipelines) {
         if (pipelines == null || pipelines.isEmpty()) {
             return ZERO.toPlainString();
@@ -365,7 +349,7 @@ public class InspectionApplicationPushEventListener {
     private void setMaintenanceInfo(InspectionEquipData equipData, String equipUnicode) {
         QueryWrapper<IdxBizJgMaintenanceRecordInfo> queryWrapper = new QueryWrapper<>();
         queryWrapper.lambda().eq(IdxBizJgMaintenanceRecordInfo::getRecord, equipUnicode)
-                .select(IdxBizJgMaintenanceRecordInfo::getRecord,IdxBizJgMaintenanceRecordInfo::getMeUnitCreditCode, IdxBizJgMaintenanceRecordInfo::getMeUnitName)
+                .select(IdxBizJgMaintenanceRecordInfo::getRecord, IdxBizJgMaintenanceRecordInfo::getMeUnitCreditCode, IdxBizJgMaintenanceRecordInfo::getMeUnitName)
                 .orderByDesc(IdxBizJgMaintenanceRecordInfo::getRecDate).last("limit 1");
         IdxBizJgMaintenanceRecordInfo maintenanceInfo = maintenanceInfoMapper.selectOne(queryWrapper);
         if (maintenanceInfo != null) {
@@ -417,7 +401,7 @@ public class InspectionApplicationPushEventListener {
                     .last("limit 1");
             JgUseRegistrationManage manage = jgUseRegistrationManageMapper.selectOne(queryWrapper);
             if (manage != null) {
-                equipData.setRegUnitName(companyCodeRegNameMap.getOrDefault(manage.getReceiveCompanyCode(), manage.getReceiveOrgName()));
+                equipData.setRegUnitName(manage.getReceiveOrgName());
             }
         }
     }

amos-boot-system-tzs/amos-boot-module-jyjc/amos-boot-module-jyjc-biz/src/main/java/com/yeejoin/amos/boot/module/jyjc/biz/kafka/KafkaProducer.java

 package com.yeejoin.amos.boot.module.jyjc.biz.kafka;
 
+import com.yeejoin.amos.boot.module.jyjc.api.common.BizCommonConstant;
+import com.yeejoin.amos.boot.module.jyjc.api.entity.JyjcInspectionApplicationPushLog;
+import com.yeejoin.amos.boot.module.jyjc.biz.service.impl.JyjcInspectionApplicationPushLogServiceImpl;
+import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.kafka.support.SendResult;
 import org.springframework.stereotype.Component;
 import org.springframework.util.concurrent.ListenableFuture;
-import org.springframework.util.concurrent.ListenableFutureCallback;
 
 /**
  * @author Administrator
  */
 @Component
 @Slf4j
+@RequiredArgsConstructor
 public class KafkaProducer {
 
     private final KafkaTemplate<String, String> kafkaTemplate;
 
-    public KafkaProducer(KafkaTemplate<String, String> kafkaTemplate) {
-        this.kafkaTemplate = kafkaTemplate;
-    }
+    private final JyjcInspectionApplicationPushLogServiceImpl pushLogService;
+
 
     /**
      * 发送消息（异步）
@@ -26,23 +29,40 @@ public class KafkaProducer {
      * @param topic   主题
      * @param message 消息内容
      */
-    public void sendMessage(String topic, String message) {
+    public void sendMessageWithLog(String topic, String message, JyjcInspectionApplicationPushLog pushLog) {
         if (log.isInfoEnabled()) {
             log.info("kafka开始发送数据：主题:{}，消息{}", topic, message);
         }
-        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
-        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
-            @Override
-            public void onFailure(Throwable throwable) {
-                log.error("发送消息（异步） failure! topic : {}, message: {}", topic, message);
-            }
-
-            @Override
-            public void onSuccess(SendResult<String, String> stringStringSendResult) {
+        try {
+            ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
+            future.addCallback(result -> {
                 if (log.isInfoEnabled()) {
-                    log.info("发送消息（异步） success!");
+                    if (result != null) {
+                        log.info("Kafka发送成功! 主题: {}, 分区: {}, Offset: {}", topic, result.getRecordMetadata().partition(), result.getRecordMetadata().offset()
+                        );
+                    }
                 }
-            }
-        });
+                updatePushLogStatus(pushLog, BizCommonConstant.PUSH_STATUS_SUCCESS, null);
+            }, ex -> {
+                log.error("Kafka发送失败! 主题: {}, 消息: {}", topic, message, ex);
+                updatePushLogStatus(pushLog, BizCommonConstant.PUSH_STATUS_FAILED, ex.getMessage());
+                throw new RuntimeException("Kafka消息发送失败", ex);  // 触发事务回滚（需配合事务配置）
+            });
+        } catch (Exception e) {
+            // 同步阶段即失败（如Topic不存在、Broker宕机）
+            log.error("Kafka同步发送失败! 主题: {}, 消息丢弃", topic, e);
+            updatePushLogStatus(pushLog, BizCommonConstant.PUSH_STATUS_FAILED, e.getMessage());
+        }
+    }
+
+    private void updatePushLogStatus(JyjcInspectionApplicationPushLog pushLog, String status, String errorMsg) {
+        try {
+            pushLog.setPushStatus(status);
+            pushLog.setRemark(errorMsg);
+            pushLogService.updateById(pushLog);
+        } catch (Exception e) {
+            log.error("更新推送日志状态失败! ID: {}", pushLog.getSequenceNbr(), e);
+            throw e;
+        }
     }
 }
\ No newline at end of file