结束时间已实现

# Conflicts:
#	couplet-modules/couplet-analyze/couplet-analyze-msg/pom.xml

couplet-common/couplet-common-business/src/main/java/com/couplet/remote/RemoteTroubleService.java

@@ -8,6 +8,7 @@ import org.springframework.cloud.openfeign.FeignClient;
 import org.springframework.web.bind.annotation.PostMapping;
 import org.springframework.web.bind.annotation.RequestBody;
 
+
 @FeignClient(contextId = "remoteTroubleService" ,
         value = ServiceNameConstants.BUSINESS_SERVICE,
         fallbackFactory = RemoteTroubleFallbackFactory.class
@@ -22,4 +23,11 @@ public interface RemoteTroubleService {
     @PostMapping("/trouble/newFaultData")
     public Result<?> newFaultData(@RequestBody CoupletTroubleCode code);
 
+    /**
+     * 修改故障码
+     * @param trouble
+     * @return
+     */
+    @PostMapping("/trouble/updEndTime")
+    public int updEndTime(@RequestBody CoupletTroubleCode trouble);
 }

couplet-common/couplet-common-business/src/main/java/com/couplet/remote/factory/RemoteTroubleFallbackFactory.java

@@ -8,6 +8,7 @@ import org.slf4j.LoggerFactory;
 import org.springframework.cloud.openfeign.FallbackFactory;
 import org.springframework.stereotype.Component;
 
+
 /**
  * @author fufanrui
  * @version 1.0
@@ -26,9 +27,13 @@ public class RemoteTroubleFallbackFactory implements FallbackFactory<RemoteTroub
 
             @Override
             public Result<?> newFaultData(CoupletTroubleCode code) {
-                return Result.error("调用失败...."+cause.getMessage());
+                return Result.error("调用失败...." + cause.getMessage());
             }
 
+            @Override
+            public int updEndTime(CoupletTroubleCode trouble) {
+                return 0;
+            }
         };
     }
 }

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

@@ -80,6 +80,7 @@
             <groupId>com.couplet</groupId>
             <artifactId>couplet-common-swagger</artifactId>
         </dependency>
+
         <!-- 事件核心配置 -->
         <dependency>
             <groupId>com.couplet</groupId>
@@ -98,7 +99,6 @@
             <artifactId>couplet-common-business</artifactId>
         </dependency>
 
-
     </dependencies>
 
 </project>

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

@@ -20,7 +20,7 @@ public class CoupletMsgApplication {
     public static void main(String[] args) {
         SpringApplication.run(CoupletMsgApplication.class);
         System.out.println("解析系统启动成功");
-        new ModelsKafkaMessage().initKafkaConsumer();
+//        new ModelsKafkaMessage().initKafkaConsumer();
     }
 
 

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

@@ -38,7 +38,7 @@ import static java.lang.Thread.sleep;
 @Component
 @Slf4j
 public class ModelsKafkaMessage {
-    private static final String TOPIC_NAME = "topic_lyh";
+    private static final String TOPIC_NAME = "topic_dxd";
     private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
 
 
@@ -52,23 +52,20 @@ 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, "group");
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "fbab");
         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");
 
         //消费者
         consumer = new KafkaConsumer<>(props);
-        //订阅主题
-        consumer.subscribe(Collections.singletonList(TOPIC_NAME));
+
+        this.consumerMessages();
     }
 
 
@@ -86,6 +83,8 @@ public class ModelsKafkaMessage {
 
     public void consumer() {
         try {
+            //订阅主题
+            consumer.subscribe(Collections.singletonList(TOPIC_NAME));
 
             //持续消费消息
             while (true) {

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

@@ -1,6 +1,5 @@
 package com.couplet.analyze.msg.service.impl;
 
-import com.alibaba.fastjson.JSON;
 import com.couplet.analyze.common.contents.AnalyzeEventContents;
 import com.couplet.analyze.msg.contents.StateConstant;
 import com.couplet.analyze.msg.domain.CoupletMsgData;
@@ -80,13 +79,18 @@ public class BreakdownServiceImpl extends KeyExpirationEventMessageListener impl
             long expireTime = 30;
             redisService.expire(key, expireTime, TimeUnit.MINUTES);
             scheduledRedis();
+//            long timeMillis = System.currentTimeMillis();
+//            log.info("故障事件结束时间："+timeMillis);
+//            log.info("故障事件检测结束.....");
+        }else {
             long timeMillis = System.currentTimeMillis();
+            CoupletTroubleCode troubleCode = new CoupletTroubleCode();
+            troubleCode.setTroubleEndTime(new Date());
+            troubleCode.setTroubleVin(coupletMsgData.getVin());
+            remoteTroubleService.updEndTime(troubleCode);
             log.info("故障事件结束时间："+timeMillis);
-            log.info("故障事件检测结束.....");
+            log.info("故障事件结束.....");
         }
-        long timeMillis = System.currentTimeMillis();
-        log.info("故障事件结束时间："+timeMillis);
-        log.info("故障事件结束.....");
     }
 
     /**
@@ -105,6 +109,7 @@ public class BreakdownServiceImpl extends KeyExpirationEventMessageListener impl
             for (CoupletMsgData member : members) {
                 Set<String> breakdownIds = redisService.getCacheSet(member.getVin()+":"+key);
                 if (breakdownIds.size()==0){
+                    //异步执行任务的方法，它可以在后台执行一个无返回值的异步操作。该方法接受一个 Runnable 类型的参数，用于指定要执行的任务
                     CompletableFuture.runAsync(() -> {
                     CoupletTroubleCode troubleCode = new CoupletTroubleCode();
                     troubleCode.setTroubleStartTime(new Date());
@@ -251,7 +256,7 @@ public class BreakdownServiceImpl extends KeyExpirationEventMessageListener impl
                                 break;
                         }
                     remoteTroubleService.newFaultData(troubleCode);
-                });
+                    });
                     redisService.setCacheSet(member.getVin()+":"+key, member.getVin()+":"+member);
                     long expireTime = 30;
                     redisService.expire(member.getVin()+":"+key, expireTime, TimeUnit.MINUTES);

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

@@ -49,7 +49,7 @@ public class ElectronicFenceServiceImpl implements IncidentService {
             log.info("电子围栏事件redis存在.......");
             for (Fence fence : cacheSet) {
                 String fenceLongitudeLatitude = fence.getFenceLongitudeLatitude();
-                if (StringUtils.isEmpty(fenceLongitudeLatitude)){
+                if (!StringUtils.isEmpty(fenceLongitudeLatitude)){
                     /**
                      * 先通过；后切割
                      */

couplet-modules/couplet-analyze/couplet-analyze-msg/src/main/java/com/couplet/analyze/msg/utils/MsgUtils.java

@@ -5,6 +5,9 @@ import lombok.extern.slf4j.Slf4j;
 
 import java.math.BigDecimal;
 import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
 import java.util.ArrayList;
 import java.util.Date;
 import java.util.List;
@@ -19,6 +22,16 @@ import java.util.Random;
 @Slf4j
 public class MsgUtils {
 
+     public static Date convertMillisToDate(long millis) {
+        Instant instant = Instant.ofEpochMilli(millis);
+        LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
+        return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant());
+    }
+
+    /**
+     * 生成GTA
+     * @return
+     */
     public static String generateGTA() {
         // 生成四位以"GTA"开头的字符串
         String prefix = "GTA";

couplet-modules/couplet-analyze/couplet-analyze-msg/src/test/java/com/couplet/msg/Main.java

@@ -1,5 +1,10 @@
 package com.couplet.msg;
 
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.format.DateTimeFormatter;
+import java.util.Date;
 import java.util.Random;
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -12,17 +17,14 @@ import java.util.regex.Pattern;
  */
 public class Main {
     public static void main(String[] args) {
-        // 创建一个字符串数组存储三个状态
-        String[] statuses = {"电池故障", "车体故障", "车尾故障","抽轮故障"};
-
-        // 生成随机数对象
-        Random rand = new Random();
-
-        // 生成一个范围在 0 到 2 之间的随机整数
-        int randomIndex = rand.nextInt(4);
-
-        // 随机选择一个字符串并输出
-        String randomStatus = statuses[randomIndex];
-        System.out.println("随机输出的字符串：" + randomStatus);
+        long timeMillis = System.currentTimeMillis();
+        Date date = new Date(timeMillis);
+        System.out.println(date);
     }
+
+//    private static Date convertMillisToDate(long millis) {
+//        Instant instant = Instant.ofEpochMilli(millis);
+//        LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
+//        return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant());
+//    }
 }

couplet-modules/couplet-business/src/main/java/com/couplet/business/server/controller/SysTroubleController.java

@@ -14,6 +14,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.validation.annotation.Validated;
 import org.springframework.web.bind.annotation.*;
 
+import java.util.Date;
 import java.util.List;
 
 /**
@@ -97,9 +98,9 @@ public class SysTroubleController extends BaseController {
     /**
      * 查询故障的状态并处理
      */
-    @PostMapping("/getUpdState")
-    public Result<?> getUpdState(@RequestBody CoupletTroubleCode code) {
-        int updateState = troubleService.updateState(code);
-        return Result.success(updateState);
+    @PostMapping("/updEndTime")
+    public int updEndTime(@RequestBody CoupletTroubleCode trouble) {
+        int i = troubleService.updEndTime(trouble);
+        return i;
     }
 }

couplet-modules/couplet-business/src/main/java/com/couplet/business/server/mapper/SysTroubleMapper.java

@@ -6,7 +6,9 @@ import com.couplet.common.domain.CoupletTroubleGrade;
 
 import com.couplet.common.domain.request.TroubleResp;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
 
+import java.util.Date;
 import java.util.List;
 
 /**
@@ -30,5 +32,7 @@ public interface SysTroubleMapper extends BaseMapper<CoupletTroubleCode> {
 
     void cleanTroubleCode();
 
-    int updateState(CoupletTroubleCode code);
+    int updEndTime(CoupletTroubleCode trouble);
+
+    List<CoupletTroubleCode> selectMaxTime();
 }

couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/SysTroubleService.java

@@ -6,6 +6,7 @@ import com.couplet.common.domain.CoupletTroubleCode;
 import com.couplet.common.domain.CoupletTroubleGrade;
 import com.couplet.common.domain.request.TroubleResp;
 
+import java.util.Date;
 import java.util.List;
 
 /**
@@ -27,5 +28,5 @@ public interface SysTroubleService extends IService<CoupletTroubleCode> {
 
     void cleanTroubleCode();
 
-    int updateState(CoupletTroubleCode code);
+    int updEndTime(CoupletTroubleCode trouble);
 }

couplet-modules/couplet-business/src/main/java/com/couplet/business/server/service/impl/SysTroubleServiceImpl.java

@@ -12,6 +12,7 @@ import com.github.pagehelper.PageInfo;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
+import java.util.Date;
 import java.util.List;
 
 /**
@@ -60,11 +61,17 @@ public class SysTroubleServiceImpl extends ServiceImpl<SysTroubleMapper, Couplet
 
     /**
      * 获取故障状态
-     * @param code
+     * @param trouble
      * @return
      */
     @Override
-    public int updateState(CoupletTroubleCode code) {
+    public int updEndTime(CoupletTroubleCode trouble) {
+        // 查询表中最大时间
+        List<CoupletTroubleCode> maxTime = sysTroubleMapper.selectMaxTime();
+        // 然后将结束时间改成当前时间
+        if (maxTime != null) {
+            sysTroubleMapper.updEndTime(trouble);
+        }
         return 0;
     }
 

couplet-modules/couplet-business/src/main/resources/mapper/business/SysTroubleMapper.xml

@@ -26,10 +26,11 @@
     <update id="cleanTroubleCode">
         truncate table couplet_trouble_code
     </update>
-    <update id="updateState">
-        update couplet_trouble_code set processing_state = 1 where thourble_id = #{troubleId}
+    <update id="updEndTime">
+        update couplet_trouble_code set trouble_end_time = #{troubleEndTime} where trouble_vin = #{troubleVin}
     </update>
 
+
     <select id="selectTroubleList" parameterType="com.couplet.business.server.mapper.SysTroubleMapper" resultMap="SysTroubleResult">
         <include refid="selectTroubleVo"/>
         <where>
@@ -45,5 +46,8 @@
     <select id="selectTroubleListByGrade" resultType="com.couplet.common.domain.CoupletTroubleGrade">
         select * from couplet_trouble_grade
     </select>
+    <select id="selectMaxTime" resultType="com.couplet.common.domain.CoupletTroubleCode">
+        select max(trouble_start_time) from couplet_trouble_code
+    </select>
 
 </mapper>

couplet-modules/couplet-modules-mq/src/main/java/com/couplet/mq/controller/KafkaTest.java

@@ -1,93 +1,93 @@
-//package com.couplet.mq.controller;
-//
-//import lombok.extern.slf4j.Slf4j;
-//import org.apache.kafka.clients.consumer.ConsumerConfig;
-//import org.apache.kafka.clients.consumer.ConsumerRecords;
-//import org.apache.kafka.clients.consumer.KafkaConsumer;
-//import org.apache.kafka.clients.producer.KafkaProducer;
-//import org.apache.kafka.clients.producer.ProducerConfig;
-//import org.apache.kafka.clients.producer.ProducerRecord;
-//
-//import javax.annotation.PostConstruct;
-//import java.time.Duration;
-//import java.util.Collections;
-//import java.util.Properties;
-//
-///**
-// * @ProjectName: five-groups-couplet
-// * @Author: LiuYunHu
-// * @CreateTime: 2024/4/4
-// * @Description: kafka测试类
-// */
-//
-//@Slf4j
-//public class KafkaTest {
-//    private static final String TOPIC_NAME = "online";
-//    private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
-//
-//    public static void main(String[] args) {
-//        //生产者示例
-////        produceMessage();
-//
-//        //消费者示例
-////        consumerMessages();
-//
-//    }
-//
-//    //生产者
-//    @PostConstruct
-//    private static void produceMessage() {
-//        Properties props = new Properties();
-//        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
-//        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
-//        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
-//
-//        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
-//        //创建生产者
-//        try {
-//
-//            //发送消息
-//            for (int i = 0; i < 10000; i++) {
-//                String message = "佳佳来喽" + (i + 1);
-//                producer.send(new ProducerRecord<>(TOPIC_NAME, message));
-//
-//                System.out.println("发送消息：" + message);
-//            }
-//        } catch (Exception e) {
-//            e.printStackTrace();
-//        } finally {
-//            producer.close();
-//        }
-//    }
-//
-//    //消费者
-////    private static void consumerMessages() {
-////        Properties props = new Properties();
-////        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
-////        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-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");
-////
-////        //创建消费者
-////        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
-////
-////        try {
-////
-////            //订阅主题
-////            consumer.subscribe(Collections.singletonList(TOPIC_NAME));
-////
-////            //持续消费消息
-////            while (true) {
-////                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
-////                records.forEach(record -> {
-////                    System.out.println("消费者接受到的消息值：" + record.value());
-////                });
-////            }
-////        } catch (Exception e) {
-////            e.printStackTrace();
-////        } finally {
-////            consumer.close();
-////        }
-////    }
-//
-//}
+package com.couplet.mq.controller;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerConfig;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.clients.producer.ProducerRecord;
+
+import javax.annotation.PostConstruct;
+import java.time.Duration;
+import java.util.Collections;
+import java.util.Properties;
+
+/**
+ * @ProjectName: five-groups-couplet
+ * @Author: LiuYunHu
+ * @CreateTime: 2024/4/4
+ * @Description: kafka测试类
+ */
+
+@Slf4j
+public class KafkaTest {
+    private static final String TOPIC_NAME = "topic_dxd";
+    private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
+
+    public static void main(String[] args) {
+        //生产者示例
+//        produceMessage();
+
+        //消费者示例
+        consumerMessages();
+
+    }
+
+    //生产者
+    @PostConstruct
+    private static void produceMessage() {
+        Properties props = new Properties();
+        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
+        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
+
+        KafkaProducer<String, String> producer = new KafkaProducer<>(props);
+        //创建生产者
+        try {
+
+            //发送消息
+            for (int i = 0; i < 10000; i++) {
+                String message = "佳佳来喽" + (i + 1);
+                producer.send(new ProducerRecord<>(TOPIC_NAME, message));
+
+                System.out.println("发送消息：" + message);
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            producer.close();
+        }
+    }
+
+    //消费者
+    private static void consumerMessages() {
+        Properties props = new Properties();
+        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
+        props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-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");
+
+        //创建消费者
+        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
+
+        try {
+
+            //订阅主题
+            consumer.subscribe(Collections.singletonList(TOPIC_NAME));
+
+            //持续消费消息
+            while (true) {
+                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
+                records.forEach(record -> {
+                    System.out.println("消费者接受到的消息值：" + record.value());
+                });
+            }
+        } catch (Exception e) {
+            e.printStackTrace();
+        } finally {
+            consumer.close();
+        }
+    }
+
+}

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