result = producer.send(new ProducerRecord<>(topic, key, value));
+ RecordMetadata meta = null;
+ try {
+ meta = result.get();
+ } catch (InterruptedException | ExecutionException e) {
+ e.printStackTrace();
+ }
+ return meta;
+ }
+
+ /**
+ * This method is called when the connection to the server is lost.
+ *
+ * @param cause the reason behind the loss of connection.
+ */
+ @Override
+ public void connectionLost (Throwable cause) {
+
+ }
+
+ /**
+ * 当消息从服务器到达时，将调用此方法。
+ * 该方法由MQTT客户端同步调用。确认不会发送回服务器，直到此方法干净地返回。
+ * 如果此方法的实现抛出异常，则客户端将被关闭。当客户端下一次重新连接时，任何QoS 1或2消息将由服务器重新传递。
+ * 在运行此方法的实现时到达的任何其他消息将在内存中建立，然后将在网络上备份。
+ * 如果应用程序需要持久化数据，那么它应该确保在从该方法返回之前持久化数据，因为在从该方法返回之后，消息被认为已被传递，并且将不可再现。
+ * 可以在此回调的实现中发送新消息 (例如，对此消息的响应)，但实现不能断开客户端的连接，因为不可能发送正在处理的消息的确认，并且会发生死锁。
+ *
+ * @param topic 消息上的主题名称已发布到
+ * @param message 真正的信息。
+ *
+ * @throws Exception 如果发生了终端错误，客户端应该关闭。
+ */
+ @Override
+ public void messageArrived (String topic, MqttMessage message) throws Exception {
+ String msg = new String(message.getPayload());
+ log.info("topic: [{}], Qos: [{}], message content: [{}]", topic, message.getQos(), msg);
+ send(KafkaContent.TOPIC, String.valueOf(msg.hashCode()), msg);
+ }
+
+ /**
+ * Called when delivery for a message has been completed, and all
+ * acknowledgments have been received. For QoS 0 messages it is
+ * called once the message has been handed to the network for
+ * delivery. For QoS 1 it is called when PUBACK is received and
+ * for QoS 2 when PUBCOMP is received. The token will be the same
+ * token as that returned when the message was published.
+ *
+ * @param token the delivery token associated with the message.
+ */
+ @Override
+ public void deliveryComplete (IMqttDeliveryToken token) {
+
+ }
+}
diff --git a/src/main/java/com/muyu/parsing/ParsingService.java b/src/main/java/com/muyu/parsing/ParsingService.java
new file mode 100644
index 0000000..a90e13d
--- /dev/null
+++ b/src/main/java/com/muyu/parsing/ParsingService.java
@@ -0,0 +1,49 @@
+package com.muyu.parsing;
+
+import com.muyu.domain.VehicleData;
+import com.muyu.parsing.service.event.EventCon;
+import lombok.AllArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.springframework.stereotype.Component;
+import org.springframework.stereotype.Service;
+
+import javax.annotation.PostConstruct;
+import java.time.Duration;
+import java.util.Arrays;
+
+/**
+ * @author DongZl
+ * @description: 解析业务
+ * @Date 2023-11-24 下午 02:58
+ */
+@Log4j2
+@Component
+@AllArgsConstructor
+public class ParsingService {
+
+ private final KafkaConsumer consumer;
+
+ @PostConstruct
+ public void init(){
+ new Thread(() -> {
+ while (true){
+ ConsumerRecords records = null;
+ try {
+ records = consumer.poll(Duration.ofMillis(5));
+ for (ConsumerRecord record : records) {
+ VehicleData vehicleData = VehicleData.msgBuild(record.value());
+ EventCon.execute(vehicleData);
+ log.info("offset: [{}]， key:[{}], value:[{}]", record.offset(), record.key(), record.value());
+ }
+ }catch (Exception e){
+ log.info("records： {}", records);
+ log.error(e);
+ }
+ }
+ }).start();
+ }
+
+}
diff --git a/src/main/java/com/muyu/parsing/TestTread.java b/src/main/java/com/muyu/parsing/TestTread.java
new file mode 100644
index 0000000..317c690
--- /dev/null
+++ b/src/main/java/com/muyu/parsing/TestTread.java
@@ -0,0 +1,20 @@
+package com.muyu.parsing;
+
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Random;
+import java.util.UUID;
+
+/**
+ * @author DongZl
+ * @description: 测试线程
+ * @Date 2023-12-11 下午 02:37
+ */
+public class TestTread {
+}
diff --git a/src/main/java/com/muyu/parsing/config/KafkaConsumerConfig.java b/src/main/java/com/muyu/parsing/config/KafkaConsumerConfig.java
new file mode 100644
index 0000000..26d549d
--- /dev/null
+++ b/src/main/java/com/muyu/parsing/config/KafkaConsumerConfig.java
@@ -0,0 +1,34 @@
+package com.muyu.parsing.config;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.List;
+import java.util.Properties;
+
+import static com.muyu.kafka.contents.KafkaContent.KAFKA_CON;
+import static com.muyu.kafka.contents.KafkaContent.TOPIC;
+
+/**
+ * @author DongZl
+ * @description: 解析卡夫卡
+ * @Date 2023-11-24 下午 02:55
+ */
+@Configuration
+public class KafkaConsumerConfig {
+
+ @Bean
+ public KafkaConsumer consumerInit(){
+ Properties props = new Properties();
+ props.put("bootstrap.servers", KAFKA_CON);
+ props.put("group.id", "group01");
+ props.put("enable.auto.commit", "true");
+ props.put("auto.commit.interval.ms", "1000");
+ props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+ props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
+ KafkaConsumer consumer = new KafkaConsumer<>(props);
+ consumer.subscribe(List.of(TOPIC));
+ return consumer;
+ }
+}
diff --git a/src/main/java/com/muyu/parsing/service/EventControl.java b/src/main/java/com/muyu/parsing/service/EventControl.java
new file mode 100644
index 0000000..367037b
--- /dev/null
+++ b/src/main/java/com/muyu/parsing/service/EventControl.java
@@ -0,0 +1,20 @@
+package com.muyu.parsing.service;
+
+import com.muyu.domain.VehicleData;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author DongZl
+ * @description: 事件控制
+ * @Date 2023-11-24 下午 03:05
+ */
+@Service
+public interface EventControl {
+
+ // 注册事件
+ public void registerEvent(String vin, String eventId);
+ // 注销事件
+ public void logoffEvent(String vin, String eventId);
+ // 执行事件
+ public void execute(VehicleData vehicleData);
+}
diff --git a/src/main/java/com/muyu/parsing/service/event/EventCon.java b/src/main/java/com/muyu/parsing/service/event/EventCon.java
new file mode 100644
index 0000000..99ff35c
--- /dev/null
+++ b/src/main/java/com/muyu/parsing/service/event/EventCon.java
@@ -0,0 +1,58 @@
+package com.muyu.parsing.service.event;
+
+import com.muyu.domain.VehicleData;
+import com.muyu.utils.SpringUtils;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * @author DongZl
+ * @description: 事件控制器
+ * @Date 2023-11-26 下午 08:58
+ */
+//@Component
+public class EventCon {
+
+ public static final Map> eventMap = new ConcurrentHashMap<>();
+ // ["event-storage", "event-realTimeTrajectory"]
+
+ /**
+ * 注册事件
+ */
+ public static void registerEvent (String vin, String eventId) {
+ getEvent(vin).add(eventId);
+ }
+
+
+ /**
+ * 注销事件
+ */
+ public static void logoffEvent (String vin, String eventId) {
+ getEvent(vin).remove(eventId);
+ }
+
+ /**
+ * 执行事件
+ * @param vehicleData
+ */
+ public static void execute (VehicleData vehicleData) {
+ List eventList = getEvent(vehicleData.getVin());
+ eventList.forEach(eventId -> {
+ // ["event-storage", "event-realTimeTrajectory"]
+ VehicleEventService vehicleEventService = SpringUtils.getBean(eventId);
+ vehicleEventService.execute(vehicleData);
+ });
+ }
+
+
+ /**
+ * 根据VIN获取事件
+ */
+ public static List getEvent(String vin){
+ return eventMap.computeIfAbsent(vin, k -> new ArrayList<>());
+ }
+}
+
diff --git a/src/main/java/com/muyu/parsing/service/event/VehicleEventService.java b/src/main/java/com/muyu/parsing/service/event/VehicleEventService.java
new file mode 100644
index 0000000..e938390
--- /dev/null
+++ b/src/main/java/com/muyu/parsing/service/event/VehicleEventService.java
@@ -0,0 +1,20 @@
+package com.muyu.parsing.service.event;
+
+import com.muyu.domain.VehicleData;
+
+/**
+ * @author DongZl
+ * @description: 车辆事件
+ * @Date 2023-11-24 下午 03:07
+ */
+public interface VehicleEventService {
+
+ public void execute(VehicleData vehicleData);
+
+ /**
+ * 事件名称
+ * @return
+ */
+ String getEventName ();
+
+}
diff --git a/src/main/java/com/muyu/parsing/service/event/impl/Guzhang.java b/src/main/java/com/muyu/parsing/service/event/impl/Guzhang.java
new file mode 100644
index 0000000..a272ec9
--- /dev/null
+++ b/src/main/java/com/muyu/parsing/service/event/impl/Guzhang.java
@@ -0,0 +1,46 @@
+package com.muyu.parsing.service.event.impl;
+
+import com.muyu.domain.VehicleData;
+import com.muyu.parsing.service.event.VehicleEventService;
+import lombok.extern.log4j.Log4j2;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.stereotype.Service;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Random;
+import java.util.UUID;
+
+/**
+ * @author DongZl
+ * @description:
+ * @Date 2023-12-1 下午 03:04
+ */
+@Log4j2
+@Service("guzhang")
+public class Guzhang implements VehicleEventService {
+ @Override
+ public void execute (VehicleData vehicleData) {
+ log.info("开始故障");
+ try {
+ Thread.sleep(1000);
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ log.info("结束故障");
+ }
+
+ /**
+ * 事件名称
+ *
+ * @return
+ */
+ @Override
+ public String getEventName () {
+ return "guzhang";
+ }
+}
diff --git a/src/main/java/com/muyu/parsing/service/event/impl/RealTimeTrajectoryEventImplService.java b/src/main/java/com/muyu/parsing/service/event/impl/RealTimeTrajectoryEventImplService.java
new file mode 100644
index 0000000..635d27f
--- /dev/null
+++ b/src/main/java/com/muyu/parsing/service/event/impl/RealTimeTrajectoryEventImplService.java
@@ -0,0 +1,71 @@
+package com.muyu.parsing.service.event.impl;
+
+import com.muyu.domain.VehicleData;
+import com.muyu.parsing.service.event.VehicleEventService;
+import lombok.extern.log4j.Log4j2;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.stereotype.Service;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Random;
+import java.util.UUID;
+
+/**
+ * @author DongZl
+ * @description: 实时轨迹
+ * @Date 2023-11-24 下午 03:08
+ */
+@Log4j2
+@Service("event-realTimeTrajectory")
+public class RealTimeTrajectoryEventImplService implements VehicleEventService {
+ @Override
+ public void execute (VehicleData vehicleData) {
+ log.info("开始实时轨迹");
+ String broker = "tcp://fluxmq.muyu.icu:1883", topic = "test",clientId = UUID.randomUUID().toString();
+ int qos = 0;
+ try {
+ MqttClient client = new MqttClient(broker, clientId, new MemoryPersistence());
+ // 连接参数
+ MqttConnectOptions options = new MqttConnectOptions();
+ // 连接
+ client.connect(options);
+ int nextInt = new Random().nextInt(1, 10);
+ log.info("实时轨迹：[{}]次", nextInt);
+ if (nextInt >= 8){
+ Thread.sleep(8 * 10000);
+ }
+ for (int i = 0 ; i < nextInt ; i++) {
+ String content = new SimpleDateFormat("yyyy年MM月dd日 HH时mm分ss秒").format(new Date()) + " - 实时轨迹 MQTT" + clientId;
+ // 创建消息并设置 QoS
+ MqttMessage message = new MqttMessage(content.getBytes());
+ message.setQos(qos);
+ // 发布消息
+ client.publish(topic, message);
+ log.info("[{}] - [{}]", Thread.currentThread().getName(), content);
+ Thread.sleep(100);
+ }
+ // 关闭连接
+ client.disconnect();
+ // 关闭客户端
+ client.close();
+ log.info("结束实时轨迹");
+ } catch (MqttException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * 事件名称
+ *
+ * @return
+ */
+ @Override
+ public String getEventName () {
+ return "event-realTimeTrajectory";
+ }
+}
diff --git a/src/main/java/com/muyu/parsing/service/event/impl/StorageEventServiceImpl.java b/src/main/java/com/muyu/parsing/service/event/impl/StorageEventServiceImpl.java
new file mode 100644
index 0000000..4369982
--- /dev/null
+++ b/src/main/java/com/muyu/parsing/service/event/impl/StorageEventServiceImpl.java
@@ -0,0 +1,71 @@
+package com.muyu.parsing.service.event.impl;
+
+import com.muyu.domain.VehicleData;
+import com.muyu.parsing.service.event.VehicleEventService;
+import lombok.extern.log4j.Log4j2;
+import org.eclipse.paho.client.mqttv3.MqttClient;
+import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
+import org.eclipse.paho.client.mqttv3.MqttException;
+import org.eclipse.paho.client.mqttv3.MqttMessage;
+import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;
+import org.springframework.stereotype.Service;
+
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.Random;
+import java.util.UUID;
+
+/**
+ * @author DongZl
+ * @description: 存储事件
+ * @Date 2023-11-24 下午 03:08
+ */
+@Log4j2
+@Service("event-storage")
+public class StorageEventServiceImpl implements VehicleEventService {
+ @Override
+ public void execute (VehicleData vehicleData) {
+ log.info("开始存储");
+ String broker = "tcp://fluxmq.muyu.icu:1883", topic = "test",clientId = UUID.randomUUID().toString();
+ int qos = 0;
+ try {
+ MqttClient client = new MqttClient(broker, clientId, new MemoryPersistence());
+ // 连接参数
+ MqttConnectOptions options = new MqttConnectOptions();
+ // 连接
+ client.connect(options);
+ int nextInt = new Random().nextInt(1, 10);
+ log.info("本地存储：[{}]次", nextInt);
+ if (nextInt >= 8){
+ Thread.sleep(8 * 10000);
+ }
+ for (int i = 0 ; i < nextInt ; i++) {
+ String content = new SimpleDateFormat("yyyy年MM月dd日 HH时mm分ss秒").format(new Date()) + " - 存储 MQTT" + clientId;
+ // 创建消息并设置 QoS
+ MqttMessage message = new MqttMessage(content.getBytes());
+ message.setQos(qos);
+ // 发布消息
+ client.publish(topic, message);
+ log.info("[{}] - [{}]", Thread.currentThread().getName(), content);
+ Thread.sleep(100);
+ }
+ // 关闭连接
+ client.disconnect();
+ // 关闭客户端
+ client.close();
+ log.info("结束存储");
+ } catch (MqttException | InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * 事件名称
+ *
+ * @return
+ */
+ @Override
+ public String getEventName () {
+ return "event-storage";
+ }
+}
diff --git a/src/main/java/com/muyu/parsing/service/event/impl/ZhiBiaoYuJing.java b/src/main/java/com/muyu/parsing/service/event/impl/ZhiBiaoYuJing.java
new file mode 100644
index 0000000..471bd2d
--- /dev/null
+++ b/src/main/java/com/muyu/parsing/service/event/impl/ZhiBiaoYuJing.java
@@ -0,0 +1,28 @@
+package com.muyu.parsing.service.event.impl;
+
+import com.muyu.domain.VehicleData;
+import com.muyu.parsing.service.event.VehicleEventService;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author DongZl
+ * @description: 指标预警
+ * @Date 2024-3-31 上午 11:14
+ */
+@Service("event-zhibiaoyujing")
+public class ZhiBiaoYuJing implements VehicleEventService {
+ @Override
+ public void execute (VehicleData vehicleData) {
+
+ }
+
+ /**
+ * 事件名称
+ *
+ * @return
+ */
+ @Override
+ public String getEventName () {
+ return "指标预警";
+ }
+}
diff --git a/src/main/java/com/muyu/parsing/storage/LocalStorage.java b/src/main/java/com/muyu/parsing/storage/LocalStorage.java
new file mode 100644
index 0000000..826acc0
--- /dev/null
+++ b/src/main/java/com/muyu/parsing/storage/LocalStorage.java
@@ -0,0 +1,45 @@
+package com.muyu.parsing.storage;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+/**
+ * @author DongZl
+ * @description: 本地通道
+ * @Date 2023-12-12 上午 09:57
+ */
+public class LocalStorage {
+
+ public volatile AtomicBoolean checkQueue = new AtomicBoolean();
+
+ public LinkedBlockingQueue trueQueue = new LinkedBlockingQueue<>();
+
+ public LinkedBlockingQueue falseQueue = new LinkedBlockingQueue<>();
+
+ private LinkedBlockingQueue getQueue(){
+ return checkQueue.get() ? trueQueue : falseQueue;
+ }
+
+ private void checkQueue(){
+ checkQueue.set(!checkQueue.get());
+ }
+
+ /**
+ * 存放
+ */
+ public void push(String data){
+ getQueue().offer(data);
+ }
+ /**
+ * 获取
+ */
+ public String[] getDataList(){
+ LinkedBlockingQueue dataQueue = getQueue();
+ this.checkQueue();
+ return dataQueue.toArray(new String[]{});
+ }
+
+}
diff --git a/src/main/java/com/muyu/utils/FixedThreadPool.java b/src/main/java/com/muyu/utils/FixedThreadPool.java
new file mode 100644
index 0000000..ed9ea1a
--- /dev/null
+++ b/src/main/java/com/muyu/utils/FixedThreadPool.java
@@ -0,0 +1,36 @@
+package com.muyu.utils;
+
+import java.util.concurrent.*;
+
+/**
+ * @author DongZl
+ * @description: 可控最大并发数线程池
+ * @Date 2023-12-5 下午 01:51
+ */
+public class FixedThreadPool {
+
+ /**
+ * 可重用固定个数的线程池
+ */
+ private final static ExecutorService fixedThreadPool =
+ new ThreadPoolExecutor(2,
+ 2,
+ 0L,TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue(45));
+
+
+ /**
+ * 线程池提交任务
+ * @param thread 线程
+ */
+ public static Future submit(Thread thread){
+ return fixedThreadPool.submit(thread);
+ }
+
+ /**
+ * 关闭线程池
+ */
+ public static void shutDown(){
+ fixedThreadPool.shutdown();
+ }
+}
diff --git a/src/main/java/com/muyu/utils/ScheduledThreadPool.java b/src/main/java/com/muyu/utils/ScheduledThreadPool.java
new file mode 100644
index 0000000..3d0d282
--- /dev/null
+++ b/src/main/java/com/muyu/utils/ScheduledThreadPool.java
@@ -0,0 +1,37 @@
+package com.muyu.utils;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author DongZl
+ * @description: 线程池
+ * @Date 2023-11-17 上午 09:16
+ */
+public class ScheduledThreadPool {
+
+ /**
+ * 周期性线程池 CPU 数量 * 2 + 1
+ */
+ private static final ScheduledExecutorService scheduledThreadPool = Executors.newScheduledThreadPool(
+ Runtime.getRuntime().availableProcessors() * 2 + 1);
+
+ public static ScheduledFuture submit (Runnable thread){
+ // 参数分别是： 任务， 多久后开始执行， 每隔多久执行一次（周期），时间单位
+ return submit(thread, 1);
+ }
+
+ public static ScheduledFuture submit (Runnable thread, long period){
+ // 参数分别是： 任务， 多久后开始执行， 每隔多久执行一次（周期），时间单位
+ return scheduledThreadPool.scheduleAtFixedRate(thread, 0, period, TimeUnit.SECONDS);
+ }
+
+ /**
+ * 关闭线程池
+ */
+ public static void shutdown() {
+ scheduledThreadPool.shutdown();
+ }
+}
diff --git a/src/main/java/com/muyu/utils/SpringUtils.java b/src/main/java/com/muyu/utils/SpringUtils.java
new file mode 100644
index 0000000..2f7ba24
--- /dev/null
+++ b/src/main/java/com/muyu/utils/SpringUtils.java
@@ -0,0 +1,114 @@
+package com.muyu.utils;
+
+import org.springframework.aop.framework.AopContext;
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+import org.springframework.beans.factory.config.BeanFactoryPostProcessor;
+import org.springframework.beans.factory.config.ConfigurableListableBeanFactory;
+import org.springframework.stereotype.Component;
+
+/**
+ * spring工具类 方便在非spring管理环境中获取bean
+ *
+ * @author muyu
+ */
+@Component
+public final class SpringUtils implements BeanFactoryPostProcessor {
+ /**
+ * Spring应用上下文环境
+ */
+ private static ConfigurableListableBeanFactory beanFactory;
+
+ /**
+ * 获取对象
+ *
+ * @param name
+ *
+ * @return Object 一个以所给名字注册的bean的实例
+ *
+ * @throws org.springframework.beans.BeansException
+ */
+ @SuppressWarnings("unchecked")
+ public static T getBean (String name) throws BeansException {
+ return (T) beanFactory.getBean(name);
+ }
+
+ /**
+ * 获取类型为requiredType的对象
+ *
+ * @param clz
+ *
+ * @return
+ *
+ * @throws org.springframework.beans.BeansException
+ */
+ public static T getBean (Class clz) throws BeansException {
+ T result = (T) beanFactory.getBean(clz);
+ return result;
+ }
+
+ /**
+ * 如果BeanFactory包含一个与所给名称匹配的bean定义，则返回true
+ *
+ * @param name
+ *
+ * @return boolean
+ */
+ public static boolean containsBean (String name) {
+ return beanFactory.containsBean(name);
+ }
+
+ /**
+ * 判断以给定名字注册的bean定义是一个singleton还是一个prototype。 如果与给定名字相应的bean定义没有被找到，将会抛出一个异常（NoSuchBeanDefinitionException）
+ *
+ * @param name
+ *
+ * @return boolean
+ *
+ * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
+ */
+ public static boolean isSingleton (String name) throws NoSuchBeanDefinitionException {
+ return beanFactory.isSingleton(name);
+ }
+
+ /**
+ * @param name
+ *
+ * @return Class 注册对象的类型
+ *
+ * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
+ */
+ public static Class getType (String name) throws NoSuchBeanDefinitionException {
+ return beanFactory.getType(name);
+ }
+
+ /**
+ * 如果给定的bean名字在bean定义中有别名，则返回这些别名
+ *
+ * @param name
+ *
+ * @return
+ *
+ * @throws org.springframework.beans.factory.NoSuchBeanDefinitionException
+ */
+ public static String[] getAliases (String name) throws NoSuchBeanDefinitionException {
+ return beanFactory.getAliases(name);
+ }
+
+ /**
+ * 获取aop代理对象
+ *
+ * @param invoker
+ *
+ * @return
+ */
+ @SuppressWarnings("unchecked")
+ public static T getAopProxy (T invoker) {
+ return (T) AopContext.currentProxy();
+ }
+
+ @Override
+ public void postProcessBeanFactory (ConfigurableListableBeanFactory beanFactory) throws BeansException {
+ SpringUtils.beanFactory = beanFactory;
+ }
+}
diff --git a/src/main/resources/application.yml b/src/main/resources/application.yml
new file mode 100644
index 0000000..235c984
--- /dev/null
+++ b/src/main/resources/application.yml
@@ -0,0 +1,5 @@
+mqtt:
+ config:
+ broker: tcp://fluxmq.muyu.icu:1883
+ topic: test
+
diff --git a/src/test/java/com/test/GuZhangTest.java b/src/test/java/com/test/GuZhangTest.java
new file mode 100644
index 0000000..3317f3d
--- /dev/null
+++ b/src/test/java/com/test/GuZhangTest.java
@@ -0,0 +1,130 @@
+package com.test;
+
+import com.muyu.MqttApplication;
+import com.muyu.domain.VehicleData;
+import com.muyu.parsing.service.event.VehicleEventService;
+import com.muyu.parsing.service.event.impl.Guzhang;
+import com.muyu.utils.FixedThreadPool;
+import com.muyu.utils.SpringUtils;
+import lombok.extern.log4j.Log4j2;
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.test.context.SpringBootTest;
+
+import java.util.*;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * @author DongZl
+ * @description: 测试
+ * @Date 2023-12-1 下午 03:04
+ */
+@Log4j2
+@SpringBootTest(classes = MqttApplication.class)
+public class GuZhangTest {
+
+ @Autowired
+ private Guzhang guzhang;
+
+ private Map eventServiceMap = new HashMap<>();
+
+ public void initEvent () {
+ eventServiceMap.put("guzhang", SpringUtils.getBean("guzhang"));
+ eventServiceMap.put("event-realTimeTrajectory", SpringUtils.getBean("event-realTimeTrajectory"));
+ eventServiceMap.put("event-storage", SpringUtils.getBean("event-storage"));
+ }
+
+ private Map> vinEventMap = new HashMap<>();
+
+ public void initVinEvent () {
+ List eventList = new ArrayList<>() {{
+ add("guzhang");
+ add("event-realTimeTrajectory");
+ add("event-storage");
+ }};
+ vinEventMap.put("VIN123456798123100", eventList);
+ vinEventMap.put("VIN123456798123101", eventList);
+ vinEventMap.put("VIN123456798123102", eventList);
+ vinEventMap.put("VIN123456798123103", eventList);
+ vinEventMap.put("VIN123456798123104", eventList);
+ vinEventMap.put("VIN123456798123105", eventList);
+ vinEventMap.put("VIN123456798123106", eventList);
+ vinEventMap.put("VIN123456798123107", eventList);
+ vinEventMap.put("VIN123456798123108", eventList);
+ vinEventMap.put("VIN123456798123109", eventList);
+ vinEventMap.put("VIN123456798123110", eventList);
+ vinEventMap.put("VIN123456798123111", eventList);
+ vinEventMap.put("VIN123456798123112", eventList);
+ vinEventMap.put("VIN123456798123113", eventList);
+ vinEventMap.put("VIN123456798123114", eventList);
+ vinEventMap.put("VIN123456798123115", eventList);
+ vinEventMap.put("VIN123456798123116", eventList);
+ vinEventMap.put("VIN123456798123117", eventList);
+ vinEventMap.put("VIN123456798123118", eventList);
+ vinEventMap.put("VIN123456798123119", eventList);
+ vinEventMap.put("VIN123456798123120", eventList);
+ vinEventMap.put("VIN123456798123121", eventList);
+ vinEventMap.put("VIN123456798123122", eventList);
+ vinEventMap.put("VIN123456798123123", eventList);
+ vinEventMap.put("VIN123456798123124", eventList);
+ vinEventMap.put("VIN123456798123125", eventList);
+ vinEventMap.put("VIN123456798123126", eventList);
+ vinEventMap.put("VIN123456798123127", eventList);
+ vinEventMap.put("VIN123456798123128", eventList);
+ vinEventMap.put("VIN123456798123129", eventList);
+ }
+
+ @Test
+ public void ceshi () {
+ this.initEvent();
+ this.initVinEvent();
+ Set vinList = this.vinEventMap.keySet();
+ CountDownLatch vinCountDownLatch = new CountDownLatch(vinList.size());
+ for (String vin : vinList) {
+ // vin - 主线程
+ new Thread(() -> {
+ try {
+ Map eventMap = new HashMap<>();
+ List eventList = this.vinEventMap.get(vin);
+ CountDownLatch vehicleEventCountDownLatch = new CountDownLatch(eventList.size());
+ // event - 子线程
+ eventList.stream().map(eventId -> this.eventServiceMap.get(eventId))
+ .forEach(vehicleEventService -> {
+ String threadName = vin + vehicleEventService.getEventName();
+ Thread thread = new Thread(() -> {
+ vehicleEventService.execute(
+ VehicleData.builder()
+ .vin(vin)
+ .build()
+ );
+ vehicleEventCountDownLatch.countDown();
+ eventMap.remove(Thread.currentThread().getName());
+ }, threadName);
+ FixedThreadPool.submit(thread);
+ eventMap.put(threadName, thread);
+ });
+ boolean isAwait;
+ int awaitSize = 1, maxAwaitSize = 3;
+ do {
+ // true / false -》 latch == 0 返回 true | false latch != 0, 但是等待时间到了
+ isAwait = vehicleEventCountDownLatch.await(1000, TimeUnit.MILLISECONDS);
+ log.info("[{}]等待[{}]次, 线程数量：[{}]", vin, awaitSize, eventMap.size());
+ }while (isAwait || awaitSize++ < maxAwaitSize);
+ eventMap.values().forEach(Thread::interrupt);
+ vinCountDownLatch.countDown();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ System.out.println("单个VIN执行结束");
+ }).start();
+ }
+
+ try {
+ vinCountDownLatch.await();
+ } catch (InterruptedException e) {
+ throw new RuntimeException(e);
+ }
+ System.out.println("整体执行结束");
+ }
+}