1.openapi发出消息，气瓶服务订阅消息进行业务处理

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

 package com.yeejoin.amos.api.openapi.face.service;
 
+import com.alibaba.fastjson.JSON;
+import com.baomidou.dynamic.datasource.annotation.DS;
 import com.google.common.collect.Lists;
 import com.yeejoin.amos.api.openapi.face.model.*;
 import com.yeejoin.amos.api.openapi.face.orm.dao.ESCylinderFillingRecordRepository;
@@ -7,32 +9,20 @@ import com.yeejoin.amos.api.openapi.face.orm.dao.ESCylinderInfoRepository;
 import com.yeejoin.amos.boot.biz.common.utils.DateUtils;
 import com.yeejoin.amos.boot.module.cylinder.api.entity.ESCylinderFillingRecordDto;
 import com.yeejoin.amos.boot.module.cylinder.api.entity.ESCylinderInfoDto;
-import com.yeejoin.amos.boot.module.cylinder.flc.api.dto.*;
+import com.yeejoin.amos.boot.module.cylinder.flc.api.dto.CylinderInfoDto;
+import com.yeejoin.amos.boot.module.cylinder.flc.api.entity.*;
 import com.yeejoin.amos.boot.module.cylinder.flc.api.mapper.*;
-
 import lombok.extern.slf4j.Slf4j;
+import org.eclipse.paho.client.mqttv3.MqttException;
 import org.springframework.beans.BeanUtils;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Component;
 import org.springframework.util.ObjectUtils;
 import org.springframework.util.StopWatch;
+import org.typroject.tyboot.component.emq.EmqKeeper;
 import org.typroject.tyboot.core.foundation.utils.Bean;
 
-import com.baomidou.dynamic.datasource.annotation.DS;
-import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
-import com.baomidou.mybatisplus.extension.service.IService;
-import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
-import com.yeejoin.amos.boot.module.cylinder.flc.api.dto.CylinderInfoDto;
-import com.yeejoin.amos.boot.module.cylinder.flc.api.entity.CylinderFilling;
-import com.yeejoin.amos.boot.module.cylinder.flc.api.entity.CylinderFillingCheck;
-import com.yeejoin.amos.boot.module.cylinder.flc.api.entity.CylinderFillingExamine;
-import com.yeejoin.amos.boot.module.cylinder.flc.api.entity.CylinderFillingRecord;
-import com.yeejoin.amos.boot.module.cylinder.flc.api.entity.CylinderInfo;
-import com.yeejoin.amos.boot.module.cylinder.flc.api.entity.CylinderInspection;
-import com.yeejoin.amos.boot.module.cylinder.flc.api.entity.CylinderTags;
-import com.yeejoin.amos.boot.module.cylinder.flc.api.entity.CylinderUnit;
-import com.yeejoin.amos.boot.module.cylinder.flc.api.entity.CylinderUnitVideo;
-
 import java.text.ParseException;
 import java.util.ArrayList;
 import java.util.List;
@@ -41,8 +31,7 @@ import java.util.stream.Collectors;
 @Slf4j
 @Component
 @DS("tzs")
-public class SyncCylinderDataService
-{
+public class SyncCylinderDataService {
     /**
      * 气瓶企业信息
      */
@@ -54,11 +43,6 @@ public class SyncCylinderDataService
     @Autowired
     private CylinderInfoMapper cylinderInfoMapper;
     /**
-   * 气瓶充装信息--充装前检查
-   */
-  @Autowired
-  private CylinderFillingMapper  cylinderFillingMapper;
-  /**
      * 液化气体气瓶充装信息-充装后复查
      */
     @Autowired
@@ -90,16 +74,19 @@ public class SyncCylinderDataService
     @Autowired
     ESCylinderInfoRepository esCylinderInfoRepository;
 
+    @Autowired
+    EmqKeeper emqKeeper;
+
+    @Value("${cylinder.filling.insert.topic:cylinder/filling/insert/topic}")
+    private String insertTopic;
+
     public void syncCylinderUnit(List<TmCylinderUnitModel> cylinderUnitDto) {
-//    List<CylinderUnit> cylinderUnits = Bean.toModels(cylinderUnitDto,CylinderUnit.class);
-//    LambdaQueryWrapper<CylinderUnit> wrapper = new LambdaQueryWrapper<>();
-//    List<CylinderUnit> cylinderUnitList = this.getBaseMapper().selectList(wrapper);
-    List<CylinderUnit> cylinderUnitList = Bean.toModels(cylinderUnitDto,CylinderUnit.class);
+        List<CylinderUnit> cylinderUnitList = Bean.toModels(cylinderUnitDto, CylinderUnit.class);
         cylinderUnitMapper.saveOrUpdateBatch(cylinderUnitList);
     }
 
     public void syncCylinderInfo(final List<TmCylinderInfoModel> cylinderInfoDto) {
-    List<CylinderInfo> cylinderUnitList = Bean.toModels(cylinderInfoDto,CylinderInfo.class);
+        List<CylinderInfo> cylinderUnitList = Bean.toModels(cylinderInfoDto, CylinderInfo.class);
         cylinderInfoMapper.saveOrUpdateBatch(cylinderUnitList);
     }
 
@@ -176,8 +163,23 @@ public class SyncCylinderDataService
         }
         esCylinderFillingRecordRepository.saveAll(records);
         cylinderFillingRecordMapper.updateCylinderFillingToEsStatus(ids);
+        this.publishMsg2CyService(records);
+
     }
 
+    /**
+     * 发布记录创建消息
+     *
+     * @param records 所有的记录
+     */
+    private void publishMsg2CyService(List<ESCylinderFillingRecordDto> records) {
+        try {
+            log.info("开始发送气瓶充装记录数据消息:{}", JSON.toJSONString(records));
+            emqKeeper.getMqttClient().publish(insertTopic, JSON.toJSONString(records).getBytes(), 2, false);
+        } catch (MqttException e) {
+            log.error("发送气瓶充装记录数据消息失败:{}", e.getMessage());
+        }
+    }
 
 
     public void createCylinderInfo2ES(TmCylinderInfoModel cylinderInfoModel) {

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

@@ -2,6 +2,7 @@ package com.yeejoin.amos.boot.module.cylinder.biz.event.listener;
 
 
 import cn.hutool.core.bean.BeanUtil;
+import com.alibaba.fastjson.JSON;
 import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
 import com.baomidou.mybatisplus.core.toolkit.Sequence;
 import com.yeejoin.amos.boot.module.cylinder.api.dto.CylinderQuestionInfoDto;
@@ -63,6 +64,7 @@ public class CylinderQuestionCreateEventListener implements ApplicationListener<
 
     @Override
     public void onApplicationEvent(CylinderQuestionCreateEvent event) {
+        log.info("2.收到问题创建消息：{}", JSON.toJSONString(event.getCylinderQuestionInfoDto()));
         // 按照一定的规则将不同的问题对象放到不同的队列里，保证多线程不会同时操作不同的问题对象数据，减少了线程的冲突，提高执行效率
         int queueIndex = Math.abs(event.getCylinderQuestionInfoDto().getQuestionObjectId().hashCode()) % threadNumber;
         hashCodeBlockingQueues.get(queueIndex).add(event.getCylinderQuestionInfoDto());
@@ -117,6 +119,7 @@ public class CylinderQuestionCreateEventListener implements ApplicationListener<
             return cylinderQuestionInfo;
         }).collect(Collectors.toList());
         cylinderQuestionInfoService.saveBatch(cylinderQuestionInfos);
+        log.info("3.问题入库创建成功：{}", JSON.toJSONString(cylinderQuestionInfos));
     }
 
     private String getQuestionObjectName(String sequenceCode, String appId) {

amos-boot-system-tzs/amos-boot-module-cylinder/amos-boot-module-cylinder-biz/src/main/java/com/yeejoin/amos/boot/module/cylinder/biz/listener/CylinderFillingInsertEventListener.java

 package com.yeejoin.amos.boot.module.cylinder.biz.listener;
 
-import com.alibaba.fastjson.JSONObject;
+import com.alibaba.fastjson.JSONArray;
 import com.yeejoin.amos.boot.module.cylinder.api.dto.CylinderQuestionInfoDto;
 import com.yeejoin.amos.boot.module.cylinder.api.enums.QuestionTypeEnum;
 import com.yeejoin.amos.boot.module.cylinder.biz.event.CylinderQuestionCreateEvent;
@@ -16,6 +16,7 @@ import org.typroject.tyboot.component.emq.EmqxListener;
 
 import javax.annotation.PostConstruct;
 import java.nio.charset.StandardCharsets;
+import java.util.List;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -95,9 +96,9 @@ public class CylinderFillingInsertEventListener extends EmqxListener {
 
     @Override
     public void processMessage(String topic, MqttMessage message) {
-        log.info("收到充装信息插入消息:{}", JSONObject.toJSONString(message));
         String msg = new String(message.getPayload(), StandardCharsets.UTF_8);
-        CylinderFillingRecordDto cylinderFillingRecordDto = JSONObject.parseObject(msg, CylinderFillingRecordDto.class);
-        blockingQueue.add(cylinderFillingRecordDto);
+        log.info("1.收到充装信息插入消息:{}", msg);
+        List<CylinderFillingRecordDto> cylinderFillingRecordDtos = JSONArray.parseArray(msg, CylinderFillingRecordDto.class);
+        blockingQueue.addAll(cylinderFillingRecordDtos);
     }
 }