修改bug

# Conflicts:
#	couplet-modules/couplet-analyze/couplet-analyze-msg/pom.xml
#	couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/model/ModelsKafkaMessage.java
#	couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/resources/bootstrap.yml
#	couplet-modules/couplet-modules-onLine/src/main/resources/bootstrap.yml

couplet-modules/couplet-analyze/couplet-analyze-msg/pom.xml

@@ -80,33 +80,24 @@
             <groupId>com.couplet</groupId>
             <artifactId>couplet-common-swagger</artifactId>
         </dependency>
-
-        <!-- mqttx依赖 -->
-        <dependency>
-            <groupId>org.eclipse.paho</groupId>
-            <artifactId>org.eclipse.paho.client.mqttv3</artifactId>
-            <version>1.2.5</version>
-        </dependency>
         <!-- 事件核心配置 -->
         <dependency>
             <groupId>com.couplet</groupId>
             <artifactId>couplet-common-event</artifactId>
         </dependency>
 
+        <!-- Kafka依赖-->
         <dependency>
-            <groupId>org.springframework.kafka</groupId>
+            <groupId>org.apache.kafka</groupId>
-            <artifactId>spring-kafka</artifactId>
+            <artifactId>kafka-clients</artifactId>
+            <version>2.8.0</version>
         </dependency>
+
         <dependency>
             <groupId>com.couplet</groupId>
             <artifactId>couplet-common-business</artifactId>
         </dependency>
 
-        <!-- RabbitMQ依赖-->
-        <dependency>
-            <groupId>org.springframework.boot</groupId>
-            <artifactId>spring-boot-starter-amqp</artifactId>
-        </dependency>
 
     </dependencies>
 

couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/CoupletMsgApplication.java

@@ -1,5 +1,7 @@
 package com.couplet.analyze.msg;
 
+import com.couplet.analyze.msg.model.ModelsKafkaMessage;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.openfeign.EnableFeignClients;
@@ -15,8 +17,13 @@ import org.springframework.scheduling.annotation.EnableScheduling;
 @EnableScheduling
 @EnableFeignClients(basePackages = "com.couplet")
 public class CoupletMsgApplication {
+
+
     public static void main(String[] args) {
         SpringApplication.run(CoupletMsgApplication.class);
         System.out.println("解析系统启动成功");
+//        new ModelsKafkaMessage().initKafkaConsumer();
     }
+
+
 }

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

@@ -1,5 +1,6 @@
 package com.couplet.analyze.msg.model;
 
+
 import com.couplet.analyze.common.event.AnalyzeEventCache;
 import com.couplet.analyze.msg.domain.CoupletMsgData;
 import com.couplet.analyze.msg.service.IncidentService;
@@ -10,6 +11,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.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Component;
 
@@ -36,7 +38,7 @@ import static java.lang.Thread.sleep;
 @Component
 @Slf4j
 public class ModelsKafkaMessage {
-    private static final String TOPIC_NAME = "online";
+    private static final String TOPIC_NAME = "ljy";
     private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
 
 
@@ -50,13 +52,16 @@ public class ModelsKafkaMessage {
     private AnalyzeEventCache analyzeEventCache;
 
 
+
+
+
     //kafka消费者初始化
     @PostConstruct
     public void initKafkaConsumer() {
         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");
 
@@ -86,7 +91,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

@@ -41,19 +41,18 @@ public class RealTimeDataServiceImpl implements IncidentService {
     public void incident(CoupletMsgData coupletMsgData) {
         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())){
 //            log.info("有实时数据，值为：[{}]开始传输实时数据", coupletMsgData.getVin());
              //判断数据是否一致，
 //             if (RealTimeJudge.addRealTime(cacheObject)) {
                  log.info("[{}]有缓存数据，值为：[{}]，且缓存数据与实时数据一致，开始传输实时数据", coupletMsgData.getVin(), cacheObject);
-                 CoupletMsgData query = incidentMapper.queryByIncident(coupletMsgData.getVin());
+                 redisService.setCacheSet("vin:query:" + coupletMsgData.getVin(), coupletMsgData);
-                 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

@@ -23,7 +23,6 @@ spring:
         # 共享配置
         shared-configs:
           - application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
-
   main:
     allow-bean-definition-overriding: true
   rabbitmq:
@@ -49,4 +48,3 @@ logging:
 mybatis-plus:
   configuration:
     map-underscore-to-camel-case: true
-

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 = "ljy";
     private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
 
     //线程池，用于异步处理消息到来时的业务逻辑

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

@@ -36,7 +36,7 @@ mqtt:
     #    broker: mqtt://115.159.47.13:1883
     username:
     password:
-    clientId: xiaoYao
+    clientId: aaaaaaa
     qos: 0
     topic: xiaoYao
 

pom.xml

@@ -34,6 +34,8 @@
         <jjwt.version>0.9.1</jjwt.version>
         <minio.version>8.2.2</minio.version>
         <poi.version>4.1.2</poi.version>
+        <Kafka.version>2.8.0</Kafka.version>
+        <dubbo.version>3.0.8</dubbo.version>
         <transmittable-thread-local.version>2.14.3</transmittable-thread-local.version>
     </properties>
 
@@ -242,7 +244,19 @@
                 <artifactId>couplet-common-event</artifactId>
                 <version>${couplet.version}</version>
             </dependency>
+            <!-- Kafka  -->
+            <dependency>
+                <groupId>org.apache.kafka</groupId>
+                <artifactId>kafka-clients</artifactId>
+                <version>${Kafka.version}</version>
+            </dependency>
 
+            <dependency>
+                <groupId>org.apache.dubbo</groupId>
+                <artifactId>dubbo</artifactId>
+                <version>${dubbo.version}</version>
+                <scope>compile</scope>
+            </dependency>
         </dependencies>
     </dependencyManagement>