feat：合并
chentaisen
parent
eb59a2400a
commit
5bb57ecbb2
|
|
@ -0,0 +1,35 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common</artifactId>
|
||||
<version>3.6.3</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>cloud-common-caffeine</artifactId>
|
||||
<description>
|
||||
cloud-common-caffeine caffeine缓存模块
|
||||
</description>
|
||||
<properties>
|
||||
<maven.compiler.source>17</maven.compiler.source>
|
||||
<maven.compiler.target>17</maven.compiler.target>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common-redis</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>com.github.ben-manes.caffeine</groupId>
|
||||
<artifactId>caffeine</artifactId>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
|
|
@ -0,0 +1,51 @@
|
|||
package com.muyu.common.caffeine.bean;
|
||||
|
||||
|
||||
import com.muyu.common.caffeine.enums.CacheNameEnums;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import com.github.benmanes.caffeine.cache.Caffeine;
|
||||
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.ArrayList;
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
|
||||
/**
|
||||
* Caffeine管理器
|
||||
* @Author: 胡杨
|
||||
* @Name: CaffeineCacheConfig
|
||||
* @Description: Caffeine管理器
|
||||
* @CreatedDate: 2024/9/26 上午11:52
|
||||
* @FilePath: com.muyu.common.caffeine.config
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class CaffeineManager {
|
||||
|
||||
/**
|
||||
* 创建缓存管理器
|
||||
* @return 缓存管理器实例
|
||||
*/
|
||||
@Bean
|
||||
public SimpleCacheManager simpleCacheManager() {
|
||||
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;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,17 @@
|
|||
package com.muyu.common.caffeine.constents;
|
||||
|
||||
/**
|
||||
* Caffeine常量
|
||||
* @Author: 胡杨
|
||||
* @Name: CaffeineContent
|
||||
* @Description: Caffeine常量
|
||||
* @CreatedDate: 2024/9/26 下午12:06
|
||||
* @FilePath: com.muyu.common.caffeine.constents
|
||||
*/
|
||||
|
||||
public class CaffeineContent {
|
||||
|
||||
public static final String CAR_VIN_KEY = "car:vin";
|
||||
|
||||
public static final String VIN = "vin";
|
||||
}
|
||||
|
|
@ -0,0 +1,68 @@
|
|||
package com.muyu.common.caffeine.enums;
|
||||
|
||||
import lombok.Getter;
|
||||
|
||||
import java.util.Arrays;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 缓存分区枚举
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: CacheNameEnums
|
||||
* @Description: 缓存分区枚举
|
||||
* @CreatedDate: 2024/10/2 上午9:17
|
||||
* @FilePath: com.muyu.common.caffeine.enums
|
||||
*/
|
||||
|
||||
@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();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,97 @@
|
|||
package com.muyu.common.caffeine.utils;
|
||||
|
||||
|
||||
import com.github.benmanes.caffeine.cache.Cache;
|
||||
import com.github.benmanes.caffeine.cache.Caffeine;
|
||||
import com.muyu.common.caffeine.constents.CaffeineContent;
|
||||
import com.muyu.common.caffeine.enums.CacheNameEnums;
|
||||
import com.muyu.common.redis.service.RedisService;
|
||||
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.cache.caffeine.CaffeineCache;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Collection;
|
||||
|
||||
/**
|
||||
* Caffeine缓存工具
|
||||
* @Author: 胡杨
|
||||
* @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;
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,2 @@
|
|||
com.muyu.common.caffeine.utils.CaffeineCacheUtils
|
||||
com.muyu.common.caffeine.bean.CaffeineManager
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common</artifactId>
|
||||
<version>3.6.3</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>cloud-common-iotdb</artifactId>
|
||||
<description>
|
||||
cloud-common-iotdb 时序数据库模块
|
||||
</description>
|
||||
<properties>
|
||||
<maven.compiler.source>17</maven.compiler.source>
|
||||
<maven.compiler.target>17</maven.compiler.target>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
</properties>
|
||||
|
||||
|
||||
<dependencies>
|
||||
<dependency>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common-core</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.apache.iotdb</groupId>
|
||||
<artifactId>iotdb-session</artifactId>
|
||||
<version>1.3.2</version>
|
||||
</dependency>
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common</artifactId>
|
||||
<version>3.6.3</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>cloud-common-kafka</artifactId>
|
||||
<description>
|
||||
cloud-common-kafka kafka中间件模块
|
||||
</description>
|
||||
<properties>
|
||||
<maven.compiler.source>17</maven.compiler.source>
|
||||
<maven.compiler.target>17</maven.compiler.target>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
|
||||
<dependency>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common-redis</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.apache.kafka</groupId>
|
||||
<artifactId>kafka-clients</artifactId>
|
||||
<version>3.0.0</version>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
|
||||
</project>
|
||||
|
|
@ -21,6 +21,10 @@
|
|||
<module>cloud-common-xxl</module>
|
||||
<module>cloud-common-rabbit</module>
|
||||
<module>cloud-common-saas</module>
|
||||
<module>cloud-common-cache</module>
|
||||
<module>cloud-common-caffeine</module>
|
||||
<module>cloud-common-iotdb</module>
|
||||
<module>cloud-common-kafka</module>
|
||||
</modules>
|
||||
|
||||
<artifactId>cloud-common</artifactId>
|
||||
|
|
|
|||
|
|
@ -0,0 +1,155 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0"
|
||||
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
|
||||
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
<parent>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-server</artifactId>
|
||||
<version>3.6.3</version>
|
||||
</parent>
|
||||
|
||||
<artifactId>cloud-modules-data-processing</artifactId>
|
||||
|
||||
<description>
|
||||
cloud-data-processing 数据处理模块
|
||||
</description>
|
||||
|
||||
<properties>
|
||||
<maven.compiler.source>17</maven.compiler.source>
|
||||
<maven.compiler.target>17</maven.compiler.target>
|
||||
<project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
|
||||
</properties>
|
||||
<dependencies>
|
||||
|
||||
<dependency>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common-kafka</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common-caffeine</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common-rabbit</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common-iotdb</artifactId>
|
||||
<version>3.6.3</version>
|
||||
</dependency>
|
||||
|
||||
<!-- SpringCloud Alibaba Nacos -->
|
||||
<dependency>
|
||||
<groupId>com.alibaba.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-alibaba-nacos-discovery</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- SpringCloud Alibaba Nacos Config -->
|
||||
<dependency>
|
||||
<groupId>com.alibaba.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- SpringCloud Alibaba Sentinel -->
|
||||
<dependency>
|
||||
<groupId>com.alibaba.cloud</groupId>
|
||||
<artifactId>spring-cloud-starter-alibaba-sentinel</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- SpringBoot Actuator -->
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-actuator</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-tomcat</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- Mysql Connector -->
|
||||
<dependency>
|
||||
<groupId>com.mysql</groupId>
|
||||
<artifactId>mysql-connector-j</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- MuYu Common DataScope -->
|
||||
<dependency>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common-datascope</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>com.muyu</groupId>
|
||||
<artifactId>cloud-common-datasource</artifactId>
|
||||
</dependency>
|
||||
|
||||
<!-- <dependency>-->
|
||||
<!-- <groupId>org.apache.iotdb</groupId>-->
|
||||
<!-- <artifactId>iotdb-jdbc</artifactId>-->
|
||||
<!-- <version>0.12.1</version>-->
|
||||
<!-- </dependency>-->
|
||||
<!-- <dependency>-->
|
||||
<!-- <groupId>com.alibaba</groupId>-->
|
||||
<!-- <artifactId>druid-spring-boot-starter</artifactId>-->
|
||||
<!-- <version>1.2.20</version>-->
|
||||
<!-- </dependency>-->
|
||||
|
||||
|
||||
<!-- <dependency>-->
|
||||
<!-- <groupId>org.mybatis</groupId>-->
|
||||
<!-- <artifactId>mybatis-spring</artifactId>-->
|
||||
<!-- <version>3.0.3</version>-->
|
||||
<!-- </dependency>-->
|
||||
<!-- <dependency>-->
|
||||
<!-- <groupId>com.baomidou</groupId>-->
|
||||
<!-- <artifactId>mybatis-plus-boot-starter</artifactId>-->
|
||||
<!-- <version>3.5.5</version>-->
|
||||
<!-- </dependency>-->
|
||||
|
||||
<!--<!– <dependency>–>-->
|
||||
<!--<!– <groupId>org.apache.iotdb</groupId>–>-->
|
||||
<!--<!– <artifactId>iotdb-session</artifactId>–>-->
|
||||
<!--<!– <version>1.3.2</version>–>-->
|
||||
<!--<!– </dependency>–>-->
|
||||
|
||||
<!-- <!– Druid –>-->
|
||||
<!-- <dependency>-->
|
||||
<!-- <groupId>com.alibaba</groupId>-->
|
||||
<!-- <artifactId>druid-spring-boot-3-starter</artifactId>-->
|
||||
<!-- <version>${druid.version}</version>-->
|
||||
<!-- </dependency>-->
|
||||
|
||||
<!-- <!– Dynamic DataSource –>-->
|
||||
<!-- <dependency>-->
|
||||
<!-- <groupId>com.baomidou</groupId>-->
|
||||
<!-- <artifactId>dynamic-datasource-spring-boot3-starter</artifactId>-->
|
||||
<!-- <version>${dynamic-ds.version}</version>-->
|
||||
<!-- </dependency>-->
|
||||
|
||||
|
||||
</dependencies>
|
||||
|
||||
<build>
|
||||
<finalName>${project.artifactId}</finalName>
|
||||
<plugins>
|
||||
<plugin>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-maven-plugin</artifactId>
|
||||
<executions>
|
||||
<execution>
|
||||
<goals>
|
||||
<goal>repackage</goal>
|
||||
</goals>
|
||||
</execution>
|
||||
</executions>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
|
|
@ -0,0 +1,32 @@
|
|||
package com.muyu.data.processing;
|
||||
|
||||
import com.muyu.common.kafka.constants.KafkaConstants;
|
||||
|
||||
import com.muyu.common.security.annotation.EnableCustomConfig;
|
||||
import com.muyu.common.security.annotation.EnableMyFeignClients;
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
|
||||
import org.springframework.boot.SpringApplication;
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
|
||||
|
||||
/**
|
||||
* 数据处理模块启动器
|
||||
* @Author: 胡杨
|
||||
* @Name: MyData
|
||||
* @Description: 数据处理模块启动器
|
||||
* @CreatedDate: 2024/9/26 下午7:31
|
||||
* @FilePath: com.muyu.data.processing
|
||||
*/
|
||||
@EnableRabbit
|
||||
@EnableCustomConfig
|
||||
@EnableMyFeignClients
|
||||
@SpringBootApplication(scanBasePackages = {"com.muyu"})
|
||||
public class MyDataApplication {
|
||||
public static void main(String[] args) {
|
||||
SpringApplication.run(MyDataApplication.class, args);
|
||||
|
||||
System.out.println("MyData 模块启动成功!");
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,100 @@
|
|||
package com.muyu.data.processing.controller;
|
||||
|
||||
import com.muyu.common.core.domain.Result;
|
||||
import com.muyu.common.security.utils.SecurityUtils;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import com.muyu.data.processing.domain.IotDbData;
|
||||
import com.muyu.data.processing.service.DataProcessingService;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
|
||||
import org.springframework.web.bind.annotation.*;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.util.Date;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 数据处理控制层
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: DataProcessing
|
||||
* @Description: 数据处理控制层
|
||||
* @CreatedDate: 2024/9/28 下午3:53
|
||||
* @FilePath: com.muyu.data.processing.controller
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@RestController
|
||||
@RequestMapping("/DataProcessing")
|
||||
public class DataProcessingController {
|
||||
@Resource
|
||||
private DataProcessingService service;
|
||||
|
||||
/**
|
||||
* 查看数据库有多少组
|
||||
*
|
||||
* @return
|
||||
*/
|
||||
@GetMapping("/selectStorageGroup")
|
||||
public Result selectStorageGroup() {
|
||||
List<String> v = service.selectStorageGroup();if (v.size() > 0) {v.forEach(x -> {
|
||||
System.out.println("group------------------" + x.toString());
|
||||
});
|
||||
return Result.success(v);
|
||||
} else {
|
||||
return Result.error(false);
|
||||
}
|
||||
}
|
||||
|
||||
@GetMapping("/selectCarData")
|
||||
public Result selectCarData(@RequestParam("vin") String vin) {
|
||||
// String firmCode = SecurityUtils.getSaasKey();
|
||||
String firmCode = "firm01";
|
||||
return Result.success(service.selectCarData(firmCode,vin));
|
||||
}
|
||||
|
||||
@PostMapping("/addCarData")
|
||||
public Result addCarData(@RequestBody IotDbData data) {
|
||||
HashMap<String, BasicData> hashMap = new HashMap<>();
|
||||
hashMap.put("timestamp", BasicData
|
||||
.builder()
|
||||
.key("timestamp")
|
||||
.label("时间戳")
|
||||
.value(String.valueOf(data.getTimestamp()))
|
||||
.type("string")
|
||||
.build());
|
||||
hashMap.put("vin", BasicData
|
||||
.builder()
|
||||
.key("vin")
|
||||
.label("VIN码")
|
||||
.value(data.getVin())
|
||||
.type("string")
|
||||
.build());
|
||||
hashMap.put("latitude", BasicData
|
||||
.builder()
|
||||
.key("latitude")
|
||||
.label("纬度")
|
||||
.value(data.getLatitude())
|
||||
.type("long")
|
||||
.build());
|
||||
hashMap.put("longitude", BasicData
|
||||
.builder()
|
||||
.key("longitude")
|
||||
.label("经度")
|
||||
.value(data.getLongitude())
|
||||
.type("long")
|
||||
.build());
|
||||
hashMap.put("firmCode", BasicData
|
||||
.builder()
|
||||
.key("firmCode")
|
||||
.label("企业编码")
|
||||
.value("firm01")
|
||||
.type("string")
|
||||
.build());
|
||||
return Result.success(service.addCarData(hashMap));
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,165 @@
|
|||
package com.muyu.data.processing.controller;
|
||||
|
||||
|
||||
import com.github.benmanes.caffeine.cache.Caffeine;
|
||||
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.constants.KafkaConstants;
|
||||
import com.muyu.common.rabbit.constants.RabbitConstants;
|
||||
import jakarta.annotation.Resource;
|
||||
import org.apache.kafka.clients.producer.KafkaProducer;
|
||||
import org.apache.kafka.clients.producer.ProducerRecord;
|
||||
import org.springframework.amqp.rabbit.core.RabbitTemplate;
|
||||
import org.springframework.cache.Cache;
|
||||
import org.springframework.cache.caffeine.CaffeineCache;
|
||||
import org.springframework.cache.support.SimpleCacheManager;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.web.bind.annotation.*;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
|
||||
import java.util.ArrayList;
|
||||
|
||||
/**
|
||||
* 测试控制层
|
||||
* @Author: 胡杨
|
||||
* @Name: Test
|
||||
* @Description:
|
||||
* @CreatedDate: 2024/9/27 上午10:54
|
||||
* @FilePath: com.muyu.data.processing.controller
|
||||
*/
|
||||
@Slf4j
|
||||
@RestController
|
||||
@RequestMapping("/Test")
|
||||
public class TestController {
|
||||
@Resource
|
||||
private KafkaProducer<String,String> kafkaProducer;
|
||||
@Resource
|
||||
private RabbitTemplate rabbitTemplate;
|
||||
@Resource
|
||||
private IotDBSessionConfig iotDBSessionConfig;
|
||||
@Resource
|
||||
private RedisTemplate<String,String> redisTemplate;
|
||||
// @Resource
|
||||
// private CaffeineCacheUtils cacheUtils;
|
||||
|
||||
@Resource
|
||||
private SimpleCacheManager cacheManager;
|
||||
|
||||
@GetMapping("/testKafka")
|
||||
public void sendMsg() {
|
||||
try {
|
||||
// 测试数据
|
||||
String jsonString = """
|
||||
[{
|
||||
"key": "vin",
|
||||
"label": "VIN码",
|
||||
"type": "String",
|
||||
"value": "vin999999"
|
||||
},{
|
||||
"key": "timestamp",
|
||||
"label": "时间戳",
|
||||
"type": "long",
|
||||
"value": "1727534036893"
|
||||
},{
|
||||
"key": "latitude",
|
||||
"label": "纬度",
|
||||
"type": "int",
|
||||
"value": "66.898"
|
||||
},{
|
||||
"key": "longitude",
|
||||
"label": "经度",
|
||||
"type": "int",
|
||||
"value": "99.12"
|
||||
}]""";
|
||||
ProducerRecord<String, String> producerRecord = new ProducerRecord<>(KafkaConstants.KafkaTopic, jsonString);
|
||||
kafkaProducer.send(producerRecord);
|
||||
System.out.println("同步消息发送成功: " + jsonString);
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
System.out.println("同步消息发送失败");
|
||||
}
|
||||
}
|
||||
|
||||
@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);
|
||||
iotDBSessionConfig.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("无缓存");
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@GetMapping("/testAddCache")
|
||||
public void testAddCache(@RequestParam("vin") String vin) {
|
||||
ArrayList<CaffeineCache> caches = new ArrayList<>();
|
||||
caches.add(new CaffeineCache(vin, Caffeine.newBuilder().recordStats().build()));
|
||||
cacheManager.setCaches(caches);
|
||||
log.info("缓存管理器创建新分区: {}", vin);
|
||||
}
|
||||
|
||||
@GetMapping("/testGetCacheNames")
|
||||
public void testGetCacheNames() {
|
||||
cacheManager.initializeCaches();
|
||||
log.info("缓存分区列表: {}", cacheManager.getCacheNames());
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
package com.muyu.data.processing.domain;
|
||||
|
||||
import lombok.*;
|
||||
|
||||
import java.io.Serializable;
|
||||
|
||||
/**
|
||||
* 报文信息 时序实体类
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: DataProcessing
|
||||
* @Description: 报文信息 时序实体类
|
||||
* @CreatedDate: 2024/9/28 下午3:48
|
||||
* @FilePath: com.muyu.data.processing.domain
|
||||
*/
|
||||
|
||||
@Data
|
||||
@ToString
|
||||
@Builder
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class BasicData implements Serializable {
|
||||
|
||||
private String key;
|
||||
private String label;
|
||||
private String value;
|
||||
private String type;
|
||||
|
||||
// public void setValueClass() {
|
||||
// Class<?> info = ClassType.getInfo(type);
|
||||
// if (info.isInstance(value)){
|
||||
// value = info.cast(value);
|
||||
// }
|
||||
// }
|
||||
}
|
||||
|
||||
|
|
@ -0,0 +1,25 @@
|
|||
package com.muyu.data.processing.domain;
|
||||
|
||||
import lombok.*;
|
||||
|
||||
/**
|
||||
* 车辆信息
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: CarData
|
||||
* @Description: 车辆信息
|
||||
* @CreatedDate: 2024/10/2 下午2:34
|
||||
* @FilePath: com.muyu.data.processing.domain
|
||||
*/
|
||||
|
||||
@Data
|
||||
@Builder
|
||||
@ToString
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class CarData {
|
||||
private String vin;
|
||||
private long timestamp;
|
||||
private String latitude;
|
||||
private String longitude;
|
||||
}
|
||||
|
|
@ -0,0 +1,35 @@
|
|||
package com.muyu.data.processing.domain;
|
||||
|
||||
import com.muyu.common.core.web.domain.BaseEntity;
|
||||
import lombok.*;
|
||||
import lombok.experimental.SuperBuilder;
|
||||
|
||||
import java.util.Date;
|
||||
|
||||
/**
|
||||
* 报文信息 时序实体类
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: DataProcessing
|
||||
* @Description: 报文信息 时序实体类
|
||||
* @CreatedDate: 2024/9/28 下午3:48
|
||||
* @FilePath: com.muyu.data.processing.domain
|
||||
*/
|
||||
|
||||
@EqualsAndHashCode(callSuper = true)
|
||||
@Data
|
||||
@ToString
|
||||
@SuperBuilder
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class IotDbData extends BaseEntity {
|
||||
private long timestamp;
|
||||
|
||||
private String vin;
|
||||
|
||||
private String latitude;
|
||||
private String longitude;
|
||||
|
||||
|
||||
}
|
||||
|
||||
|
|
@ -0,0 +1,22 @@
|
|||
package com.muyu.data.processing.domain;
|
||||
|
||||
import lombok.*;
|
||||
|
||||
/**
|
||||
* 临时类2
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: Temporary2
|
||||
* @Description: 临时类2
|
||||
* @CreatedDate: 2024/9/30 下午7:27
|
||||
* @FilePath: com.muyu.data.processing.domain
|
||||
*/
|
||||
|
||||
@Data
|
||||
@ToString
|
||||
@Builder
|
||||
@NoArgsConstructor
|
||||
@AllArgsConstructor
|
||||
public class Temporary2 {
|
||||
private String test;
|
||||
}
|
||||
|
|
@ -0,0 +1,71 @@
|
|||
package com.muyu.data.processing.kafka;
|
||||
|
||||
|
||||
import cn.hutool.core.thread.ThreadUtil;
|
||||
import cn.hutool.json.JSONUtil;
|
||||
import com.alibaba.nacos.shaded.com.google.common.collect.Lists;
|
||||
import com.muyu.common.kafka.constants.KafkaConstants;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import com.muyu.data.processing.service.DataProcessingService;
|
||||
import com.muyu.data.processing.strategy.core.StartStrategy;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecord;
|
||||
import org.apache.kafka.clients.consumer.ConsumerRecords;
|
||||
import org.apache.kafka.clients.consumer.KafkaConsumer;
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Collection;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* kafka消费者
|
||||
* @Author: 胡杨
|
||||
* @Name: KafkaConsumerService
|
||||
* @Description: kafka消费者
|
||||
* @CreatedDate: 2024/9/27 上午9:27
|
||||
* @FilePath: com.muyu.data.processing.kafka
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class KafkaConsumerService 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对象
|
||||
List<BasicData> dataList = JSONUtil.toList(originalMsg, BasicData.class);
|
||||
log.info("从Kafka中消费的实体数据: " + dataList);
|
||||
// 执行策略
|
||||
startStrategy.applyStrategy(getDataMap(dataList));
|
||||
}
|
||||
}
|
||||
});
|
||||
thread.start();
|
||||
}
|
||||
|
||||
private HashMap<String, BasicData> getDataMap(List<BasicData> dataList) {
|
||||
HashMap<String, BasicData> basicDataHashMap = new HashMap<>();
|
||||
dataList.forEach(data -> basicDataHashMap.put(data.getKey(), data));
|
||||
return basicDataHashMap;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,35 @@
|
|||
package com.muyu.data.processing.mapper;
|
||||
|
||||
import com.muyu.data.processing.domain.CarData;
|
||||
import com.muyu.data.processing.domain.IotDbData;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import org.apache.ibatis.annotations.Mapper;
|
||||
import org.apache.ibatis.annotations.Param;
|
||||
import org.springframework.stereotype.Repository;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 数据处理持久层
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: DataPeocessingMapper
|
||||
* @Description: 数据处理持久层
|
||||
* @CreatedDate: 2024/9/28 下午3:47
|
||||
* @FilePath: com.muyu.data.processing.mapper
|
||||
*/
|
||||
|
||||
@Repository
|
||||
@Mapper
|
||||
public interface DataProcessingMapper{
|
||||
|
||||
List<String> selectStorageGroup();
|
||||
|
||||
Integer insIotDbData(@Param("key") String key, @Param("value") String value);
|
||||
|
||||
void strategyCheck(@Param("dataList") List<BasicData> dataList);
|
||||
|
||||
Integer insIotDbDataVo(IotDbData build);
|
||||
|
||||
List<CarData> selectCarData(@Param("tableName") String tableName);
|
||||
}
|
||||
|
|
@ -0,0 +1,74 @@
|
|||
package com.muyu.data.processing.rebbit;
|
||||
|
||||
|
||||
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.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;
|
||||
import java.util.HashSet;
|
||||
|
||||
/**
|
||||
* 下线事件监听
|
||||
* @Author: 胡杨
|
||||
* @Name: DownlineRabbitConsumer
|
||||
* @Description: 车辆下线监听器
|
||||
* @CreatedDate: 2024/9/26 下午8:21
|
||||
* @FilePath: com.muyu.data.processing.rebbit
|
||||
*/
|
||||
@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);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,74 @@
|
|||
package com.muyu.data.processing.rebbit;
|
||||
|
||||
|
||||
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.beans.factory.annotation.Value;
|
||||
import org.springframework.cache.CacheManager;
|
||||
import org.springframework.data.redis.core.RedisTemplate;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 上线事件监听
|
||||
* @Author: 胡杨
|
||||
* @Name: GoOnlineRabbitConsumer
|
||||
* @Description: 上线事件
|
||||
* @CreatedDate: 2024/9/26 下午7:38
|
||||
* @FilePath: com.muyu.data.processing.rebbit
|
||||
*/
|
||||
@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);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
package com.muyu.data.processing.service;
|
||||
|
||||
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import com.muyu.data.processing.domain.CarData;
|
||||
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 数据处理业务层
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: DataProcessing
|
||||
* @Description: 数据处理业务层
|
||||
* @CreatedDate: 2024/9/28 下午3:52
|
||||
* @FilePath: com.muyu.data.processing.server
|
||||
*/
|
||||
|
||||
public interface DataProcessingService{
|
||||
|
||||
/**
|
||||
* 选择存储组
|
||||
*
|
||||
* @return {@link List }<{@link String }>
|
||||
*/
|
||||
List<String> selectStorageGroup();
|
||||
|
||||
void strategyCheck(List<BasicData> dataList);
|
||||
|
||||
Integer insIotDbData(String key, String value);
|
||||
|
||||
List<CarData> selectCarData(String firmCode, String vin);
|
||||
|
||||
Object addCarData(HashMap<String, BasicData> hashMap);
|
||||
}
|
||||
|
|
@ -0,0 +1,148 @@
|
|||
package com.muyu.data.processing.service.impl;
|
||||
|
||||
|
||||
import javax.annotation.Resource;
|
||||
|
||||
import com.muyu.common.iotdb.config.IotDBSessionConfig;
|
||||
import com.muyu.data.processing.domain.CarData;
|
||||
import com.muyu.data.processing.domain.IotDbData;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import org.apache.iotdb.isession.SessionDataSet;
|
||||
import org.apache.iotdb.isession.pool.SessionDataSetWrapper;
|
||||
import org.apache.iotdb.rpc.IoTDBConnectionException;
|
||||
import org.apache.iotdb.rpc.StatementExecutionException;
|
||||
import org.apache.iotdb.session.pool.SessionPool;
|
||||
import org.apache.iotdb.tsfile.read.common.Field;
|
||||
import org.apache.iotdb.tsfile.read.common.RowRecord;
|
||||
import org.springframework.stereotype.Service;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import com.muyu.data.processing.mapper.DataProcessingMapper;
|
||||
import com.muyu.data.processing.service.DataProcessingService;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 数据处理实现层
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: DataProcessing
|
||||
* @Description: 数据处理实现层
|
||||
* @CreatedDate: 2024/9/28 下午3:52
|
||||
* @FilePath: com.muyu.data.processing.server.impl
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Service
|
||||
public class DataProcessingServiceImpl implements DataProcessingService {
|
||||
@Resource
|
||||
private DataProcessingMapper mapper;
|
||||
@Resource
|
||||
private SessionPool sessionPool;
|
||||
|
||||
|
||||
@Override
|
||||
public List<String> selectStorageGroup() {
|
||||
return mapper.selectStorageGroup();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void strategyCheck(List<BasicData> dataList) {
|
||||
HashMap<String, BasicData> kafkaDataHashMap = new HashMap<>();
|
||||
dataList.forEach(data -> kafkaDataHashMap.put(data.getKey(), data));
|
||||
// Result<String[]> result = rootStrategy.applyStrategy(kafkaDataHashMap);
|
||||
// String[] data = result.getData();
|
||||
// insIotDbData(data[0],data[1]);
|
||||
IotDbData build = IotDbData.builder()
|
||||
.vin(kafkaDataHashMap.get("vin").getValue())
|
||||
.timestamp(Long.parseLong(kafkaDataHashMap.get("timestamp").getValue()))
|
||||
.latitude(kafkaDataHashMap.get("latitude").getValue())
|
||||
.longitude(kafkaDataHashMap.get("longitude").getValue())
|
||||
.build();
|
||||
mapper.insIotDbDataVo(build);
|
||||
// dataList.forEach(KafkaData::setValueClass);
|
||||
// mapper.strategyCheck(dataList);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Integer insIotDbData(String key, String value) {
|
||||
return mapper.insIotDbData(key, value);
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<CarData> selectCarData(String firmCode, String vin) {
|
||||
ArrayList<CarData> carDataList = new ArrayList<>();
|
||||
String sql = "select * from root.one."+firmCode+"."+vin;
|
||||
try {
|
||||
SessionDataSetWrapper dataSetWrapper = sessionPool.executeQueryStatement(sql);
|
||||
List<String> columnNames = dataSetWrapper.getColumnNames();
|
||||
while (dataSetWrapper.hasNext()){
|
||||
RowRecord next = dataSetWrapper.next();
|
||||
CarData data = getCarData(vin, next, columnNames);
|
||||
carDataList.add(data);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
return carDataList;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object addCarData(HashMap<String, BasicData> hashMap) {
|
||||
// StringBuilder sql = new StringBuilder("insert into root.one.");
|
||||
// sql.append(hashMap.get("firmCode").getValue())
|
||||
// .append(".")
|
||||
// .append(hashMap.get("vin").getValue())
|
||||
// .append("(");
|
||||
// hashMap.remove("firmCode");
|
||||
// hashMap.remove("vin");
|
||||
// StringBuilder keys = new StringBuilder();
|
||||
// StringBuilder values = new StringBuilder();
|
||||
// hashMap.keySet().forEach(key -> {
|
||||
// if (hashMap.get(key) != null) {
|
||||
// keys.append(key).append(",");
|
||||
// if ("String".equals(hashMap.get(key).getType())) {
|
||||
// values.append("'")
|
||||
// .append(hashMap.get(key).getValue())
|
||||
// .append("'")
|
||||
// .append(",");
|
||||
// }else {
|
||||
// values.append(hashMap.get(key).getValue())
|
||||
// .append(",");
|
||||
// }
|
||||
// }
|
||||
// });
|
||||
// sql.append(keys.substring(0, keys.length() - 1))
|
||||
// .append(") values (")
|
||||
// .append(values.substring(0, values.length() - 1))
|
||||
// .append(")");
|
||||
// try {
|
||||
// sessionPool.executeNonQueryStatement(sql.toString());
|
||||
// } catch (StatementExecutionException e) {
|
||||
// throw new RuntimeException(e);
|
||||
// } catch (IoTDBConnectionException e) {
|
||||
// throw new RuntimeException(e);
|
||||
// }
|
||||
// log.info("成功执行sql语句: [{}]", sql);
|
||||
// return sql;
|
||||
return null;
|
||||
}
|
||||
|
||||
private static CarData getCarData(String vin, RowRecord next, List<String> columnNames) {
|
||||
List<Field> fields = next.getFields();
|
||||
CarData data = new CarData();
|
||||
data.setVin(vin);
|
||||
data.setTimestamp(next.getTimestamp());
|
||||
for (int i = 0; i < columnNames.size(); i++) {
|
||||
if (columnNames.get(i).contains("latitude")) {
|
||||
data.setLatitude(fields.get(i-1).getStringValue());
|
||||
}else if (columnNames.get(i).contains("longitude")) {
|
||||
data.setLongitude(fields.get(i-1).getStringValue());
|
||||
}
|
||||
}
|
||||
return data;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,24 @@
|
|||
package com.muyu.data.processing.strategy;
|
||||
|
||||
import com.muyu.data.processing.strategy.core.EndStrategy;
|
||||
|
||||
/**
|
||||
* 策略控制者接口
|
||||
* @Author: 胡杨
|
||||
* @Name: StrategyHandler
|
||||
* @Description: 策略控制者接口
|
||||
* @CreatedDate: 2024/9/28 上午9:35
|
||||
* @FilePath: com.muyu.data.processing.strategy
|
||||
*/
|
||||
public interface StrategyHandler<T,R> {
|
||||
|
||||
@SuppressWarnings("rawtypes")
|
||||
StrategyHandler DEFAULT = param -> new EndStrategy();
|
||||
|
||||
/**
|
||||
* 执行方法
|
||||
* @param t 入参
|
||||
* @return 返回结果
|
||||
*/
|
||||
R apply(T t);
|
||||
}
|
||||
|
|
@ -0,0 +1,60 @@
|
|||
package com.muyu.data.processing.strategy;
|
||||
|
||||
import jakarta.annotation.PostConstruct;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.Objects;
|
||||
/**
|
||||
* 抽象策略路由
|
||||
* @Author: 胡杨
|
||||
* @Name: abstractStrategyRouter
|
||||
* @Description: 抽象策略路由
|
||||
* @CreatedDate: 2024/9/28 上午9:26
|
||||
* @FilePath: com.muyu.data.processing.strategy
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
public abstract class abstractStrategyRouter<T,R> {
|
||||
|
||||
/**
|
||||
* 策略映射器, 指定入参与出参以决定策略处理者
|
||||
* @param <T> 策略入参
|
||||
* @param <R> 策略出参
|
||||
*/
|
||||
public interface StrategyMapper<T,R>{
|
||||
// 通过入参获取对应策略处理方法,使用Map实现
|
||||
StrategyHandler<T,R> getHandler(T param);
|
||||
}
|
||||
|
||||
/**
|
||||
* 选择下级策略
|
||||
* @return
|
||||
*/
|
||||
protected abstract StrategyMapper<T,R> registerStrategy();
|
||||
|
||||
/**
|
||||
* 默认策略处理者
|
||||
*/
|
||||
@SuppressWarnings("unchecked")
|
||||
private StrategyHandler<T,R> defaultStrategyHandler = StrategyHandler.DEFAULT;
|
||||
|
||||
|
||||
/**
|
||||
* 选择策略处理者
|
||||
* @param param 入参
|
||||
* @return 策略处理结果
|
||||
*/
|
||||
public R applyStrategy(T param) {
|
||||
StrategyMapper<T, R> trStrategyMapper = registerStrategy();
|
||||
if (trStrategyMapper == null) {
|
||||
return defaultStrategyHandler.apply(param);
|
||||
}
|
||||
final StrategyHandler<T,R> strategyHandler = trStrategyMapper.getHandler(param);
|
||||
if (strategyHandler != null) {
|
||||
return strategyHandler.apply(param);
|
||||
}
|
||||
// 使用默认策略处理者
|
||||
return defaultStrategyHandler.apply(param);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,63 @@
|
|||
package com.muyu.data.processing.strategy.branch;
|
||||
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.core.EndStrategy;
|
||||
import com.muyu.data.processing.strategy.leaves.DataStorageStrategy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 持久化数据处理
|
||||
* 数据持久化之前的数据调整
|
||||
* @Author: 胡杨
|
||||
* @Name: BasicStrategy
|
||||
* @Description: 责任树 - 数据持久化数据处理
|
||||
* @CreatedDate: 2024/9/30 下午7:24
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class DataStorageProcessStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
// 必要参数
|
||||
private final static HashMap<String,String> NECESSARY_PARAM = new HashMap<>();
|
||||
static {
|
||||
NECESSARY_PARAM.put("VIN","VIN码");
|
||||
NECESSARY_PARAM.put("timestamp","时间戳");
|
||||
NECESSARY_PARAM.put("longitude","经度");
|
||||
NECESSARY_PARAM.put("latitude","纬度");
|
||||
}
|
||||
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param-> {
|
||||
// 判断是否存在问题
|
||||
if (param.containsKey("DataStorageProcessStrategy")) {
|
||||
log.error("持久化流程错误，缺少必要参数： {}", param.get("DataStorageProcessStrategy").getKey());
|
||||
param.remove("DataStorageProcessStrategy");
|
||||
return new EndStrategy();
|
||||
}
|
||||
log.info("持久化数据处理节点已通过。。。");
|
||||
return new DataStorageStrategy();
|
||||
};
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("持久化数据处理节点开始处理。。。");
|
||||
// 判断是否缺少必要参数,如果有,记录
|
||||
NECESSARY_PARAM.keySet().forEach(key->{
|
||||
if (!basicDataMap.containsKey(key)) {
|
||||
basicDataMap.put("DataStorageProcessStrategy", BasicData.builder().key(NECESSARY_PARAM.get(key)).build());
|
||||
}
|
||||
});
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
package com.muyu.data.processing.strategy.branch;
|
||||
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.leaves.DataStorageStrategy;
|
||||
import com.muyu.data.processing.strategy.leaves.FaultAlarmStrategy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 故障数据判断
|
||||
* 判断是否故障
|
||||
* @Author: 胡杨
|
||||
* @Name: BasicStrategy
|
||||
* @Description: 责任树 - 故障数据判断
|
||||
* @CreatedDate: 2024/9/30 下午7:24
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class FaultJudgmentStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param-> new FaultAlarmStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("故障判断节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,37 @@
|
|||
package com.muyu.data.processing.strategy.branch;
|
||||
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.leaves.DataStorageStrategy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 故障数据处理
|
||||
* 调整故障判断规则
|
||||
* @Author: 胡杨
|
||||
* @Name: FaultProcessingStrategy
|
||||
* @Description: 故障参数处理
|
||||
* @CreatedDate: 2024/9/30 下午7:47
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class FaultProcessingStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param-> new FaultJudgmentStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("故障数据处理节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
package com.muyu.data.processing.strategy.branch;
|
||||
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.leaves.FaultAlarmStrategy;
|
||||
import com.muyu.data.processing.strategy.leaves.FenceAlarmStrategy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 围栏数据判断
|
||||
* 判断是否围栏违规
|
||||
* @Author: 胡杨
|
||||
* @Name: BasicStrategy
|
||||
* @Description: 责任树 - 围栏数据判断
|
||||
* @CreatedDate: 2024/9/30 下午7:24
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class FenceJudgmentStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param-> new FenceAlarmStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("围栏数据判断节点通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
package com.muyu.data.processing.strategy.branch;
|
||||
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 围栏数据处理
|
||||
* 调整围栏判断规则
|
||||
* @Author: 胡杨
|
||||
* @Name: FaultProcessingStrategy
|
||||
* @Description: 围栏参数处理
|
||||
* @CreatedDate: 2024/9/30 下午7:47
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class FenceProcessingStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param-> new FenceJudgmentStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("围栏数据处理节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
package com.muyu.data.processing.strategy.branch;
|
||||
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.leaves.RealTimeAlarmStrategy;
|
||||
import com.muyu.data.processing.strategy.leaves.WarningAlarmStrategy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 实时数据判断
|
||||
* 判断实时数据情况
|
||||
* @Author: 胡杨
|
||||
* @Name: BasicStrategy
|
||||
* @Description: 责任树 - 实时数据判断
|
||||
* @CreatedDate: 2024/9/30 下午7:24
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class RealTimeJudgmentStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param-> new RealTimeAlarmStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("实时数据判断节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
package com.muyu.data.processing.strategy.branch;
|
||||
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 实时数据处理
|
||||
* 调整实时数据
|
||||
* @Author: 胡杨
|
||||
* @Name: FaultProcessingStrategy
|
||||
* @Description: 实时数据理
|
||||
* @CreatedDate: 2024/9/30 下午7:47
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class RealTimeProcessingStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param-> new RealTimeJudgmentStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("实时数据处理节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
package com.muyu.data.processing.strategy.branch;
|
||||
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.leaves.FenceAlarmStrategy;
|
||||
import com.muyu.data.processing.strategy.leaves.WarningAlarmStrategy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 预警数据判断
|
||||
* 判断预警策略
|
||||
* @Author: 胡杨
|
||||
* @Name: BasicStrategy
|
||||
* @Description: 责任树 - 预警数据判断
|
||||
* @CreatedDate: 2024/9/30 下午7:24
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class WarningJudgmentStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param-> new WarningAlarmStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("预警数据判断节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,36 @@
|
|||
package com.muyu.data.processing.strategy.branch;
|
||||
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 预警数据处理
|
||||
* 调整预警使用策略
|
||||
* @Author: 胡杨
|
||||
* @Name: FaultProcessingStrategy
|
||||
* @Description: 预警数据处理
|
||||
* @CreatedDate: 2024/9/30 下午7:47
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class WarningProcessingStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param-> new WarningJudgmentStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("预警数据处理节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,41 @@
|
|||
package com.muyu.data.processing.strategy.core;
|
||||
|
||||
import com.muyu.common.caffeine.enums.CacheNameEnums;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 基础校验节点
|
||||
* 负责基础校验
|
||||
* @Author: 胡杨
|
||||
* @Name: BasicStrategy
|
||||
* @Description: 责任树 - 基础校验节点
|
||||
* @CreatedDate: 2024/9/30 下午7:24
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class BasicStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param -> new RoutingStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("开始执行基础校验节点。。。");
|
||||
basicDataMap.put(CacheNameEnums.STORAGE.getCode(), null);
|
||||
basicDataMap.put(CacheNameEnums.FAULT.getCode(), null);
|
||||
basicDataMap.put(CacheNameEnums.REALTIME.getCode(), null);
|
||||
log.info("基础校验节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
package com.muyu.data.processing.strategy.core;
|
||||
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 结束节点
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: EndStrategy
|
||||
* @Description: 策略树 - 结束节点
|
||||
* @CreatedDate: 2024/9/30 下午7:13
|
||||
* @FilePath: com.muyu.data.processing.strategy.leaves
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class EndStrategy implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("结束节点已通过。。。");
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,59 @@
|
|||
package com.muyu.data.processing.strategy.core;
|
||||
|
||||
import com.muyu.common.caffeine.enums.CacheNameEnums;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.branch.*;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 路由节点
|
||||
* 根据条件重新导向对应节点
|
||||
* @Author: 胡杨
|
||||
* @Name: RoutingStrategy
|
||||
* @Description: 路由节点
|
||||
* @CreatedDate: 2024/9/30 下午7:37
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class RoutingStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
|
||||
private final static HashMap<String, StrategyHandler<HashMap<String, BasicData>, Temporary2>> map = new HashMap<>();
|
||||
static{
|
||||
map.put(CacheNameEnums.WARMING.getCode(), new WarningProcessingStrategy());
|
||||
map.put(CacheNameEnums.REALTIME.getCode(), new RealTimeProcessingStrategy());
|
||||
map.put(CacheNameEnums.FENCE.getCode(), new FenceProcessingStrategy());
|
||||
map.put(CacheNameEnums.FAULT.getCode(), new FaultProcessingStrategy());
|
||||
map.put(CacheNameEnums.STORAGE.getCode(), new DataStorageProcessStrategy());
|
||||
}
|
||||
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param -> {
|
||||
// 编写路由规则
|
||||
List<String> codes = CacheNameEnums.getCodes();
|
||||
for (String code : codes) {
|
||||
if(param.containsKey(code)){
|
||||
param.remove(code);
|
||||
return map.get(code);
|
||||
}
|
||||
}
|
||||
// 默认返回结束节点
|
||||
return new EndStrategy();
|
||||
};
|
||||
};
|
||||
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> stringListHashMap) {
|
||||
log.info("路由节点已通过。。。");
|
||||
return applyStrategy(stringListHashMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,47 @@
|
|||
package com.muyu.data.processing.strategy.core;
|
||||
|
||||
|
||||
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* 开始节点
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: StartStrategy
|
||||
* @Description: 策略路由实现
|
||||
* @CreatedDate: 2024/9/28 上午10:39
|
||||
* @FilePath: com.muyu.data.processing.strategy.impl
|
||||
* 开始节点
|
||||
* ↓
|
||||
* 基础校验节点
|
||||
* ↓
|
||||
* 路由节点 ← ← ← ← ← ←
|
||||
* ↙ ↙ ↓ ↘ ↘
|
||||
* 数据处理节点 预警处理节点 故障处理节点 围栏处理节点 实时数据数据处理节点 ↑
|
||||
* ↓ ↓ ↓ ↓ ↓
|
||||
* 数据持久化处理节点 预警处理节点 故障处理节点 围栏处理节点 实时数据数据处理节点 ↑
|
||||
* ↓ ↓ ↓ ↓ ↓
|
||||
* ↓ 预警通知节点 故障通知节点 围栏通知节点 实时数据处理节点 ↑
|
||||
* ↓ ↘ ↓ ↙ ↙
|
||||
* ↓ ↓ ↑
|
||||
* → → → 路由节点 → → → → → →
|
||||
* ↓
|
||||
* 结束节点
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class StartStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param -> new BasicStrategy();
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,90 @@
|
|||
package com.muyu.data.processing.strategy.leaves;
|
||||
|
||||
import com.muyu.common.caffeine.enums.CacheNameEnums;
|
||||
import com.muyu.common.iotdb.config.IotDBSessionConfig;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.service.DataProcessingService;
|
||||
import com.muyu.data.processing.service.impl.DataProcessingServiceImpl;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.core.RoutingStrategy;
|
||||
import jakarta.annotation.Resource;
|
||||
import lombok.Setter;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.apache.iotdb.rpc.IoTDBConnectionException;
|
||||
import org.apache.iotdb.rpc.StatementExecutionException;
|
||||
import org.apache.iotdb.session.pool.SessionPool;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 数据持久化
|
||||
* 车辆数据进行持久化
|
||||
* @Author: 胡杨
|
||||
* @Name: BasicStrategy
|
||||
* @Description: 责任树 - 数据持久化
|
||||
* @CreatedDate: 2024/9/30 下午7:24
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class DataStorageStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param -> {
|
||||
log.info("数据持久化节点已通过。。。");
|
||||
return new RoutingStrategy();
|
||||
};
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
// 执行持久化方法
|
||||
addCarData(basicDataMap);
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
|
||||
private void addCarData(HashMap<String, BasicData> hashMap) {
|
||||
StringBuilder sql = new StringBuilder("insert into root.one.");
|
||||
sql.append(hashMap.get("firmCode").getValue())
|
||||
.append(".")
|
||||
.append(hashMap.get("VIN").getValue())
|
||||
.append("(");
|
||||
hashMap.remove("firmCode");
|
||||
hashMap.remove("VIN");
|
||||
StringBuilder keys = new StringBuilder();
|
||||
StringBuilder values = new StringBuilder();
|
||||
hashMap.keySet().forEach(key -> {
|
||||
if (hashMap.get(key) != null) {
|
||||
keys.append(key).append(",");
|
||||
if ("String".equals(hashMap.get(key).getType())) {
|
||||
values.append("'")
|
||||
.append(hashMap.get(key).getValue())
|
||||
.append("'")
|
||||
.append(",");
|
||||
}else {
|
||||
values.append(hashMap.get(key).getValue())
|
||||
.append(",");
|
||||
}
|
||||
}
|
||||
});
|
||||
sql.append(keys.substring(0, keys.length() - 1))
|
||||
.append(") values (")
|
||||
.append(values.substring(0, values.length() - 1))
|
||||
.append(")");
|
||||
try {
|
||||
new IotDBSessionConfig().getSessionPool().executeNonQueryStatement(sql.toString());
|
||||
} catch (StatementExecutionException e) {
|
||||
throw new RuntimeException(e);
|
||||
} catch (IoTDBConnectionException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
log.info("成功执行sql语句: [{}]", sql);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
package com.muyu.data.processing.strategy.leaves;
|
||||
|
||||
import com.muyu.common.caffeine.enums.CacheNameEnums;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.core.RoutingStrategy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 故障报警
|
||||
* 故障数据记录并报警
|
||||
* @Author: 胡杨
|
||||
* @Name: BasicStrategy
|
||||
* @Description: 责任树 - 故障报警
|
||||
* @CreatedDate: 2024/9/30 下午7:24
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class FaultAlarmStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param -> new RoutingStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("故障报警节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
package com.muyu.data.processing.strategy.leaves;
|
||||
|
||||
import com.muyu.common.caffeine.enums.CacheNameEnums;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.core.RoutingStrategy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 围栏报警
|
||||
* 围栏数据记录并报警
|
||||
* @Author: 胡杨
|
||||
* @Name: BasicStrategy
|
||||
* @Description: 责任树 - 围栏报警
|
||||
* @CreatedDate: 2024/9/30 下午7:24
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class FenceAlarmStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param -> new RoutingStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("围栏报警节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
package com.muyu.data.processing.strategy.leaves;
|
||||
|
||||
import com.muyu.common.caffeine.enums.CacheNameEnums;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.core.RoutingStrategy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 实时数据
|
||||
* 处理实时数据事件
|
||||
* @Author: 胡杨
|
||||
* @Name: BasicStrategy
|
||||
* @Description: 责任树 - 实时数据
|
||||
* @CreatedDate: 2024/9/30 下午7:24
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class RealTimeAlarmStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param -> new RoutingStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("实时数据处理节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
package com.muyu.data.processing.strategy.leaves;
|
||||
|
||||
import com.muyu.common.caffeine.enums.CacheNameEnums;
|
||||
import com.muyu.data.processing.domain.BasicData;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import com.muyu.data.processing.domain.Temporary2;
|
||||
import com.muyu.data.processing.strategy.StrategyHandler;
|
||||
import com.muyu.data.processing.strategy.abstractStrategyRouter;
|
||||
import com.muyu.data.processing.strategy.core.RoutingStrategy;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 预警报警
|
||||
* 预警数据记录并提示
|
||||
* @Author: 胡杨
|
||||
* @Name: BasicStrategy
|
||||
* @Description: 责任树 - 预警提示
|
||||
* @CreatedDate: 2024/9/30 下午7:24
|
||||
* @FilePath: com.muyu.data.processing.strategy.branch
|
||||
*/
|
||||
|
||||
@Slf4j
|
||||
@Component
|
||||
public class WarningAlarmStrategy extends abstractStrategyRouter<HashMap<String, BasicData>, Temporary2>
|
||||
implements StrategyHandler<HashMap<String, BasicData>, Temporary2> {
|
||||
@Override
|
||||
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
|
||||
return param -> new RoutingStrategy();
|
||||
}
|
||||
|
||||
@Override
|
||||
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
|
||||
log.info("预警报警节点已通过。。。");
|
||||
return applyStrategy(basicDataMap);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,28 @@
|
|||
package com.muyu.data.processing.utils;
|
||||
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
/**
|
||||
* 数据处理工具类
|
||||
*
|
||||
* @Author: 胡杨
|
||||
* @Name: DataUtils
|
||||
* @Description: 数据处理工具类
|
||||
* @CreatedDate: 2024/9/29 上午10:15
|
||||
* @FilePath: com.muyu.data.processing.utils
|
||||
*/
|
||||
|
||||
@Component
|
||||
public class DataUtils {
|
||||
|
||||
/**
|
||||
* 类型转换
|
||||
* @param data 转换值
|
||||
* @param type 转换类型
|
||||
* @return 转换结果
|
||||
* @param <T> 返回类型
|
||||
*/
|
||||
public static <T> T convert(Object data, Class<T> type) {
|
||||
return type.cast(data);
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,2 @@
|
|||
Spring Boot Version: ${spring-boot.version}
|
||||
Spring Application Name: ${spring.application.name}
|
||||
|
|
@ -0,0 +1,72 @@
|
|||
# Tomcat
|
||||
server:
|
||||
port: 9711
|
||||
|
||||
# nacos线上地址
|
||||
nacos:
|
||||
addr: 47.116.173.119:8848
|
||||
user-name: nacos
|
||||
password: nacos
|
||||
namespace: one-saas
|
||||
# SPRING_AMQP_DESERIALIZATION_TRUST_ALL=true spring.amqp.deserialization.trust.all
|
||||
# Spring
|
||||
spring:
|
||||
iotdb:
|
||||
ip: 47.116.173.119
|
||||
port: 6667
|
||||
user: root
|
||||
password: root
|
||||
fetchSize: 10000
|
||||
maxActive: 10
|
||||
amqp:
|
||||
deserialization:
|
||||
trust:
|
||||
all: true
|
||||
main:
|
||||
allow-bean-definition-overriding: true
|
||||
application:
|
||||
# 应用名称
|
||||
name: cloud-data-processing
|
||||
profiles:
|
||||
# 环境配置
|
||||
active: dev
|
||||
cloud:
|
||||
nacos:
|
||||
discovery:
|
||||
# 服务注册地址
|
||||
server-addr: ${nacos.addr}
|
||||
# nacos用户名
|
||||
username: ${nacos.user-name}
|
||||
# nacos密码
|
||||
password: ${nacos.password}
|
||||
# 命名空间
|
||||
namespace: ${nacos.namespace}
|
||||
config:
|
||||
# 服务注册地址
|
||||
server-addr: ${nacos.addr}
|
||||
# nacos用户名
|
||||
username: ${nacos.user-name}
|
||||
# nacos密码
|
||||
password: ${nacos.password}
|
||||
# 命名空间
|
||||
namespace: ${nacos.namespace}
|
||||
# 配置文件格式
|
||||
file-extension: yml
|
||||
# 共享配置
|
||||
shared-configs:
|
||||
# 系统共享配置
|
||||
- application-${spring.profiles.active}.${spring.cloud.nacos.config.file-extension}
|
||||
# 系统环境Config共享配置
|
||||
- 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}
|
||||
|
||||
logging:
|
||||
level:
|
||||
com.muyu.system.mapper: DEBUG
|
||||
|
||||
cacheNames: fault,fence,warming
|
||||
|
|
@ -0,0 +1,74 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<configuration scan="true" scanPeriod="60 seconds" debug="false">
|
||||
<!-- 日志存放路径 -->
|
||||
<property name="log.path" value="logs/cloud-data-processing"/>
|
||||
<!-- 日志输出格式 -->
|
||||
<property name="log.pattern" value="%d{HH:mm:ss.SSS} [%thread] %-5level %logger{20} - [%method,%line] - %msg%n"/>
|
||||
|
||||
<!-- 控制台输出 -->
|
||||
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
|
||||
<encoder>
|
||||
<pattern>${log.pattern}</pattern>
|
||||
</encoder>
|
||||
</appender>
|
||||
|
||||
<!-- 系统日志输出 -->
|
||||
<appender name="file_info" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||
<file>${log.path}/info.log</file>
|
||||
<!-- 循环政策：基于时间创建日志文件 -->
|
||||
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
|
||||
<!-- 日志文件名格式 -->
|
||||
<fileNamePattern>${log.path}/info.%d{yyyy-MM-dd}.log</fileNamePattern>
|
||||
<!-- 日志最大的历史 60天 -->
|
||||
<maxHistory>60</maxHistory>
|
||||
</rollingPolicy>
|
||||
<encoder>
|
||||
<pattern>${log.pattern}</pattern>
|
||||
</encoder>
|
||||
<filter class="ch.qos.logback.classic.filter.LevelFilter">
|
||||
<!-- 过滤的级别 -->
|
||||
<level>INFO</level>
|
||||
<!-- 匹配时的操作：接收（记录） -->
|
||||
<onMatch>ACCEPT</onMatch>
|
||||
<!-- 不匹配时的操作：拒绝（不记录） -->
|
||||
<onMismatch>DENY</onMismatch>
|
||||
</filter>
|
||||
</appender>
|
||||
|
||||
<appender name="file_error" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||
<file>${log.path}/error.log</file>
|
||||
<!-- 循环政策：基于时间创建日志文件 -->
|
||||
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
|
||||
<!-- 日志文件名格式 -->
|
||||
<fileNamePattern>${log.path}/error.%d{yyyy-MM-dd}.log</fileNamePattern>
|
||||
<!-- 日志最大的历史 60天 -->
|
||||
<maxHistory>60</maxHistory>
|
||||
</rollingPolicy>
|
||||
<encoder>
|
||||
<pattern>${log.pattern}</pattern>
|
||||
</encoder>
|
||||
<filter class="ch.qos.logback.classic.filter.LevelFilter">
|
||||
<!-- 过滤的级别 -->
|
||||
<level>ERROR</level>
|
||||
<!-- 匹配时的操作：接收（记录） -->
|
||||
<onMatch>ACCEPT</onMatch>
|
||||
<!-- 不匹配时的操作：拒绝（不记录） -->
|
||||
<onMismatch>DENY</onMismatch>
|
||||
</filter>
|
||||
</appender>
|
||||
|
||||
<!-- 系统模块日志级别控制 -->
|
||||
<logger name="com.muyu" level="info"/>
|
||||
<!-- Spring日志级别控制 -->
|
||||
<logger name="org.springframework" level="warn"/>
|
||||
|
||||
<root level="info">
|
||||
<appender-ref ref="console"/>
|
||||
</root>
|
||||
|
||||
<!--系统操作日志-->
|
||||
<root level="info">
|
||||
<appender-ref ref="file_info"/>
|
||||
<appender-ref ref="file_error"/>
|
||||
</root>
|
||||
</configuration>
|
||||
|
|
@ -0,0 +1,81 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<configuration scan="true" scanPeriod="60 seconds" debug="false">
|
||||
<!-- 日志存放路径 -->
|
||||
<property name="log.path" value="logs/cloud-data-processing"/>
|
||||
<!-- 日志输出格式 -->
|
||||
<property name="log.pattern" value="%d{HH:mm:ss.SSS} [%thread] %-5level %logger{20} - [%method,%line] - %msg%n"/>
|
||||
<property name="log.sky.pattern" value="%d{HH:mm:ss.SSS} %yellow([%tid]) [%thread] %-5level %logger{20} - [%method,%line] - %msg%n"/>
|
||||
|
||||
<!-- 控制台输出 -->
|
||||
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
|
||||
<encoder>
|
||||
<pattern>${log.sky.pattern}</pattern>
|
||||
</encoder>
|
||||
</appender>
|
||||
|
||||
<!-- 系统日志输出 -->
|
||||
<appender name="file_info" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||
<file>${log.path}/info.log</file>
|
||||
<!-- 循环政策：基于时间创建日志文件 -->
|
||||
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
|
||||
<!-- 日志文件名格式 -->
|
||||
<fileNamePattern>${log.path}/info.%d{yyyy-MM-dd}.log</fileNamePattern>
|
||||
<!-- 日志最大的历史 60天 -->
|
||||
<maxHistory>60</maxHistory>
|
||||
</rollingPolicy>
|
||||
|
||||
<filter class="ch.qos.logback.classic.filter.LevelFilter">
|
||||
<!-- 过滤的级别 -->
|
||||
<level>INFO</level>
|
||||
<!-- 匹配时的操作：接收（记录） -->
|
||||
<onMatch>ACCEPT</onMatch>
|
||||
<!-- 不匹配时的操作：拒绝（不记录） -->
|
||||
<onMismatch>DENY</onMismatch>
|
||||
</filter>
|
||||
</appender>
|
||||
|
||||
<appender name="file_error" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||
<file>${log.path}/error.log</file>
|
||||
<!-- 循环政策：基于时间创建日志文件 -->
|
||||
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
|
||||
<!-- 日志文件名格式 -->
|
||||
<fileNamePattern>${log.path}/error.%d{yyyy-MM-dd}.log</fileNamePattern>
|
||||
<!-- 日志最大的历史 60天 -->
|
||||
<maxHistory>60</maxHistory>
|
||||
</rollingPolicy>
|
||||
|
||||
<filter class="ch.qos.logback.classic.filter.LevelFilter">
|
||||
<!-- 过滤的级别 -->
|
||||
<level>ERROR</level>
|
||||
<!-- 匹配时的操作：接收（记录） -->
|
||||
<onMatch>ACCEPT</onMatch>
|
||||
<!-- 不匹配时的操作：拒绝（不记录） -->
|
||||
<onMismatch>DENY</onMismatch>
|
||||
</filter>
|
||||
</appender>
|
||||
|
||||
<!-- 使用gRpc将日志发送到skywalking服务端 -->
|
||||
<appender name="GRPC_LOG" class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.log.GRPCLogClientAppender">
|
||||
<encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder">
|
||||
<layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.TraceIdPatternLogbackLayout">
|
||||
<Pattern>${log.sky.pattern}</Pattern>
|
||||
</layout>
|
||||
</encoder>
|
||||
</appender>
|
||||
|
||||
<!-- 系统模块日志级别控制 -->
|
||||
<logger name="com.muyu" level="info"/>
|
||||
<!-- Spring日志级别控制 -->
|
||||
<logger name="org.springframework" level="warn"/>
|
||||
|
||||
<root level="info">
|
||||
<appender-ref ref="GRPC_LOG"/>
|
||||
<appender-ref ref="console"/>
|
||||
</root>
|
||||
|
||||
<!--系统操作日志-->
|
||||
<root level="info">
|
||||
<appender-ref ref="file_info"/>
|
||||
<appender-ref ref="file_error"/>
|
||||
</root>
|
||||
</configuration>
|
||||
|
|
@ -0,0 +1,81 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<configuration scan="true" scanPeriod="60 seconds" debug="false">
|
||||
<!-- 日志存放路径 -->
|
||||
<property name="log.path" value="logs/cloud-data-processing"/>
|
||||
<!-- 日志输出格式 -->
|
||||
<property name="log.pattern" value="%d{HH:mm:ss.SSS} [%thread] %-5level %logger{20} - [%method,%line] - %msg%n"/>
|
||||
<property name="log.sky.pattern" value="%d{HH:mm:ss.SSS} %yellow([%tid]) [%thread] %-5level %logger{20} - [%method,%line] - %msg%n"/>
|
||||
|
||||
<!-- 控制台输出 -->
|
||||
<appender name="console" class="ch.qos.logback.core.ConsoleAppender">
|
||||
<encoder>
|
||||
<pattern>${log.sky.pattern}</pattern>
|
||||
</encoder>
|
||||
</appender>
|
||||
|
||||
<!-- 系统日志输出 -->
|
||||
<appender name="file_info" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||
<file>${log.path}/info.log</file>
|
||||
<!-- 循环政策：基于时间创建日志文件 -->
|
||||
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
|
||||
<!-- 日志文件名格式 -->
|
||||
<fileNamePattern>${log.path}/info.%d{yyyy-MM-dd}.log</fileNamePattern>
|
||||
<!-- 日志最大的历史 60天 -->
|
||||
<maxHistory>60</maxHistory>
|
||||
</rollingPolicy>
|
||||
|
||||
<filter class="ch.qos.logback.classic.filter.LevelFilter">
|
||||
<!-- 过滤的级别 -->
|
||||
<level>INFO</level>
|
||||
<!-- 匹配时的操作：接收（记录） -->
|
||||
<onMatch>ACCEPT</onMatch>
|
||||
<!-- 不匹配时的操作：拒绝（不记录） -->
|
||||
<onMismatch>DENY</onMismatch>
|
||||
</filter>
|
||||
</appender>
|
||||
|
||||
<appender name="file_error" class="ch.qos.logback.core.rolling.RollingFileAppender">
|
||||
<file>${log.path}/error.log</file>
|
||||
<!-- 循环政策：基于时间创建日志文件 -->
|
||||
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
|
||||
<!-- 日志文件名格式 -->
|
||||
<fileNamePattern>${log.path}/error.%d{yyyy-MM-dd}.log</fileNamePattern>
|
||||
<!-- 日志最大的历史 60天 -->
|
||||
<maxHistory>60</maxHistory>
|
||||
</rollingPolicy>
|
||||
|
||||
<filter class="ch.qos.logback.classic.filter.LevelFilter">
|
||||
<!-- 过滤的级别 -->
|
||||
<level>ERROR</level>
|
||||
<!-- 匹配时的操作：接收（记录） -->
|
||||
<onMatch>ACCEPT</onMatch>
|
||||
<!-- 不匹配时的操作：拒绝（不记录） -->
|
||||
<onMismatch>DENY</onMismatch>
|
||||
</filter>
|
||||
</appender>
|
||||
|
||||
<!-- 使用gRpc将日志发送到skywalking服务端 -->
|
||||
<appender name="GRPC_LOG" class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.log.GRPCLogClientAppender">
|
||||
<encoder class="ch.qos.logback.core.encoder.LayoutWrappingEncoder">
|
||||
<layout class="org.apache.skywalking.apm.toolkit.log.logback.v1.x.TraceIdPatternLogbackLayout">
|
||||
<Pattern>${log.sky.pattern}</Pattern>
|
||||
</layout>
|
||||
</encoder>
|
||||
</appender>
|
||||
|
||||
<!-- 系统模块日志级别控制 -->
|
||||
<logger name="com.muyu" level="info"/>
|
||||
<!-- Spring日志级别控制 -->
|
||||
<logger name="org.springframework" level="warn"/>
|
||||
|
||||
<root level="info">
|
||||
<appender-ref ref="GRPC_LOG"/>
|
||||
<appender-ref ref="console"/>
|
||||
</root>
|
||||
|
||||
<!--系统操作日志-->
|
||||
<root level="info">
|
||||
<appender-ref ref="file_info"/>
|
||||
<appender-ref ref="file_error"/>
|
||||
</root>
|
||||
</configuration>
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
<?xml version="1.0" encoding="UTF-8"?>
|
||||
<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd">
|
||||
<mapper namespace="com.muyu.data.processing.mapper.DataProcessingMapper">
|
||||
|
||||
<select id="selectStorageGroup" resultType="java.lang.String">
|
||||
show storage group
|
||||
</select>
|
||||
|
||||
<select id="selectCarData" resultType="com.muyu.data.processing.domain.CarData">
|
||||
select * from ${tableName};
|
||||
</select>
|
||||
|
||||
<insert id="insIotDbData">
|
||||
insert into root.one.data(${key}) values(${value});
|
||||
</insert>
|
||||
<insert id="strategyCheck">
|
||||
insert into root.one.data
|
||||
(
|
||||
<foreach collection="dataList" item="data" separator=",">
|
||||
${data.key}
|
||||
</foreach>
|
||||
) values
|
||||
(
|
||||
<foreach collection="dataList" item="data" separator=",">
|
||||
#{data.value}
|
||||
</foreach>
|
||||
)
|
||||
|
||||
</insert>
|
||||
<insert id="insIotDbDataVo">
|
||||
insert into
|
||||
root.one.data
|
||||
(timestamp, vin, latitude,longitude)
|
||||
values (#{timestamp}, #{vin}, #{latitude}, #{longitude})
|
||||
</insert>
|
||||
|
||||
|
||||
</mapper>
|
||||
|
|
@ -13,6 +13,7 @@
|
|||
<module>cloud-modules-gen</module>
|
||||
<module>cloud-modules-file</module>
|
||||
<module>cloud-modules-openbusiness</module>
|
||||
<module>cloud-modules-data-processing</module>
|
||||
</modules>
|
||||
|
||||
<artifactId>cloud-modules</artifactId>
|
||||
|
|
|
|||
Loading…
Reference in New Issue