one-YI
/
cloud-server
5
2
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat(): 优化Kafka逻辑,新增缓存工具类

dev
面包骑士 2024-10-09 17:04:11 +08:00
parent ed3a542066
commit 465d39a4bf
26 changed files with 237 additions and 750 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -3,16 +3,14 @@ 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.context.annotation.Configuration;
import org.springframework.stereotype.Component;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@ -26,7 +24,7 @@ import java.util.List;
*/
@Slf4j
@Component
@Configuration
public class CaffeineManager {
/**
@ -34,7 +32,7 @@ public class CaffeineManager {
* @return 缓存管理器实例
*/
@Bean
public SimpleCacheManager simpleCacheManager() {
public CacheManager cacheManager() {
SimpleCacheManager cacheManager = new SimpleCacheManager();
List<String> cacheNames = CacheNameEnums.getCodes();
cacheManager.setCaches(cacheNames.stream()

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

@ -1,97 +0,0 @@
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;
}
}

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

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

2
cloud-modules/cloud-modules-car-gateway/src/main/resources/bootstrap.yml
View File

@ -7,7 +7,7 @@ nacos:
addr: 47.116.173.119:8848
user-name: nacos
password: nacos
namespace: one
namespace: oneone
# SPRING_AMQP_DESERIALIZATION_TRUST_ALL=true spring.amqp.deserialization.trust.all
# Spring
spring:

45
cloud-modules/cloud-modules-data-processing/pom.xml
View File

@ -90,51 +90,6 @@
<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>-->
<!--&lt;!&ndash; <dependency>&ndash;&gt;-->
<!--&lt;!&ndash; <groupId>org.apache.iotdb</groupId>&ndash;&gt;-->
<!--&lt;!&ndash; <artifactId>iotdb-session</artifactId>&ndash;&gt;-->
<!--&lt;!&ndash; <version>1.3.2</version>&ndash;&gt;-->
<!--&lt;!&ndash; </dependency>&ndash;&gt;-->
<!-- &lt;!&ndash; Druid &ndash;&gt;-->
<!-- <dependency>-->
<!-- <groupId>com.alibaba</groupId>-->
<!-- <artifactId>druid-spring-boot-3-starter</artifactId>-->
<!-- <version>${druid.version}</version>-->
<!-- </dependency>-->
<!-- &lt;!&ndash; Dynamic DataSource &ndash;&gt;-->
<!-- <dependency>-->
<!-- <groupId>com.baomidou</groupId>-->
<!-- <artifactId>dynamic-datasource-spring-boot3-starter</artifactId>-->
<!-- <version>${dynamic-ds.version}</version>-->
<!-- </dependency>-->
</dependencies>
<build>

4
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/CloudVehicleEventApplication.java
View File

@ -5,6 +5,7 @@ import com.muyu.common.security.annotation.EnableMyFeignClients;
import org.springframework.amqp.rabbit.annotation.EnableRabbit;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.context.annotation.ComponentScan;
/**
* 数据处理模块启动器
@ -17,7 +18,8 @@ import org.springframework.boot.autoconfigure.SpringBootApplication;
@EnableRabbit
@EnableCustomConfig
@EnableMyFeignClients
@SpringBootApplication(scanBasePackages = {"com.muyu"})
@ComponentScan(basePackages = {"com.muyu"})
@SpringBootApplication
public class CloudVehicleEventApplication {
public static void main(String[] args) {
SpringApplication.run(CloudVehicleEventApplication.class, args);

65
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/controller/DataProcessingController.java
View File

@ -32,69 +32,4 @@ 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));
}
}

165
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/controller/TestController.java
View File

@ -1,165 +0,0 @@
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());
}
}

6
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/domain/BasicData.java
View File

@ -26,11 +26,5 @@ public class BasicData implements Serializable {
private String value;
private String type;
// public void setValueClass() {
// Class<?> info = ClassType.getInfo(type);
// if (info.isInstance(value)){
// value = info.cast(value);
// }
// }
}

13
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/kafka/KafkaConsumerService.java
View File

@ -39,13 +39,14 @@ public class KafkaConsumerService implements InitializingBean {
private StartStrategy startStrategy;
@Override
public void afterPropertiesSet() throws Exception {
Thread thread = new Thread(() -> {
public void afterPropertiesSet() {
new Thread(() -> {
log.info("启动线程监听Topic: {}", KafkaConstants.KafkaTopic);
ThreadUtil.sleep(1000);
Collection<String> topics = Lists.newArrayList(KafkaConstants.KafkaTopic);
kafkaConsumer.subscribe(topics);
while (true) {
try {
ThreadUtil.sleep(1000);
System.out.println("开始消费数据,等待中...");
ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord consumerRecord : consumerRecords) {
@ -58,9 +59,11 @@ public class KafkaConsumerService implements InitializingBean {
// 执行策略
startStrategy.applyStrategy(getDataMap(dataList));
}
}catch (Exception e) {
log.error("kafka执行异常：" + e.getMessage());
}
});
thread.start();
}
}).start();
}
private HashMap<String, BasicData> getDataMap(List<BasicData> dataList) {

9
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/mapper/DataProcessingMapper.java
View File

@ -23,13 +23,4 @@ import java.util.List;
@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);
}

146
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/rebbit/DownlineRabbitConsumer.java
View File

@ -1,74 +1,72 @@
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);
}
}
//package com.muyu.data.processing.rebbit;
//
//
//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: 胡杨
// * @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);
// }
//}

146
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/rebbit/GoOnlineRabbitConsumer.java
View File

@ -1,74 +1,72 @@
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);
}
}
//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.cache.CacheManager;
//import org.springframework.data.redis.core.RedisTemplate;
//import org.springframework.stereotype.Component;
//
//import java.io.IOException;
//
///**
// * 上线事件监听
// * @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);
// }
//}

14
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/service/DataProcessingService.java
View File

@ -19,18 +19,4 @@ import java.util.List;
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);
}

119
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/service/impl/DataProcessingServiceImpl.java
View File

@ -3,26 +3,14 @@ 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.cache.Cache;
import org.springframework.cache.CacheManager;
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;
/**
* 数据处理实现层
*
@ -40,109 +28,10 @@ public class DataProcessingServiceImpl implements DataProcessingService {
private DataProcessingMapper mapper;
@Resource
private SessionPool sessionPool;
@Resource
private CacheManager cacheManager;
@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;
}
}

4
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/abstractStrategyRouter.java
View File

@ -1,7 +1,9 @@
package com.muyu.data.processing.strategy;
import com.github.yulichang.toolkit.SpringContentUtils;
import jakarta.annotation.PostConstruct;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cache.CacheManager;
import org.springframework.stereotype.Component;
import java.util.Objects;
@ -17,6 +19,8 @@ import java.util.Objects;
@Component
public abstract class abstractStrategyRouter<T,R> {
protected static CacheManager cacheManager = SpringContentUtils.getBean(CacheManager.class);
/**
* 策略映射器, 指定入参与出参以决定策略处理者
* @param <T> 策略入参

11
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/BasicStrategy.java
View File

@ -1,13 +1,17 @@
package com.muyu.data.processing.strategy.core;
import com.github.yulichang.toolkit.SpringContentUtils;
import com.muyu.common.caffeine.enums.CacheNameEnums;
import com.muyu.common.core.utils.StringUtils;
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.impl.DataProcessingServiceImpl;
import com.muyu.data.processing.strategy.StrategyHandler;
import com.muyu.data.processing.strategy.abstractStrategyRouter;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cache.CacheManager;
import org.springframework.stereotype.Component;
/**
@ -24,6 +28,7 @@ import org.springframework.stereotype.Component;
@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();
@ -33,8 +38,10 @@ public class BasicStrategy extends abstractStrategyRouter<HashMap<String, BasicD
public Temporary2 apply(HashMap<String, BasicData> basicDataMap) {
log.info("开始执行基础校验节点。。。");
basicDataMap.put(CacheNameEnums.STORAGE.getCode(), null);
basicDataMap.put(CacheNameEnums.FAULT.getCode(), null);
String string = cacheManager.getCache(CacheNameEnums.REALTIME.getCode()).get(basicDataMap.get("VIN").getValue(), String.class);
if (StringUtils.isNotEmpty(string)){
basicDataMap.put(CacheNameEnums.REALTIME.getCode(), null);
}
log.info("基础校验节点已通过。。。");
return applyStrategy(basicDataMap);
}

3
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/RoutingStrategy.java
View File

@ -38,8 +38,7 @@ public class RoutingStrategy extends abstractStrategyRouter<HashMap<String, Ba
protected StrategyMapper<HashMap<String, BasicData>, Temporary2> registerStrategy() {
return param -> {
// 编写路由规则
List<String> codes = CacheNameEnums.getCodes();
for (String code : codes) {
for (String code : map.keySet()) {
if(param.containsKey(code)){
param.remove(code);
return map.get(code);

9
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/DataStorageStrategy.java
View File

@ -1,23 +1,16 @@
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;
/**

3
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/FaultAlarmStrategy.java
View File

@ -1,9 +1,8 @@
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;

3
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/FenceAlarmStrategy.java
View File

@ -1,9 +1,8 @@
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;

3
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/RealTimeAlarmStrategy.java
View File

@ -1,9 +1,8 @@
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;

3
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/WarningAlarmStrategy.java
View File

@ -1,9 +1,8 @@
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;

42
cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/utils/CacheUtils.java 100644
View File

@ -0,0 +1,42 @@
package com.muyu.data.processing.utils;
import com.github.yulichang.toolkit.SpringContentUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cache.Cache;
import org.springframework.cache.CacheManager;
import org.springframework.stereotype.Component;
/**
* 缓存工具类
* @Author: 胡杨
* @Name: CacheUtils
* @Description: 缓存工具类
* @CreatedDate: 2024/10/9 下午4:21
* @FilePath: com.muyu.data.processing.utils
*/
@Slf4j
@Component
public class CacheUtils {
private static CacheManager cacheManager = SpringContentUtils.getBean(CacheManager.class);
public Cache getCache(String cacheName) {
Cache cache = cacheManager.getCache(cacheName);
if (cache == null){
throw new RuntimeException("缓存分区: "+cacheName+" 不存在");
}
return cache;
}
public void setCacheValue(String cacheName, String key, Object value) {
getCache(cacheName).put(key, value);
log.info("缓存存储成功:分区-{}, 键-{}, 值-{}", cacheName, key, value);
}
public <T> T getCacheValue(String cacheName, String key, Class<T> type) {
T value = getCache(cacheName).get(key, type);
log.info("获取缓存信息:分区-{}, 键-{}, 值-{}", cacheName, key, value);
return value;
}
}

4
cloud-modules/cloud-modules-data-processing/src/main/resources/bootstrap.yml
View File

@ -7,7 +7,7 @@ nacos:
addr: 47.116.173.119:8848
user-name: nacos
password: nacos
namespace: one-saas
namespace: oneone
# SPRING_AMQP_DESERIALIZATION_TRUST_ALL=true spring.amqp.deserialization.trust.all
# Spring
spring:
@ -68,5 +68,3 @@ spring:
logging:
level:
com.muyu.system.mapper: DEBUG
cacheNames: fault,fence,warming

38
cloud-modules/cloud-modules-data-processing/src/main/resources/mapper/processing/DataProcessingMapper.xml
View File

@ -1,38 +0,0 @@
<?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>