diff --git a/cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/bean/CaffeineManager.java b/cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/bean/CaffeineManager.java index 8ccc11f..d4463d2 100644 --- a/cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/bean/CaffeineManager.java +++ b/cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/bean/CaffeineManager.java @@ -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 cacheNames = CacheNameEnums.getCodes(); cacheManager.setCaches(cacheNames.stream() diff --git a/cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/utils/CaffeineCacheUtils.java b/cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/utils/CaffeineCacheUtils.java deleted file mode 100644 index 9064bf9..0000000 --- a/cloud-common/cloud-common-caffeine/src/main/java/com/muyu/common/caffeine/utils/CaffeineCacheUtils.java +++ /dev/null @@ -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 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 getCarCache(String cacheName, String key, Class 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; - - } - -} diff --git a/cloud-common/cloud-common-caffeine/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports b/cloud-common/cloud-common-caffeine/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports index 0b7acd4..42fbf86 100644 --- a/cloud-common/cloud-common-caffeine/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports +++ b/cloud-common/cloud-common-caffeine/src/main/resources/META-INF/spring/org.springframework.boot.autoconfigure.AutoConfiguration.imports @@ -1,2 +1 @@ -com.muyu.common.caffeine.utils.CaffeineCacheUtils com.muyu.common.caffeine.bean.CaffeineManager diff --git a/cloud-modules/cloud-modules-car-gateway/src/main/resources/bootstrap.yml b/cloud-modules/cloud-modules-car-gateway/src/main/resources/bootstrap.yml index e33efb2..2b120b0 100644 --- a/cloud-modules/cloud-modules-car-gateway/src/main/resources/bootstrap.yml +++ b/cloud-modules/cloud-modules-car-gateway/src/main/resources/bootstrap.yml @@ -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: diff --git a/cloud-modules/cloud-modules-data-processing/pom.xml b/cloud-modules/cloud-modules-data-processing/pom.xml index 6db1f80..572d4d5 100644 --- a/cloud-modules/cloud-modules-data-processing/pom.xml +++ b/cloud-modules/cloud-modules-data-processing/pom.xml @@ -90,51 +90,6 @@ com.muyu cloud-common-datasource - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/CloudVehicleEventApplication.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/CloudVehicleEventApplication.java index e2519a3..751ae62 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/CloudVehicleEventApplication.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/CloudVehicleEventApplication.java @@ -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); diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/controller/DataProcessingController.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/controller/DataProcessingController.java index 8ab8328..d44f2ba 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/controller/DataProcessingController.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/controller/DataProcessingController.java @@ -32,69 +32,4 @@ public class DataProcessingController { @Resource private DataProcessingService service; - /** - * 查看数据库有多少组 - * - * @return - */ - @GetMapping("/selectStorageGroup") - public Result selectStorageGroup() { - List 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 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)); - } - - } diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/controller/TestController.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/controller/TestController.java deleted file mode 100644 index 5704ff7..0000000 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/controller/TestController.java +++ /dev/null @@ -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 kafkaProducer; - @Resource - private RabbitTemplate rabbitTemplate; - @Resource - private IotDBSessionConfig iotDBSessionConfig; - @Resource - private RedisTemplate 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 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 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()); - } -} diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/domain/BasicData.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/domain/BasicData.java index 706b2cf..b4cae78 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/domain/BasicData.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/domain/BasicData.java @@ -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); -// } -// } } diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/kafka/KafkaConsumerService.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/kafka/KafkaConsumerService.java index b4b2196..ff1eaf5 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/kafka/KafkaConsumerService.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/kafka/KafkaConsumerService.java @@ -39,28 +39,31 @@ 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 topics = Lists.newArrayList(KafkaConstants.KafkaTopic); kafkaConsumer.subscribe(topics); while (true) { - System.out.println("开始消费数据,等待中..."); - ConsumerRecords consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000)); - for (ConsumerRecord consumerRecord : consumerRecords) { - //1.从ConsumerRecord中获取消费数据 - String originalMsg = (String) consumerRecord.value(); - log.info("从Kafka中消费的原始数据: " + originalMsg); - //2.把消费数据转换为DTO对象 - List dataList = JSONUtil.toList(originalMsg, BasicData.class); - log.info("从Kafka中消费的实体数据: " + dataList); - // 执行策略 - startStrategy.applyStrategy(getDataMap(dataList)); + try { + ThreadUtil.sleep(1000); + System.out.println("开始消费数据,等待中..."); + ConsumerRecords consumerRecords = kafkaConsumer.poll(Duration.ofMillis(1000)); + for (ConsumerRecord consumerRecord : consumerRecords) { + //1.从ConsumerRecord中获取消费数据 + String originalMsg = (String) consumerRecord.value(); + log.info("从Kafka中消费的原始数据: " + originalMsg); + //2.把消费数据转换为DTO对象 + List dataList = JSONUtil.toList(originalMsg, BasicData.class); + log.info("从Kafka中消费的实体数据: " + dataList); + // 执行策略 + startStrategy.applyStrategy(getDataMap(dataList)); + } + }catch (Exception e) { + log.error("kafka执行异常：" + e.getMessage()); } } - }); - thread.start(); + }).start(); } private HashMap getDataMap(List dataList) { diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/mapper/DataProcessingMapper.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/mapper/DataProcessingMapper.java index fdd1cfe..eaea598 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/mapper/DataProcessingMapper.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/mapper/DataProcessingMapper.java @@ -23,13 +23,4 @@ import java.util.List; @Mapper public interface DataProcessingMapper{ - List selectStorageGroup(); - - Integer insIotDbData(@Param("key") String key, @Param("value") String value); - - void strategyCheck(@Param("dataList") List dataList); - - Integer insIotDbDataVo(IotDbData build); - - List selectCarData(@Param("tableName") String tableName); } diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/rebbit/DownlineRabbitConsumer.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/rebbit/DownlineRabbitConsumer.java index c3a38b6..97a496b 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/rebbit/DownlineRabbitConsumer.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/rebbit/DownlineRabbitConsumer.java @@ -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 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 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); +// } +//} diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/rebbit/GoOnlineRabbitConsumer.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/rebbit/GoOnlineRabbitConsumer.java index 05bae21..e74815e 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/rebbit/GoOnlineRabbitConsumer.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/rebbit/GoOnlineRabbitConsumer.java @@ -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 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 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); +// } +//} diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/service/DataProcessingService.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/service/DataProcessingService.java index 67cacf3..03cf4a0 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/service/DataProcessingService.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/service/DataProcessingService.java @@ -19,18 +19,4 @@ import java.util.List; public interface DataProcessingService{ - /** - * 选择存储组 - * - * @return {@link List }<{@link String }> - */ - List selectStorageGroup(); - - void strategyCheck(List dataList); - - Integer insIotDbData(String key, String value); - - List selectCarData(String firmCode, String vin); - - Object addCarData(HashMap hashMap); } diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/service/impl/DataProcessingServiceImpl.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/service/impl/DataProcessingServiceImpl.java index 8d824bb..67abfcb 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/service/impl/DataProcessingServiceImpl.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/service/impl/DataProcessingServiceImpl.java @@ -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 selectStorageGroup() { - return mapper.selectStorageGroup(); - } - - @Override - public void strategyCheck(List dataList) { - HashMap kafkaDataHashMap = new HashMap<>(); - dataList.forEach(data -> kafkaDataHashMap.put(data.getKey(), data)); -// Result 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 selectCarData(String firmCode, String vin) { - ArrayList carDataList = new ArrayList<>(); - String sql = "select * from root.one."+firmCode+"."+vin; - try { - SessionDataSetWrapper dataSetWrapper = sessionPool.executeQueryStatement(sql); - List 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 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 columnNames) { - List 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; - } } diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/abstractStrategyRouter.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/abstractStrategyRouter.java index 9da9691..3a788da 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/abstractStrategyRouter.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/abstractStrategyRouter.java @@ -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 { + protected static CacheManager cacheManager = SpringContentUtils.getBean(CacheManager.class); + /** * 策略映射器, 指定入参与出参以决定策略处理者 * @param 策略入参 diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/BasicStrategy.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/BasicStrategy.java index d67112b..886f2b3 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/BasicStrategy.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/BasicStrategy.java @@ -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, Temporary2> implements StrategyHandler, Temporary2> { + @Override protected StrategyMapper, Temporary2> registerStrategy() { return param -> new RoutingStrategy(); @@ -33,8 +38,10 @@ public class BasicStrategy extends abstractStrategyRouter basicDataMap) { log.info("开始执行基础校验节点。。。"); basicDataMap.put(CacheNameEnums.STORAGE.getCode(), null); - basicDataMap.put(CacheNameEnums.FAULT.getCode(), null); - basicDataMap.put(CacheNameEnums.REALTIME.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); } diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/RoutingStrategy.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/RoutingStrategy.java index ce46f71..c09cca3 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/RoutingStrategy.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/core/RoutingStrategy.java @@ -38,8 +38,7 @@ public class RoutingStrategy extends abstractStrategyRouter, Temporary2> registerStrategy() { return param -> { // 编写路由规则 - List codes = CacheNameEnums.getCodes(); - for (String code : codes) { + for (String code : map.keySet()) { if(param.containsKey(code)){ param.remove(code); return map.get(code); diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/DataStorageStrategy.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/DataStorageStrategy.java index a57c426..3b6e005 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/DataStorageStrategy.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/DataStorageStrategy.java @@ -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; /** diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/FaultAlarmStrategy.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/FaultAlarmStrategy.java index 2ebad77..febb0d6 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/FaultAlarmStrategy.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/FaultAlarmStrategy.java @@ -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; diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/FenceAlarmStrategy.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/FenceAlarmStrategy.java index 6845011..a5b43eb 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/FenceAlarmStrategy.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/FenceAlarmStrategy.java @@ -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; diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/RealTimeAlarmStrategy.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/RealTimeAlarmStrategy.java index 0a4cbc3..696736b 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/RealTimeAlarmStrategy.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/RealTimeAlarmStrategy.java @@ -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; diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/WarningAlarmStrategy.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/WarningAlarmStrategy.java index 8026940..71c8c0c 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/WarningAlarmStrategy.java +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/strategy/leaves/WarningAlarmStrategy.java @@ -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; diff --git a/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/utils/CacheUtils.java b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/utils/CacheUtils.java new file mode 100644 index 0000000..9682206 --- /dev/null +++ b/cloud-modules/cloud-modules-data-processing/src/main/java/com/muyu/data/processing/utils/CacheUtils.java @@ -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 getCacheValue(String cacheName, String key, Class type) { + T value = getCache(cacheName).get(key, type); + log.info("获取缓存信息:分区-{}, 键-{}, 值-{}", cacheName, key, value); + return value; + } +} diff --git a/cloud-modules/cloud-modules-data-processing/src/main/resources/bootstrap.yml b/cloud-modules/cloud-modules-data-processing/src/main/resources/bootstrap.yml index b40171b..8bf82eb 100644 --- a/cloud-modules/cloud-modules-data-processing/src/main/resources/bootstrap.yml +++ b/cloud-modules/cloud-modules-data-processing/src/main/resources/bootstrap.yml @@ -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 diff --git a/cloud-modules/cloud-modules-data-processing/src/main/resources/mapper/processing/DataProcessingMapper.xml b/cloud-modules/cloud-modules-data-processing/src/main/resources/mapper/processing/DataProcessingMapper.xml deleted file mode 100644 index 7766db8..0000000 --- a/cloud-modules/cloud-modules-data-processing/src/main/resources/mapper/processing/DataProcessingMapper.xml +++ /dev/null @@ -1,38 +0,0 @@ - - - - - - - - - - insert into root.one.data(${key}) values(${value}); - - - insert into root.one.data - ( - - ${data.key} - - ) values - ( - - #{data.value} - - ) - - - - insert into - root.one.data - (timestamp, vin, latitude,longitude) - values (#{timestamp}, #{vin}, #{latitude}, #{longitude}) - - - -