From 00fffd0aef51cce32a3db652c169ea474ce52853 Mon Sep 17 00:00:00 2001 From: gaoyutao Date: Fri, 27 Feb 2026 17:50:43 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=86=85=E5=AD=98=E8=AF=A6?= =?UTF-8?q?=E7=BB=86=E9=80=BB=E8=BE=91=E3=80=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../consumer/TimeRangeReConsumer.java | 2 +- .../rocketmq/handler/MessageHandler.java | 57 +++++++++++++++++++ 2 files changed, 58 insertions(+), 1 deletion(-) 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 8748751..a2fe786 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 @@ -218,7 +218,7 @@ public class TimeRangeReConsumer implements DisposableBean { String dataType = message.getDataType(); String clientId = message.getClientId(); // 只处理"网络上报重试"类型的消息 - if (MsgEnum.网络上报重试.getValue().equals(dataType) && !"a7948be4439e40bf09acf48164b336a9".equals(clientId)) { + if (MsgEnum.网络上报重试.getValue().equals(dataType)) { receivedCount.incrementAndGet(); // 只缓存消息，不处理 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 aacd1a0..e76e1a1 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 @@ -190,12 +190,69 @@ public class MessageHandler { Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 List detailsVoList = memoryVo.getDetails(); if(detailsVoList != null && !detailsVoList.isEmpty()){ + // 关键：每个clientId有自己独立的key + String memoryCountKey = "memory:count:" + clientId; + + // 1. 给这个客户端的所有内存次数+1 + Map memoryCountMap = redisTemplate.opsForHash().entries(memoryCountKey); + for (Map.Entry entry : memoryCountMap.entrySet()) { + String memoryName = (String) entry.getKey(); + String countStr = (String) entry.getValue(); + + try { + int count = Integer.parseInt(countStr) + 1; + redisTemplate.opsForHash().put(memoryCountKey, memoryName, String.valueOf(count)); + } catch (NumberFormatException e) { + redisTemplate.opsForHash().put(memoryCountKey, memoryName, "1"); + } + } + // 2. 处理本次上报的内存 + Set reportedMemories = new HashSet<>(); for (InitialMemoryDetailsInfo detailsInfo : detailsVoList) { + String memoryName = detailsInfo.getName(); // 假设内存有name字段 + reportedMemories.add(memoryName); + + // 本次上报的内存，次数重置为0 + redisTemplate.opsForHash().put(memoryCountKey, memoryName, "0"); + detailsInfo.setTotal(memoryVo.getTotal()); detailsInfo.setClientId(clientId); detailsInfo.setCreateTime(createTime); + // 存在 + detailsInfo.setRemark("1"); initialMemoryDetailsInfoService.insertInitialMemoryDetailsInfo(detailsInfo); } + // 3. 检查次数≥3的内存 + memoryCountMap = redisTemplate.opsForHash().entries(memoryCountKey); + List memoriesToRemove = new ArrayList<>(); + + for (Map.Entry entry : memoryCountMap.entrySet()) { + String memoryName = (String) entry.getKey(); + String countStr = (String) entry.getValue(); + + try { + int count = Integer.parseInt(countStr); + // 如果次数≥3且本次没上报 + if (count >= 3 && !reportedMemories.contains(memoryName)) { + // 创建内存缺失记录（根据你的实际业务对象调整） + InitialMemoryDetailsInfo missingRecord = new InitialMemoryDetailsInfo(); + // 丢失 + missingRecord.setRemark("0"); + missingRecord.setClientId(clientId); + missingRecord.setName(memoryName); + initialMemoryDetailsInfoService.insertInitialMemoryDetailsInfo(missingRecord); + memoriesToRemove.add(memoryName); + + } + } catch (NumberFormatException e) { + memoriesToRemove.add(memoryName); + } + } + + // 4. 删除已处理的内存记录 + if (!memoriesToRemove.isEmpty()) { + redisTemplate.opsForHash().delete(memoryCountKey, memoriesToRemove.toArray()); + } } } }