diff --git a/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DataCleanServiceImpl.java b/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DataCleanServiceImpl.java index 720c846..58e1b67 100644 --- a/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DataCleanServiceImpl.java +++ b/algorithm/algorithm-boot/src/main/java/com/njcn/algorithm/serviceimpl/line/DataCleanServiceImpl.java @@ -69,6 +69,23 @@ public class DataCleanServiceImpl implements IDataCleanService { @Override public void dataVCleanHandler(CalculatedParam calculatedParam) { + Runtime runtime = Runtime.getRuntime(); + // 获取 JVM 最大可用内存（以字节为单位） + long maxMemory = runtime.maxMemory(); + // 获取 JVM 当前已分配的内存（以字节为单位） + long totalMemory = runtime.totalMemory(); + // 获取 JVM 当前空闲内存（以字节为单位） + long freeMemory = runtime.freeMemory(); + // 计算已使用的内存 + long usedMemory = totalMemory - freeMemory; + + System.out.println("最大可用内存: " + maxMemory / (1024 * 1024) + " MB"); + System.out.println("当前已分配的内存: " + totalMemory / (1024 * 1024) + " MB"); + System.out.println("当前空闲内存: " + freeMemory / (1024 * 1024) + " MB"); + System.out.println("已使用的内存: " + usedMemory / (1024 * 1024) + " MB"); + System.out.println("第一次分析结束-----------------------------------------"); + + logger.info("{},dataV表异常数据算法执行=====》", LocalDateTime.now()); List result = new ArrayList<>(); //获取标准 diff --git a/algorithm/algorithm-boot/src/main/resources/bootstrap.yml b/algorithm/algorithm-boot/src/main/resources/bootstrap.yml index 6d0bdba..152a842 100644 --- a/algorithm/algorithm-boot/src/main/resources/bootstrap.yml +++ b/algorithm/algorithm-boot/src/main/resources/bootstrap.yml @@ -31,6 +31,8 @@ spring: refresh: true - data-Id: share-config-datasource-db.yaml refresh: true + - data-Id: data-platform.yaml + refresh: true main: allow-bean-definition-overriding: true diff --git a/analysis/event/event-boot/src/main/resources/bootstrap.yml b/analysis/event/event-boot/src/main/resources/bootstrap.yml index da86ebe..2a90fcb 100644 --- a/analysis/event/event-boot/src/main/resources/bootstrap.yml +++ b/analysis/event/event-boot/src/main/resources/bootstrap.yml @@ -29,7 +29,7 @@ spring: shared-configs: - data-id: share-config.yaml refresh: true - - data-Id: algorithm-config.yaml + - data-Id: share-config-datasource-db.yaml refresh: true main: allow-bean-definition-overriding: true diff --git a/analysis/rt/rt-boot/src/main/resources/bootstrap.yml b/analysis/rt/rt-boot/src/main/resources/bootstrap.yml index 4e273c4..b919c3f 100644 --- a/analysis/rt/rt-boot/src/main/resources/bootstrap.yml +++ b/analysis/rt/rt-boot/src/main/resources/bootstrap.yml @@ -29,7 +29,7 @@ spring: shared-configs: - data-id: share-config.yaml refresh: true - - data-Id: algorithm-config.yaml + - data-Id: share-config-datasource-db.yaml refresh: true main: allow-bean-definition-overriding: true diff --git a/analysis/stat/stat-boot/src/main/resources/bootstrap.yml b/analysis/stat/stat-boot/src/main/resources/bootstrap.yml index 6dde5df..d66e6d9 100644 --- a/analysis/stat/stat-boot/src/main/resources/bootstrap.yml +++ b/analysis/stat/stat-boot/src/main/resources/bootstrap.yml @@ -29,7 +29,7 @@ spring: shared-configs: - data-id: share-config.yaml refresh: true - - data-Id: data-platform.yaml + - data-Id: share-config-datasource-db.yaml refresh: true main: allow-bean-definition-overriding: true diff --git a/data-processing/data-processing-boot/src/main/resources/bootstrap.yml b/data-processing/data-processing-boot/src/main/resources/bootstrap.yml index c32d6f1..69fc72b 100644 --- a/data-processing/data-processing-boot/src/main/resources/bootstrap.yml +++ b/data-processing/data-processing-boot/src/main/resources/bootstrap.yml @@ -30,6 +30,8 @@ spring: shared-configs: - data-id: share-config.yaml refresh: true + - data-Id: share-config-datasource-db.yaml + refresh: true - data-Id: data-platform.yaml refresh: true main: diff --git a/message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumerTest.java b/message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumerTest.java new file mode 100644 index 0000000..2daacca --- /dev/null +++ b/message/message-boot/src/main/java/com/njcn/message/consumer/FrontDataConsumerTest.java @@ -0,0 +1,173 @@ +//package com.njcn.message.consumer; +// +//import com.alibaba.fastjson.JSONObject; +//import com.njcn.message.constant.MessageStatus; +//import com.njcn.message.messagedto.MessageDataDTO; +//import com.njcn.middle.rocket.constant.EnhanceMessageConstant; +//import com.njcn.middle.rocket.handler.EnhanceConsumerMessageHandler; +//import com.njcn.redis.pojo.enums.RedisKeyEnum; +//import com.njcn.redis.utils.RedisUtil; +//import com.njcn.stat.api.MessAnalysisFeignClient; +//import com.njcn.system.api.RocketMqLogFeignClient; +//import com.njcn.system.pojo.po.RocketmqMsgErrorLog; +//import lombok.extern.slf4j.Slf4j; +//import org.apache.rocketmq.spring.annotation.RocketMQMessageListener; +//import org.apache.rocketmq.spring.core.RocketMQListener; +//import org.springframework.beans.factory.annotation.Autowired; +//import org.springframework.stereotype.Component; +// +//import javax.annotation.Resource; +//import java.util.ArrayList; +//import java.util.List; +//import java.util.Objects; +// +// +///** +// * 类的介绍： +// * +// * @author xuyang +// * @version 1.0.0 +// * @createTime 2023/8/11 15:32 +// */ +//@Component +//@RocketMQMessageListener( +// topic = "Test_Topic", +// consumerGroup = "Test_consumer", +// selectorExpression = "Test_Tag||Test_Keys", +// consumeThreadNumber = 10, +// enableMsgTrace = true +//) +//@Slf4j +//public class FrontDataConsumerTest extends EnhanceConsumerMessageHandler implements RocketMQListener { +// +// @Autowired +// private MessAnalysisFeignClient messAnalysisFeignClient; +// +// +// +// @Resource +// private RedisUtil redisUtil; +// +// @Resource +// private RocketMqLogFeignClient rocketMqLogFeignClient; +// +// +// private List messageList = new ArrayList<>(1); +// +// @Override +// public void onMessage(String baseMessage) { +// MessageDataDTO messageDataDTO = JSONObject.parseObject(baseMessage,MessageDataDTO.class); +// super.dispatchMessage(messageDataDTO); +// +// } +// +// /*** +// * 通过redis分布式锁判断当前消息所处状态 +// * 1、null 查不到该key的数据，属于第一次消费，放行 +// * 2、fail 上次消息消费时发生异常，放行 +// * 3、being processed 正在处理，打回去 +// * 4、success 最近72小时消费成功，避免重复消费，打回去 +// */ +// @Override +// public boolean filter(MessageDataDTO message) { +// String keyStatus = redisUtil.getStringByKey(message.getKey()); +// if (Objects.isNull(keyStatus) || keyStatus.equalsIgnoreCase(MessageStatus.FAIL)) { +// redisUtil.saveByKeyWithExpire(message.getKey(), MessageStatus.BEING_PROCESSED, 60L); +// return false; +// } +// return true; +// } +// /** +// * 消费成功，缓存到redis72小时，避免重复消费 +// */ +// @Override +// protected void consumeSuccess(MessageDataDTO message) { +// redisUtil.saveByKeyWithExpire(message.getKey(), MessageStatus.SUCCESS, RedisKeyEnum.ROCKET_MQ_KEY.getTime()); +// } +// +// +// @Override +// protected void handleMessage(MessageDataDTO message) { +// synchronized (messageList) { +// messageList.add(message); +// if (messageList.size() >= 1) { +// saveToDatabase(); +// } +// } +// } +// +// +// +// +// /** +// * 发生异常时，进行错误信息入库保存 +// * 默认没有实现类，子类可以实现该方法，调用feign接口进行入库保存 +// */ +// @Override +// protected void saveExceptionMsgLog(MessageDataDTO message, String identity, Exception exception) { +// redisUtil.saveByKeyWithExpire(message.getKey(), MessageStatus.FAIL, RedisKeyEnum.ROCKET_MQ_KEY.getTime()); +// RocketmqMsgErrorLog rocketmqMsgErrorLog = new RocketmqMsgErrorLog(); +// rocketmqMsgErrorLog.setMsgKey(message.getKey()); +// rocketmqMsgErrorLog.setResource(message.getSource()); +// if (identity.equalsIgnoreCase(EnhanceMessageConstant.IDENTITY_SINGLE)) { +// //数据库字段配置长度200，避免插入失败，大致分析异常原因 +// String exceptionMsg = exception.getMessage(); +// if(exceptionMsg.length() > 200){ +// exceptionMsg = exceptionMsg.substring(0,180); +// } +// rocketmqMsgErrorLog.setRecord(exceptionMsg); +// //如果是当前消息重试的则略过 +// if(!message.getSource().startsWith(EnhanceMessageConstant.RETRY_PREFIX)){ +// //单次消费异常 +// rocketMqLogFeignClient.add(rocketmqMsgErrorLog); +// } +// } else { +// rocketmqMsgErrorLog.setRecord("重试消费" + super.getMaxRetryTimes() + "次，依旧消费失败。"); +// //重试N次后，依然消费异常 +// rocketMqLogFeignClient.add(rocketmqMsgErrorLog); +// } +// } +// +// +// /*** +// * 处理失败后，是否重试 +// * 一般开启 +// */ +// @Override +// protected boolean isRetry() { +// return false; +// } +// +// +// /*** +// * 消费失败是否抛出异常，抛出异常后就不再消费了 +// */ +// @Override +// protected boolean throwException() { +// return false; +// } +// +// +// +// //50个消息做一组插入数据库 +// public void saveToDatabase(){ +// try { +// long start = System.currentTimeMillis(); +// +// messAnalysisFeignClient.analysis(messageList); +// +// long end = System.currentTimeMillis(); +// log.info("处理120条消息所需时间------------"+(end-start)); +// }catch (Exception e){{ +// log.info(e.toString()); +// } +// }finally{ +// messageList.clear(); +// } +// +// } +// +// +// +// +//} diff --git a/message/message-boot/src/main/resources/bootstrap.yml b/message/message-boot/src/main/resources/bootstrap.yml index 3898c94..1f93eda 100644 --- a/message/message-boot/src/main/resources/bootstrap.yml +++ b/message/message-boot/src/main/resources/bootstrap.yml @@ -29,7 +29,7 @@ spring: shared-configs: - data-id: share-config.yaml refresh: true - - data-Id: algorithm-config.yaml + - data-Id: data-platform.yaml refresh: true main: allow-bean-definition-overriding: true