From 9fdafe542d2ce43337c75b8602c64ecf0aee4993 Mon Sep 17 00:00:00 2001 From: gaoyutao Date: Thu, 26 Feb 2026 14:07:37 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=86=85=E5=AD=98=E8=AF=A6?= =?UTF-8?q?=E7=BB=86=E5=A4=84=E7=90=86=E3=80=81=E4=BC=98=E5=8C=96=E5=8E=86?= =?UTF-8?q?=E5=8F=B2=E6=B6=88=E6=81=AF=E5=A4=84=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../tongran/common/core/enums/MsgEnum.java | 2 + .../RmResourceRegistrationController.java | 1 + .../consumer/TimeRangeReConsumer.java | 138 +++++++++----- .../InitialMemoryDetailsInfoController.java | 97 ++++++++++ .../domain/InitialMemoryDetailsInfo.java | 79 ++++++++ .../rocketmq/domain/vo/MemoryDetailsVo.java | 21 +++ .../tongran/rocketmq/domain/vo/MemoryVo.java | 17 ++ .../rocketmq/handler/MessageHandler.java | 24 +++ .../InitialMemoryDetailsInfoMapper.java | 61 +++++++ .../IInitialMemoryDetailsInfoService.java | 61 +++++++ .../InitialMemoryDetailsInfoServiceImpl.java | 96 ++++++++++ .../InitialMemoryDetailsInfoMapper.xml | 171 ++++++++++++++++++ 12 files changed, 718 insertions(+), 50 deletions(-) create mode 100644 tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/InitialMemoryDetailsInfoController.java create mode 100644 tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/InitialMemoryDetailsInfo.java create mode 100644 tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/MemoryDetailsVo.java create mode 100644 tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/MemoryVo.java create mode 100644 tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialMemoryDetailsInfoMapper.java create mode 100644 tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialMemoryDetailsInfoService.java create mode 100644 tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialMemoryDetailsInfoServiceImpl.java create mode 100644 tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialMemoryDetailsInfoMapper.xml diff --git a/tongran-common/tongran-common-core/src/main/java/com/tongran/common/core/enums/MsgEnum.java b/tongran-common/tongran-common-core/src/main/java/com/tongran/common/core/enums/MsgEnum.java index bbea723..5a39dc8 100644 --- a/tongran-common/tongran-common-core/src/main/java/com/tongran/common/core/enums/MsgEnum.java +++ b/tongran-common/tongran-common-core/src/main/java/com/tongran/common/core/enums/MsgEnum.java @@ -23,6 +23,8 @@ public enum MsgEnum { tcpdump结果上报("TCPDUMP_RESULT"), + 内存详情上报("MEMORY_DETAILS"), + 获取最新策略("GET_POLICY"), 获取最新策略应答("GET_POLICY_RSP"), diff --git a/tongran-modules/tongran-system/src/main/java/com/tongran/system/controller/RmResourceRegistrationController.java b/tongran-modules/tongran-system/src/main/java/com/tongran/system/controller/RmResourceRegistrationController.java index b86ad90..dc64242 100644 --- a/tongran-modules/tongran-system/src/main/java/com/tongran/system/controller/RmResourceRegistrationController.java +++ b/tongran-modules/tongran-system/src/main/java/com/tongran/system/controller/RmResourceRegistrationController.java @@ -74,6 +74,7 @@ public class RmResourceRegistrationController extends BaseController @PostMapping("/export") public void export(HttpServletResponse response, @RequestBody RmResourceRegistration rmResourceRegistration) { + rmResourceRegistration.setDelFlag(0); List list = rmResourceRegistrationService.selectRmResourceRegistrationList(rmResourceRegistration); ExcelUtil util = new ExcelUtil(RmResourceRegistration.class); util.showColumn(rmResourceRegistration.getProperties()); diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/TimeRangeReConsumer.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/TimeRangeReConsumer.java index afd8ec0..b9937f1 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/TimeRangeReConsumer.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/TimeRangeReConsumer.java @@ -18,6 +18,7 @@ import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently; import org.apache.rocketmq.common.consumer.ConsumeFromWhere; import org.apache.rocketmq.common.message.MessageExt; import org.apache.rocketmq.common.message.MessageQueue; +import org.springframework.beans.factory.DisposableBean; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.stereotype.Component; @@ -30,15 +31,9 @@ import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; -/** - * 时间范围重消费消费者（两阶段优化版） - * 第一阶段：并行消费并缓存所有消息 - * 第二阶段：按设备顺序处理缓存的消息 - */ @Slf4j @Component -public class TimeRangeReConsumer { - +public class TimeRangeReConsumer implements DisposableBean { @Autowired(required = false) private MessageHistoryDataHandler messageHandler; @@ -48,15 +43,15 @@ public class TimeRangeReConsumer { private static final String END_TIME = "2026-02-02 12:00:00"; // 统计信息 - private final AtomicLong receivedCount = new AtomicLong(0); // 已接收消息数 + private final AtomicLong receivedCount = new AtomicLong(0); private final AtomicLong skippedCount = new AtomicLong(0); - private final AtomicLong processedCount = new AtomicLong(0); // 已处理消息数 + private final AtomicLong processedCount = new AtomicLong(0); private final Map queueCurrentOffsets = new ConcurrentHashMap<>(); private final Set completedQueues = ConcurrentHashMap.newKeySet(); // 控制是否继续消费 private AtomicBoolean shouldContinue = new AtomicBoolean(true); - private AtomicBoolean isProcessingPhase = new AtomicBoolean(false); // 第二阶段处理标记 + private AtomicBoolean isProcessingPhase = new AtomicBoolean(false); // 消息缓存：按clientId分组，按时间戳排序 private final ConcurrentHashMap> messageCache = @@ -78,8 +73,8 @@ public class TimeRangeReConsumer { private String msgId; private DeviceMessage message; private MessageExt messageExt; - private long timestamp; // 从data中解析的时间戳 - private long bornTimestamp; // 消息产生时间 + private long timestamp; + private long bornTimestamp; @Override public int compareTo(MessageWrapper other) { @@ -87,11 +82,25 @@ public class TimeRangeReConsumer { } } - - // 添加一个标识字段 private final AtomicBoolean isRunning = new AtomicBoolean(false); - // 暴露启动方法 + // 添加销毁方法 + @Override + public void destroy() throws Exception { + log.info("正在关闭TimeRangeReConsumer线程池..."); + if (sequentialProcessor != null && !sequentialProcessor.isShutdown()) { + sequentialProcessor.shutdown(); + try { + if (!sequentialProcessor.awaitTermination(10, TimeUnit.SECONDS)) { + sequentialProcessor.shutdownNow(); + } + } catch (InterruptedException e) { + sequentialProcessor.shutdownNow(); + Thread.currentThread().interrupt(); + } + } + log.info("TimeRangeReConsumer线程池已关闭"); + } public String startReconsumeTask(String startTime, String endTime, String namesrvAddr) { if (isRunning.compareAndSet(false, true)) { new Thread(() -> { @@ -107,6 +116,7 @@ public class TimeRangeReConsumer { } return "已有任务正在运行"; } + public void startTimeRangeConsumer(String startTime, String endTime, String namesrvAddr) { new Thread(() -> { try { @@ -118,6 +128,20 @@ public class TimeRangeReConsumer { } private void doReconsume(String startTime, String endTime, String namesrvAddr) throws Exception { + // 重置关键状态 + shouldContinue.set(true); + isProcessingPhase.set(false); + + // 清空缓存 + messageCache.clear(); + processingFutures.clear(); + + // 重置计数器 + receivedCount.set(0); + skippedCount.set(0); + processedCount.set(0); + queueCurrentOffsets.clear(); + completedQueues.clear(); SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); long startTimestamp = sdf.parse(startTime).getTime(); long endTimestamp = sdf.parse(endTime).getTime(); @@ -166,7 +190,7 @@ public class TimeRangeReConsumer { completedQueues.add(queueId); // 检查是否所有队列都完成 - if (completedQueues.size() >= 4) { // 假设有4个队列 + if (completedQueues.size() >= 4) { log.info("所有队列均已超过结束时间，第一阶段完成"); shouldContinue.set(false); } @@ -178,8 +202,7 @@ public class TimeRangeReConsumer { String dataType = message.getDataType(); String clientId = message.getClientId(); // 只处理"网络上报重试"类型的消息 - if(MsgEnum.网络上报重试.getValue().equals(dataType)){ - + if (MsgEnum.网络上报重试.getValue().equals(dataType) && !"a7948be4439e40bf09acf48164b336a9".equals(clientId)) { receivedCount.incrementAndGet(); // 只缓存消息，不处理 @@ -272,7 +295,7 @@ public class TimeRangeReConsumer { private long extractTimestampFromData(String data) { try { List interfaces = JsonDataParser.parseJsonData(data, InitialBandwidthTraffic.class); - if(!interfaces.isEmpty()) { + if (!interfaces.isEmpty()) { // 时间戳转换 long timestamp = interfaces.get(0).getTimestamp(); return timestamp; @@ -280,7 +303,7 @@ public class TimeRangeReConsumer { } catch (Exception e) { log.warn("解析data时间戳失败，使用默认值", e); } - return System.currentTimeMillis(); // 解析失败使用当前时间 + return System.currentTimeMillis(); } /** @@ -417,8 +440,6 @@ public class TimeRangeReConsumer { } catch (Exception e) { log.error("处理设备{}消息失败: msgId={}", clientId, wrapper.getMsgId(), e); - // 根据业务决定是否重试或跳过 - // 这里选择跳过，记录错误 } } } @@ -433,24 +454,30 @@ public class TimeRangeReConsumer { } /** - * 等待第二阶段处理完成 + * 等待第二阶段处理完成 - 添加进度监控 */ private void waitForPhase2Completion() { log.info("等待所有设备顺序处理完成..."); + // ==================== 新增：启动第二阶段进度监控 ==================== + ScheduledExecutorService phase2Monitor = Executors.newSingleThreadScheduledExecutor(); + phase2Monitor.scheduleAtFixedRate(() -> { + if (isProcessingPhase.get() && !processingFutures.isEmpty()) { + printPhase2Progress(); + } + }, 5, 10, TimeUnit.SECONDS); // 5秒后开始，每10秒输出一次进度 + CompletableFuture allFutures = CompletableFuture.allOf( processingFutures.values().toArray(new CompletableFuture[0]) ); try { // 等待所有任务完成 - allFutures.get(2, TimeUnit.HOURS); // 最多等待2小时 - + allFutures.get(2, TimeUnit.HOURS); log.info("✓ 所有设备顺序处理完成"); } catch (TimeoutException e) { log.warn("第二阶段处理超时（2小时），强制结束"); - // 可以记录未完成的任务 List unfinishedDevices = new ArrayList<>(); for (Map.Entry> entry : processingFutures.entrySet()) { if (!entry.getValue().isDone()) { @@ -462,9 +489,21 @@ public class TimeRangeReConsumer { } catch (Exception e) { log.error("等待第二阶段处理异常", e); } finally { + // 关闭进度监控 + phase2Monitor.shutdownNow(); + try { + phase2Monitor.awaitTermination(3, TimeUnit.SECONDS); + } catch (InterruptedException e) { + log.warn("进度监控关闭异常", e); + Thread.currentThread().interrupt(); + } + shutdownResources(); } + // 最后输出一次进度 + printPhase2Progress(); + // 最后统计 int remainingMessages = messageCache.values().stream() .mapToInt(Queue::size) @@ -477,12 +516,11 @@ public class TimeRangeReConsumer { messageCache.clear(); } } - /** - * 处理完成后关闭资源 - 添加状态重置 - */ - private void shutdownResources() { - log.info("重消费完成，正在关闭资源..."); + private void shutdownResources() { + log.info("重消费完成，正在清理资源..."); + + // 不要关闭线程池，只重置状态 shouldContinue.set(false); isProcessingPhase.set(false); isRunning.set(false); @@ -491,28 +529,21 @@ public class TimeRangeReConsumer { messageCache.clear(); processingFutures.clear(); - // 关闭线程池 - sequentialProcessor.shutdown(); - try { - if (!sequentialProcessor.awaitTermination(10, TimeUnit.SECONDS)) { - sequentialProcessor.shutdownNow(); - } - } catch (InterruptedException e) { - sequentialProcessor.shutdownNow(); - Thread.currentThread().interrupt(); - } - - // 必须重置计数器和集合 + // 重置计数器和集合 receivedCount.set(0); processedCount.set(0); skippedCount.set(0); queueCurrentOffsets.clear(); completedQueues.clear(); - log.info("✓ 重消费者资源已完全释放"); + // 注意：不要关闭线程池！它会被复用 + // sequentialProcessor.shutdown(); // 注释掉这行 + + log.info("✓ 重消费者状态已重置"); } + /** - * 查找大致的起始offset + * 查找大致的起始offset - 修复版（处理异常） */ private void findApproximateStartOffset(DefaultMQPushConsumer consumer, long startTimestamp) throws Exception { SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); @@ -547,13 +578,20 @@ public class TimeRangeReConsumer { log.info("队列{}: offset={} (范围: {}-{})", queue.getQueueId(), offset, minOffset, maxOffset); } else { - // 使用最小offset - long minOffset = consumer.minOffset(queue); - queueCurrentOffsets.put(queue.getQueueId(), minOffset); - log.info("队列{}: 使用最小offset={}", queue.getQueueId(), minOffset); + // 使用最小offset - 使用新方法避免过时API + try { + long minOffset = consumer.minOffset(queue); + queueCurrentOffsets.put(queue.getQueueId(), minOffset); + log.info("队列{}: 使用最小offset={}", queue.getQueueId(), minOffset); + } catch (Exception e) { + // 如果minOffset也失败，使用0作为兜底 + log.warn("队列{}获取最小offset失败，使用0作为兜底: {}", queue.getQueueId(), e.getMessage()); + queueCurrentOffsets.put(queue.getQueueId(), 0L); + } } } catch (Exception e) { - log.warn("队列{}设置offset失败: {}", queue.getQueueId(), e.getMessage()); + log.warn("队列{}设置offset失败，使用0作为兜底: {}", queue.getQueueId(), e.getMessage()); + queueCurrentOffsets.put(queue.getQueueId(), 0L); } } diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/InitialMemoryDetailsInfoController.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/InitialMemoryDetailsInfoController.java new file mode 100644 index 0000000..c1f8c0e --- /dev/null +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/InitialMemoryDetailsInfoController.java @@ -0,0 +1,97 @@ +package com.tongran.rocketmq.controller; + +import com.tongran.common.core.utils.poi.ExcelUtil; +import com.tongran.common.core.web.controller.BaseController; +import com.tongran.common.core.web.domain.AjaxResult; +import com.tongran.common.core.web.page.PageDomain; +import com.tongran.common.core.web.page.TableDataInfo; +import com.tongran.common.log.annotation.Log; +import com.tongran.common.log.enums.BusinessType; +import com.tongran.common.security.annotation.RequiresPermissions; +import com.tongran.rocketmq.domain.InitialMemoryDetailsInfo; +import com.tongran.rocketmq.service.IInitialMemoryDetailsInfoService; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.*; + +import javax.servlet.http.HttpServletResponse; +import java.util.List; + +/** + * 内存详细信息Controller + * + * @author gyt + * @date 2026-02-24 + */ +@RestController +@RequestMapping("/initialMemoryDetailsInfo") +@RequiresPermissions("rocketmq:traffic") +public class InitialMemoryDetailsInfoController extends BaseController +{ + @Autowired + private IInitialMemoryDetailsInfoService initialMemoryDetailsInfoService; + + /** + * 查询内存详细信息列表 + */ + @PostMapping("/list") + public TableDataInfo list(@RequestBody InitialMemoryDetailsInfo initialMemoryDetailsInfo) + { + PageDomain pageDomain = new PageDomain(); + pageDomain.setPageNum(initialMemoryDetailsInfo.getPageNum()); + pageDomain.setPageSize(initialMemoryDetailsInfo.getPageSize()); + startPage(pageDomain); + List list = initialMemoryDetailsInfoService.selectInitialMemoryDetailsInfoList(initialMemoryDetailsInfo); + return getDataTable(list); + } + + /** + * 导出内存详细信息列表 + */ + @Log(title = "内存详细信息", businessType = BusinessType.EXPORT) + @PostMapping("/export") + public void export(HttpServletResponse response, InitialMemoryDetailsInfo initialMemoryDetailsInfo) + { + List list = initialMemoryDetailsInfoService.selectInitialMemoryDetailsInfoList(initialMemoryDetailsInfo); + ExcelUtil util = new ExcelUtil(InitialMemoryDetailsInfo.class); + util.exportExcel(response, list, "内存详细信息数据"); + } + + /** + * 获取内存详细信息详细信息 + */ + @GetMapping(value = "/{id}") + public AjaxResult getInfo(@PathVariable("id") Long id) + { + return success(initialMemoryDetailsInfoService.selectInitialMemoryDetailsInfoById(id)); + } + + /** + * 新增内存详细信息 + */ + @Log(title = "内存详细信息", businessType = BusinessType.INSERT) + @PostMapping + public AjaxResult add(@RequestBody InitialMemoryDetailsInfo initialMemoryDetailsInfo) + { + return toAjax(initialMemoryDetailsInfoService.insertInitialMemoryDetailsInfo(initialMemoryDetailsInfo)); + } + + /** + * 修改内存详细信息 + */ + @Log(title = "内存详细信息", businessType = BusinessType.UPDATE) + @PutMapping + public AjaxResult edit(@RequestBody InitialMemoryDetailsInfo initialMemoryDetailsInfo) + { + return toAjax(initialMemoryDetailsInfoService.updateInitialMemoryDetailsInfo(initialMemoryDetailsInfo)); + } + + /** + * 删除内存详细信息 + */ + @Log(title = "内存详细信息", businessType = BusinessType.DELETE) + @DeleteMapping("/{ids}") + public AjaxResult remove(@PathVariable Long[] ids) + { + return toAjax(initialMemoryDetailsInfoService.deleteInitialMemoryDetailsInfoByIds(ids)); + } +} diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/InitialMemoryDetailsInfo.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/InitialMemoryDetailsInfo.java new file mode 100644 index 0000000..39d9661 --- /dev/null +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/InitialMemoryDetailsInfo.java @@ -0,0 +1,79 @@ +package com.tongran.rocketmq.domain; + +import com.tongran.common.core.annotation.Excel; +import com.tongran.common.core.web.domain.BaseEntity; +import lombok.Data; + +/** + * 内存详细信息对象 initial_memory_details_info + * + * @author gyt + * @date 2026-02-24 + */ +@Data +public class InitialMemoryDetailsInfo extends BaseEntity +{ + private static final long serialVersionUID = 1L; + + /** 主键ID */ + private Long id; + + /** 客户端ID */ + @Excel(name = "客户端ID") + private String clientId; + + /** 插槽总数 */ + @Excel(name = "插槽总数") + private Integer total; + + /** 名称（如DIMM000） */ + @Excel(name = "名称") + private String name; + + /** 位置（如mainboard） */ + @Excel(name = "位置") + private String location; + + /** 厂商（如Samsung） */ + @Excel(name = "厂商") + private String manufacturer; + + /** 容量（MB，如32768） */ + @Excel(name = "容量") + private String capacity; + + /** 主频（MHz，如2133） */ + @Excel(name = "主频") + private String frequency; + + /** 序列号（如0x32A1EA07） */ + @Excel(name = "序列号") + private String serialNumber; + + /** 类型（如DDR4） */ + @Excel(name = "类型") + private String type; + + /** 最小电压（mV，如1200） */ + @Excel(name = "最小电压") + private String minVoltage; + + /** RANK列（如4 rank） */ + @Excel(name = "RANK列") + private String rank; + + /** 位宽（如72 bit） */ + @Excel(name = "位宽") + private String bitWidth; + + /** 技术（如Synchronous|Registered） */ + @Excel(name = "技术") + private String technology; + + /** 部件编码（如M386A4G40DM0-CPB） */ + @Excel(name = "部件编码") + private String partNumber; + /** 健康状态 0配置错误 1正常*/ + private String healthStatus; + +} diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/MemoryDetailsVo.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/MemoryDetailsVo.java new file mode 100644 index 0000000..2c872b8 --- /dev/null +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/MemoryDetailsVo.java @@ -0,0 +1,21 @@ +package com.tongran.rocketmq.domain.vo; + +import lombok.Data; + +@Data +public class MemoryDetailsVo { + private String clientId; + private Integer total; + private String name; // 名称（如 DIMM000） + private String location; // 位置（mainboard） + private String manufacturer; // 厂商（Samsung） + private Long capacity; // 容量（MB，如 32768） + private Integer frequency; // 主频（MHz，如 2133） + private String serialNumber; // 序列号（如 0x32A1EA07） + private String type; // 类型（DDR4） + private Integer minVoltage; // 最小电压（mV，如 1200） + private String rank; // RANK(列)（如 4 rank） + private String bitWidth; // 位宽（如 72 bit） + private String technology; // 技术（Synchronous|Registered (B N/A） + private String partNumber; // 部件编码（如 M386A4G40DM0-CPB） +} diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/MemoryVo.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/MemoryVo.java new file mode 100644 index 0000000..18e7a98 --- /dev/null +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/MemoryVo.java @@ -0,0 +1,17 @@ +package com.tongran.rocketmq.domain.vo; + +import com.tongran.rocketmq.domain.InitialMemoryDetailsInfo; +import lombok.Data; + +import java.util.List; + +@Data +public class MemoryVo { + + + private List details; + + private Integer total; + + private Integer timestamp; +} diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java index fc68392..aacd1a0 100644 --- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java @@ -147,6 +147,8 @@ public class MessageHandler { private SendAlarmPushUtil sendAlarmPushUtil; @Autowired private IRmAlarmThresholdService rmAlarmThresholdService; + @Autowired + private IInitialMemoryDetailsInfoService initialMemoryDetailsInfoService; /** @@ -174,6 +176,28 @@ public class MessageHandler { registerHandler(MsgEnum.macvlan状态上报.getValue(), this::handleMacvlanMessage); registerHandler(MsgEnum.iops结果上报.getValue(), this::handleIopsResultMessage); registerHandler(MsgEnum.tcpdump结果上报.getValue(), this::handleTcpdumpResultMessage); + registerHandler(MsgEnum.内存详情上报.getValue(), this::handleMemoryDetailsMessage); + } + + private void handleMemoryDetailsMessage(DeviceMessage message) { + List memoryVoList = JsonDataParser.parseJsonData(message.getData(), MemoryVo.class); + String clientId = message.getClientId(); + if(memoryVoList != null && !memoryVoList.isEmpty()){ + MemoryVo memoryVo = memoryVoList.get(0); + // 时间戳转换 + long timestamp = memoryVo.getTimestamp(); + long millis = timestamp * 1000; + Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 + List detailsVoList = memoryVo.getDetails(); + if(detailsVoList != null && !detailsVoList.isEmpty()){ + for (InitialMemoryDetailsInfo detailsInfo : detailsVoList) { + detailsInfo.setTotal(memoryVo.getTotal()); + detailsInfo.setClientId(clientId); + detailsInfo.setCreateTime(createTime); + initialMemoryDetailsInfoService.insertInitialMemoryDetailsInfo(detailsInfo); + } + } + } } private void handleTcpdumpResultMessage(DeviceMessage message) { diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialMemoryDetailsInfoMapper.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialMemoryDetailsInfoMapper.java new file mode 100644 index 0000000..0caa986 --- /dev/null +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialMemoryDetailsInfoMapper.java @@ -0,0 +1,61 @@ +package com.tongran.rocketmq.mapper; + +import java.util.List; +import com.tongran.rocketmq.domain.InitialMemoryDetailsInfo; + +/** + * 内存详细信息Mapper接口 + * + * @author gyt + * @date 2026-02-24 + */ +public interface InitialMemoryDetailsInfoMapper +{ + /** + * 查询内存详细信息 + * + * @param id 内存详细信息主键 + * @return 内存详细信息 + */ + public InitialMemoryDetailsInfo selectInitialMemoryDetailsInfoById(Long id); + + /** + * 查询内存详细信息列表 + * + * @param initialMemoryDetailsInfo 内存详细信息 + * @return 内存详细信息集合 + */ + public List selectInitialMemoryDetailsInfoList(InitialMemoryDetailsInfo initialMemoryDetailsInfo); + + /** + * 新增内存详细信息 + * + * @param initialMemoryDetailsInfo 内存详细信息 + * @return 结果 + */ + public int insertInitialMemoryDetailsInfo(InitialMemoryDetailsInfo initialMemoryDetailsInfo); + + /** + * 修改内存详细信息 + * + * @param initialMemoryDetailsInfo 内存详细信息 + * @return 结果 + */ + public int updateInitialMemoryDetailsInfo(InitialMemoryDetailsInfo initialMemoryDetailsInfo); + + /** + * 删除内存详细信息 + * + * @param id 内存详细信息主键 + * @return 结果 + */ + public int deleteInitialMemoryDetailsInfoById(Long id); + + /** + * 批量删除内存详细信息 + * + * @param ids 需要删除的数据主键集合 + * @return 结果 + */ + public int deleteInitialMemoryDetailsInfoByIds(Long[] ids); +} diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialMemoryDetailsInfoService.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialMemoryDetailsInfoService.java new file mode 100644 index 0000000..6a52fde --- /dev/null +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialMemoryDetailsInfoService.java @@ -0,0 +1,61 @@ +package com.tongran.rocketmq.service; + +import java.util.List; +import com.tongran.rocketmq.domain.InitialMemoryDetailsInfo; + +/** + * 内存详细信息Service接口 + * + * @author gyt + * @date 2026-02-24 + */ +public interface IInitialMemoryDetailsInfoService +{ + /** + * 查询内存详细信息 + * + * @param id 内存详细信息主键 + * @return 内存详细信息 + */ + public InitialMemoryDetailsInfo selectInitialMemoryDetailsInfoById(Long id); + + /** + * 查询内存详细信息列表 + * + * @param initialMemoryDetailsInfo 内存详细信息 + * @return 内存详细信息集合 + */ + public List selectInitialMemoryDetailsInfoList(InitialMemoryDetailsInfo initialMemoryDetailsInfo); + + /** + * 新增内存详细信息 + * + * @param initialMemoryDetailsInfo 内存详细信息 + * @return 结果 + */ + public int insertInitialMemoryDetailsInfo(InitialMemoryDetailsInfo initialMemoryDetailsInfo); + + /** + * 修改内存详细信息 + * + * @param initialMemoryDetailsInfo 内存详细信息 + * @return 结果 + */ + public int updateInitialMemoryDetailsInfo(InitialMemoryDetailsInfo initialMemoryDetailsInfo); + + /** + * 批量删除内存详细信息 + * + * @param ids 需要删除的内存详细信息主键集合 + * @return 结果 + */ + public int deleteInitialMemoryDetailsInfoByIds(Long[] ids); + + /** + * 删除内存详细信息信息 + * + * @param id 内存详细信息主键 + * @return 结果 + */ + public int deleteInitialMemoryDetailsInfoById(Long id); +} diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialMemoryDetailsInfoServiceImpl.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialMemoryDetailsInfoServiceImpl.java new file mode 100644 index 0000000..e8cb045 --- /dev/null +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialMemoryDetailsInfoServiceImpl.java @@ -0,0 +1,96 @@ +package com.tongran.rocketmq.service.impl; + +import com.tongran.common.core.utils.DateUtils; +import com.tongran.rocketmq.domain.InitialMemoryDetailsInfo; +import com.tongran.rocketmq.mapper.InitialMemoryDetailsInfoMapper; +import com.tongran.rocketmq.service.IInitialMemoryDetailsInfoService; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.stereotype.Service; + +import java.util.List; + +/** + * 内存详细信息Service业务层处理 + * + * @author gyt + * @date 2026-02-24 + */ +@Service +public class InitialMemoryDetailsInfoServiceImpl implements IInitialMemoryDetailsInfoService +{ + @Autowired + private InitialMemoryDetailsInfoMapper initialMemoryDetailsInfoMapper; + + /** + * 查询内存详细信息 + * + * @param id 内存详细信息主键 + * @return 内存详细信息 + */ + @Override + public InitialMemoryDetailsInfo selectInitialMemoryDetailsInfoById(Long id) + { + return initialMemoryDetailsInfoMapper.selectInitialMemoryDetailsInfoById(id); + } + + /** + * 查询内存详细信息列表 + * + * @param initialMemoryDetailsInfo 内存详细信息 + * @return 内存详细信息 + */ + @Override + public List selectInitialMemoryDetailsInfoList(InitialMemoryDetailsInfo initialMemoryDetailsInfo) + { + return initialMemoryDetailsInfoMapper.selectInitialMemoryDetailsInfoList(initialMemoryDetailsInfo); + } + + /** + * 新增内存详细信息 + * + * @param initialMemoryDetailsInfo 内存详细信息 + * @return 结果 + */ + @Override + public int insertInitialMemoryDetailsInfo(InitialMemoryDetailsInfo initialMemoryDetailsInfo) + { + return initialMemoryDetailsInfoMapper.insertInitialMemoryDetailsInfo(initialMemoryDetailsInfo); + } + + /** + * 修改内存详细信息 + * + * @param initialMemoryDetailsInfo 内存详细信息 + * @return 结果 + */ + @Override + public int updateInitialMemoryDetailsInfo(InitialMemoryDetailsInfo initialMemoryDetailsInfo) + { + initialMemoryDetailsInfo.setUpdateTime(DateUtils.getNowDate()); + return initialMemoryDetailsInfoMapper.updateInitialMemoryDetailsInfo(initialMemoryDetailsInfo); + } + + /** + * 批量删除内存详细信息 + * + * @param ids 需要删除的内存详细信息主键 + * @return 结果 + */ + @Override + public int deleteInitialMemoryDetailsInfoByIds(Long[] ids) + { + return initialMemoryDetailsInfoMapper.deleteInitialMemoryDetailsInfoByIds(ids); + } + + /** + * 删除内存详细信息信息 + * + * @param id 内存详细信息主键 + * @return 结果 + */ + @Override + public int deleteInitialMemoryDetailsInfoById(Long id) + { + return initialMemoryDetailsInfoMapper.deleteInitialMemoryDetailsInfoById(id); + } +} diff --git a/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialMemoryDetailsInfoMapper.xml b/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialMemoryDetailsInfoMapper.xml new file mode 100644 index 0000000..b7230f8 --- /dev/null +++ b/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialMemoryDetailsInfoMapper.xml @@ -0,0 +1,171 @@ + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + select id, client_id, total, name, location, manufacturer, capacity, frequency, serial_number, type, min_voltage, rank, bit_width, technology, part_number, health_status, create_by, create_time, update_by, update_time, remark from initial_memory_details_info + + + + + + + + insert into initial_memory_details_info + + client_id, + total, + name, + location, + manufacturer, + capacity, + frequency, + serial_number, + type, + min_voltage, + rank, + bit_width, + technology, + part_number, + + health_status, + create_by, + create_time, + update_by, + update_time, + remark, + + + #{clientId}, + #{total}, + #{name}, + #{location}, + #{manufacturer}, + #{capacity}, + #{frequency}, + #{serialNumber}, + #{type}, + #{minVoltage}, + #{rank}, + #{bitWidth}, + #{technology}, + #{partNumber}, + + #{healthStatus}, + #{createBy}, + #{createTime}, + #{updateBy}, + #{updateTime}, + #{remark}, + + on duplicate key update + + total = values(total), + name = values(name), + location = values(location), + manufacturer = values(manufacturer), + capacity = values(capacity), + frequency = values(frequency), + serial_number = values(serial_number), + `type` = values(`type`), + min_voltage = values(min_voltage), + rank = values(rank), + bit_width = values(bit_width), + technology = values(technology), + part_number = values(part_number), + + health_status = values(health_status), + update_time = now(), + remark = values(remark), + + + + + update initial_memory_details_info + + client_id = #{clientId}, + total = #{total}, + name = #{name}, + location = #{location}, + manufacturer = #{manufacturer}, + capacity = #{capacity}, + frequency = #{frequency}, + serial_number = #{serialNumber}, + type = #{type}, + min_voltage = #{minVoltage}, + rank = #{rank}, + bit_width = #{bitWidth}, + technology = #{technology}, + part_number = #{partNumber}, + + health_status = #{healthStatus}, + create_by = #{createBy}, + create_time = #{createTime}, + update_by = #{updateBy}, + update_time = #{updateTime}, + remark = #{remark}, + + where id = #{id} + + + + delete from initial_memory_details_info where id = #{id} + + + + delete from initial_memory_details_info where id in + + #{id} + + + \ No newline at end of file