five-groups
/
five-groups-couplet
4
0
Fork 0
Code Issues Pull Requests 2 Packages Projects Releases Wiki Activity

结束时间实现

server_five_dongxiaodong
dongxiaodong 2024-04-11 09:43:44 +08:00
parent c5fbb65b62
commit 81632a5679
7 changed files with 140 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@ -38,7 +38,7 @@ import static java.lang.Thread.sleep;
@Component @Component
@Slf4j @Slf4j
public class ModelsKafkaMessage { 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"; private static final String BOOTSTRAP_SERVERS = "39.103.133.136:9092";
@ -58,7 +58,7 @@ public class ModelsKafkaMessage {
Properties props = new Properties(); Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS); props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
// props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-consumer-group"); // 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.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer"); props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
@ -76,7 +76,7 @@ public class ModelsKafkaMessage {
* @return * @return
*/ */
// @Scheduled(fixedDelay = 50) // @Scheduled(fixedDelay = 50)
// @PostConstruct @PostConstruct
public void consumerMessages() { public void consumerMessages() {
executorService.execute(this::consumer); executorService.execute(this::consumer);
} }

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

@ -80,13 +80,17 @@ public class BreakdownServiceImpl extends KeyExpirationEventMessageListener impl
long expireTime = 30; long expireTime = 30;
redisService.expire(key, expireTime, TimeUnit.MINUTES); redisService.expire(key, expireTime, TimeUnit.MINUTES);
scheduledRedis(); scheduledRedis();
// long timeMillis = System.currentTimeMillis();
// log.info("故障事件结束时间："+timeMillis);
// log.info("故障事件检测结束.....");
}else {
long timeMillis = System.currentTimeMillis(); long timeMillis = System.currentTimeMillis();
log.info("故障事件结束时间："+timeMillis); Date date = MsgUtils.convertMillisToDateTimeSering(timeMillis);
log.info("故障事件检测结束....."); CoupletTroubleCode troubleCode = new CoupletTroubleCode();
troubleCode.setTroubleEndTime(date);
log.info("故障事件结束时间："+date);
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) { for (CoupletMsgData member : members) {
Set<String> breakdownIds = redisService.getCacheSet(member.getVin()+":"+key); Set<String> breakdownIds = redisService.getCacheSet(member.getVin()+":"+key);
if (breakdownIds.size()==0){ if (breakdownIds.size()==0){
//异步执行任务的方法，它可以在后台执行一个无返回值的异步操作。该方法接受一个 Runnable 类型的参数，用于指定要执行的任务
CompletableFuture.runAsync(() -> { CompletableFuture.runAsync(() -> {
CoupletTroubleCode troubleCode = new CoupletTroubleCode(); CoupletTroubleCode troubleCode = new CoupletTroubleCode();
troubleCode.setTroubleStartTime(new Date()); troubleCode.setTroubleStartTime(new Date());
@ -251,7 +256,7 @@ public class BreakdownServiceImpl extends KeyExpirationEventMessageListener impl
break; break;
} }
remoteTroubleService.newFaultData(troubleCode); remoteTroubleService.newFaultData(troubleCode);
}); });
redisService.setCacheSet(member.getVin()+":"+key, member.getVin()+":"+member); redisService.setCacheSet(member.getVin()+":"+key, member.getVin()+":"+member);
long expireTime = 30; long expireTime = 30;
redisService.expire(member.getVin()+":"+key, expireTime, TimeUnit.MINUTES); redisService.expire(member.getVin()+":"+key, expireTime, TimeUnit.MINUTES);

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

@ -5,6 +5,10 @@ import lombok.extern.slf4j.Slf4j;
import java.math.BigDecimal; import java.math.BigDecimal;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Date; import java.util.Date;
import java.util.List; import java.util.List;
@ -19,6 +23,20 @@ import java.util.Random;
@Slf4j @Slf4j
public class MsgUtils { public class MsgUtils {
/**
* 转换时间戳
* @return
*/
public static Date convertMillisToDateTimeSering(Long millis) {
Instant instant = Instant.ofEpochMilli(millis);
LocalDateTime localDateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
return Date.from(localDateTime.atZone(ZoneId.systemDefault()).toInstant());
}
/***
* 随机生成故障码
* @return
*/
public static String generateGTA() { public static String generateGTA() {
// 生成四位以"GTA"开头的字符串 // 生成四位以"GTA"开头的字符串
String prefix = "GTA"; String prefix = "GTA";

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

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

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

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

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

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