fix:气瓶充装对接日志调整

amos-boot-data/amos-boot-data-openapi/src/main/java/com/yeejoin/amos/api/openapi/face/service/CylinderService.java

@@ -17,8 +17,6 @@ import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Component;
-import org.springframework.transaction.PlatformTransactionManager;
-import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.ObjectUtils;
 import org.typroject.tyboot.component.cache.Redis;
 import org.typroject.tyboot.component.emq.EmqKeeper;
@@ -26,10 +24,15 @@ import org.typroject.tyboot.core.foundation.context.RequestContext;
 import org.typroject.tyboot.core.foundation.utils.ValidationUtil;
 import org.typroject.tyboot.core.restful.exception.instance.BadRequest;
 
+import javax.annotation.PostConstruct;
 import java.text.SimpleDateFormat;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  * 气瓶service
@@ -48,11 +51,11 @@ public class CylinderService {
     private final TmCylinderFillingRecordService cylinderFillingRecordService;
     private final TmCylinderFillingService cylinderFillingService;
 
-    private final PlatformTransactionManager transactionManager;
+    private final SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
 
-    private SimpleDateFormat simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+    private final BlockingQueue<CylinderFillingMessageModel> blockingQueue = new LinkedBlockingQueue<>();
 
-    public CylinderService(RedisTemplate redisTemplate, CylinderFillingDataValidationService cylinderFillingDataValidationService, TmCylinderFillingMessageService cylinderFillingMessageService, EmqKeeper emqKeeper, TmCylinderFillingCheckService cylinderFillingCheckService, TmCylinderFillingRecordService cylinderFillingRecordService, TmCylinderFillingService cylinderFillingService, PlatformTransactionManager transactionManager) {
+    public CylinderService(RedisTemplate redisTemplate, CylinderFillingDataValidationService cylinderFillingDataValidationService, TmCylinderFillingMessageService cylinderFillingMessageService, EmqKeeper emqKeeper, TmCylinderFillingCheckService cylinderFillingCheckService, TmCylinderFillingRecordService cylinderFillingRecordService, TmCylinderFillingService cylinderFillingService) {
         this.redisTemplate = redisTemplate;
         this.cylinderFillingDataValidationService = cylinderFillingDataValidationService;
         this.cylinderFillingMessageService = cylinderFillingMessageService;
@@ -60,7 +63,6 @@ public class CylinderService {
         this.cylinderFillingCheckService = cylinderFillingCheckService;
         this.cylinderFillingRecordService = cylinderFillingRecordService;
         this.cylinderFillingService = cylinderFillingService;
-        this.transactionManager = transactionManager;
     }
 
     @DSTransactional
@@ -101,8 +103,8 @@ public class CylinderService {
         emqKeeper.getMqttClient().publish("openapi/cylinder/filling", JSONObject.fromObject(cylinderFillingMessage).toString().getBytes(), 2, false);
     }
 
-    @Transactional
-    public JSONArray validateFillingData(JSONObject jsonobject, CylinderFillingMessage cylinderFillingMessage) throws MqttException {
+//    @Transactional
+    public JSONArray validateFillingData(JSONObject jsonobject, CylinderFillingMessage cylinderFillingLogMessage) throws MqttException {
         CylinderFillingDataValidationResultModel validateResult = null;
 
         try {
@@ -110,13 +112,13 @@ public class CylinderService {
         } catch (Exception e) {
             e.printStackTrace();
             logger.error("液化气体气瓶充装信息上报，数据校验失败");
-            cylinderFillingMessage.setMessage(e.getMessage());
+            cylinderFillingLogMessage.setMessage(e.getMessage());
         }
 
         JSONArray errorData = new JSONArray();
 
         if (!ObjectUtils.isEmpty(validateResult)) {
-            cylinderFillingMessage.setCylinderNumber(validateResult.getCylinderNumber());
+            cylinderFillingLogMessage.setCylinderNumber(validateResult.getCylinderNumber());
             List<String> message = new ArrayList<>();
             int errorNumber = 0;
             JSONObject error = new JSONObject();
@@ -144,23 +146,46 @@ public class CylinderService {
             if (!ObjectUtils.isEmpty(validateResult.getSeqCodeErrorData())) {
                 errorNumber += validateResult.getSeqCodeErrorData().size();
                 error.put("气瓶信息不存在：", validateResult.getSeqCodeErrorData());
-                message.add("气瓶信息不存在数量：" + validateResult.getSeqCodeErrorData().size());
+                message.add("气瓶信息不存在数量：" + validateResult.getSeqCodeErrorData().size() + "个");
                 errorData.add(error);
             }
 
             if (errorNumber <= 0) {
-                message.add("液化气体气瓶充装信息成功数：" + validateResult.getSuccessCylinderNumber() + "条");
+                cylinderFillingLogMessage.setMessage("液化气体气瓶充装信息成功数：" + validateResult.getSuccessCylinderNumber() + "条");
             } else {
-                cylinderFillingMessage.setMessage(String.join("条; ", message) + "个");
-                CylinderFillingMessageModel cylinderFillingMessageModel = new CylinderFillingMessageModel();
-                BeanUtils.copyProperties(cylinderFillingMessage, cylinderFillingMessageModel);
-                cylinderFillingMessageService.createWithModel(cylinderFillingMessageModel);
-                emqKeeper.getMqttClient().publish("openapi/cylinder/filling", JSONObject.fromObject(cylinderFillingMessage).toString().getBytes(), 2, false);
-                throw new BadRequest(error.toString());
+                cylinderFillingLogMessage.setMessage(String.join("条; ", message));
+//                cylinderFillingMessageService.createWithModel(cylinderFillingMessageModel);
+                emqKeeper.getMqttClient().publish("openapi/cylinder/filling", JSONObject.fromObject(cylinderFillingLogMessage).toString().getBytes(), 2, false);
             }
+            CylinderFillingMessageModel cylinderFillingMessageModel = new CylinderFillingMessageModel();
+            BeanUtils.copyProperties(cylinderFillingLogMessage, cylinderFillingMessageModel);
+
+            // 充装信息概览放入blockingQueue
+            blockingQueue.add(cylinderFillingMessageModel);
         } else {
             errorData.add(new JSONObject().put("数据校验失败", "数据校验失败"));
         }
         return errorData;
     }
+
+    @PostConstruct
+    void init() {
+        ExecutorService executorService = Executors.newFixedThreadPool(3);
+        for (int i = 0; i < 3; i++) {
+            executorService.execute(() -> {
+                try {
+                    while (true) {
+                        logger.info("处理气瓶对接错误信息入库~开始");
+                        CylinderFillingMessageModel cylinderFillingMessageModel = blockingQueue.take();
+                        cylinderFillingMessageService.createWithModel(cylinderFillingMessageModel);
+                        logger.info("处理气瓶对接错误信息入库~完成");
+                    }
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                    logger.error("处理气瓶对接错误信息入库异常", e);
+                    throw new RuntimeException(e);
+                }
+            });
+        }
+    }
 }