cloud-common/cloud-common-kafka/pom.xml

@@ -33,5 +33,9 @@
             <groupId>com.muyu</groupId>
             <artifactId>cloud-common-core</artifactId>
         </dependency>
+
+
+
+
     </dependencies>
 </project>

cloud-modules/cloud-event/src/main/java/com/muyu/event/consumer/MessageConsumer.java

@@ -1,7 +1,7 @@
 package com.muyu.event.consumer;
+
 import com.alibaba.fastjson2.JSONObject;
 import com.muyu.event.basic.EventPublisher;
-import com.muyu.event.service.IncidentService;
 import lombok.extern.log4j.Log4j2;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -9,10 +9,13 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.ApplicationRunner;
 import org.springframework.stereotype.Component;
+
 import java.time.Duration;
 import java.util.Collections;
 import java.util.List;
+
 import static org.bouncycastle.asn1.x500.style.RFC4519Style.l;
+
 /**
  * kafka监听
  * @author 刘武
@@ -26,18 +29,16 @@ public class MessageConsumer implements ApplicationRunner {
 
     @Autowired
     public KafkaConsumer consumer;
-
+    @Autowired
     private EventPublisher eventPublisher;
 
-    private final String topic="kafka-topic";
-
-    @Autowired
-    private IncidentService incidentService;
+    private final String topic="four_car";
 
     @Override
     public void run(ApplicationArguments args) throws Exception {
         List<String> list = Collections.singletonList(topic);
         consumer.subscribe(list);
+
         while (true){
             ConsumerRecords<String,String> consumerRecords = consumer.poll(Duration.ofMillis(100));
             consumerRecords.forEach(record -> {
@@ -45,12 +46,6 @@ public class MessageConsumer implements ApplicationRunner {
                 JSONObject jsonObject = JSONObject.parseObject(value);
                 log.info("value:{}",value);
                 eventPublisher.publishEvent(jsonObject);
-                try {
-                    incidentService.warnEventProcess(jsonObject);
-                } catch (Exception e) {
-                    throw new RuntimeException(e);
-                }
-
             });
         }
     }

cloud-modules/cloud-event/src/main/java/com/muyu/event/listener/AddDatabaseListener.java

@@ -1,9 +1,13 @@
 package com.muyu.event.listener;
+
 import com.alibaba.fastjson2.JSONObject;
 import com.muyu.event.basic.EventCustom;
 import com.muyu.event.basic.EventListener;
+
+
 import java.util.ArrayList;
 import java.util.List;
+
 /**
  * 添加数据库事件
  * @program: cloud-server
@@ -25,6 +29,9 @@ public class AddDatabaseListener implements EventListener {
             values.add((String) value);
         });
 
+
+
+
     }
 
     @Override

cloud-modules/cloud-event/src/main/java/com/muyu/event/service/IncidentService.java

@@ -12,6 +12,5 @@ public interface IncidentService {
 
    void warnEventProcess(JSONObject jsonObject) throws Exception;
 
-   void eventAlarmProcessing(JSONObject jsonObject) throws Exception;
 
 }

cloud-modules/cloud-event/src/main/java/com/muyu/event/service/impl/IncidentServiceImpl.java

@@ -1,9 +1,11 @@
 package com.muyu.event.service.impl;
 import com.alibaba.fastjson2.JSONObject;
-import com.muyu.cache.*;
+import com.muyu.cache.MessageTemplateTypeCacheService;
+import com.muyu.cache.SysCarCacheService;
+import com.muyu.cache.WarnRuleCacheService;
+import com.muyu.cache.WarnStrategyCacheService;
 import com.muyu.common.domain.MessageTemplateType;
 import com.muyu.common.domain.SysCar;
-import com.muyu.common.domain.SysCarFault;
 import com.muyu.common.domain.resp.SysCarVo;
 import com.muyu.common.domain.resp.WarnRuleResp;
 import com.muyu.common.domain.resp.WarnStrategyResp;
@@ -36,7 +38,6 @@ import java.util.stream.Collectors;
 @Log4j2
 @Service
 public class IncidentServiceImpl implements IncidentService {
-
     private static  int DURATION_SECONDS = 5;
     private static List<JSONObject> receivedStrings = new ArrayList<>();
     private static ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);
@@ -60,23 +61,27 @@ public class IncidentServiceImpl implements IncidentService {
     //报文模版
     @Resource
     private MessageTemplateTypeCacheService messageTemplateTypeCacheService;
-    //kafka的主题名称
-    private final String topic="kafka-topic";
+
+    private final String topic="four_car";
+
     @Autowired
-    private SysCarFaultCacheService sysCarFaultCacheService;
+    private MessageConsumer messageConsumer;
+
+    @Autowired
+    public KafkaConsumer consumer;
+
     @Override
     public void warnEventProcess(JSONObject jsonObject) throws Exception {
+
         receivedStrings.add(jsonObject);
         //协议解析:每秒穿过来一个JSONObject jsonObject;  添加进receivedStrings
         //根据这个车辆VIN查询出他对应的车辆类型
         String carVin=null;
         //报文模版的ID
         Integer templateId=null;
-        //取出这辆车的carVin
         for (JSONObject receivedString : receivedStrings) {
             carVin = (String) receivedString.get("carVin");
         }
-        //这辆车的信息
         SysCar carByVin = null;
         List<SysCarVo> carVoList = sysCarCacheService.get(sysCarCacheService.keyPre());
         Map<String, SysCarVo> carMap = carVoList.stream()
@@ -89,7 +94,6 @@ public class IncidentServiceImpl implements IncidentService {
         List<WarnStrategyResp> carWithWarnStrategyList=null;
         List<WarnStrategyResp> warnStrategyResps = warnStrategyCacheService.get(warnStrategyCacheService.keyPre());
         for (WarnStrategyResp warnStrategyResp : warnStrategyResps) {
-            //策略中有绑定的车辆ID
             if(warnStrategyResp.getCarTypeId()==carByVin.getCarTypeId()){
                 carWithWarnStrategyList.add(warnStrategyResp);
             }
@@ -150,29 +154,7 @@ public class IncidentServiceImpl implements IncidentService {
             }
         }
     }
-    @Override
-    public void eventAlarmProcessing(JSONObject jsonObject) throws Exception {
-        String carVin = (String) jsonObject.get("carVin");
-        //车辆缓存配置
-        List<SysCarVo> carVoList = sysCarCacheService.get(sysCarCacheService.keyPre());
-        //这辆车的信息
-        SysCar carByVin = null;
-        Map<String, SysCarVo> carMap = carVoList.stream()
-                .collect(Collectors.toMap(SysCarVo::getCarVin, Function.identity()));
-        //获取到了这个车辆的信息
-        carByVin = carMap.get(carVin);
-        Long carTypeId = carByVin.getCarTypeId();
-        //车辆绑定的报警规则
-        List<SysCarFault> sysCarFaults = sysCarFaultCacheService.get(sysCarFaultCacheService.keyPre());
-        //车辆的所有报警规则都重新存储在sysCarFaultList里面
-        List<SysCarFault> sysCarFaultList=null;
-        SysCar finalCarByVin = carByVin;
-        sysCarFaults.forEach(sysCarFault -> {
-            if(sysCarFault.getCarTypeId().equals(carTypeId)){
-                sysCarFaultList.add(sysCarFault);
-            }
-        });
-        //报文模版
 
-    }
+
+
 }

cloud-modules/cloud-modules-template/src/main/java/com/muyu/template/config/MqttConfigure.java

@@ -160,6 +160,7 @@ public class MqttConfigure {
         log.info("发送kafka成功");
         return jsonObject;
     }
+
     //kafka发送消息
     public void sendKafka(JSONObject jsonObject){
         ProducerRecord<String, String> stringStringProducerRecord = new ProducerRecord<>("four_car", jsonObject.toString());

cloud-modules/saas/saas-common/src/main/java/com/muyu/common/domain/SysCar.java

@@ -1,4 +1,5 @@
 package com.muyu.common.domain;
+
 import com.baomidou.mybatisplus.annotation.IdType;
 import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
@@ -7,6 +8,7 @@ import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.EqualsAndHashCode;
 import lombok.NoArgsConstructor;
+
 /**
  * 车辆管理表
  * @author sx
@@ -20,7 +22,6 @@ import lombok.NoArgsConstructor;
 @EqualsAndHashCode(callSuper = true)
 @TableName(value = "sys_car",autoResultMap = true)
 public class SysCar extends BaseEntity {
-
     @TableId(value = "id",type = IdType.AUTO)
     private Long id;
     private String carVin;