fix(): 修复时间处理基础架构

cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/basic/BasicEvent.java

@@ -0,0 +1,37 @@
+package com.muyu.event.process.basic;
+
+import org.springframework.context.ApplicationEvent;
+
+/**
+ * @Author: zi run
+ * @Date 2024/9/30 15:11
+ * @Description 基础事件
+ */
+public class BasicEvent<T> extends ApplicationEvent {
+
+    /**
+     * 事件携带的数据
+     */
+    private final T data;
+
+    /**
+     * 构造函数，初始化事件源和数据
+     *
+     * @param source 事件源对象
+     * @param data 事件携带的数据
+     */
+    public BasicEvent(Object source, T data) {
+        super(source);
+        this.data = data;
+    }
+
+
+    /**
+     * 获取事件携带的数据
+     *
+     * @return 事件数据
+     */
+    public T getData() {
+        return data;
+    }
+}

cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/basic/BasicEventHandler.java

@@ -0,0 +1,37 @@
+package com.muyu.event.process.basic;
+
+import org.springframework.context.ApplicationListener;
+import org.springframework.stereotype.Component;
+
+/**
+ * @Author: zi run
+ * @Date 2024/9/30 15:37
+ * @Description 基础事件处理器
+ */
+@Component
+public class BasicEventHandler<T> implements ApplicationListener<BasicEvent<T>> {
+
+    /**
+     * 具体事件监听器
+     */
+    private final BasicEventListener<T> listener;
+
+    /**
+     * 构造函数，用于注入具体事件监听器
+     *
+     * @param listener 具体事件监听器
+     */
+    public BasicEventHandler(BasicEventListener<T> listener) {
+        this.listener = listener;
+    }
+
+    /**
+     * 处理应用事件
+     *
+     * @param event 事件对象
+     */
+    @Override
+    public void onApplicationEvent(BasicEvent<T> event) {
+        listener.onEvent(event);
+    }
+}

cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/basic/BasicEventListener.java

@@ -0,0 +1,16 @@
+package com.muyu.event.process.basic;
+
+/**
+ * @Author: zi run
+ * @Date 2024/9/30 15:35
+ * @Description 基础事件监听器
+ */
+public interface BasicEventListener<T> {
+
+    /**
+     * 处理事件的方法
+     *
+     * @param event 事件对象
+     */
+    void onEvent(BasicEvent<T> event);
+}

cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/basic/CustomEvent.java

@@ -1,37 +0,0 @@
-package com.muyu.event.process.basic;
-
-import com.alibaba.fastjson2.JSONObject;
-import org.springframework.context.ApplicationEvent;
-
-/**
- * @Author: zi run
- * @Date 2024/9/29 21:19
- * @Description 自定义事件
- */
-public class CustomEvent extends ApplicationEvent {
-
-    /**
-     * 存储与事件相关联的数据
-     */
-    private final JSONObject data;
-
-    /**
-     * 创建一个新的自定义事件
-     *
-     * @param source  事件源，表示触发此事件的对象
-     * @param data    事件携带的数据，以JSON格式存储
-     */
-    public CustomEvent(Object source, JSONObject data) {
-        super(source);
-        this.data = data;
-    }
-
-    /**
-     * 获取与此事件相关联的数据。
-     *
-     * @return 包含事件数据的JSONObject
-     */
-    public JSONObject getData() {
-        return data;
-    }
-}

cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/basic/EventListener.java

@@ -1,18 +0,0 @@
-package com.muyu.event.process.basic;
-
-import org.springframework.context.ApplicationListener;
-
-/**
- * @Author: zi run
- * @Date 2024/9/29 21:29
- * @Description 事件监听基准
- */
-public interface EventListener extends ApplicationListener<CustomEvent> {
-
-    /**
-     * 处理接收到的自定义事件。
-     *
-     * @param event 已发布的自定义事件实例，包含事件的源和相关数据
-     */
-    void onEvent(CustomEvent event);
-}

cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/basic/EventPublisher.java

@@ -1,6 +1,5 @@
 package com.muyu.event.process.basic;
 
-import com.alibaba.fastjson2.JSONObject;
 import org.springframework.context.ApplicationEventPublisher;
 import org.springframework.context.ApplicationEventPublisherAware;
 import org.springframework.stereotype.Component;
@@ -16,24 +15,24 @@ public class EventPublisher implements ApplicationEventPublisherAware {
     /**
      * 应用程序事件发布者，用于发布事件
      */
-    private ApplicationEventPublisher applicationEventPublisher;
+    private ApplicationEventPublisher publisher;
 
     /**
-     * 设置应用程序事件发布者。
+     * 设置应用程序事件发布者
      *
-     * @param applicationEventPublisher 应用程序事件发布者实例
+     * @param publisher 应用程序事件发布者实例
      */
     @Override
-    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
+    public void setApplicationEventPublisher(ApplicationEventPublisher publisher) {
-        this.applicationEventPublisher = applicationEventPublisher;
+        this.publisher = publisher;
     }
 
     /**
-     * 发布自定义事件。
+     * 发布事件
-     *
+     * @param event 要发布的事件
-     * @param messages 事件携带的数据，以JSON格式传递
+     * @param <T> 事件数据类型
      */
-    public void publishEvent(JSONObject messages) {
+    public <T> void publish(BasicEvent<T> event) {
-        applicationEventPublisher.publishEvent(new CustomEvent(this, messages));
+        publisher.publishEvent(event);
     }
 }

cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/config/EventListenerConfig.java

@@ -1,19 +0,0 @@
-package com.muyu.event.process.config;
-
-import com.muyu.event.process.listener.AddDatabaseListener;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-/**
- * @Author: zi run
- * @Date 2024/9/29 21:29
- * @Description 事件监听配置
- */
-@Configuration
-public class EventListenerConfig {
-
-    @Bean
-    public AddDatabaseListener addDatabaseListener() {
-       return new AddDatabaseListener();
-    }
-}

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

@@ -1,24 +1,24 @@
 package com.muyu.event.process.consumer;
 
-import cn.hutool.core.thread.ThreadUtil;
-import com.alibaba.fastjson2.JSONObject;
-import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
 import com.muyu.event.process.basic.EventPublisher;
+import com.muyu.event.process.event.IoTDBInsertDataEvent;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.springframework.beans.factory.InitializingBean;
 import org.springframework.boot.ApplicationArguments;
 import org.springframework.boot.ApplicationRunner;
-import org.springframework.scheduling.annotation.Async;
+import org.springframework.context.ApplicationListener;
+import org.springframework.context.event.ContextClosedEvent;
 import org.springframework.stereotype.Component;
 
 import java.time.Duration;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.List;
-import java.util.stream.Collectors;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 /**
  * @Author: zi run
@@ -28,7 +28,7 @@ import java.util.stream.Collectors;
 @Slf4j
 @Component
 @RequiredArgsConstructor
-public class VehicleConsumer implements InitializingBean {
+public class VehicleConsumer implements ApplicationRunner, ApplicationListener<ContextClosedEvent> {
 
     /**
      * kafka消费者
@@ -43,40 +43,51 @@ public class VehicleConsumer implements InitializingBean {
     /**
      * 协议解析报文传递数据(队列名称)
      */
-    public final static String MESSAGE_PARSING = "test-topic";
+    public final static String MESSAGE_PARSING = "MessageParsing";
 
-//    @Override
+    /**
-//    public void run(ApplicationArguments args) throws Exception {
+     * 设定固定大小的线程池，线程数量与当前可用的处理器核心数相同
-//        log.info("开始监听kafka-topic：{}", MESSAGE_PARSING);
+     */
-//        List<String> topicList = Collections.singletonList(MESSAGE_PARSING);
+    private final ExecutorService executorService =
-//        kafkaConsumer.subscribe(topicList);
+            Executors.newFixedThreadPool(10);
-//
-//        while (true) {
-//            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(100));
-//            consumerRecords.forEach(record -> {
-//                String value = record.value();
-//                log.info("接收到车辆报文数据，内容：{}", value);
-//                eventPublisher.publishEvent(JSONObject.parseObject(value));
-//            });
-//        }
-//    }
 
-    @Async
     @Override
-    public void afterPropertiesSet() throws Exception {
+    public void run(ApplicationArguments args) throws Exception {
-        new Thread(() -> {
         log.info("启动线程监听Topic: {}", MESSAGE_PARSING);
-            ThreadUtil.sleep(100);
+        List<String> topics = Collections.singletonList(MESSAGE_PARSING);
-            Collection<String> topics = Lists.newArrayList(MESSAGE_PARSING);
         kafkaConsumer.subscribe(topics);
         while (true) {
             ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(100));
-                consumerRecords.forEach(consumerRecord -> {
+            consumerRecords.forEach(consumerRecord -> executorService.submit(() -> handleRecord(consumerRecord)));
+        }
+    }
+
+    private void handleRecord(ConsumerRecord<String, String> consumerRecord) {
         String message = consumerRecord.value();
         log.info("接收到车辆报文数据，内容：{}", message);
-                    eventPublisher.publishEvent(JSONObject.parseObject(message));
+        log.info("------------------------------------------------");
-                });
+        eventPublisher.publish(new IoTDBInsertDataEvent(this, message));
+    }
+
+    @Override
+    public void onApplicationEvent(ContextClosedEvent event) {
+        log.info("关闭线程池和Kafka消费者");
+
+        try {
+            executorService.shutdown();
+            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
+                executorService.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            log.error("线程池关闭被中断，强制关闭", e);
+            executorService.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+
+        try {
+            kafkaConsumer.close(); // 关闭Kafka消费者
+        } catch (Exception e) {
+            log.error("关闭Kafka消费者时发生错误", e);
         }
-        }).start();
     }
 }

cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/controller/TestEventController.java

@@ -50,9 +50,9 @@ public class TestEventController extends BaseController {
     @GetMapping(value = "/sendKafka")
     public Result<String> senKafka() {
         JSONObject jsonObject = new JSONObject();
-        jsonObject.put("id",1);
+        jsonObject.put("id","1");
         jsonObject.put("name","张三");
-        jsonObject.put("age",18);
+        jsonObject.put("age","18");
         jsonObject.put("sex","男");
         ProducerRecord<String, String> producerRecord = new ProducerRecord<>(kafkaTopicName, jsonObject.toJSONString());
         kafkaProducer.send(producerRecord);

cloud-modules/cloud-modules-event-process/src/main/java/com/muyu/event/process/event/IoTDBInsertDataEvent.java

@@ -0,0 +1,20 @@
+package com.muyu.event.process.event;
+
+import com.muyu.event.process.basic.BasicEvent;
+
+/**
+ * @Author: zi run
+ * @Date 2024/9/29 21:19
+ * @Description 向IoTDB插入数据事件
+ */
+public class IoTDBInsertDataEvent extends BasicEvent<String> {
+
+    /**
+     * 构造函数，向IoTDB插入数据创建事件
+     *
+     * @param messsge 消息
+     */
+    public IoTDBInsertDataEvent(Object source, String messsge) {
+        super(source, messsge);
+    }
+}

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

@@ -1,43 +0,0 @@
-package com.muyu.event.process.listener;
-
-import com.alibaba.fastjson2.JSONObject;
-import com.muyu.event.process.basic.CustomEvent;
-import com.muyu.event.process.basic.EventListener;
-import com.muyu.event.process.iotdb.basic.service.IService;
-import com.muyu.event.process.iotdb.service.IoTDBService;
-import lombok.RequiredArgsConstructor;
-import org.springframework.beans.factory.annotation.Autowired;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * @Author: zi run
- * @Date 2024/9/29 22:12
- * @Description 添加数据库事件监听器
- */
-public class AddDatabaseListener implements EventListener {
-
-    /**
-     * IoTDB业务层
-     */
-    @Autowired
-    private IoTDBService ioTDBService;
-
-    @Override
-    public void onEvent(CustomEvent event) {
-        JSONObject data = event.getData();
-        List<String> keyList = new ArrayList<>();
-        List<String> valueList = new ArrayList<>();
-        data.forEach((key, value) -> {
-            keyList.add(key);
-            valueList.add((String) value);
-        });
-        ioTDBService.insertStringRecord("root.vehicle", System.currentTimeMillis(), keyList, valueList);
-    }
-
-    @Override
-    public void onApplicationEvent(CustomEvent event) {
-        onEvent(event);
-    }
-}

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

@@ -0,0 +1,68 @@
+package com.muyu.event.process.listener;
+
+import com.alibaba.fastjson2.JSONObject;
+import com.muyu.event.process.basic.BasicEvent;
+import com.muyu.event.process.basic.BasicEventListener;
+import com.muyu.event.process.event.IoTDBInsertDataEvent;
+import com.muyu.event.process.iotdb.service.IoTDBService;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+
+/**
+ * @Author: zi run
+ * @Date 2024/9/29 22:12
+ * @Description 向IoTDB插入数据事件监听器
+ */
+@Component
+@RequiredArgsConstructor
+public class IoTDBInsertDataListener implements BasicEventListener<String> {
+
+    /**
+     * IoTDB业务层
+     */
+    private final IoTDBService ioTDBService;
+
+    /**
+     * 设备名（表名）
+     */
+    private static final String DEVICE_ID = "root.vehicle";
+
+    /**
+     * 处理接收到的事件，将数据插入到 IoTDB
+     *
+     * @param event 接收到的事件，包含需要插入的数据
+     */
+    @Override
+    public void onEvent(BasicEvent<String> event) {
+        JSONObject data = JSONObject.parseObject(event.getData());
+        List<String> keyList = extractKeys(data);
+        List<String> valueList = extractValues(data);
+        ioTDBService.insertStringRecord(DEVICE_ID, System.currentTimeMillis(), keyList, valueList);
+    }
+
+    /**
+     * 从给定的JSONObject中提取所有的键
+     *
+     * @param data 要提取键的JSONObject
+     * @return 键的列表
+     */
+    private List<String> extractKeys(JSONObject data) {
+        return data.keySet().stream().collect(Collectors.toList());
+    }
+
+    /**
+     * 从给定的 JSONObject 中提取所有的值，并将其转换为字符串
+     *
+     * @param data 要提取值的JSONObject
+     * @return 值的列表，以字符串形式表示
+     */
+    private List<String> extractValues(JSONObject data) {
+        return data.values().stream()
+                .map(Object::toString)
+                .collect(Collectors.toList());
+    }
+}