fix():重新构建数据处理结构

cloud-common/cloud-common-cache/src/main/java/com/muyu/common/domain/InformationData.java

@@ -1,11 +1,17 @@
 package com.muyu.common.domain;
 
+import com.muyu.common.system.domain.SysDept;
 import com.muyu.domain.SysCar;
+import com.muyu.domain.WarnLogs;
+import com.muyu.domain.resp.FaultConditionTotalListResp;
+import com.muyu.domain.resp.FenceGroupResp;
 import lombok.AllArgsConstructor;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
 
+import java.util.List;
+
 /**
  * @Author WangXin
  * @Data 2024/9/30
@@ -17,5 +23,24 @@ import lombok.experimental.SuperBuilder;
 @AllArgsConstructor
 @NoArgsConstructor
 public class InformationData {
+    /**
+     * 车辆基本信息
+     */
     private SysCar sysCar;
+    /**
+     * 公司信息
+     */
+    private SysDept sysDept;
+    /**
+     * 车辆电子绑定电子围栏信息
+     */
+    private FenceGroupResp fenceGroupResp;
+    /**
+     * 故障规则数据总数列表
+     */
+    private FaultConditionTotalListResp faultConditionTotalListResp;
+    /**
+     * 预警日志对象
+     */
+    private List<WarnLogs> warnLogsList;
 }

cloud-common/cloud-common-caffeine/pom.xml

@@ -22,10 +22,16 @@
     </properties>
 
     <dependencies>
+
         <dependency>
             <groupId>com.github.ben-manes.caffeine</groupId>
             <artifactId>caffeine</artifactId>
         </dependency>
+
+        <dependency>
+            <groupId>com.muyu</groupId>
+            <artifactId>cloud-common-redis</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/bean/CaffeineManager.java

@@ -0,0 +1,44 @@
+package com.muyu.common.caffeine.bean;
+
+import com.github.benmanes.caffeine.cache.Caffeine;
+import com.muyu.common.caffeine.enums.CacheNameEnums;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.cache.CacheManager;
+import org.springframework.cache.caffeine.CaffeineCache;
+import org.springframework.cache.support.SimpleCacheManager;
+import org.springframework.context.annotation.Bean;
+import org.springframework.stereotype.Component;
+
+import java.util.List;
+
+/**
+ * Caffeine管理器
+ * @Author: WangXin
+ * @Name: CaffeineCacheConfig
+ * @Description: Caffeine管理器
+ */
+
+@Slf4j
+@Component
+public class CaffeineManager {
+
+    /**
+     * 创建缓存管理器
+     * @return  缓存管理器实例
+     */
+    @Bean
+    public CacheManager cacheManager() {
+        SimpleCacheManager cacheManager = new SimpleCacheManager();
+        List<String> cacheNames = CacheNameEnums.getCodes();
+        cacheManager.setCaches(cacheNames.stream()
+                .map(name -> new CaffeineCache(
+                        name,
+                        Caffeine.newBuilder()
+                                .recordStats()
+                                .build()))
+                .toList());
+        log.info("缓存管理器初始化完成,缓存分区:{}", cacheNames);
+        return cacheManager;
+    }
+
+}

cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/constents/CaffeineContent.java

@@ -0,0 +1,15 @@
+package com.muyu.common.caffeine.constents;
+
+/**
+ * Caffeine常量
+ * @Author: WangXin
+ * @Name: CaffeineContent
+ * @Description: Caffeine常量
+ */
+
+public class CaffeineContent {
+
+    public static final String CAR_VIN_KEY = "car:vin";
+
+    public static final String VIN = "vin";
+}

cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/enums/CacheNameEnums.java

@@ -0,0 +1,66 @@
+package com.muyu.common.caffeine.enums;
+
+import lombok.Getter;
+
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * 缓存分区枚举
+ *
+ * @Author: WangXin
+ * @Name: CacheNameEnums
+ * @Description: 缓存分区枚举
+ */
+
+@Getter
+public enum CacheNameEnums {
+    STORAGE("storage", "持久化"),
+    FAULT("fault", "故障"),
+    FENCE("fence", "围栏"),
+    WARMING("warming", "预警"),
+    REALTIME("realTime", "实时信息");
+
+    private final String code;
+    private final String info;
+
+    CacheNameEnums(String code, String info) {
+        this.code = code;
+        this.info = info;
+    }
+
+    /**
+     * 鉴别参数是否是枚举的值
+     *
+     * @param code 需鉴别参数
+     * @return 如果存在返回结果turn, 否则返回false
+     */
+    public static boolean isCode(String code) {
+        return Arrays.stream(values())
+                .map(CacheNameEnums::getCode)
+                .anyMatch(c -> c.equals(code));
+    }
+
+    /**
+     * 获取枚举Value
+     * @param code 编码
+     * @return  Value
+     */
+    public static String getInfo(String code) {
+        return Arrays.stream(values())
+                .filter(c -> c.getCode().equals(code))
+                .map(CacheNameEnums::getInfo)
+                .findFirst()
+                .orElse("");
+    }
+
+    /**
+     * 获取所有code
+     * @return code集合
+     */
+    public static List<String> getCodes() {
+        return Arrays.stream(values())
+                .map(CacheNameEnums::getCode)
+                .toList();
+    }
+}

cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/utils/CaffeineCacheUtils.java

@@ -0,0 +1,89 @@
+package com.muyu.common.caffeine.utils;
+
+
+import com.muyu.common.caffeine.enums.CacheNameEnums;
+import jakarta.annotation.Resource;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.ObjectUtils;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.cache.CacheManager;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+/**
+ * Caffeine缓存工具
+ * @Author: WangXin
+ * @Name: CaffeineUtils
+ * @Description: 缓存工具类
+ * @CreatedDate: 2024/9/26 下午2:53
+ * @FilePath: com.muyu.common.caffeine
+ */
+@Slf4j
+@Component
+public class CaffeineCacheUtils {
+    @Resource
+    private CacheManager cacheManager;
+    @Resource
+    private RedisTemplate<String, String> redisTemplate;
+
+
+    /**
+     * 车辆上线 - 新增缓存
+     */
+    public void addCarCache(String vin) {
+        // 从Redis中获取缓存信息
+        for (String name : CacheNameEnums.getCodes()) {
+            String value = redisTemplate.opsForValue().get(name+":"+vin);
+            cacheManager.getCache(name).put(vin, value);
+            log.info("存储缓存, 缓存分区:[{}], 车辆编码:[{}], 存储值:[{}]", name, vin, value);
+        }
+        log.info("车辆编码:{},本地缓存完成...",vin);
+    }
+
+    /**
+     * 车辆下线 - 删除缓存
+     */
+    public void deleteCarCache(String cacheName) {
+        if (!hasCarVinCache(cacheName,null)) {
+            log.warn("车辆编码:{},本地缓存不存在该车辆信息...", cacheName);
+            return;
+        }
+        cacheManager.getCache(cacheName).invalidate();
+        log.info("车辆编码:{},本地缓存删除完成...", cacheName);
+    }
+
+    /**
+     * 获取车辆信息缓存
+     */
+    public Object getCarCache(String cacheName, String key) {
+        if (!hasCarVinCache(cacheName, key)){
+            log.warn("车辆编码:{},本地缓存不存在该车辆信息...",cacheName);
+            return null;
+        }
+        return cacheManager.getCache(cacheName).get(key).get();
+    }
+
+    /**
+     * 获取车辆信息缓存
+     */
+    public <T> T getCarCache(String cacheName, String key, Class<T> type) {
+        if (!hasCarVinCache(cacheName,key)){
+            log.warn("车辆编码:{},本地缓存不存在该车辆信息...",cacheName);
+            return null;
+        }
+        return cacheManager.getCache(cacheName).get(key, type);
+    }
+
+    /**
+     * 判断缓存存在与否
+     */
+    public Boolean hasCarVinCache(String cacheName,String key) {
+        boolean notEmpty = ObjectUtils.isNotEmpty(cacheManager.getCache(cacheName));
+        if (notEmpty && StringUtils.isNotEmpty(key)){
+            return ObjectUtils.isNotEmpty(cacheManager.getCache(cacheName).get(key).get());
+        }
+        return notEmpty;
+
+    }
+
+}

cloud-common/cloud-common-caffeine/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1,2 @@
+com.muyu.common.caffeine.utils.CaffeineCacheUtils
+com.muyu.common.caffeine.bean.CaffeineManager

cloud-common/cloud-common-core/src/main/java/com/muyu/common/core/exception/ServiceException.java

@@ -5,7 +5,7 @@ package com.muyu.common.core.exception;
  *
  * @author muyu
  */
-public final class ServiceException extends RuntimeException {
+public class ServiceException extends RuntimeException {
     private static final long serialVersionUID = 1L;
 
     /**

cloud-common/cloud-common-iotdb/pom.xml

@@ -34,11 +34,6 @@
             <groupId>org.projectlombok</groupId>
             <artifactId>lombok</artifactId>
         </dependency>
-        <dependency>
-            <groupId>org.projectlombok</groupId>
-            <artifactId>lombok</artifactId>
-            <scope>provided</scope>
-        </dependency>
     </dependencies>
 
 </project>

cloud-common/cloud-common-iotdb/src/main/java/com/muyu/common/iotdb/config/IotDBSessionConfig.java

@@ -12,6 +12,7 @@ import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
 import org.apache.iotdb.tsfile.read.common.Field;
 import org.apache.iotdb.tsfile.read.common.RowRecord;
 import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.stereotype.Component;
 
@@ -84,6 +85,12 @@ public class IotDBSessionConfig {
 
     }
 
+    /**
+     * sql查数据
+     * @param sessionPool
+     * @param sql
+     * @return
+     */
     public SessionDataSet selectRecord(SessionPool sessionPool,String sql) {
         log.info("iotdb数据查询：sql:[{}]",sql);
         SessionDataSetWrapper sessionDataSetWrapper = null;

cloud-common/cloud-common-iotdb/src/main/java/com/muyu/common/iotdb/constant/IotdbConstant.java

@@ -9,5 +9,6 @@ package com.muyu.common.iotdb.constant;
 public interface IotdbConstant {
 
     String ROOT_DATA_DATAJSON = "root.car.data.datajson";
+
     String SELECT_ROOT_DATA_DATAJSON_DATASOURCE = "select * from root.car.data.datajson";
 }

cloud-common/cloud-common-iotdb/src/main/java/com/muyu/common/iotdb/domain/EventActuate.java

@@ -1,29 +0,0 @@
-package com.muyu.common.iotdb.domain;
-
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-import lombok.experimental.SuperBuilder;
-
-import java.util.List;
-
-/**
- * @Author WangXin
- * @Data 2024/9/29
- * @Description 事件驱动对象
- * @Version 1.0.0
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@NoArgsConstructor
-public class EventActuate {
-    /**
-     * json数据
-     */
-    private String jsonData;
-    /**
-     * 事件驱动key集合
-     */
-    private List<String> eventKeys;
-}

cloud-common/cloud-common-iotdb/src/main/java/com/muyu/common/iotdb/domain/ResultEntity.java

@@ -1,20 +0,0 @@
-package com.muyu.common.iotdb.domain;
-
-
-import com.muyu.common.iotdb.domain.dto.IotDbRecordAble;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-
-@EqualsAndHashCode(callSuper = true)
-@Data
-public class ResultEntity extends IotDbRecordAble {
-
-    private Float temperature;
-
-    private String hardware;
-
-    private Boolean status;
-
-    private String time;
-
-}

cloud-common/cloud-common-iotdb/src/main/java/com/muyu/common/iotdb/domain/TestDataType.java

@@ -1,18 +0,0 @@
-package com.muyu.common.iotdb.domain;
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-@Data
-@Builder
-@AllArgsConstructor
-@NoArgsConstructor
-public class TestDataType {
-    private Float temperature;
-    private String hardware;
-    private Boolean status;
-    private Double testDouble;
-    private Long testLong;
-}

cloud-common/cloud-common-iotdb/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -1 +1 @@
-com.muyu.data.iotdb.config.IotDBSessionConfig
+com.muyu.common.iotdb.config.IotDBSessionConfig

cloud-common/cloud-common-kafka/src/main/java/com/muyu/common/kafka/constant/KafkaConstants.java

@@ -0,0 +1,14 @@
+package com.muyu.common.kafka.constant;
+
+/**
+ * @Author: WangXin
+ * @date: 2024/7/10
+ * @Description: kafka常量
+ * @Version 1.0.0
+ */
+public class KafkaConstants {
+
+    public final static String KafkaTopic = "kafka_topic";
+
+    public final static String KafkaGrop = "kafka_grop";
+}

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/config/ConfirmCallbackConfig.java

@@ -0,0 +1,48 @@
+package com.muyu.common.rabbit.config;
+
+import lombok.extern.log4j.Log4j2;
+import org.springframework.amqp.rabbit.connection.CorrelationData;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.PostConstruct;
+
+/**
+ * 消息发送确认配置
+ * 消息发送到交换机的回调
+ */
+@Component
+@Log4j2
+public class ConfirmCallbackConfig implements RabbitTemplate.ConfirmCallback {
+
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
+
+    /**
+     * @PostContruct是spring框架的注解，在⽅法上加该注解会在项⽬启动的时候执⾏该⽅法，也可以理解为在spring容器初始化的时候执
+     */
+    @PostConstruct
+    public void init() {
+        rabbitTemplate.setConfirmCallback(this);
+    }
+
+    /**
+     * 交换机不管是否收到消息的一个回调方法
+     *
+     * @param correlationData 消息相关数据
+     * @param ack             交换机是否收到消息
+     * @param cause           失败原因
+     */
+    @Override
+    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
+        if (!ack) {
+            String exchange = correlationData.getReturned().getExchange();
+            String message = correlationData.getReturned().getMessage().getBody().toString();
+            // 发送异常
+            log.error("消息：{}，发送到交换机：{}失败，原因是：{}", message, exchange, cause);
+            // TODO 可以把异常信息 以及 消息的内容直接添加到 MYSQL
+        }
+    }
+
+}

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/config/DelayedQueueConfig.java

@@ -1,83 +0,0 @@
-package com.muyu.common.rabbit.config;
-
-
-import com.muyu.common.rabbit.constants.RabbitmqConstants;
-import org.springframework.amqp.core.*;
-import org.springframework.amqp.rabbit.core.RabbitAdmin;
-import org.springframework.beans.factory.annotation.Qualifier;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-import javax.annotation.Resource;
-import java.util.HashMap;
-
-/**
- * @ClassName: DelayedQueueConfig
- * @Description: 延迟队列配置类
- */
-@Configuration
-public class DelayedQueueConfig {
-
-
-    @Resource
-    private RabbitAdmin rabbitAdmin;
-
-    /**
-     * 声明队列
-     * @return 返回队列
-     */
-    @Bean
-    public Queue delayedQueue() {
-        Queue queue = new Queue(RabbitmqConstants.DELAYED_QUEUE_NAME);
-        rabbitAdmin.declareQueue(queue);
-        return queue;
-    }
-
-    /**
-     * 声明交换机
-     * @return 返回交换机
-     */
-    @Bean
-    public Exchange delayedExchange() {
-        HashMap<String, Object> arguments = new HashMap<>(3);
-
-        arguments.put("x-delayed-type", "direct");
-
-        /**
-         * 声明自定义交换机
-         * 第一个参数：交换机的名称
-         * 第二个参数：交换机的类型
-         * 第三个参数：是否需要持久化
-         * 第四个参数：是否自动删除
-         * 第五个参数：其他参数
-         */
-        CustomExchange customExchange = new CustomExchange(
-                RabbitmqConstants.DELAYED_EXCHANGE_NAME,
-                "x-delayed-message",
-                true,
-                false,
-                arguments);
-        rabbitAdmin.declareExchange(customExchange);
-        return customExchange;
-    }
-
-    /**
-     * 绑定交换机
-     * @param delayedQueue 队列对象
-     * @param delayedExchange 交换机对象
-     */
-    @Bean
-    public Binding delayedQueueBindingDelayedExchange(
-            @Qualifier("delayedQueue") Queue delayedQueue,
-            @Qualifier("delayedExchange") Exchange delayedExchange) {
-
-        Binding noargs = BindingBuilder.bind(delayedQueue)
-                .to(delayedExchange)
-                .with(RabbitmqConstants.DELAYED_ROUTING_KEY)
-                .noargs();
-        rabbitAdmin.declareBinding(noargs);
-        return noargs;
-    }
-
-
-}

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/config/MyConfirmCallback.java

@@ -1,47 +0,0 @@
-package com.muyu.common.rabbit.config;
-
-import lombok.AllArgsConstructor;
-import org.springframework.amqp.rabbit.connection.CorrelationData;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.stereotype.Component;
-
-import javax.annotation.PostConstruct;
-
-/**
- * @ClassName:
- * @Description: 消息发送到 交换机的确认  回调方法
- */
-@Component
-@AllArgsConstructor
-public class MyConfirmCallback implements RabbitTemplate.ConfirmCallback {
-
-
-    private RabbitTemplate rabbitTemplate;
-
-//    public MyConfirmCallback(RabbitTemplate rabbitTemplate) {
-//        this.rabbitTemplate = rabbitTemplate;
-//        // 设置 消息发送到交换机成功 的回调
-//        this.rabbitTemplate.setConfirmCallback(this);
-//    }
-
-    @PostConstruct
-    public void init() {
-        this.rabbitTemplate.setConfirmCallback(this);
-    }
-
-    /**
-     * 发送消息到交换机的回调方法  消息发送成功或者失败都会执行
-     *
-     * @param correlationData correlation data for the callback.  消息的元数据
-     * @param ack             true for ack, false for nack
-     * @param cause           An optional cause, for nack, when available, otherwise null.
-     */
-    @Override
-    public void confirm(CorrelationData correlationData, boolean ack, String cause) {
-        if (ack) {
-            System.out.println("消息发送到交换机成功~");
-        } else {
-            System.out.println("消息发送到交换机失败，失败的原因：" + cause);
-        }
-    }
-}

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/config/RabbitAdminConfig.java

@@ -1,49 +0,0 @@
-package com.muyu.common.rabbit.config;
-
-
-
-import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
-import org.springframework.amqp.rabbit.connection.ConnectionFactory;
-import org.springframework.amqp.rabbit.core.RabbitAdmin;
-import org.springframework.beans.factory.annotation.Value;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-/**
- * @ClassName: RabbitAdminConfig
- * @Description: RabbitAdmin配置类
- */
-@Configuration
-public class RabbitAdminConfig {
-    @Value("${spring.rabbitmq.host}")
-    private String host;
-    @Value("${spring.rabbitmq.username}")
-    private String username;
-    @Value("${spring.rabbitmq.password}")
-    private String password;
-    @Value("${spring.rabbitmq.virtualhost}")
-    private String virtualHost;
-
-    @Bean
-    public ConnectionFactory connectionFactory() {
-        CachingConnectionFactory cachingConnectionFactory = new CachingConnectionFactory();
-        cachingConnectionFactory.setHost(host);
-        cachingConnectionFactory.setUsername(username);
-        cachingConnectionFactory.setPassword(password);
-        cachingConnectionFactory.setVirtualHost(virtualHost);
-        return cachingConnectionFactory;
-
-    }
-
-    @Bean
-    public RabbitAdmin rabbitAdmin(ConnectionFactory connectionFactory) {
-        RabbitAdmin rabbitAdmin = new RabbitAdmin(connectionFactory);
-        rabbitAdmin.setAutoStartup(true);
-        return rabbitAdmin;
-    }
-
-}
-
-
-
-

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/config/RabbitListenerConfigurer.java

@@ -0,0 +1,42 @@
+package com.muyu.common.rabbit.config;
+
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.listener.RabbitListenerEndpointRegistrar;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.messaging.converter.MappingJackson2MessageConverter;
+import org.springframework.messaging.handler.annotation.support.DefaultMessageHandlerMethodFactory;
+
+import javax.annotation.Resource;
+
+@Configuration
+public class RabbitListenerConfigurer implements org.springframework.amqp.rabbit.annotation.RabbitListenerConfigurer {
+
+    static {
+        System.setProperty("spring.amqp.deserialization.trust.all", "true");
+    }
+
+    //以下配置RabbitMQ消息服务
+    @Resource
+    public ConnectionFactory connectionFactory;
+
+
+    /**
+     * 处理器方法工厂
+     * @return
+     */
+    @Bean
+    public DefaultMessageHandlerMethodFactory handlerMethodFactory() {
+        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
+        // 这里的转换器设置实现了 通过 @Payload 注解 自动反序列化message body
+        factory.setMessageConverter(new MappingJackson2MessageConverter());
+        return factory;
+    }
+
+    @Override
+    public void configureRabbitListeners(RabbitListenerEndpointRegistrar rabbitListenerEndpointRegistrar) {
+        rabbitListenerEndpointRegistrar.setMessageHandlerMethodFactory(handlerMethodFactory());
+    }
+
+}
+

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/config/RabbitmqConfig.java

@@ -1,20 +0,0 @@
-package com.muyu.common.rabbit.config;
-
-import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
-import org.springframework.amqp.support.converter.MessageConverter;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-/**
- * JSON  消息转换器  自动将发送的消息转换成 json 字符串 并且 消费者接收到消息的时候自动反序列化 成需要的对象
- */
-@Configuration
-public class RabbitmqConfig {
-
-
-    // 消息转换配置
-    @Bean
-    public MessageConverter jsonMessageConverter() {
-        return new Jackson2JsonMessageConverter();
-    }
-}

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/config/ReturnCallbackConfig.java

@@ -1,8 +1,9 @@
 package com.muyu.common.rabbit.config;
 
-import lombok.AllArgsConstructor;
+import lombok.extern.log4j.Log4j2;
 import org.springframework.amqp.core.ReturnedMessage;
 import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
 import javax.annotation.PostConstruct;
@@ -11,27 +12,30 @@ import javax.annotation.PostConstruct;
  * 消息发送到队列的确认
  */
 @Component
-@AllArgsConstructor
+@Log4j2
 public class ReturnCallbackConfig implements RabbitTemplate.ReturnsCallback {
 
+    @Autowired
+    private RabbitTemplate rabbitTemplate;
 
-    private final RabbitTemplate rabbitTemplate;
+    /**
-
+     * @PostContruct是spring框架的注解，在⽅法上加该注解会在项⽬启动的时候执⾏该⽅法，也可以理解为在spring容器初始化的时候执
-    @PostConstruct // @PostContruct是spring框架的注解，在⽅法上加该注解会在项⽬启动的时候执⾏该⽅法，也可以理解为在spring容器初始化的时候执
+     */
+    @PostConstruct
     public void init() {
         rabbitTemplate.setReturnsCallback(this);
     }
 
     /**
-     * 消息发送到 队列失败的时候执行
+     * 消息发送失败 则会执行这个方法
      *
      * @param returnedMessage the returned message and metadata.
      */
     @Override
     public void returnedMessage(ReturnedMessage returnedMessage) {
-        System.out.println("消息" + returnedMessage.getMessage().toString() +
+        log.error("消息：{}，被交换机：{} 回退！退回原因为：{}",
-                "被交换机" + returnedMessage.getExchange() + "回退！"
+                returnedMessage.getMessage().toString(), returnedMessage.getExchange(), returnedMessage.getReplyText());
-                + "退回原因为：" + returnedMessage.getReplyText());
+        // TODO 回退了所有的信息，可做补偿机制
-        // 回退了所有的信息，可做补偿机制  记录发送的日志
     }
+
 }

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/config/TopicConfig.java

@@ -1,75 +0,0 @@
-package com.muyu.common.rabbit.config;
-
-import org.springframework.amqp.core.*;
-import org.springframework.context.annotation.Bean;
-import org.springframework.context.annotation.Configuration;
-
-/**
- * @Author: WangXin
- * @Time: 2024/4/22 11:55
- * @Description: 主题模式配置
- */
-@Configuration
-public class TopicConfig {
-
-    /**
-     * 主题模式交换机
-     * @return exchange
-     */
-    @Bean(name = "topicExchange")
-    public Exchange getTopicExchange(){
-        return ExchangeBuilder
-                .topicExchange("exchange_topic")
-                .build();
-    }
-
-    /**
-     * 主题队列 01
-     * @return queue
-     */
-    @Bean(name = "topicQueue01")
-    public Queue getTopicQueue01(){
-        return QueueBuilder
-                .durable("queue_topic_01")
-                .build();
-    }
-
-    /**
-     * 主题队列 02
-     * @return queue
-     */
-    @Bean(name = "topicQueue02")
-    public Queue getTopicQueue02(){
-        return QueueBuilder
-                .durable("queue_topic_02")
-                .build();
-    }
-
-    /**
-     * 绑定队列 01
-     * @return binding
-     */
-    @Bean
-    public Binding getTopicBinding01(){
-        return BindingBuilder
-                .bind(getTopicQueue01())
-                .to(getTopicExchange())
-                //路由键 队列1接收debug级别的消息
-                .with("front.#")
-                .noargs();
-    }
-
-    /**
-     * 绑定队列 02
-     * @return binding
-     */
-    @Bean
-    public Binding getTopicBinding02(){
-        return BindingBuilder
-                .bind(getTopicQueue02())
-                .to(getTopicExchange())
-                // 路由键 队列2接收info级别的消息
-                .with("back.order.*")
-                .noargs();
-    }
-}

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/constants/RabbitConstants.java

@@ -0,0 +1,15 @@
+package com.muyu.common.rabbit.constants;
+
+/**
+ * rabbit常量
+ * @Author: WangXin
+ * @date: 2024/7/10
+ * @Description: rabbit常量
+ * @Version 1.0.0
+ */
+public class RabbitConstants {
+
+    public final static String GO_ONLINE_QUEUE= "GoOnline";
+
+    public final static String DOWNLINE_QUEUE= "Downline";
+}

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/constants/RabbitmqConstants.java

@@ -1,34 +0,0 @@
-package com.muyu.common.rabbit.constants;
-
-/**
- * @Author: WangXin
- * @date: 2024/7/10
- * @Description: rabbitmq常量
- * @Version 1.0.0
- */
-public interface RabbitmqConstants {
-
-    //普通队列
-    String BASIC_QUEUE_NAME = "BASIC_QUEUE_NAME";
-
-    String lOG_QUEUE_NAME = "LOG_QUEUE_NAME";
-    //延迟队列
-    //队列名称
-    String DELAYED_QUEUE_NAME = "delayed_queue";
-    //交换机名称
-    String DELAYED_EXCHANGE_NAME = "DELAYED_EXCHANGE";
-    //交换机
-    String DELAYED_ROUTING_KEY = "delayed";
-    /**
-     * 上下线监听交换机
-     */
-    String TOP_BOTTOM_STITCHING = "top_bottom_stitching";
-    /**
-     * 上线规则
-     */
-    String TOP_RULE = "car.top.data";
-    /**
-     * 车辆下线规则
-     */
-    String BOTTOM_RULE = "car.bottom.data";
-}

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/consumer/RabbitMQConsumerUtil.java

@@ -1,140 +0,0 @@
-package com.muyu.common.rabbit.consumer;
-
-import com.alibaba.fastjson2.JSONObject;
-import com.muyu.common.redis.service.RedisService;
-import com.rabbitmq.client.Channel;
-import lombok.AllArgsConstructor;
-import lombok.extern.log4j.Log4j2;
-import org.springframework.amqp.core.Message;
-import org.springframework.stereotype.Component;
-
-import java.io.IOException;
-import java.util.Date;
-
-/**
- * @ClassName: RabbitMQConsumerUtil
- * @Description: rabbitmq消费者
- */
-@Component
-@Log4j2
-@AllArgsConstructor
-public class RabbitMQConsumerUtil {
-
-    private final RedisService redisService;
-
-
-    /**
-     * 普通消费者
-     * @param data 数据类型
-     * @param message
-     * @param channel
-     */
-    public void rabbitMQBasicConsumer(Object data ,Message message , Channel channel) {
-        log.info("当前时间：{} ：RabbitMQConsumerUtil : {}", new Date(), message);
-        try {
-            // 获取到消息 开始消费
-            log.info("消息消费者接收到消息，消息内容：{}", JSONObject.toJSONString(data));
-
-
-            Long add = redisService.redisTemplate.opsForSet().add(data, message.getMessageProperties().getMessageId());
-
-            if (add != 1) {
-                return;
-            }
-
-
-            /**
-             * -----------------------------------以下为异步业务操作----------------------------
-             */
-
-            /**
-             * ------------------------------------------------------------------------------
-             */
-            // 消费消息成功之后需要确认
-            // long deliveryTag  消息投递序号  自增的数字  在整个队列中唯一 拿到这个序号就相当于拿到这条消息
-            // boolean multiple  是否批量确认  true 批量  确认小于等于当前投递序号的消息   false 单个确认
-            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
-            log.info("xxx消费者接收到消息，消息内容：{}，消费成功...", message);
-
-        } catch (Exception e) {
-            log.error("xxx消费者接收到消息，消息内容：{}，消费消息异常，异常信息：{}", message, e);
-            // 消息回退  拒绝消费消息
-            // long deliveryTag  消息投递序号  自增的数字  在整个队列中唯一 拿到这个序号就相当于拿到这条消息
-            // boolean requeue   是否回到原来的队列
-            try {
-                channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
-//                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
-            } catch (IOException ex) {
-                log.error("xxx消费者接收到消息，消息内容：{}，回退消息异常，异常信息：{}", message, ex);
-            }
-        }finally {
-            try {
-                channel.close();
-            } catch (Exception e) {
-                log.error("xxx消费者关闭Channel异常，消息内容：{}，异常信息：{}", message, e);
-            }
-        }
-    }
-
-
-    /**
-     * 普通消费者
-     * @param data 数据类型
-     * @param message
-     * @param channel
-     */
-    public void carUpConsumer(String data,Message message , Channel channel) {
-        log.info("当前时间：{} ：RabbitMQConsumerUtil : {}", new Date(), message);
-        try {
-            // 获取到消息 开始消费
-            log.info("消息消费者接收到消息，消息内容：{}", JSONObject.toJSONString(data));
-
-
-            Long add = redisService.redisTemplate.opsForSet().add(data, message.getMessageProperties().getMessageId());
-
-            if (add != 1) {
-                return;
-            }
-
-
-            /**
-             * -----------------------------------以下为异步业务操作----------------------------
-             */
-            log.info("[ 根据vin拿到缓存 ] vin为 --》 {}",data);
-            log.info("[ 存入本地缓存 ] 数据为 --》 {}",data);
-            log.info("[ 存入本地缓存 ] 数据为 --》 {}",data);
-            /**
-             * ------------------------------------------------------------------------------
-             */
-            // 消费消息成功之后需要确认
-            // long deliveryTag  消息投递序号  自增的数字  在整个队列中唯一 拿到这个序号就相当于拿到这条消息
-            // boolean multiple  是否批量确认  true 批量  确认小于等于当前投递序号的消息   false 单个确认
-            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
-            log.info("xxx消费者接收到消息，消息内容：{}，消费成功...", message);
-
-        } catch (Exception e) {
-            log.error("xxx消费者接收到消息，消息内容：{}，消费消息异常，异常信息：{}", message, e);
-            // 消息回退  拒绝消费消息
-            // long deliveryTag  消息投递序号  自增的数字  在整个队列中唯一 拿到这个序号就相当于拿到这条消息
-            // boolean requeue   是否回到原来的队列
-            try {
-                channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
-//                channel.basicNack(message.getMessageProperties().getDeliveryTag(), false, true);
-            } catch (IOException ex) {
-                log.error("xxx消费者接收到消息，消息内容：{}，回退消息异常，异常信息：{}", message, ex);
-            }
-        }finally {
-            try {
-                channel.close();
-            } catch (Exception e) {
-                log.error("xxx消费者关闭Channel异常，消息内容：{}，异常信息：{}", message, e);
-            }
-        }
-    }
-
-
-
-
-
-
-}

cloud-common/cloud-common-rabbit/src/main/java/com/muyu/common/rabbit/producer/RabbitMQProducerUtil.java

@@ -1,174 +0,0 @@
-package com.muyu.common.rabbit.producer;
-
-import com.muyu.common.core.domain.Result;
-import com.muyu.common.rabbit.constants.RabbitmqConstants;
-import lombok.AllArgsConstructor;
-import lombok.extern.log4j.Log4j2;
-import org.springframework.amqp.core.MessageProperties;
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.stereotype.Component;
-
-import java.util.UUID;
-
-/**
- * @ClassName: RabbitMQProducer
- * @Description: rabbitmq生产者
- */
-@Component
-@AllArgsConstructor
-@Log4j2
-public class RabbitMQProducerUtil {
-    //redis工具类对象
-
-    //rabbit
-    private final RabbitTemplate rabbitTemplate;
-
-
-    /**
-     * 简单模型
-     *
-     * @param param 传递的消息 （如果是对象需要序列化）
-     * @return 结果集
-     * 一对一消费，只有一个消费者能接收到
-     */
-    public Result<?> basicSendMessage(String queueName, Object param, String msg) {
-
-        log.info("【简单模型mq】 : method: 【 basicSendMessage 】 - ages: 【 String : {}, Object : {}, String : {} 】 ---> 【 消息发送中。。。 】", RabbitmqConstants.BASIC_QUEUE_NAME, param, msg);
-        // 发送简单模型消息
-        // 第一个参数： 绑定规则 相当于 队列名称
-        // 第二个参数：消息内容
-        rabbitTemplate.convertAndSend(queueName, param, message -> {
-             message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
-             return message;
-        } );
-
-        log.info("【简单模型mq】 : method: 【 basicSendMessage 】- queue: 【 {} 】 ---> 【 消息发送成功 】", RabbitmqConstants.BASIC_QUEUE_NAME);
-
-        return Result.success(msg!=null?msg:"消息发送成功");
-    }
-
-    /**
-     * Work queue 工作模型
-     *
-     * @param obj 传递的消息 （如果是对象需要序列化）
-     * @return 结果集
-     * 多个消费者，你一个我一个分配消费消息，有预取机制，默认公平消费，可配置 能者多劳模式（），谁完成的快，谁多做一点
-     */
-    public Result<?> workSendMessage(String queueName, Object obj, String msg) {
-
-        log.info("【工作模型mq】 : method: 【 workSendMessage 】 - ages: 【 String : {}, Object : {}, String : {} 】 ---> 【 消息发送中。。。 】", queueName, obj, msg);
-        // 发送简单模型消息
-        // 第一个参数： 绑定规则 相当于 队列名称
-        // 第二个参数：消息内容
-        rabbitTemplate.convertAndSend(queueName, obj, message -> {
-            message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
-            return message;
-        } );
-
-        log.info("【工作模型mq】 : method: 【 workSendMessage 】- queue: 【 {} 】 ---> 【 消息发送成功 】", queueName);
-
-        return Result.success("消息发送成功");
-    }
-
-    /**
-     * Publish/Subscribe  发布订阅者模型
-     * 多个消费者，多个消费者可以同时接收到消息  有交换机  类型  fanout
-     *
-     * @param exchange 交换机名称
-     * @param obj      发送的消息Object
-     * @param msg      响应的内容
-     * @return 结果集
-     */
-    public Result<?> publishSubscribeSendMessage(String exchange, Object obj, String msg) {
-
-        log.info("【订阅模型mq】 : method: 【 workSendMessage 】 - ages: 【 String : {}, Object : {}, String : {} 】 ---> 【 消息发送中。。。 】", exchange, obj, msg);
-        // 发送简单模型消息
-        // 第一个参数： exchange 交换机的名称
-        // 第二个参数： 绑定规则 发布订阅者模型 不写  默认 "" 只要绑定就行 不需要规则
-        // 第三个参数：消息内容
-        rabbitTemplate.convertAndSend(exchange, "", obj, message -> {
-            message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
-            return message;
-        } );
-
-        log.info("【订阅模型mq】 : method: 【 workSendMessage 】- exchange: 【 {} 】 ---> 【 消息发送成功 】", exchange);
-
-        return Result.success("消息发送成功");
-    }
-
-    /**
-     * Routing路由模型
-     * 使用的是 Direct 类型的交换机，会将接收到的消息根据 规则 路由到指定的Queue(队列)，因此称为路由模式
-     *
-     * @param exchange 交换机名称
-     * @param rule     绑定规则  一个字符串即可
-     * @param obj      发送的消息Object
-     * @param msg      响应的内容
-     * @return 结果集
-     */
-    public Result<?> routingSendMessage(String exchange, String rule, Object obj, String msg) {
-
-        log.info("【路由模型mq】 : method: 【 workSendMessage 】 - ages: 【 String : {}, Object : {}, String : {} 】 ---> 【 消息发送中。。。 】", exchange, obj, msg);
-        // 发送简单模型消息
-        // 第一个参数： 绑定规则 相当于 队列名称
-        // 第二个参数：消息内容
-        rabbitTemplate.convertAndSend(exchange, rule, obj, message -> {
-            message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
-            return message;
-        } );
-
-        log.info("【路由模型mq】 : method: 【 workSendMessage 】- exchange: 【 {} 】 ---> 【 消息发送成功 】", exchange);
-
-        return Result.success("消息发送成功");
-    }
-
-
-    /**
-     * Topic主题模型模型
-     * 使用的是 topic 类型的交换机
-     *
-     * @param exchange 交换机名称
-     * @param rule     绑定规则  可以绑定多个单词以 . 拼接 也可以使用 #(匹配 零个 一个 或 多个 单词) 或 *(匹配 一个 单词) 通配符（例如：name.msg, *.msg, age.# ）
-     * @param obj      发送的消息Object
-     * @param msg      响应的内容
-     * @return 结果集
-     */
-    public Result<?> topicSendMessage(String exchange, String rule, Object obj) {
-
-        log.info("【主题模型mq】 : method: 【 workSendMessage 】 - ages: 【 String : {}, Object : {} 】 ---> 【 消息发送中。。。 】", exchange, obj);
-        // 发送简单模型消息
-        // 第一个参数： 绑定规则 相当于 队列名称
-        // 第二个参数：消息内容
-        rabbitTemplate.convertAndSend(exchange, rule, obj, message -> {
-            message.getMessageProperties().setMessageId(UUID.randomUUID().toString());
-            return message;
-        } );
-
-        log.info("【主题模型mq】 : method: 【 workSendMessage 】- exchange: 【 {} 】 ---> 【 消息发送成功 】", exchange);
-
-        return Result.success(obj,"消息发送成功");
-    }
-
-
-    /**
-     * 延迟队列模型
-     * @param param 传输内容
-     * @param delayTime 延迟时间
-     * @return 结果集
-     */
-    public Result<?> delayedSendMessage(Long delayTime,  Object param) {
-        log.info("【延迟队列模型】 : method: 【 delayedSendMessage 】 消息内容:{}---> 【 消息发送中。。。 】",param);
-
-        rabbitTemplate.convertAndSend(RabbitmqConstants.DELAYED_EXCHANGE_NAME, RabbitmqConstants.DELAYED_ROUTING_KEY,param, message -> {
-            MessageProperties messageProperties = message.getMessageProperties();
-            messageProperties.setMessageId(UUID.randomUUID().toString());
-            messageProperties.setDelayLong(delayTime);
-            return message;
-        });
-        log.info("【延迟队列模型】 : method: 【 delayedSendMessage 】 消息内容:{}---> 【 消息发送成功 】",param);
-
-        return Result.success(param,"消息发送成功");
-
-    }
-
-}

cloud-common/cloud-common-rabbit/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -1,7 +1,3 @@
-com.muyu.rabbitmq.producer.RabbitMQProducerUtil
+com.muyu.common.rabbit.config.RabbitListenerConfigurer
-com.muyu.rabbitmq.consumer.RabbitMQConsumerUtil
+com.muyu.common.rabbit.config.ConfirmCallbackConfig
-com.muyu.rabbitmq.config.RabbitmqConfig
+com.muyu.common.rabbit.config.ReturnCallbackConfig
-com.muyu.rabbitmq.config.MyConfirmCallback
-com.muyu.rabbitmq.config.DelayedQueueConfig
-com.muyu.rabbitmq.config.RabbitAdminConfig
-com.muyu.rabbitmq.config.ReturnCallbackConfig

cloud-common/cloud-common-saas/pom.xml

@@ -5,11 +5,11 @@
     <modelVersion>4.0.0</modelVersion>
     <parent>
         <groupId>com.muyu</groupId>
-        <artifactId>cloud-modules-enterprise</artifactId>
+        <artifactId>cloud-common</artifactId>
         <version>3.6.3</version>
     </parent>
 
-    <artifactId>enterprise-cache</artifactId>
+    <artifactId>cloud-common-saas</artifactId>
 
     <properties>
         <maven.compiler.source>17</maven.compiler.source>
@@ -18,21 +18,16 @@
     </properties>
 
     <dependencies>
+        <!-- 多数据源依赖 -->
         <dependency>
             <groupId>com.muyu</groupId>
-            <artifactId>enterpise-common</artifactId>
+            <artifactId>cloud-common-datasource</artifactId>
         </dependency>
 
+        <!-- 鉴权依赖 -->
         <dependency>
             <groupId>com.muyu</groupId>
-            <artifactId>cloud-common-cache</artifactId>
+            <artifactId>cloud-common-security</artifactId>
-            <version>3.6.3</version>
-        </dependency>
-        <dependency>
-            <groupId>com.muyu</groupId>
-            <artifactId>cloud-modules-vehicle-gateway</artifactId>
-            <version>3.6.3</version>
-            <scope>compile</scope>
         </dependency>
     </dependencies>
 

cloud-common/cloud-common-saas/src/main/java/com/muyu/cloud/common/many/datasource/ManyDataSource.java

@@ -0,0 +1,111 @@
+package com.muyu.cloud.common.many.datasource;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.alibaba.fastjson2.JSON;
+import com.baomidou.mybatisplus.autoconfigure.MybatisPlusAutoConfiguration;
+import com.muyu.cloud.common.many.datasource.constents.DatasourceContent;
+import com.muyu.cloud.common.saas.domain.model.EntInfo;
+import com.muyu.cloud.common.many.datasource.factory.DruidDataSourceFactory;
+import com.muyu.cloud.common.many.datasource.domain.model.DataSourceInfo;
+import com.muyu.cloud.common.many.datasource.role.DynamicDataSource;
+import com.muyu.cloud.common.saas.exception.SaaSException;
+import com.muyu.common.core.constant.SecurityConstants;
+import com.muyu.common.core.constant.UserConstants;
+import com.muyu.common.core.domain.Result;
+import com.muyu.common.core.utils.SpringUtils;
+import com.muyu.common.system.domain.SysDept;
+import com.muyu.common.system.domain.SysUser;
+import com.muyu.common.system.remote.RemoteUserService;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.boot.ApplicationArguments;
+import org.springframework.boot.ApplicationRunner;
+import org.springframework.boot.autoconfigure.AutoConfiguration;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * @Author: DongZeLiang
+ * @date: 2024/6/3
+ * @Description: 多数据源
+ * @Version: 1.0
+ */
+@Log4j2
+@Configuration
+@AutoConfiguration(before = MybatisPlusAutoConfiguration.class)
+public class ManyDataSource implements ApplicationRunner {
+
+
+    private List<EntInfo> dataSourceInfoList(){
+        RemoteUserService RemoteUserService = SpringUtils.getBean(RemoteUserService.class);
+        Result<List<String>> listResult = RemoteUserService.selectFirmDatabaseList(SecurityConstants.INNER);
+        if (listResult==null){
+            throw new SaaSException("saas远调数据源错误");
+        }
+        List<String> data = listResult.getData();
+        if (listResult.getCode() ==Result.SUCCESS && data !=null){
+            List<EntInfo> list = new ArrayList<>();
+            for (String row : data) {
+                list.add(
+                        EntInfo.builder()
+                                .entCode(row)
+                                .ip(DatasourceContent.IP)
+                                .port(DatasourceContent.PORT)
+                                .build()
+                );
+            }
+            return list;
+        }else {
+            log.error("远调数据源错误,远调数据为：{}", JSON.toJSONString(data));
+            throw new RuntimeException("远调数据源错误,远调--》 companyList ");
+        }
+    }
+//    private List<EntInfo> dataPrimarySourceInfoList(){
+//        List<EntInfo> list = new ArrayList<>();
+//        list.add(
+//                EntInfo.builder()
+//                        .entCode()
+//                        .ip(DatasourceContent.IP)
+//                        .port(DatasourceContent.PORT)
+//                        .build()
+//        );
+//        return list;
+//    }
+
+    @Bean
+    public DynamicDataSource dynamicDataSource(DruidDataSourceFactory druidDataSourceFactory) {
+        // 企业列表 企业CODE，端口，IP
+        Map<Object, Object> dataSourceMap = new HashMap<>();
+        dataSourceInfoList()
+                .stream()
+                .map(entInfo -> DataSourceInfo.hostAndPortBuild(entInfo.getEntCode(), entInfo.getIp(), entInfo.getPort()))
+                .forEach(dataSourceInfo -> {
+                    dataSourceMap.put(dataSourceInfo.getKey(), druidDataSourceFactory.create(dataSourceInfo));
+                });
+        //设置动态数据源
+        DynamicDataSource dynamicDataSource = new DynamicDataSource();
+//        dynamicDataSource.setDefaultTargetDataSource(masterDataSource());
+        dynamicDataSource.setTargetDataSources(dataSourceMap);
+        //将数据源信息备份在defineTargetDataSources中
+        dynamicDataSource.setDefineTargetDataSources(dataSourceMap);
+        return dynamicDataSource;
+    }
+
+    @Override
+    public void run(ApplicationArguments args) throws Exception {
+        DruidDataSourceFactory druidDataSourceFactory = SpringUtils.getBean(DruidDataSourceFactory.class);
+        DynamicDataSource dynamicDataSource = SpringUtils.getBean(DynamicDataSource.class);
+        for (EntInfo entInfo : dataSourceInfoList()) {
+            DataSourceInfo dataSourceInfo = DataSourceInfo.hostAndPortBuild(
+                    entInfo.getEntCode(), entInfo.getIp(), entInfo.getPort()
+            );
+            DruidDataSource druidDataSource = druidDataSourceFactory.create(dataSourceInfo);
+            dynamicDataSource.put(dataSourceInfo.getKey(), druidDataSource);
+            log.info("存储数据连接池为：key:{}",dataSourceInfo.getKey());
+        }
+    }
+}

cloud-common/cloud-common-saas/src/main/java/com/muyu/cloud/common/many/datasource/constents/DatasourceContent.java

@@ -0,0 +1,19 @@
+package com.muyu.cloud.common.many.datasource.constents;
+
+/**
+ * @author DongZl
+ * @description: 数据源常量
+ * @Date 2023-8-1 上午 11:02
+ */
+public class DatasourceContent {
+
+    public final static String DATASOURCE_URL = "jdbc:mysql://{}:{}/{}?useUnicode=true&characterEncoding=utf8&zeroDateTimeBehavior=convertToNull&useSSL=true&serverTimezone=GMT%2B8";
+
+    public final static String USER_NAME = "root";
+
+    public final static String PASSWORD = "wx0713101x";
+
+    public final static String IP = "127.0.0.1";
+
+    public final static Integer PORT = 3307;
+}

cloud-common/cloud-common-saas/src/main/java/com/muyu/cloud/common/many/datasource/domain/model/DataSourceInfo.java

@@ -0,0 +1,50 @@
+package com.muyu.cloud.common.many.datasource.domain.model;
+
+import com.muyu.cloud.common.many.datasource.constents.DatasourceContent;
+import com.muyu.common.core.utils.StringUtils;
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * @author DongZl
+ * @description: 数据源实体类
+ * @Date 2023-8-1 上午 11:15
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class DataSourceInfo {
+
+    /**
+     * 键
+     */
+    private String key;
+
+    /**
+     * 地址
+     */
+    private String url;
+
+    /**
+     * 用户名
+     */
+    private String userName;
+
+    /**
+     * 密码
+     */
+    private String password;
+
+
+    public static DataSourceInfo hostAndPortBuild(String key, String host, Integer port) {
+        return DataSourceInfo.builder()
+                .key(key)
+                .url(StringUtils.format(DatasourceContent.DATASOURCE_URL, host, port, key))
+                .password(DatasourceContent.PASSWORD)
+                .userName(DatasourceContent.USER_NAME)
+                .build();
+    }
+}

cloud-common/cloud-common-saas/src/main/java/com/muyu/cloud/common/many/datasource/factory/DruidDataSourceFactory.java

@@ -0,0 +1,39 @@
+package com.muyu.cloud.common.many.datasource.factory;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.muyu.cloud.common.many.datasource.domain.model.DataSourceInfo;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.stereotype.Component;
+
+import java.sql.SQLException;
+
+/**
+ * @Author: DongZeLiang
+ * @date: 2024/6/3
+ * @Description: Druid工厂
+ * @Version: 1.0
+ */
+@Log4j2
+@Component
+public class DruidDataSourceFactory {
+
+    /**
+     * @Description: 根据传递的数据源信息测试数据库连接
+     */
+    public DruidDataSource create(DataSourceInfo dataSourceInfo) {
+        DruidDataSource druidDataSource = new DruidDataSource();
+        druidDataSource.setUrl(dataSourceInfo.getUrl());
+        druidDataSource.setUsername(dataSourceInfo.getUserName());
+        druidDataSource.setPassword(dataSourceInfo.getPassword());
+        druidDataSource.setBreakAfterAcquireFailure(true);
+        druidDataSource.setConnectionErrorRetryAttempts(0);
+        try {
+            druidDataSource.getConnection(2000);
+            log.info("{} -> 数据源连接成功", dataSourceInfo.getKey());
+            return druidDataSource;
+        } catch (SQLException throwables) {
+            log.error("数据源 {} 连接失败,用户名：{}，密码 {}",dataSourceInfo.getUrl(),dataSourceInfo.getUserName(),dataSourceInfo.getPassword());
+            return null;
+        }
+    }
+}

cloud-common/cloud-common-saas/src/main/java/com/muyu/cloud/common/many/datasource/holder/DynamicDataSourceHolder.java

@@ -0,0 +1,42 @@
+package com.muyu.cloud.common.many.datasource.holder;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.util.Assert;
+
+/**
+ * 数据源切换处理
+ *
+ * @author Dongzl
+ */
+@Slf4j
+public class DynamicDataSourceHolder {
+    /**
+     * 保存动态数据源名称
+     */
+    private static final ThreadLocal<String> DYNAMIC_DATASOURCE_KEY = new ThreadLocal<>();
+
+    /**
+     * 设置/切换数据源，决定当前线程使用哪个数据源
+     */
+    public static void setDynamicDataSourceKey(String key){
+        log.info("数据源切换为：{}",key);
+        DYNAMIC_DATASOURCE_KEY.set(key);
+    }
+
+    /**
+     * 获取动态数据源名称，默认使用mater数据源
+     */
+    public static String getDynamicDataSourceKey(){
+        String key = DYNAMIC_DATASOURCE_KEY.get();
+        Assert.notNull(key, "请携带数据标识");
+        return key;
+    }
+
+    /**
+     * 移除当前数据源
+     */
+    public static void removeDynamicDataSourceKey(){
+        log.info("移除数据源：{}",DYNAMIC_DATASOURCE_KEY.get());
+        DYNAMIC_DATASOURCE_KEY.remove();
+    }
+}

cloud-common/cloud-common-saas/src/main/java/com/muyu/cloud/common/many/datasource/role/DynamicDataSource.java

@@ -0,0 +1,55 @@
+package com.muyu.cloud.common.many.datasource.role;
+
+import com.alibaba.druid.pool.DruidDataSource;
+import com.muyu.cloud.common.many.datasource.holder.DynamicDataSourceHolder;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.EqualsAndHashCode;
+import lombok.NoArgsConstructor;
+import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
+
+import java.util.Map;
+
+/**
+ * 动态数据源
+ * 调用AddDefineDataSource组件的addDefineDynamicDataSource（）方法，获取原来targetdatasources的map，并将新的数据源信息添加到map中，并替换targetdatasources中的map
+ * 切换数据源时可以使用@DataSource(value = "数据源名称")，或者DynamicDataSourceContextHolder.setContextKey("数据源名称")
+ * @author Dongzl
+ */
+@EqualsAndHashCode(callSuper = true)
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class DynamicDataSource extends AbstractRoutingDataSource {
+    /**
+     * 备份所有数据源信息 备份的是个 指针 ！！！
+     */
+    private Map<Object, Object> defineTargetDataSources;
+
+    /**
+     * 判定键是否出站了
+     * @param key 键
+     * @return 存在结果 true存在 false不存在
+     */
+    public boolean hashKey(String key){
+        return defineTargetDataSources.containsKey(key);
+    }
+
+    /**
+     * 添加数据库
+     * @param key 键
+     * @param value 数据源
+     */
+    public void put(String  key, DruidDataSource value) {
+        defineTargetDataSources.put(key, value);
+        this.afterPropertiesSet();
+    }
+
+    /**
+     * 决定当前线程使用哪个数据源
+     */
+    @Override
+    protected Object determineCurrentLookupKey() {
+        return DynamicDataSourceHolder.getDynamicDataSourceKey();
+    }
+}

cloud-common/cloud-common-saas/src/main/java/com/muyu/cloud/common/saas/contents/SaaSConstant.java

@@ -0,0 +1,12 @@
+package com.muyu.cloud.common.saas.contents;
+
+/**
+ * @Author: DongZeLiang
+ * @date: 2024/6/3
+ * @Description: SAAS常量
+ * @Version: 1.0
+ */
+public class SaaSConstant {
+
+    public final static String SAAS_KEY = "ent-code";
+}

cloud-common/cloud-common-saas/src/main/java/com/muyu/cloud/common/saas/domain/model/EntInfo.java

@@ -0,0 +1,25 @@
+package com.muyu.cloud.common.saas.domain.model;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * @Author: DongZeLiang
+ * @date: 2024/6/3
+ * @Description: 企业信息
+ * @Version: 1.0
+ */
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class EntInfo {
+
+    private String entCode;
+
+    private String ip;
+
+    private Integer port;
+}

cloud-common/cloud-common-saas/src/main/java/com/muyu/cloud/common/saas/exception/SaaSException.java

@@ -0,0 +1,28 @@
+package com.muyu.cloud.common.saas.exception;
+
+
+import com.muyu.common.core.exception.ServiceException;
+
+/**
+ * @Author: DongZeLiang
+ * @date: 2024/6/3
+ * @Description: SaaS异常类
+ * @Version: 1.0
+ */
+public class SaaSException extends ServiceException {
+
+    public SaaSException (String message, Integer code) {
+        super(message, code);
+    }
+
+    public SaaSException (String message) {
+        super(message);
+    }
+
+    /**
+     * 空构造方法，避免反序列化问题
+     */
+    public SaaSException () {
+        super();
+    }
+}

cloud-common/cloud-common-saas/src/main/java/com/muyu/cloud/common/saas/interceptor/SaaSInterceptor.java

@@ -0,0 +1,66 @@
+package com.muyu.cloud.common.saas.interceptor;
+
+import com.alibaba.fastjson2.JSONObject;
+import com.muyu.cloud.common.saas.contents.SaaSConstant;
+import com.muyu.cloud.common.many.datasource.holder.DynamicDataSourceHolder;
+import com.muyu.cloud.common.saas.exception.SaaSException;
+import com.muyu.cloud.common.many.datasource.role.DynamicDataSource;
+import com.muyu.common.core.utils.ServletUtils;
+import com.muyu.common.core.utils.SpringUtils;
+import jakarta.servlet.http.HttpServletRequest;
+import jakarta.servlet.http.HttpServletResponse;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.web.method.HandlerMethod;
+import org.springframework.web.servlet.AsyncHandlerInterceptor;
+
+
+/**
+ * @Author: DongZeLiang
+ * @date: 2024/6/3
+ * @Description: SAAS拦截器
+ * @Version: 1.0
+ */
+@Log4j2
+public class SaaSInterceptor implements AsyncHandlerInterceptor {
+
+    private Boolean flag = false;
+
+    /**
+     * 之前
+     */
+    @Override
+    public boolean preHandle (HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
+        if (!(handler instanceof HandlerMethod)) {
+            return true;
+        }
+        String SaaSKey = ServletUtils.getHeader(request, SaaSConstant.SAAS_KEY);
+        if (SaaSKey == null) {
+            throw new SaaSException("SaaS非法访问");
+        }
+        if (SaaSKey.equals("-")){
+            log.info("使用 [ nacos ] 配置数据库 ");
+            flag = true;
+            return true;
+        }
+
+        DynamicDataSource dynamicDataSource = SpringUtils.getBean(DynamicDataSource.class);
+        if (!dynamicDataSource.hashKey(SaaSKey)){
+            throw new SaaSException("SaaS非法访问");
+        }
+
+        DynamicDataSourceHolder.setDynamicDataSourceKey(SaaSKey);
+        return true;
+    }
+
+    /**
+     * 之后
+     */
+    @Override
+    public void afterConcurrentHandlingStarted (HttpServletRequest request, HttpServletResponse response, Object handler) throws Exception {
+        if (flag){
+            DynamicDataSourceHolder.removeDynamicDataSourceKey();
+            flag = false;
+        }
+
+    }
+}

cloud-common/cloud-common-saas/src/main/java/com/muyu/cloud/common/saas/interceptor/WebMvcSaaSConfig.java

@@ -0,0 +1,31 @@
+package com.muyu.cloud.common.saas.interceptor;
+
+import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
+import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
+
+/**
+ * 拦截器配置
+ *
+ * @author muyu
+ */
+public class WebMvcSaaSConfig implements WebMvcConfigurer {
+    /**
+     * 不需要拦截的地址
+     */
+    public static final String[] excludeUrls = {"/login", "/logout", "/refresh"};
+
+    @Override
+    public void addInterceptors (InterceptorRegistry registry) {
+        registry.addInterceptor(getHeaderInterceptor())
+                .addPathPatterns("/**")
+                .excludePathPatterns(excludeUrls)
+                .order(-10);
+    }
+
+    /**
+     * 自定义请求头拦截器
+     */
+    public SaaSInterceptor getHeaderInterceptor () {
+        return new SaaSInterceptor();
+    }
+}

cloud-common/cloud-common-saas/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -0,0 +1,3 @@
+com.muyu.cloud.common.saas.interceptor.WebMvcSaaSConfig
+com.muyu.cloud.common.many.datasource.ManyDataSource
+com.muyu.cloud.common.many.datasource.factory.DruidDataSourceFactory

cloud-common/cloud-common-system/pom.xml

@@ -25,5 +25,9 @@
             <groupId>com.muyu</groupId>
             <artifactId>cloud-common-core</artifactId>
         </dependency>
+        <dependency>
+            <groupId>io.swagger.core.v3</groupId>
+            <artifactId>swagger-annotations-jakarta</artifactId>
+        </dependency>
     </dependencies>
 </project>

cloud-common/cloud-common-system/src/main/java/com/muyu/common/system/domain/SysFirmDatasource.java

@@ -0,0 +1,64 @@
+package com.muyu.common.system.domain;
+
+import com.baomidou.mybatisplus.annotation.IdType;
+import com.baomidou.mybatisplus.annotation.TableId;
+import com.baomidou.mybatisplus.annotation.TableName;
+import io.swagger.v3.oas.annotations.media.Schema;
+import io.swagger.v3.oas.annotations.tags.Tag;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import lombok.experimental.SuperBuilder;
+
+
+/**
+ * @Author WangXin
+ * @Data 2024/10/2
+ * @Description 数据源信息
+ * @Version 1.0.0
+ */
+@Data
+@SuperBuilder
+@AllArgsConstructor
+@NoArgsConstructor
+@Tag(name = "数据源信息")
+@TableName("sys_firm_datasource")
+public class SysFirmDatasource {
+
+    /**
+     * 数据源Id
+     */
+    @Schema(name = "数据源Id")
+    @TableId(type = IdType.AUTO)
+    private Long id;
+    /**
+     * 数据源url
+     */
+    @Schema(name = "数据源url")
+    private String datasourceUrl;
+    /**
+     * 数据源用户名
+     */
+    @Schema(name = "数据源用户名")
+    private String username;
+    /**
+     * 数据源密码
+     */
+    @Schema(name = "数据源密码")
+    private String password;
+    /**
+     * 数据源IP
+     */
+    @Schema(name = "数据源IP")
+    private String ip;
+    /**
+     * 数据源端口号
+     */
+    @Schema(name = "数据源端口号")
+    private Integer port;
+    /**
+     * 数据源库名
+     */
+    @Schema(name = "数据源库名")
+    private String databaseName;
+}

cloud-common/cloud-common-system/src/main/java/com/muyu/common/system/remote/RemoteUserService.java

@@ -9,6 +9,8 @@ import com.muyu.common.system.domain.LoginUser;
 import org.springframework.cloud.openfeign.FeignClient;
 import org.springframework.web.bind.annotation.*;
 
+import java.util.List;
+
 /**
  * 用户服务
  *
@@ -37,4 +39,7 @@ public interface RemoteUserService {
      */
     @PostMapping("/user/register")
     public Result<Boolean> registerUserInfo (@RequestBody SysUser sysUser, @RequestHeader(SecurityConstants.FROM_SOURCE) String source);
+
+    @GetMapping("/")
+    Result<List<String>> selectFirmDatabaseList(@RequestHeader(SecurityConstants.FROM_SOURCE) String inner);
 }

cloud-common/cloud-common-system/src/main/java/com/muyu/common/system/remote/factory/RemoteUserFallbackFactory.java

@@ -9,6 +9,8 @@ import org.slf4j.LoggerFactory;
 import org.springframework.cloud.openfeign.FallbackFactory;
 import org.springframework.stereotype.Component;
 
+import java.util.List;
+
 /**
  * 用户服务降级处理
  *
@@ -31,6 +33,11 @@ public class RemoteUserFallbackFactory implements FallbackFactory<RemoteUserServ
             public Result<Boolean> registerUserInfo (SysUser sysUser, String source) {
                 return Result.error("注册用户失败:" + throwable.getMessage());
             }
+
+            @Override
+            public Result<List<String>> selectFirmDatabaseList(String inner) {
+                return Result.error("注册用户失败:" + throwable.getMessage());
+            }
         };
     }
 }

cloud-common/pom.xml

@@ -24,6 +24,7 @@
         <module>cloud-common-kafka</module>
         <module>cloud-common-iotdb</module>
         <module>cloud-common-caffeine</module>
+        <module>cloud-common-saas</module>
     </modules>
 
     <artifactId>cloud-common</artifactId>

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/pom.xml

@@ -39,6 +39,10 @@
             <artifactId>lombok</artifactId>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>com.muyu</groupId>
+            <artifactId>enterpise-common</artifactId>
+        </dependency>
     </dependencies>
 
 </project>

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/basics/ConditionsBasics.java

@@ -0,0 +1,16 @@
+package com.muyu.data.basics;
+
+/**
+ * 条件基础类
+ * @Author WangXin
+ * @Data 2024/10/5
+ * @Version 1.0.0
+ */
+public interface ConditionsBasics<T> {
+    /**
+     * 判断是否执行
+     * @param t 入参
+     * @return true/false
+     */
+    boolean isExecute(T t);
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/basics/EventProcessBasics.java

@@ -1,32 +0,0 @@
-package com.muyu.data.basics;
-
-import org.springframework.context.ApplicationEvent;
-
-/**
- * @Author WangXin
- * @Data 2024/9/29
- * @Description 事件处理基础类
- * @Version 1.0.0
- */
-public abstract class EventProcessBasics {
-
-    /**
-     * 下一个事件对象
-     */
-    protected EventProcessBasics nextEvent;
-
-    /**
-     * 下一个事件
-     * @param nextHandler 下一个事件处理
-     */
-    public void setNextHandler(EventProcessBasics nextHandler) {
-        this.nextEvent = nextHandler;
-    }
-
-    /**
-     * 事件处理抽象类
-     * @param eventKey 事件唯一key
-     */
-    public abstract void handleEvent(String eventKey);
-
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/basics/EventQueueConfig.java

@@ -1,37 +0,0 @@
-package com.muyu.data.basics;
-
-
-
-import lombok.AllArgsConstructor;
-import lombok.Builder;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-
-import java.util.concurrent.LinkedBlockingDeque;
-
-/**
- * @Author WangXin
- * @Data 2024/9/29
- * @Description 事件队列配置
- * @Version 1.0.0
- */
-@Data
-@Builder
-@AllArgsConstructor
-@NoArgsConstructor
-public class EventQueueConfig {
-
-    private LinkedBlockingDeque<EventProcessBasics> taskNodeQueue = new LinkedBlockingDeque<>();
-
-    public void addEvent(EventProcessBasics obj){
-        this.taskNodeQueue.add(obj);
-    }
-
-    public boolean hashEventNext(){
-        return !taskNodeQueue.isEmpty();
-    }
-
-    private EventProcessBasics nextTaskNode(){
-        return taskNodeQueue.poll();
-    }
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/basics/EventStrategyHandler.java

@@ -0,0 +1,25 @@
+package com.muyu.data.basics;
+
+import com.muyu.data.core.EndStrategy;
+
+/**
+ * 策略处理者，所有节点（根节点除外）都必须实现这个接口
+ * @param <P> 接口入参
+ * @param <R> 接口返回值
+ * @author WangXin
+ * @date 2024/10/4
+ * @version 1.0.0
+ */
+public interface EventStrategyHandler<P,R> {
+
+    @SuppressWarnings("rawtypes")
+    EventStrategyHandler DEFAULT = t -> new EndStrategy();
+
+    /**
+     * 审核
+     * @param param 参数
+     * @return 返回值
+     */
+    R apply(P param);
+
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/basics/EventStrategyRouter.java

@@ -0,0 +1,76 @@
+package com.muyu.data.basics;
+
+/**
+ * 策略路由抽象类
+ * @Author WangXin
+ * @Data 2024/10/4
+ * @Version 1.0.0
+ * 通用的“策略树“框架，通过树形结构实现分发与委托，每层通过指定的参数进行向下分发委托，直到达到最终的执行者。
+ * 该框架包含两个类：{@code StrategyHandler} 和 {@code AbstractStrategyRouter}
+ * 其中：通过实现 {@code AbstractStrategyRouter} 抽象类完成对策略的分发，
+ * 实现 {@code StrategyHandler} 接口来对策略进行实现。
+ * 像是第二层 A、B 这样的节点，既是 Root 节点的策略实现者也是策略A1、A2、B1、B2 的分发者，这样的节点只需要
+ * 同时继承 {@code StrategyHandler} 和实现 {@code AbstractStrategyRouter} 接口就可以了。
+ *
+ * <pre>
+ *           +---------+
+ *           |  Root   |   ----------- 第 1 层策略入口
+ *           +---------+
+ *            /       \  ------------- 根据入参 P1 进行策略分发
+ *           /         \
+ *     +------+      +------+
+ *     |  A   |      |  B   |  ------- 第 2 层不同策略的实现
+ *     +------+      +------+
+ *       /  \          /  \  --------- 根据入参 P2 进行策略分发
+ *      /    \        /    \
+ *   +---+  +---+  +---+  +---+
+ *   |A1 |  |A2 |  |B1 |  |B2 |  ----- 第 3 层不同策略的实现
+ *   +---+  +---+  +---+  +---+
+ *
+ * </pre>
+ *
+ */
+public abstract class EventStrategyRouter<T, R> {
+
+    /**
+     * 策略映射器, 指定入参与出参以决定策略处理者
+     * @param <T>   策略入参
+     * @param <R>   策略出参
+     */
+    public interface StrategyMapper<T,R>{
+        // 通过入参获取对应策略处理方法,使用Map实现
+        EventStrategyHandler<T,R> getHandler(T param);
+    }
+
+    /**
+     * 选择下级策略
+     * @return
+     */
+    protected abstract StrategyMapper<T,R> registerStrategy();
+
+    /**
+     * 默认策略处理者
+     */
+    @SuppressWarnings("unchecked")
+    private EventStrategyHandler<T,R> defaultStrategyHandler = EventStrategyHandler.DEFAULT;
+
+
+    /**
+     * 选择策略处理者
+     * @param param 入参
+     * @return 策略处理结果
+     */
+    public R applyStrategy(T param) {
+        StrategyMapper<T, R> trStrategyMapper = registerStrategy();
+        if (trStrategyMapper == null) {
+            return defaultStrategyHandler.apply(param);
+        }
+        final EventStrategyHandler<T,R> strategyHandler = trStrategyMapper.getHandler(param);
+        if (strategyHandler != null) {
+            return strategyHandler.apply(param);
+        }
+        // 使用默认策略处理者
+        return defaultStrategyHandler.apply(param);
+    }
+
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/basics/StartEvent.java

@@ -1,26 +0,0 @@
-package com.muyu.data.basics;
-
-import com.muyu.data.domain.EventActuate;
-import org.springframework.context.ApplicationEvent;
-
-import java.util.List;
-
-/**
- * @Author WangXin
- * @Data 2024/9/29
- * @Description 事件启动类
- * @Version 1.0.0
- */
-public class StartEvent extends ApplicationEvent {
-
-    private EventActuate eventActuate;
-
-    public StartEvent(EventActuate source) {
-        super(source);
-        this.eventActuate = source;
-    }
-
-    public EventActuate getEventActuate() {
-        return eventActuate;
-    }
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/constant/EventConditions.java

@@ -0,0 +1,46 @@
+package com.muyu.data.constant;
+
+import com.muyu.data.basics.ConditionsBasics;
+import com.muyu.data.domain.Information;
+
+/**
+ * 事件判断枚举（包含所有事件判断条件）
+ * @Author WangXin
+ * @Data 2024/10/5
+ * @Version 1.0.0
+ */
+public enum EventConditions implements ConditionsBasics<Information> {
+
+    /**
+     * 车辆故障判断
+     */
+    FAULT_STRATEGY {
+        @Override
+        public boolean isExecute(Information information) {
+            // 实现逻辑
+            return false;
+        }
+    },
+
+    /**
+     * 电子围栏判断
+     */
+    FENCE_STRATEGY {
+        @Override
+        public boolean isExecute(Information information) {
+            // 实现逻辑
+            return false;
+        }
+    },
+
+    /**
+     * 车辆预警判断
+     */
+    WARN_STRATEGY {
+        @Override
+        public boolean isExecute(Information information) {
+            // 实现逻辑
+            return false;
+        }
+    };
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/constant/EventConstant.java

@@ -1,12 +0,0 @@
-package com.muyu.data.constant;
-
-/**
- * @Author WangXin
- * @Data 2024/9/29
- * @Description 事件常量
- * @Version 1.0.0
- */
-public interface EventConstant {
-
-    String STORAGE_EVENT = "storageEvent";
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/constant/EventEnum.java

@@ -0,0 +1,54 @@
+package com.muyu.data.constant;
+
+import com.muyu.data.core.IotdbStrategy;
+import com.muyu.data.core.event.FaultStrategy;
+import com.muyu.data.core.event.FenceStrategy;
+import com.muyu.data.core.event.WarnStrategy;
+
+/**
+ * 事件枚举
+ * @Author WangXin
+ * @Data 2024/10/6
+ * @Version 1.0.0
+ */
+public enum EventEnum {
+    /** 电子围栏事件 */
+    FENCE_EVENT("电子围栏事件", FenceStrategy.class),
+    /** 车辆报警事件 */
+    WARN_EVENT("车辆报警事件", WarnStrategy.class),
+    /** 电子围栏事件 */
+    FAULT_EVENT("车辆预警事件", FaultStrategy.class),
+    /** 存储事件 */
+    IOTDB_EVENT("存储事件", IotdbStrategy.class);
+
+    /**
+     * 事件名称
+     */
+    private String eventName;
+
+    /**
+     * 事件类型
+     */
+    private Class<?> aClass;
+
+    EventEnum(String eventName, Class<?> aClass) {
+        this.eventName = eventName;
+        this.aClass = aClass;
+    }
+
+    public String getEventName() {
+        return eventName;
+    }
+
+    public void setEventName(String eventName) {
+        this.eventName = eventName;
+    }
+
+    public Class<?> getaClass() {
+        return aClass;
+    }
+
+    public void setaClass(Class<?> aClass) {
+        this.aClass = aClass;
+    }
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/core/EndStrategy.java

@@ -0,0 +1,26 @@
+package com.muyu.data.core;
+
+import com.muyu.data.basics.EventStrategyHandler;
+import com.muyu.data.domain.Information;
+import lombok.extern.slf4j.Slf4j;
+
+/**
+ * 结束节点
+ * @Author WangXin
+ * @Data 2024/10/5
+ * @Version 1.0.0
+ */
+@Slf4j
+public class EndStrategy implements EventStrategyHandler<Information,Information> {
+
+    /**
+     * 结束节点方法
+     * @param param 参数
+     * @return null
+     */
+    @Override
+    public Information apply(Information param) {
+        log.info("[线路结束] --- >");
+        return null;
+    }
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/core/IotdbStrategy.java

@@ -0,0 +1,87 @@
+package com.muyu.data.core;
+
+import com.muyu.data.basics.EventStrategyHandler;
+import com.muyu.data.basics.EventStrategyRouter;
+import com.muyu.data.constant.EventConditions;
+import com.muyu.data.constant.EventEnum;
+import com.muyu.data.core.event.FenceStrategy;
+import com.muyu.data.domain.Information;
+
+import lombok.extern.log4j.Log4j2;
+
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * 存储事件节点（路由转发）
+ * @Author WangXin
+ * @Data 2024/10/5
+ * @Version 1.0.0
+ */
+@Log4j2
+public class IotdbStrategy
+        extends EventStrategyRouter<Information, Information>
+        implements EventStrategyHandler<Information, Information> {
+
+    private static ConcurrentHashMap<EventEnum,EventStrategyRouter<Information,Information>> map = new ConcurrentHashMap<>();
+
+    @Override
+    public Information apply(Information param) {
+        return applyStrategy(param);
+    }
+
+    @Override
+    protected StrategyMapper<Information, Information> registerStrategy() {
+        return param -> {
+            initExecuteStrategy(param);
+            if (isEventMapSize()){
+                log.info("[没有事件执行]");
+            }else {
+                ExecutorService executor = Executors.newFixedThreadPool(map.size());
+                for (EventEnum eventEnum : map.keySet()) {
+                    executor.submit(() -> {
+                        log.info("开始执行 [{}] ",eventEnum.getEventName());
+                        return map.get(eventEnum);
+                    });
+                }
+                // 关闭线程池
+                executor.shutdown();
+                try {
+                    // 等待所有任务完成
+                    if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
+                        executor.shutdownNow();
+                    }
+                } catch (InterruptedException e) {
+                    executor.shutdownNow();
+                }
+            }
+            return new EndStrategy();
+        };
+    }
+
+    /**
+     * 初始化事件map
+     * @param param 入参
+     */
+    private static void initExecuteStrategy(Information param){
+        if (EventConditions.FENCE_STRATEGY.isExecute(param)){
+            map.put(EventEnum.FENCE_EVENT,new FenceStrategy());
+        }
+        if (EventConditions.FAULT_STRATEGY.isExecute(param)){
+            map.put(EventEnum.FAULT_EVENT,new FenceStrategy());
+        }
+        if (EventConditions.WARN_STRATEGY.isExecute(param)){
+            map.put(EventEnum.WARN_EVENT,new FenceStrategy());
+        }
+    }
+
+    /**
+     * 判断事件map长度
+     * @return
+     */
+    private static boolean isEventMapSize(){
+        return map.size() > 0;
+    }
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/core/StartStrategy.java

@@ -0,0 +1,23 @@
+package com.muyu.data.core;
+
+import com.muyu.data.basics.EventStrategyRouter;
+import com.muyu.data.domain.Information;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Component;
+
+/**
+ * 根节点策略
+ * @Author WangXin
+ * @Data 2024/10/6
+ * @Version 1.0.0
+ */
+@Slf4j
+@Component
+public class StartStrategy extends EventStrategyRouter<Information,Information> {
+
+
+    @Override
+    protected StrategyMapper<Information, Information> registerStrategy() {
+        return param -> new IotdbStrategy();
+    }
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/core/event/FaultStrategy.java

@@ -0,0 +1,26 @@
+package com.muyu.data.core.event;
+
+import com.muyu.data.basics.EventStrategyHandler;
+import com.muyu.data.basics.EventStrategyRouter;
+import com.muyu.data.domain.Information;
+
+/**
+ * 车辆故障报警事件
+ * @Author WangXin
+ * @Data 2024/10/5
+ * @Version 1.0.0
+ */
+public class FaultStrategy
+        extends EventStrategyRouter<Information, Information>
+        implements EventStrategyHandler<Information, Information> {
+
+    @Override
+    public Information apply(Information param) {
+        return applyStrategy(param);
+    }
+
+    @Override
+    protected StrategyMapper<Information, Information> registerStrategy() {
+        return null;
+    }
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/core/event/FenceStrategy.java

@@ -0,0 +1,25 @@
+package com.muyu.data.core.event;
+
+import com.muyu.data.basics.EventStrategyHandler;
+import com.muyu.data.basics.EventStrategyRouter;
+import com.muyu.data.domain.Information;
+
+/**
+ * 电子围栏事件
+ * @Author WangXin
+ * @Data 2024/10/5
+ * @Version 1.0.0
+ */
+public class FenceStrategy
+        extends EventStrategyRouter<Information,Information>
+        implements EventStrategyHandler<Information,Information> {
+    @Override
+    public Information apply(Information param) {
+        return applyStrategy(param);
+    }
+
+    @Override
+    protected StrategyMapper<Information, Information> registerStrategy() {
+        return null;
+    }
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/core/event/WarnStrategy.java

@@ -0,0 +1,26 @@
+package com.muyu.data.core.event;
+
+import com.muyu.data.basics.EventStrategyHandler;
+import com.muyu.data.basics.EventStrategyRouter;
+import com.muyu.data.domain.Information;
+
+/**
+ * 车辆预警事件
+ * @Author WangXin
+ * @Data 2024/10/5
+ * @Version 1.0.0
+ */
+public class WarnStrategy
+        extends EventStrategyRouter<Information, Information>
+        implements EventStrategyHandler<Information, Information> {
+
+    @Override
+    public Information apply(Information param) {
+        return applyStrategy(param);
+    }
+
+    @Override
+    protected StrategyMapper<Information, Information> registerStrategy() {
+        return null;
+    }
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/domain/DataJSON.java

@@ -1,34 +0,0 @@
-package com.muyu.data.domain;
-
-import com.alibaba.fastjson2.JSONObject;
-import io.swagger.v3.oas.annotations.media.Schema;
-import io.swagger.v3.oas.annotations.tags.Tag;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.NoArgsConstructor;
-import lombok.experimental.SuperBuilder;
-
-/**
- * @Author WangXin
- * @Data 2024/9/30
- * @Description JSON数据对象
- * @Version 1.0.0
- */
-@Data
-@SuperBuilder
-@AllArgsConstructor
-@NoArgsConstructor
-@Tag(name = "ionDB数据源对象")
-public class DataJSON {
-    /**
-     * 时间戳
-     */
-    @Schema(name = "时间戳")
-    private Long timestamp;
-
-    /**
-     * 车辆JSON数据
-     */
-    @Schema(name = "车辆JSON数据")
-    private JSONObject datasource;
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/domain/EventActuate.java

@@ -9,9 +9,9 @@ import lombok.experimental.SuperBuilder;
 import java.util.List;
 
 /**
+ * 事件驱动对象
  * @Author WangXin
  * @Data 2024/9/29
- * @Description 事件驱动对象
  * @Version 1.0.0
  */
 @Data

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/domain/Information.java

@@ -4,12 +4,15 @@ import com.baomidou.mybatisplus.annotation.IdType;
 import com.baomidou.mybatisplus.annotation.TableId;
 import com.baomidou.mybatisplus.annotation.TableName;
 import com.fasterxml.jackson.annotation.JsonFormat;
+import com.muyu.domain.SysCar;
+import com.muyu.domain.SysCarFault;
 import io.swagger.v3.oas.annotations.media.Schema;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
 import lombok.NoArgsConstructor;
 import lombok.experimental.SuperBuilder;
+import org.springframework.boot.autoconfigure.web.WebProperties;
 
 import java.util.Date;
 /**
@@ -30,61 +33,9 @@ import java.util.Date;
 @NoArgsConstructor
 @TableName(value = "car_information",autoResultMap = true)
 public class Information {
-    //自增主键
+
-    @TableId(value = "id",type = IdType.AUTO)
+    private SysCar sysCar;
-    private Long id;
+
-    //车辆VIn马
+    private SysCarFault carFault;
-    @Schema(name = "车辆VIn马",type = "String")
-    private String carVin;
-    //车辆品牌
-    @Schema(name = "车辆品牌",type = "String")
-    private String carBrand;
-    //车辆类型外键
-    @Schema(name = "车辆类型外键",type = "Integer")
-    private Integer typeId;
-    //车辆类型名称
-    @Schema(type = "String",description = "车辆类型名称")
-    private String typeName;
-    //电子围栏外键
-    @Schema(name = "电子围栏外键",type = "String")
-    private String groupId;
-    //车辆电机厂商
-    @Schema(name = "车辆电机厂商",type = "String")
-    private String carMotorManufacturer;
-    //电机型号
-    @Schema(name = "电机型号",type = "String")
-    private String carMotorModel;
-    //车辆电池厂商
-    @Schema(name = "车辆电池厂商",type = "String")
-    private String carBatteryManufacturer;
-    //电池型号
-    @Schema(name = "电池型号",type = "String")
-    private String carBatteryModel;
-    //围栏组编码
-    @Schema(name = "围栏组编码",type = "String")
-    private String groupCode;
-    //启用状态(1.在线 2.离线 3.已断开 4.待连接 5.维修中
-    @Schema(name = "启用状态(1.在线 2.离线 3.已断开 4.待连接 5.维修中",type = "String")
-    private String state;
-    //创建人
-    @Schema(name = "创建人",type = "Integer")
-    private String createBy;
-    //创建时间
-    @Schema(name = "创建时间",type = "Date")
-    @JsonFormat(pattern="yyyy-MM-dd HH:mm:ss",timezone="GMT+8")
-    private Date createTime;
-    //更新人
-    @Schema(name = "更新人",type = "Integer")
-    private String updateBy;
-    //更新时间
-    @Schema(name = "更新时间",type = "Date")
-    @JsonFormat(pattern="yyyy-MM-dd HH:mm:ss",timezone="GMT+8")
-    private Date updateTime;
-    //备注
-    @Schema(name = "备注",type = "String")
-    private String remark;
-    //策略id
-    @Schema(name = "策略id",type = "Integer")
-    private Integer strategyId;
 
 }

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/event/AutoStartupEventListener.java

@@ -1,28 +0,0 @@
-package com.muyu.data.event;
-
-import com.alibaba.fastjson2.JSONObject;
-import com.muyu.data.basics.StartEvent;
-import com.muyu.data.domain.EventActuate;
-import com.muyu.data.event.tactics.IotdbStoreEvent;
-import org.springframework.context.ApplicationListener;
-import org.springframework.stereotype.Component;
-
-/**
- * @Author WangXin
- * @Data 2024/9/29
- * @Description 自启动事件监听器
- * @Version 1.0.0
- */
-@Component
-public class AutoStartupEventListener implements ApplicationListener<StartEvent> {
-
-    @Override
-    public void onApplicationEvent(StartEvent event) {
-
-        EventActuate eventActuate = event.getEventActuate();
-        JSONObject jsonData = eventActuate.getJsonData();
-        new IotdbStoreEvent().execute(jsonData);
-
-    }
-
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/event/EventStrategy.java

@@ -1,21 +0,0 @@
-package com.muyu.data.event;
-
-import com.alibaba.fastjson2.JSONObject;
-
-import java.util.List;
-
-/**
- * @Author WangXin
- * @Data 2024/9/29
- * @Description 事件策略接口
- * @Version 1.0.0
- */
-
-public interface EventStrategy {
-
-    /**
-     * 方法执行
-     * @param jsonObject
-     */
-    void execute(JSONObject jsonObject);
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/event/EventStrategyContext.java

@@ -1,30 +0,0 @@
-package com.muyu.data.event;
-
-import com.alibaba.fastjson2.JSONObject;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-
-/**
- * @Author WangXin
- * @Data 2024/9/29
- * @Description 事件执行上下文
- * @Version 1.0.0
- */
-@Data
-@AllArgsConstructor
-public class EventStrategyContext {
-    /**
-     * 事件接口
-     */
-    private EventStrategy eventStrategy;
-
-    /**
-     * 调用策略类中的方法
-     * @param jsonObject json对象
-     */
-    public void handleEvent(JSONObject jsonObject) {
-        if (jsonObject != null) {
-            eventStrategy.execute(jsonObject);
-        }
-    }
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/event/StorageEvent.java

@@ -1,37 +0,0 @@
-package com.muyu.data.event;
-
-import com.muyu.data.basics.EventProcessBasics;
-import com.muyu.data.constant.EventConstant;
-import lombok.AllArgsConstructor;
-import lombok.Data;
-import lombok.EqualsAndHashCode;
-import lombok.extern.log4j.Log4j2;
-
-/**
- * @Author WangXin
- * @Data 2024/9/29
- * @Description 存储事件
- * @Version 1.0.0
- */
-@EqualsAndHashCode(callSuper = true)
-@Log4j2
-@Data
-@AllArgsConstructor
-public class StorageEvent extends EventProcessBasics {
-    /**
-     * 事件名称
-     */
-    private String eventName;
-
-    @Override
-    public void handleEvent(String eventKey) {
-        if (eventKey.equals(eventName)){
-            log.info("开始执行 [{}] 事件", eventKey);
-
-        }else if (nextEvent != null){
-            nextEvent.handleEvent(eventKey);
-        }else {
-            log.info("处理结束，最后处理的事件为 [{}]", eventKey);
-        }
-    }
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/event/tactics/FenceStrategyEvent.java

@@ -1,18 +0,0 @@
-package com.muyu.data.event.tactics;
-
-import com.alibaba.fastjson2.JSONObject;
-import com.muyu.data.event.EventStrategy;
-
-/**
- * @Author WangXin
- * @Data 2024/9/30
- * @Description 电子围栏事件
- * @Version 1.0.0
- */
-public class FenceStrategyEvent implements EventStrategy {
-
-    @Override
-    public void execute(JSONObject jsonObject) {
-
-    }
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/java/com/muyu/data/event/tactics/IotdbStoreEvent.java

@@ -1,28 +0,0 @@
-package com.muyu.data.event.tactics;
-
-import com.alibaba.fastjson2.JSONObject;
-import com.muyu.data.event.EventStrategy;
-import lombok.extern.log4j.Log4j2;
-import org.springframework.stereotype.Component;
-
-import java.util.List;
-
-/**
- * @Author WangXin
- * @Data 2024/9/29
- * @Description Iotdb存储事件
- * @Version 1.0.0
- */
-@Log4j2
-@Component
-public class IotdbStoreEvent implements EventStrategy {
-
-    /**
-     * 执行存储事件
-     * @param jsonObject json对象
-     */
-    @Override
-    public void execute(JSONObject jsonObject) {
-        log.info("[存储事件] ---》 json对象：{}", jsonObject);
-    }
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-common/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports

@@ -1 +1 @@
-com.muyu.data.iotdb.config.IotDBSessionConfig
+

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/pom.xml

@@ -27,6 +27,10 @@
             <groupId>com.muyu</groupId>
             <artifactId>enterprise-cache</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.muyu</groupId>
+            <artifactId>cloud-common-caffeine</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.muyu</groupId>
             <artifactId>cloud-modules-data-process-common</artifactId>

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/DataProcessApplication.java

@@ -2,13 +2,15 @@ package com.muyu.data;
 
 import com.muyu.common.security.annotation.EnableCustomConfig;
 import com.muyu.common.security.annotation.EnableMyFeignClients;
+import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
 
 @EnableCustomConfig
 @EnableMyFeignClients
-@SpringBootApplication
+@SpringBootApplication(exclude = {DataSourceAutoConfiguration.class})
 public class DataProcessApplication {
     public static void main(String[] args) {
-        System.out.println("Hello world!");
+        SpringApplication.run(DataProcessApplication.class, args);
     }
 }

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/basic/EventCustom.java

@@ -1,24 +0,0 @@
-package com.muyu.data.basic;
-
-import com.alibaba.fastjson2.JSONObject;
-import org.springframework.context.ApplicationEvent;
-
-/**
- * 事件类型
- *
- * @program: cloud-server
- * @author: WangXin
- * @create: 2024-09-29 20:03
- **/
-public class EventCustom extends ApplicationEvent {
-
-    private JSONObject data;
-
-    public EventCustom(Object source, JSONObject data) {
-        super(source);
-        this.data = data;
-    }
-    public JSONObject getData() {
-        return data;
-    }
-}

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

@@ -1,16 +0,0 @@
-package com.muyu.data.basic;
-
-import org.springframework.context.ApplicationListener;
-
-/**
- * 事件监听接口
- *
- * @program: cloud-server
- * @author: WangXin
- * @create: 2024-09-29 20:05
- **/
-public interface EventListener  extends ApplicationListener<EventCustom> {
-
-    void onEvent(EventCustom event);
-
-}

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

@@ -1,32 +0,0 @@
-package com.muyu.data.basic;
-
-import com.alibaba.fastjson2.JSONObject;
-import org.springframework.context.ApplicationEventPublisher;
-import org.springframework.context.ApplicationEventPublisherAware;
-import org.springframework.stereotype.Component;
-
-/**
- * 策略发送事件
- *
- * @program: cloud-server
- * @author: WangXin
- * @create: 2024-09-29 17:43
- **/
-@Component
-public class EventPublisher implements ApplicationEventPublisherAware {
-
-    private ApplicationEventPublisher publisher;
-
-
-    @Override
-    public void setApplicationEventPublisher(ApplicationEventPublisher applicationEventPublisher) {
-        this.publisher = applicationEventPublisher;
-    }
-
-    public void publishEvent(JSONObject message) {
-        EventCustom event = new EventCustom(this, message);
-        publisher.publishEvent(event);
-    }
-
-
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/config/IotDBConfig.java

@@ -0,0 +1,127 @@
+package com.muyu.data.config;
+
+
+import com.muyu.data.domain.CarInformation;
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.isession.util.Version;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+import org.apache.iotdb.tsfile.read.common.Field;
+import org.apache.iotdb.tsfile.read.common.RowRecord;
+import org.springframework.context.annotation.Configuration;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * iotdb数据库 配置
+ * @author WangXIn
+ * @package：com.muyu.event.config
+ * @name：IOTDBConfig
+ * @date：2024/9/28 19:35
+ */
+@Configuration
+public class IoTDBConfig {
+
+    public static final String HOST="127.0.0.1";
+    public static final Integer PORT=6667;
+    public static final String USERNAME="root";
+    public static final String PASSWORD="root";
+    public static final String TABLENAME="root.four.car_information";
+
+
+    /**
+     * 初始化连接
+     * @return
+     */
+    public Session initIoTDB(){
+        // 初始化与连接
+        Session session = new Session.Builder()
+                .host(HOST)
+                .port(PORT)
+                .username(USERNAME)
+                .password(PASSWORD)
+                .version(Version.V_1_0)
+                .build();
+        return session;
+    };
+
+
+    /**
+     * 实时数据库添加
+     * @param list
+     */
+    public void insertIoTDB(List<String> list){
+        Session session = initIoTDB();
+        // 开启session Rpc不压缩
+        try {
+            session.open(false);
+
+            //添加字段名称
+            ArrayList<String> measurements = new ArrayList<>();
+            //添加字段类型
+            ArrayList<TSDataType> types = new ArrayList<>();
+
+            measurements.add("car_vin");
+            measurements.add("information");
+
+            session.insertRecord(TABLENAME,System.currentTimeMillis(),measurements,list);
+            //关闭连接
+            session.close();
+        } catch (IoTDBConnectionException e) {
+            throw new RuntimeException(e);
+        } catch (StatementExecutionException e) {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public CarInformation queryIoTDB(String carVin)  {
+        Session session = initIoTDB();
+        try {
+            session.open(false);
+        } catch (IoTDBConnectionException e) {
+            throw new RuntimeException(e);
+        }
+        Long timeMillis = System.currentTimeMillis();
+
+        CarInformation carInformation = new CarInformation();
+
+        try(SessionDataSet dataSet= session.executeQueryStatement("select * from root.four.car_information where car_vin='"+carVin+"'")){
+            System.out.println(dataSet.getColumnNames());
+            dataSet.setFetchSize(1024);
+            while (dataSet.hasNext()){
+//                List<Field> fields = dataSet.next().getFields();
+//                carInformation.setCarVin(String.valueOf(fields.get(0)));
+//                carInformation.setInformation(String.valueOf(fields.get(1)));
+
+                String[] fields = dataSet.next().toString().split("\t");
+                carInformation.setTime(Long.valueOf(fields[0]));
+                carInformation.setCarVin(fields[1]);
+                carInformation.setInformation(fields[2]);
+            }
+
+        } catch (IoTDBConnectionException e) {
+            throw new RuntimeException(e);
+        } catch (StatementExecutionException e) {
+            throw new RuntimeException(e);
+        }
+
+        //关闭连接
+        try {
+            session.close();
+        } catch (IoTDBConnectionException e) {
+            throw new RuntimeException(e);
+        }
+        return carInformation;
+    };
+
+
+
+
+
+
+
+
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/consumer/GoOfflineConsumer.java

@@ -1,38 +0,0 @@
-package com.muyu.data.consumer;
-
-
-import com.muyu.data.util.CacheUtil;
-import lombok.extern.log4j.Log4j2;
-import org.springframework.amqp.rabbit.annotation.Exchange;
-import org.springframework.amqp.rabbit.annotation.Queue;
-import org.springframework.amqp.rabbit.annotation.QueueBinding;
-import org.springframework.amqp.rabbit.annotation.RabbitListener;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-/**
- * 下线监听
- *
- * @program: cloud-server
- * @author: WangXin
- * @create: 2024-09-30 10:39
- **/
-@Log4j2
-@Component
-public class GoOfflineConsumer {
-
-
-    @Autowired
-    private CacheUtil cacheUtil;
-
-
-    @RabbitListener(bindings = @QueueBinding(
-            value = @Queue(value = "GO_OFFLINE", durable = "true"),
-            exchange = @Exchange(value = "OFFLINE_EXCHANGE", type = "fanout")))
-    public void online(String vin) {
-        log.info("车辆vin: {},车辆开始消费", vin);
-        cacheUtil.remove(vin);
-    }
-
-
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/consumer/GoOnlineConsumer.java

@@ -1,45 +0,0 @@
-package com.muyu.data.consumer;
-
-import com.muyu.data.domain.Information;
-import com.muyu.data.util.CacheUtil;
-import com.muyu.domain.CarInformation;
-import com.muyu.enterprise.cache.VehicleCacheService;
-
-
-import lombok.extern.log4j.Log4j2;
-import org.springframework.amqp.rabbit.annotation.Exchange;
-import org.springframework.amqp.rabbit.annotation.Queue;
-import org.springframework.amqp.rabbit.annotation.QueueBinding;
-import org.springframework.amqp.rabbit.annotation.RabbitListener;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.stereotype.Component;
-
-/**
- * 上线监听
- *
- * @program: cloud-server
- * @author: WangXin
- * @create: 2024-09-29 16:37
- **/
-@Log4j2
-@Component
-public class GoOnlineConsumer {
-
-
-    @Autowired
-    private CacheUtil cacheUtil;
-
-    @Autowired
-    private VehicleCacheService vehicleCacheService;
-
-    @RabbitListener(bindings = @QueueBinding(
-            value = @Queue(value = "GO_ONLINE", durable = "true"),
-            exchange = @Exchange(value = "ONLINE_EXCHANGE", type = "fanout")))
-    public void online(String vin) {
-        log.info("车辆vin: {},车辆开始消费", vin);
-        CarInformation carInformation = vehicleCacheService.get(vin);
-        cacheUtil.put(vin,carInformation);
-    }
-
-
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/consumer/MessageConsumer.java

@@ -0,0 +1,61 @@
+package com.muyu.data.consumer;
+
+import cn.hutool.core.thread.ThreadUtil;
+import cn.hutool.json.JSONUtil;
+import com.alibaba.fastjson2.JSONObject;
+import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
+import com.muyu.common.kafka.constant.KafkaConstants;
+import com.muyu.data.core.StartStrategy;
+import com.muyu.data.domain.Information;
+import jakarta.annotation.Resource;
+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.beans.factory.InitializingBean;
+import org.springframework.stereotype.Component;
+
+import java.time.Duration;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+
+/**
+ * kafka监听
+ * @author WangXIn
+ */
+@Component
+@Log4j2
+public class MessageConsumer implements InitializingBean {
+
+    @Resource
+    private KafkaConsumer kafkaConsumer;
+    @Resource
+    private StartStrategy startStrategy;
+
+    @Override
+    public void afterPropertiesSet() throws Exception {
+        Thread thread = new Thread(() -> {
+            log.info("启动线程监听Topic: {}", KafkaConstants.KafkaTopic);
+            ThreadUtil.sleep(1000);
+            Collection<String> topics = Lists.newArrayList(KafkaConstants.KafkaTopic);
+            kafkaConsumer.subscribe(topics);
+            while (true) {
+                System.out.println("开始消费数据,等待中...");
+                ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));
+                for (ConsumerRecord consumerRecord : consumerRecords) {
+                    //1.从ConsumerRecord中获取消费数据
+                    String originalMsg = (String) consumerRecord.value();
+                    log.info("从Kafka中消费的原始数据: " + originalMsg);
+                    //2.把消费数据转换为DTO对象
+                    JSONObject jsonObject = JSONObject.parseObject(originalMsg);
+                    Information information = jsonObject.getObject("key", Information.class);
+                    log.info("从Kafka中消费的实体数据: " + information);
+                    // 执行策略
+                    startStrategy.applyStrategy(information);
+                }
+            }
+        });
+        thread.start();
+    }
+}

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

@@ -1,72 +0,0 @@
-package com.muyu.data.consumer;
-
-import com.alibaba.fastjson2.JSONObject;
-import com.muyu.data.basic.EventPublisher;
-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.beans.factory.annotation.Autowired;
-import org.springframework.boot.ApplicationArguments;
-import org.springframework.boot.ApplicationRunner;
-import org.springframework.context.ApplicationListener;
-import org.springframework.context.event.ContextClosedEvent;
-import org.springframework.scheduling.annotation.Async;
-import org.springframework.stereotype.Component;
-
-import java.time.Duration;
-import java.util.Collections;
-import java.util.List;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
-
-/**
-* 车辆消费者
-* @program: cloud-server
-* @author: WangXin
-* @create: 2024-09-28 14:55
-**/
-@Log4j2
-@Component
-public class VehicleConsumer implements ApplicationRunner, ApplicationListener<ContextClosedEvent> {
-
-    @Autowired
-    private KafkaConsumer consumer;
-
-    @Autowired
-    private EventPublisher eventPublisher;
-
-    private final String topic = "vehicle";
-
-    private final ExecutorService executorService = Executors.newFixedThreadPool(10);
-
-
-    @Async
-    @Override
-    public void run(ApplicationArguments args) throws Exception {
-        log.info("开始监听kafka-topic：{}", topic);
-        List<String> topics = Collections.singletonList(topic);
-        consumer.subscribe(topics);
-
-        while (true) {
-            ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofMillis(100));
-            consumerRecords.forEach(record -> {
-                executorService.submit(() -> handleRecord(record));
-            });
-        }
-    }
-
-    private void handleRecord(ConsumerRecord<String, String> record) {
-        String value = record.value();
-        JSONObject jsonObject = JSONObject.parseObject(value);
-        log.info("value: {}", value);
-        eventPublisher.publishEvent(jsonObject);
-    }
-
-    @Override
-    public void onApplicationEvent(ContextClosedEvent event) {
-        log.info("关闭kafka和线程");
-        consumer.close();
-        executorService.shutdown();
-    }
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/controller/IotdbController.java

@@ -1,40 +0,0 @@
-package com.muyu.data.controller;
-
-import com.alibaba.fastjson2.JSONObject;
-import com.muyu.common.core.domain.Result;
-import com.muyu.common.iotdb.config.IotDBSessionConfig;
-import jakarta.annotation.Resource;
-import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
-import org.springframework.web.bind.annotation.*;
-
-import java.util.List;
-
-import static com.muyu.common.iotdb.constant.IotdbConstant.ROOT_DATA_DATAJSON;
-
-/**
- * iotdb测试
- *
- * @program: cloud-server
- * @author: WangXin
- * @create: 2024-09-28 19:09
- **/
-@RestController
-@RequestMapping("iotdb")
-public class IotdbController {
-
-    @Resource
-    private IotDBSessionConfig iotDBSessionConfig;
-
-    @PostMapping("/add")
-    public Result<?> addJSON(@RequestBody JSONObject jsonObject) {
-        iotDBSessionConfig.insertRecord(iotDBSessionConfig.getSessionPool(),ROOT_DATA_DATAJSON,System.currentTimeMillis(), List.of("datasource"),List.of(TSDataType.TEXT),jsonObject);
-        return Result.success();
-    }
-
-    @GetMapping("/findByDataTime/{time}")
-    public Result<?> findByDataTime(@PathVariable("time") Long time){
-        return null;
-    }
-
-
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/controller/RabbitController.java

@@ -1,32 +0,0 @@
-package com.muyu.data.controller;
-
-import org.springframework.amqp.rabbit.core.RabbitTemplate;
-import org.springframework.beans.factory.annotation.Autowired;
-import org.springframework.web.bind.annotation.GetMapping;
-import org.springframework.web.bind.annotation.RequestMapping;
-import org.springframework.web.bind.annotation.RestController;
-
-/**
- * rabbit测试
- *
- * @program: cloud-server
- * @author: WangXin
- * @create: 2024-09-30 00:04
- **/
-@RestController
-@RequestMapping("rabbit")
-public class RabbitController {
-
-    @Autowired
-    private RabbitTemplate rabbitTemplate;
-
-    @GetMapping("/send")
-    public void send() {
-        rabbitTemplate.convertAndSend("ONLINE_EXCHANGE", "", "vin123456");
-    }
-
-
-
-
-
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/controller/TestController.java

@@ -1,40 +1,133 @@
 package com.muyu.data.controller;
 
-import cn.hutool.json.JSONObject;
+import com.alibaba.fastjson2.JSONObject;
+import com.muyu.common.caffeine.enums.CacheNameEnums;
+import com.muyu.common.core.utils.uuid.UUID;
+import com.muyu.common.iotdb.config.IotDBSessionConfig;
+import com.muyu.common.kafka.constant.KafkaConstants;
+import com.muyu.common.rabbit.constants.RabbitConstants;
+import com.muyu.data.domain.Information;
+import jakarta.annotation.Resource;
+import lombok.extern.log4j.Log4j2;
 import org.apache.kafka.clients.producer.KafkaProducer;
 import org.apache.kafka.clients.producer.ProducerRecord;
-import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.cache.Cache;
+import org.springframework.cache.CacheManager;
+import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestParam;
 import org.springframework.web.bind.annotation.RestController;
 
 /**
-* 测试kafka
+* 测试
 * @program: cloud-server
 * @author: WangXin
 * @create: 2024-09-28 14:55
 **/
 @RestController
 @RequestMapping
+@Log4j2
 public class TestController {
 
-    @Autowired
+    @Resource
-    private KafkaProducer kafkaProducer;
+    private KafkaProducer<String,JSONObject> kafkaProducer;
+    @Resource
+    private RabbitTemplate rabbitTemplate;
+    @Resource
+    private IotDBSessionConfig iotDBConfig;
+    @Resource
+    private RedisTemplate<String,String> redisTemplate;
+//    @Resource
+//    private CaffeineCacheUtils cacheUtils;
 
+    @Resource
+    private CacheManager cacheManager;
 
-    private static final String topic = "vehicle";
+    @GetMapping("/testKafka")
-    @GetMapping("send")
+    public void sendMsg() {
-    public String sendKafka(){
+        try {
-
+            // 测试数据
-        JSONObject entries = new JSONObject();
+            Information information = new Information();
-        entries.set("car_vin","vin123123");
+            JSONObject from = JSONObject.from(information);
-        entries.set("car_name","奥迪");
+            ProducerRecord<String, JSONObject> producerRecord = new ProducerRecord<>(KafkaConstants.KafkaTopic, from);
-        String entriesString = entries.toString();
-        ProducerRecord<String, String> producerRecord = new ProducerRecord<String, String>(topic,entriesString);
             kafkaProducer.send(producerRecord);
+            System.out.println("同步消息发送成功: " + from);
+        } catch (Exception e) {
+            e.printStackTrace();
+            System.out.println("同步消息发送失败");
+        }
+    }
 
-        return "success";
+    @GetMapping("/testRabbit/GoOnline")
+    public void testRabbitGoOnline(@RequestParam("msg") String msg) {
+        rabbitTemplate.convertAndSend(RabbitConstants.GO_ONLINE_QUEUE, msg, message -> {
+            message.getMessageProperties().setMessageId(UUID.randomUUID().toString().replace("-",""));
+            return message;
+        });
+    }
+
+    @GetMapping("/testRabbit/Downline")
+    public void testRabbitDownline(@RequestParam("msg") String msg) {
+        rabbitTemplate.convertAndSend(RabbitConstants.DOWNLINE_QUEUE, msg, message -> {
+            message.getMessageProperties().setMessageId(UUID.randomUUID().toString().replace("-",""));
+            return message;
+        });
+    }
+
+    @GetMapping("/insertData")
+    public void insertData(@RequestParam("deviceId") String deviceId, @RequestParam("time") long time, @RequestParam("value") double value) throws Exception {
+        String sql = String.format("insert into root.one.%s(timestamp, temperature) values (%d, %f)", deviceId, time, value);
+        iotDBConfig.getSessionPool().executeNonQueryStatement(sql);
+    }
+
+    @GetMapping("/testSetRedis")
+    public void testSetRedis(@RequestParam("key") String key,@RequestParam("value") String value) {
+        redisTemplate.opsForValue().set(key,value);
+    }
+
+    @GetMapping("/testGetCache")
+    public void testGetCache(@RequestParam("cacheName") String cacheName,@RequestParam("key") String key) {
+        Cache cache = cacheManager.getCache(cacheName);
+        if (cache != null) {
+            String v = cache.get(key,String.class);
+            log.info("缓存值为: {}",v);
+        }else {
+            log.info("无缓存");
+        }
+    }
+
+    @GetMapping("/textSetCache")
+    public void textSetCache(
+            @RequestParam("cacheName") String cacheName,
+            @RequestParam("key") String key,
+            @RequestParam("value") String value) {
+        Cache cache = cacheManager.getCache(cacheName);
+        if (cache != null){
+            cache.put(key, value);
+            log.info("设置缓存成功");
+        }else {
+            log.info("无缓存");
+        }
+    }
+
+    @GetMapping("/testDelCache")
+    public void testDelCache(@RequestParam("cacheName") String cacheName) {
+        if (!CacheNameEnums.isCode(cacheName)){
+            log.info("缓存分区不存在");
+            return;
+        }
+        Cache cache = cacheManager.getCache(cacheName);
+        if (cache != null) {
+            cache.invalidate();
+            log.info("删除缓存成功");
+        }else{
+            log.info("无缓存");
+        }
     }
 
 
+
+
 }

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/domain/CarInformation.java

@@ -0,0 +1,45 @@
+package com.muyu.data.domain;
+
+import com.alibaba.fastjson.JSON;
+import com.muyu.common.core.annotation.Excel;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * 车辆实时信息实体类
+ * @author WangXIn
+ * @package：com.muyu.event.domian
+ * @name：CarInformation
+ * @date：2024/9/28 19:25
+ */
+@Data
+@AllArgsConstructor
+@NoArgsConstructor
+public class CarInformation {
+
+    /**
+     * 时间戳
+     */
+    @Excel(name = "时间戳")
+    private Long Time;
+
+    /**
+     * 车辆vin码
+     */
+    @Excel(name = "车辆vin码")
+    private String carVin;
+
+    /**
+     * 车辆实时信息
+     */
+    @Excel(name = "车辆实时信息")
+    private String information;
+
+
+
+
+
+
+
+}

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

@@ -1,42 +0,0 @@
-package com.muyu.data.listener;
-
-import com.alibaba.fastjson2.JSONObject;
-import com.muyu.data.basic.EventCustom;
-import com.muyu.data.basic.EventListener;
-import org.springframework.context.annotation.Configuration;
-
-import java.util.ArrayList;
-import java.util.List;
-
-/**
- * 添加数据库事件
- * @program: cloud-server
- * @author: WangXin
- * @create: 2024-09-29 17:34
- **/
-@Configuration
-public class AddDatabaseListener implements EventListener {
-
-
-    @Override
-    public void onEvent(EventCustom event) {
-
-        JSONObject jsonObject = event.getData();
-        List<String> keys = new ArrayList<>();
-        List<String> values = new ArrayList<>();
-
-        jsonObject.forEach((key, value) -> {
-            keys.add(key);
-            values.add((String) value);
-        });
-
-        long time = System.currentTimeMillis();
-
-
-    }
-
-    @Override
-    public void onApplicationEvent(EventCustom event) {
-        onEvent(event);
-    }
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/monitor/CatRabbitMonitor.java

@@ -1,45 +0,0 @@
-package com.muyu.data.monitor;
-
-import com.muyu.common.core.domain.Result;
-import com.muyu.common.rabbit.constants.RabbitmqConstants;
-import com.muyu.common.rabbit.consumer.RabbitMQConsumerUtil;
-import com.muyu.common.rabbit.producer.RabbitMQProducerUtil;
-import jakarta.annotation.Resource;
-import org.springframework.amqp.rabbit.annotation.RabbitListener;
-import org.springframework.stereotype.Component;
-
-/**
- * @Author WangXin
- * @Data 2024/9/30
- * @Description 车辆上下线监听器
- * @Version 1.0.0
- */
-@Component
-public class CatRabbitMonitor {
-
-    @Resource
-    private RabbitMQConsumerUtil rabbitMQConsumerUtil;
-    @Resource
-    private RabbitMQProducerUtil rabbitMQProducerUtil;
-
-    @RabbitListener(queues = {"queue_topic_01"})
-    public void topicConsumer01(String msg){
-        System.out.println("消费者 -01- 接收消息:" + msg);
-    }
-
-
-    /**
-     * 测试上线方法
-     * @param carVin
-     */
-    public void topicConsumer02(String carVin){
-        Result<?> result = rabbitMQProducerUtil.topicSendMessage(
-                RabbitmqConstants.TOP_BOTTOM_STITCHING,
-                RabbitmqConstants.TOP_RULE,
-                carVin
-        );
-        if (result.getCode() != Result.SUCCESS){
-            throw new RuntimeException(result.getMsg());
-        }
-    }
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/rabbit/DownlineRabbitConsumer.java

@@ -0,0 +1,69 @@
+package com.muyu.data.rabbit;
+
+
+import com.muyu.common.rabbit.constants.RabbitConstants;
+import com.rabbitmq.client.Channel;
+import jakarta.annotation.Resource;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.ObjectUtils;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.annotation.Queue;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.cache.Cache;
+import org.springframework.cache.CacheManager;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+
+/**
+ * 下线事件监听
+ * @Author: WangXin
+ * @Description: 车辆下线监听器
+ */
+@Slf4j
+@Component
+@Setter
+public class DownlineRabbitConsumer {
+    @Resource
+    private RedisTemplate<String,String> redisTemplate;
+    @Resource
+    private CacheManager cacheManager;
+
+    @RabbitListener(queuesToDeclare = {@Queue(RabbitConstants.DOWNLINE_QUEUE)})
+    public void downline(String vin, Message message, Channel channel) {
+        log.info("车辆 {} 下线, 配置信息准备中。。。",vin);
+        try {
+            // 重复性校验
+            Long add = redisTemplate.opsForSet().add(RabbitConstants.DOWNLINE_QUEUE, message.getMessageProperties().getMessageId());
+            if (add>0) {
+                deleteCarCache(vin);
+                log.info("车辆 {} 下线, 消息已确认。。。",vin);
+            } else {
+                log.info("车辆 {} 下线, 消息重复消费，已确认。。。",vin);
+            }
+            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+            log.info("车辆 {} 下线, 配置信息已准备完毕。。。",vin);
+        } catch (IOException e) {
+            try {
+                log.warn("车辆 {} 下线, 配置信息准备失败，返回队列，原因：{}", vin, e.getMessage());
+                channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
+            } catch (IOException ex) {
+                log.warn("车辆 {} 下线, 消息返回队列失败，原因：{}", vin, ex.getMessage());
+            }
+        }
+    }
+
+
+    /**
+     * 车辆下线 - 删除缓存
+     */
+    public void deleteCarCache(String vin) {
+        Cache cache = cacheManager.getCache(vin);
+        if (ObjectUtils.isNotEmpty(cache)){
+            cache.invalidate();
+        }
+        log.info("车辆编码:{},本地缓存删除完成...", vin);
+    }
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/rabbit/GoOnlineRabbitConsumer.java

@@ -0,0 +1,67 @@
+package com.muyu.data.rabbit;
+import com.muyu.common.caffeine.enums.CacheNameEnums;
+import com.muyu.common.rabbit.constants.RabbitConstants;
+import com.rabbitmq.client.Channel;
+import jakarta.annotation.Resource;
+import lombok.Setter;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.rabbit.annotation.Queue;
+import org.springframework.amqp.rabbit.annotation.RabbitListener;
+import org.springframework.cache.CacheManager;
+import org.springframework.data.redis.core.RedisTemplate;
+import org.springframework.stereotype.Component;
+
+import java.io.IOException;
+
+/**
+ * 上线事件监听
+ * @Author: WangXin
+ * @Name: GoOnlineRabbitConsumer
+ */
+@Slf4j
+@Component
+@Setter
+public class GoOnlineRabbitConsumer {
+    @Resource
+    private RedisTemplate<String,String> redisTemplate;
+    @Resource
+    private CacheManager cacheManager;
+
+    @RabbitListener(queuesToDeclare = {@Queue(RabbitConstants.GO_ONLINE_QUEUE)})
+    public void goOnline(String vin, Message message, Channel channel){
+        log.info("车辆 {} 上线, 配置信息准备中。。。",vin);
+        try {
+            // 重复性校验
+            Long add = redisTemplate.opsForSet().add(RabbitConstants.GO_ONLINE_QUEUE, message.getMessageProperties().getMessageId());
+            if (add>0) {
+                addCarCache(vin);
+                log.info("车辆 {} 上线, 消息已确认。。。",vin);
+            } else {
+                log.info("车辆 {} 上线, 消息重复消费，已确认。。。",vin);
+            }
+            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
+            log.info("车辆 {} 上线, 配置信息已准备完毕。。。",vin);
+        } catch (IOException e) {
+            try {
+                log.warn("车辆 {} 上线, 配置信息准备失败，返回队列，原因：{}", vin, e.getMessage());
+                channel.basicAck(message.getMessageProperties().getDeliveryTag(), true);
+            } catch (IOException ex) {
+                log.warn("车辆 {} 上线, 消息返回队列失败，原因：{}", vin, ex.getMessage());
+            }
+        }
+    }
+
+    /**
+     * 车辆上线 - 新增缓存
+     */
+    public void addCarCache(String vin) {
+        // 从Redis中获取缓存信息
+        for (String name : CacheNameEnums.getCodes()) {
+            String value = redisTemplate.opsForValue().get(name+":"+vin);
+            cacheManager.getCache(name).put(vin, value);
+            log.info("存储缓存, 缓存分区:[{}], 车辆编码:[{}], 存储值:[{}]", name, vin, value);
+        }
+        log.info("车辆编码:{},本地缓存完成...",vin);
+    }
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/service/TestService.java

@@ -0,0 +1,10 @@
+package com.muyu.data.service;
+
+/**
+ *
+ * @Author WangXin
+ * @Data 2024/10/3
+ * @Version 1.0.0
+ */
+public interface TestService {
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/service/impl/TestServiceImpl.java

@@ -0,0 +1,20 @@
+package com.muyu.data.service.impl;
+
+import com.muyu.data.service.TestService;
+import org.springframework.stereotype.Service;
+
+/**
+ * @author WangXIn
+ * @package：com.muyu.event.service.impl
+ * @name：TestServiceImpl
+ * @date：2024/9/29 21:00
+ */
+@Service
+public class TestServiceImpl implements TestService {
+
+
+
+
+
+
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/util/CacheUtil.java

@@ -1,38 +0,0 @@
-package com.muyu.data.util;
-
-import com.github.benmanes.caffeine.cache.Cache;
-import com.github.benmanes.caffeine.cache.Caffeine;
-import com.muyu.data.domain.Information;
-import org.springframework.stereotype.Component;
-
-/**
- * 缓存工具类
- *
- * @program: cloud-server
- * @author: WangXin
- * @create: 2024-09-30 10:08
- **/
-@Component
-public class CacheUtil<T> {
-
-    private final Cache<String, T> cache;
-
-    public CacheUtil() {
-        this.cache = Caffeine.newBuilder()
-                .maximumSize(500L)
-                .build();
-    }
-
-    public T get(String key) {
-        return cache.getIfPresent(key);
-    }
-
-    public void put(String key, T value) {
-        cache.put(key, value);
-    }
-
-    public void remove(String key) {
-        cache.invalidate(key);
-    }
-
-}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/util/Iotdb.java

@@ -0,0 +1,66 @@
+package com.muyu.data.util;
+
+import org.apache.iotdb.isession.SessionDataSet;
+import org.apache.iotdb.isession.util.Version;
+import org.apache.iotdb.rpc.IoTDBConnectionException;
+import org.apache.iotdb.rpc.StatementExecutionException;
+import org.apache.iotdb.session.Session;
+import org.apache.iotdb.tsfile.file.metadata.enums.TSDataType;
+
+import java.util.ArrayList;
+import java.util.List;
+
+public class Iotdb {
+    public static void main(String[] args) throws IoTDBConnectionException, StatementExecutionException {
+
+        System.out.println("测试数据库开始~~~~~");
+
+        // 初始化与连接
+        Session session = new Session.Builder()
+                .host("47.116.173.119")
+                .port(6667)
+                .username("root")
+                .password("root")
+                .version(Version.V_1_0)
+                .build();
+
+        // 开启session Rpc不压缩
+        session.open(false);
+
+        // 写入数据
+        ArrayList<Object> list = new ArrayList<>();
+        list.add(100);
+        insertRecord(session,list);
+
+        //查询数据
+        queryRecord(session);
+
+        //关闭连接
+        session.close();
+
+    }
+
+
+    private static void insertRecord(Session session, List<Object> values) throws IoTDBConnectionException, StatementExecutionException {
+        ArrayList<String> measurements = new ArrayList<>();
+        ArrayList<TSDataType> types = new ArrayList<>();
+
+        measurements.add("status");
+        types.add(TSDataType.INT32);
+        session.insertRecord("root.four.test",System.currentTimeMillis(),measurements,types,values);
+        System.out.println("————————————————写入数据成功————————————————");
+    }
+
+    private static void queryRecord(Session session) throws IoTDBConnectionException, StatementExecutionException {
+        System.out.println("————————————————查询数据开始————————————————");
+        try(SessionDataSet dataSet= session.executeQueryStatement("select * from root.four.test")){
+            System.out.println(dataSet.getColumnNames());
+            dataSet.setFetchSize(1024);
+            while (dataSet.hasNext()){
+                System.out.println(dataSet.next());
+            }
+
+        }
+        System.out.println("————————————————查询数据结束————————————————");
+    }
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/java/com/muyu/data/util/Receive.java

@@ -0,0 +1,48 @@
+package com.muyu.data.util;
+
+import org.eclipse.paho.client.mqttv3.*;
+
+public class Receive {
+
+    public static void main(String[] args) {
+        String topic = "vehicle";
+        String broker = "tcp://47.101.53.251:1883";
+        String clientId="lw";
+
+        try {
+            MqttClient mqttClient= new MqttClient(broker,clientId);
+            MqttConnectOptions connectOptions=new MqttConnectOptions();
+            connectOptions.setCleanSession(true);
+            System.out.println("Connecting to broker：" + broker);
+            mqttClient.connect(connectOptions);
+            System.out.println("已连接");
+            mqttClient.setCallback(new MqttCallback(){
+
+                @Override
+                public void connectionLost(Throwable throwable) {
+                    System.out.println("Connect lost!");
+                }
+
+                @Override
+                public void messageArrived(String s, MqttMessage mqttMessage) throws Exception {
+                    System.out.println("Message arrived. topic:"+topic);
+                }
+
+                @Override
+                public void deliveryComplete(IMqttDeliveryToken iMqttDeliveryToken) {
+
+                }
+            });
+
+            mqttClient.subscribe(topic);
+            System.out.println("Subscribed to topic " + topic);
+        } catch (MqttException e) {
+            System.out.println("reason "+e.getReasonCode());
+            System.out.println("msg " +e.getMessage());
+            System.out.println("loc " +e.getLocalizedMessage());
+            System.out.println("cause "+e.getCause());
+            System.out.println("excep "+e);
+            e.printStackTrace();
+        }
+    }
+}

cloud-modules/cloud-modules-data-process/cloud-modules-data-process-server/src/main/resources/bootstrap.yml

@@ -4,10 +4,10 @@ server:
 
 # nacos线上地址
 nacos:
-  addr: 123.57.152.124:8848
+  addr: 127.0.0.1:8848
   user-name: nacos
   password: nacos
-  namespace: five
+  namespace: wx
 # SPRING_AMQP_DESERIALIZATION_TRUST_ALL=true spring.amqp.deserialization.trust.all
 # Spring
 spring:
@@ -17,10 +17,6 @@ spring:
     port: 6667
     password: root
     maxSize: 10
-  amqp:
-    deserialization:
-      trust:
-        all: true
   main:
     allow-bean-definition-overriding: true
   application:
@@ -59,3 +55,7 @@ spring:
           - application-config-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
           # xxl-job 配置文件
           - application-xxl-config-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
+          # rabbit 配置文件
+          - application-rabbit-config-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
+          # kafka 配置文件
+          - application-kafka-config-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}

cloud-modules/cloud-modules-enterprise/enterpise-cache/pom.xml

@@ -19,8 +19,6 @@
 
     <dependencies>
 
-
-
         <dependency>
             <groupId>com.muyu</groupId>
             <artifactId>cloud-common-cache</artifactId>

cloud-modules/cloud-modules-enterprise/enterpise-cache/src/main/java/com/muyu/enterpise/cache/VehicleCacheService.java

@@ -1,4 +1,4 @@
-package com.muyu.enterprise.cache;
+package com.muyu.enterpise.cache;
 
 import com.muyu.common.cache.CacheAbsBasic;
 import com.muyu.domain.CarInformation;

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

@@ -66,6 +66,10 @@ public class SysCar extends BaseEntity{
     @Schema(name = "启用状态")
     private Integer state;
 
+    /** 车辆所属公司 */
+    @Schema(name = "公司Id")
+    private Integer deptId;
+
 
 
     @Override