feat: 优化代码

couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelsKafkaMessage.java

@@ -10,6 +10,7 @@ import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
@@ -56,7 +57,7 @@ public class ModelsKafkaMessage {
         Properties props = new Properties();
         props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
 //        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group");
-        props.put(ConsumerConfig.GROUP_ID_CONFIG, "ddd");
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
         props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
         props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
 
@@ -78,7 +79,6 @@ public class ModelsKafkaMessage {
     public void consumerMessages() {
         executorService.execute(this::consumer);
     }
-
     public void consumer() {
         try {
 
@@ -86,7 +86,7 @@ public class ModelsKafkaMessage {
             while (true) {
                 ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                 records.forEach(record -> {
-                    System.out.println("接收到的数据：" + record.value());
+                    log.info("接收到的数据：" + record.value());
                     String str = hexToString(record.value());
                     List<CoupletMsgData> coupletMsgDataList = sendMsg(str);
                     for (CoupletMsgData msgData : coupletMsgDataList) {

couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/service/impl/RealTimeDataServiceImpl.java

@@ -42,8 +42,8 @@ public class RealTimeDataServiceImpl implements IncidentService {
         log.info("实时数据事件开始.....");
         RealTimeDataRequest cacheObject = redisService.getCacheObject("vin:" + coupletMsgData.getVin());
         //判断是否有缓存数据
-        if (redisService.hasKey("vin:query:" + cacheObject.getVin())){
+        if (redisService.hasKey("vin:query:" + coupletMsgData.getVin())){
-            redisService.deleteObject("vin:query:" + cacheObject.getVin());
+            redisService.deleteObject("vin:query:" + coupletMsgData.getVin());
         }
 //         if (RealTimeJudge.isJudge(coupletMsgData.getVin())) {
         if (coupletMsgData.getVin().equals(cacheObject.getVin())){
@@ -51,9 +51,8 @@ public class RealTimeDataServiceImpl implements IncidentService {
              //判断数据是否一致，
 //             if (RealTimeJudge.addRealTime(cacheObject)) {
                  log.info("[{}]有缓存数据，值为：[{}]，且缓存数据与实时数据一致，开始传输实时数据", coupletMsgData.getVin(), cacheObject);
-                 CoupletMsgData query = incidentMapper.queryByIncident(coupletMsgData.getVin());
+                 redisService.setCacheObject("vin:query:" + coupletMsgData.getVin(), cacheObject);
-                 redisService.setCacheObject("vin:query:" + cacheObject.getVin(), query);
+                 redisService.expire("vin:"+coupletMsgData.getVin(),10, TimeUnit.MINUTES);
-                 redisService.expire("vin:"+cacheObject.getVin(),10, TimeUnit.MINUTES);
 //             } else {
 //                 log.info("[{}]有缓存数据，值为：[{}]，且缓存数据与实时数据不一致，开始传输实时数据", coupletMsgData.getVin(), cacheObject);
 //             }

couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/resources/bootstrap.yml

@@ -43,6 +43,11 @@ spring:
      template:
        # 只要消息抵达Queue，就会异步发送优先回调return firm
        mandatory: true
+  kafka:
+    bootstrap-servers: 115.159.47.13:9092
+    consumer:
+      group-id: group
+      auto-offset-reset: earliest
 logging:
   level:
     com.couplet.analyze.msg.mapper: DEBUG

couplet-modules/couplet-business/pom.xml

@@ -114,6 +114,7 @@
             <groupId>com.couplet</groupId>
             <artifactId>couplet-analyze-msg</artifactId>
         </dependency>
+
     </dependencies>
 
     <build>

couplet-modules/couplet-modules-onLine/src/main/java/com/couplet/online/utils/MqttMonitor.java

@@ -76,7 +76,7 @@ public class MqttMonitor {
 
 
     //Kafka生产者配置
-    private static final String TOPIC_NAME = "xiaoYao";
+    private static final String TOPIC_NAME = "online";
     private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
 
     //线程池，用于异步处理消息到来时的业务逻辑

couplet-modules/couplet-modules-onLine/src/main/resources/bootstrap.yml

@@ -28,7 +28,6 @@ spring:
 logging:
   level:
     com.couplet.system.mapper: DEBUG
-
 # 订阅端配置
 mqtt:
   server: