消息组件优化

amos-boot-module/amos-boot-module-biz/amos-boot-module-jcs-biz/src/main/java/com/yeejoin/amos/boot/module/jcs/biz/service/impl/SignServiceImpl.java

@@ -43,6 +43,9 @@ public class SignServiceImpl extends BaseService<SignDto,Sign,SignMapper> implem
     @Value("${mqtt.topic.person.sign}")
     private String personSign;
 
+    @Value("${mqtt.topic.person.sign.zxj}")
+    private String personSignZxj;
+
     /**
      * 分页查询
      */
@@ -138,11 +141,28 @@ public class SignServiceImpl extends BaseService<SignDto,Sign,SignMapper> implem
                     SerializerFeature.WriteMapNullValue);
             try {
                 emqKeeper.getMqttClient().publish(personSign, json.getBytes(), RuleConfig.DEFAULT_QOS, false);
+                map.put("personOfDay", ObjectUtils.isEmpty(sign.getPersonOfDay())?"":sign.getPersonOfDay());
+                map.put("type", ObjectUtils.isEmpty(sign.getType())?"":sign.getType());
+                map.put("photos", ObjectUtils.isEmpty(sign.getPhotos())?"":sign.getPhotos());
+                map.put("remarks", ObjectUtils.isEmpty(sign.getRemarks())?"":sign.getRemarks());
+                map.put("date", ObjectUtils.isEmpty(sign.getDate())?"":sign.getDate());
+                map.put("recDate", ObjectUtils.isEmpty(sign.getRecDate())?"":sign.getRecDate());
+                map.put("isDelete", ObjectUtils.isEmpty(sign.getIsDelete())?"":sign.getIsDelete());
+                map.put("source", ObjectUtils.isEmpty(sign.getSource())?"":sign.getSource());
+                String json1=JSONObject.toJSONString(map, SerializerFeature.PrettyFormat,
+                        SerializerFeature.WriteMapNullValue);
                 // 发送emq消息转kafka
                 JSONObject jsonObject = new JSONObject();
                 jsonObject.put("topic", personSign);
                 jsonObject.put("data",json);
                 emqKeeper.getMqttClient().publish("emq.sign.created",jsonObject.toString().getBytes(),1,false);
+
+                // 发送emq消息转kafka 同步业务库打卡表
+                JSONObject jsonObject1 = new JSONObject();
+                jsonObject1.put("topic", personSignZxj);
+                jsonObject1.put("data",json1);
+                emqKeeper.getMqttClient().publish("emq.bussSign.created",jsonObject1.toString().getBytes(),1,false);
+
             } catch (MqttException e) {
                 log.info(String.format("发送eqm打卡消息失败：%s", e.getMessage()));
             }

amos-boot-utils/amos-boot-utils-message/src/main/java/com/yeejoin/amos/message/kafka/KafkaConsumerService.java

@@ -34,7 +34,7 @@ public class KafkaConsumerService {
         String topic = messageObj.getString("topic");
         JSONObject data = messageObj.getJSONObject("data");
         try {
-            emqKeeper.getMqttClient().publish(topic, data.toString().getBytes("UTF-8"), 1,false);
+            emqKeeper.getMqttClient().publish(topic, data.toString().getBytes("UTF-8"), 0,false);
         } catch (MqttException e) {
             e.printStackTrace();
         } catch (UnsupportedEncodingException e) {

amos-boot-utils/amos-boot-utils-message/src/main/resources/application-dev.properties

@@ -23,7 +23,7 @@ spring.redis.password=1234560
 
 #需要监听得kafka消息主题 根据是否是中心极和站端选择需要监听得主题进行配置
 kafka.topics=null.topic
-kafka.init.topics=akka.iot.created,akka.patrol.created
+kafka.init.topics=akka.iot.created,akka.patrol.created,akka.sign.created,akka.bussSign.created,akka.user.created
 
 #需要监听得eqm消息主题 根据是否是中心极和站端选择需要监听得主题进行配置
-emq.topic=eqm.iot.created,eqm.patrol.created
+emq.topic=emq.iot.created,emq.patrol.created,emq.sign.created,emq.bussSign.created,emq.user.created
\ No newline at end of file

amos-boot-utils/amos-boot-utils-message/src/main/resources/json/topic.json

@@ -13,5 +13,15 @@
     "code": "sign",
     "emqTopic": "emq.sign.created",
     "akkaTopic": "akka.sign.created"
+  },
+  {
+    "code": "bussSign",
+    "emqTopic": "emq.bussSign.created",
+    "akkaTopic": "akka.bussSign.created"
+  },
+  {
+    "code": "user",
+    "emqTopic": "emq.user.created",
+    "akkaTopic": "akka.user.created"
   }
 ]
\ No newline at end of file