虎虎虎

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/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);
 //             }

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>