消息组件优化

amos-boot-module/amos-boot-module-biz/amos-boot-module-equip-biz/src/main/java/com/yeejoin/equipmanage/service/impl/MqttReceiveServiceImpl.java

@@ -279,9 +279,9 @@ public class MqttReceiveServiceImpl implements MqttReceiveService {
         jsonObject.put("data",message);
 
         try {
-            emqKeeper.getMqttClient().publish("eqm.iot.created",jsonObject.toString().getBytes(),1,false);
+            emqKeeper.getMqttClient().publish("emq.iot.created",jsonObject.toString().getBytes(),1,false);
         } catch (MqttException e) {
-            e.printStackTrace();
+            log.info(String.format("发送eqm转kafka消息失败：%s", e.getMessage()));
         }
 
         if (!StringUtils.isEmpty(traceId)) {

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

@@ -136,6 +136,11 @@ public class SignServiceImpl extends BaseService<SignDto,Sign,SignMapper> implem
                     SerializerFeature.WriteMapNullValue);
             try {
                 emqKeeper.getMqttClient().publish(personSign, json.getBytes(), RuleConfig.DEFAULT_QOS, false);
+                // 发送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);
             } catch (MqttException e) {
                 e.printStackTrace();
             }

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

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

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

@@ -6,7 +6,7 @@ spring.jackson.time-zone=GMT+8
 spring.jackson.date-format=yyyy-MM-dd HH:mm:ss
 spring.jackson.serialization.write-dates-as-timestamps=true
 # kafka集群信息
-spring.kafka.bootstrap-servers=127.0.0.1:9092 
+spring.kafka.bootstrap-servers=172.16.3.100:9092
 # 生产者配置
 # 设置大于0的值，则客户端会将发送失败的记录重新发送  # 重试次数
 spring.kafka.producer.retries=3

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

 [
   {
     "code": "iot",
-    "emqTopic": "eqm.iot.created",
+    "emqTopic": "emq.iot.created",
     "akkaTopic": "akka.iot.created"
   },
   {
     "code": "patrol",
-    "emqTopic": "eqm.patrol.created",
+    "emqTopic": "emq.patrol.created",
     "akkaTopic": "akka.patrol.created"
+  },
+  {
+    "code": "sign",
+    "emqTopic": "emq.sign.created",
+    "akkaTopic": "akka.sign.created"
   }
 ]
\ No newline at end of file