refactor(安全追溯): 优化消息发送和处理逻辑

amos-boot-system-tzs/amos-boot-module-jg/amos-boot-module-jg-biz/src/main/java/com/yeejoin/amos/boot/module/jg/biz/listener/SafetyProblemTopicMessage.java

@@ -2,6 +2,7 @@ package com.yeejoin.amos.boot.module.jg.biz.listener;
 
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONException;
 import com.alibaba.fastjson.JSONObject;
 import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
 import com.yeejoin.amos.boot.biz.common.utils.RedisUtils;
@@ -74,24 +75,27 @@ public class SafetyProblemTopicMessage extends EmqxListener {
 
     @PostConstruct
     void init() throws Exception {
-        log.info("订阅安全追溯问题主题:{}", buildTopic(problemEventTopic));
         emqKeeper.subscript(buildTopic(problemEventTopic), 2, this);
         executorService = Executors.newFixedThreadPool(threadNumber);
         for (int i = 0; i < threadNumber; i++) {
             executorService.execute(() -> {
                 try {
-                    while (true) {
+                    while (!Thread.currentThread().isInterrupted()) {
                         SafetyProblemEvent safetyProblemMessageEvent = blockingQueue.take();
-                        JSONArray jsonObject = JSON.parseArray(safetyProblemMessageEvent.getMessage().toString());
-                        log.info("接收到问题生产消息:{}", jsonObject);
-                        SafetyProblemEventHandler eventHandler = SafetyProblemEventHandlerFactory.createProblemHandler(safetyProblemMessageEvent.getTopic());
-                        eventHandler.handle(safetyProblemMessageEvent);
-                        log.info("处理问题生产消息完成");
+                        try {
+                            log.info("接收到问题生产原始消息:{}", safetyProblemMessageEvent.getMessage());
+                            JSONArray jsonObject = JSON.parseArray(safetyProblemMessageEvent.getMessage().toString());
+                            log.info("接收到问题生产消息:{}", jsonObject);
+                            SafetyProblemEventHandler eventHandler = SafetyProblemEventHandlerFactory.createProblemHandler(safetyProblemMessageEvent.getTopic());
+                            eventHandler.handle(safetyProblemMessageEvent);
+                            log.info("处理问题生产消息完成");
+                        } catch (JSONException | ClassCastException | IllegalArgumentException e) {
+                            log.error("处理消息异常", e);
+                        }
                     }
                 } catch (InterruptedException e) {
+                    log.error("处理线程被中断，准备退出", e);
                     Thread.currentThread().interrupt();
-                    log.error("处理问题生产消息异常", e);
-                    throw new RuntimeException(e);
                 }
             });
         }
@@ -149,7 +153,8 @@ public class SafetyProblemTopicMessage extends EmqxListener {
                     safetyProblemTracing -> new QueryWrapper<>()
                             .eq("problem_type_code", safetyProblemTracing.getProblemTypeCode())
                             .eq("source_id", safetyProblemTracing.getSourceId())
-                            .eq("problem_status_code", SafetyProblemStatusEnum.UNHANDLED.getCode()));}
+                            .eq("problem_status_code", SafetyProblemStatusEnum.UNHANDLED.getCode()));
+        }
     }
 
     public static void generateUnitProblem(JSONArray jsonArray, SafetyProblemTypeEnum problemTypeEnum, SafetyProblemTracingServiceImpl safetyProblemTracingService) {
@@ -189,7 +194,8 @@ public class SafetyProblemTopicMessage extends EmqxListener {
                     safetyProblemTracing -> new QueryWrapper<>()
                             .eq("problem_type_code", safetyProblemTracing.getProblemTypeCode())
                             .eq("source_id", safetyProblemTracing.getSourceId())
-                            .eq("problem_status_code", SafetyProblemStatusEnum.UNHANDLED.getCode()));}
+                            .eq("problem_status_code", SafetyProblemStatusEnum.UNHANDLED.getCode()));
+        }
     }
 
     public static void generatePersonnelProblem(JSONArray jsonArray, SafetyProblemTypeEnum problemTypeEnum, SafetyProblemTracingServiceImpl safetyProblemTracingService) {

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

@@ -7,6 +7,9 @@ import org.eclipse.paho.client.mqttv3.MqttException;
 import org.springframework.stereotype.Component;
 import org.typroject.tyboot.component.emq.EmqKeeper;
 
+import java.util.ArrayList;
+import java.util.List;
+
 /**
  * @author Administrator
  */
@@ -14,7 +17,7 @@ import org.typroject.tyboot.component.emq.EmqKeeper;
 @Slf4j
 public class BizEmqPublisher {
 
-    private EmqKeeper emqKeeper;
+    private final EmqKeeper emqKeeper;
 
     public BizEmqPublisher(EmqKeeper emqKeeper) {
         this.emqKeeper = emqKeeper;
@@ -24,10 +27,19 @@ public class BizEmqPublisher {
      * 检验检测结果入库消息
      */
     private static final String INSPECTION_DETECTION_SAVE_TO_DB_TOPIC_PREFIX = "safetyProblemTracing/jy/bj";
+    private static final String INSPECTION_DETECTION_SAVE_TO_DB_TOPIC_TYPE_UPDATE = "update";
 
     public void sendInspectionMsgAfterSave(InspectionDetectionInfo info, String type) {
         try {
-            emqKeeper.getMqttClient().publish(this.buildSave2DbEmqTopic(type), JSON.toJSONBytes(info), 2, false);
+            if (emqKeeper.getMqttClient() == null) {
+                return;
+            }
+            // 安全追溯目前只需要type=update的信息
+            if (INSPECTION_DETECTION_SAVE_TO_DB_TOPIC_TYPE_UPDATE.equals(type)) {
+                List<InspectionDetectionInfo> list = new ArrayList<>();
+                list.add(info);
+                emqKeeper.getMqttClient().publish(this.buildSave2DbEmqTopic(type), JSON.toJSONBytes(list), 2, false);
+            }
         } catch (MqttException e) {
             log.error("发送检验检测信息入库消息失败", e);
         }