From 180c522ca32087039fb5b74cbe19b62f4053253e Mon Sep 17 00:00:00 2001 From: gaoyutao Date: Tue, 3 Feb 2026 14:07:55 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=8E=86=E5=8F=B2=E6=B6=88?= =?UTF-8?q?=E6=81=AF=E6=B6=88=E8=B4=B9=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../system/RmResourceRegistrationMapper.xml | 2 +- .../consumer/TimeRangeReConsumer.java | 598 ++++++++++++++++++ .../controller/HistoryTrafficController.java | 29 + .../rocketmq/handler/MessageHandler.java | 23 +- .../handler/MessageHistoryDataHandler.java | 291 +++++++++ 5 files changed, 926 insertions(+), 17 deletions(-) create mode 100644 tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/TimeRangeReConsumer.java create mode 100644 tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/HistoryTrafficController.java create mode 100644 tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHistoryDataHandler.java diff --git a/tongran-modules/tongran-system/src/main/resources/mapper/system/RmResourceRegistrationMapper.xml b/tongran-modules/tongran-system/src/main/resources/mapper/system/RmResourceRegistrationMapper.xml index 73e2fd2..257218b 100644 --- a/tongran-modules/tongran-system/src/main/resources/mapper/system/RmResourceRegistrationMapper.xml +++ b/tongran-modules/tongran-system/src/main/resources/mapper/system/RmResourceRegistrationMapper.xml @@ -112,7 +112,7 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" - insert into rm_resource_registration + insert IGNORE into rm_resource_registration client_id, hardware_sn, 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 new file mode 100644 index 0000000..afd8ec0 --- /dev/null +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/TimeRangeReConsumer.java @@ -0,0 +1,598 @@ +package com.tongran.rocketmq.consumer; + +import com.alibaba.fastjson.JSON; +import com.tongran.common.core.enums.MsgEnum; +import com.tongran.rocketmq.domain.DeviceMessage; +import com.tongran.rocketmq.domain.InitialBandwidthTraffic; +import com.tongran.rocketmq.enums.MessageCodeEnum; +import com.tongran.rocketmq.handler.MessageHistoryDataHandler; +import com.tongran.rocketmq.utils.JsonDataParser; +import lombok.AllArgsConstructor; +import lombok.Data; +import lombok.NoArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext; +import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus; +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.annotation.Autowired; +import org.springframework.stereotype.Component; + +import java.nio.charset.StandardCharsets; +import java.text.SimpleDateFormat; +import java.time.Instant; +import java.util.*; +import java.util.concurrent.*; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +/** + * 时间范围重消费消费者（两阶段优化版） + * 第一阶段：并行消费并缓存所有消息 + * 第二阶段：按设备顺序处理缓存的消息 + */ +@Slf4j +@Component +public class TimeRangeReConsumer { + + + @Autowired(required = false) + private MessageHistoryDataHandler messageHandler; + + // 时间范围配置 + private static final String START_TIME = "2026-02-02 06:00:00"; + private static final String END_TIME = "2026-02-02 12:00:00"; + + // 统计信息 + private final AtomicLong receivedCount = new AtomicLong(0); // 已接收消息数 + private final AtomicLong skippedCount = 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); // 第二阶段处理标记 + + // 消息缓存：按clientId分组，按时间戳排序 + private final ConcurrentHashMap> messageCache = + new ConcurrentHashMap<>(); + + // 第二阶段顺序处理器线程池 + private final ExecutorService sequentialProcessor = Executors.newFixedThreadPool( + Math.min(Runtime.getRuntime().availableProcessors() * 4, 32) + ); + + // 第二阶段处理跟踪 + private final Map> processingFutures = new ConcurrentHashMap<>(); + private final AtomicInteger activeSequentialTasks = new AtomicInteger(0); + + @Data + @NoArgsConstructor + @AllArgsConstructor + public static class MessageWrapper implements Comparable { + private String msgId; + private DeviceMessage message; + private MessageExt messageExt; + private long timestamp; // 从data中解析的时间戳 + private long bornTimestamp; // 消息产生时间 + + @Override + public int compareTo(MessageWrapper other) { + return Long.compare(this.timestamp, other.timestamp); + } + } + + + // 添加一个标识字段 + private final AtomicBoolean isRunning = new AtomicBoolean(false); + + // 暴露启动方法 + public String startReconsumeTask(String startTime, String endTime, String namesrvAddr) { + if (isRunning.compareAndSet(false, true)) { + new Thread(() -> { + try { + doReconsume(startTime, endTime, namesrvAddr); + } catch (Exception e) { + log.error("重消费失败", e); + } finally { + isRunning.set(false); + } + }).start(); + return "重消费任务已启动"; + } + return "已有任务正在运行"; + } + public void startTimeRangeConsumer(String startTime, String endTime, String namesrvAddr) { + new Thread(() -> { + try { + doReconsume(startTime, endTime, namesrvAddr); + } catch (Exception e) { + log.error("重消费失败", e); + } + }).start(); + } + + private void doReconsume(String startTime, String endTime, String namesrvAddr) throws Exception { + SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + long startTimestamp = sdf.parse(startTime).getTime(); + long endTimestamp = sdf.parse(endTime).getTime(); + + String consumerGroup = "RECONSUME_GROUP_TEST" + Instant.now().getEpochSecond(); + DefaultMQPushConsumer consumer = new DefaultMQPushConsumer(consumerGroup); + consumer.setNamesrvAddr(namesrvAddr); + + // ================== 第一阶段：开始消息缓存 ================== + log.info("===== 第一阶段：开始消息缓存 ====="); + + consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_TIMESTAMP); + consumer.setConsumeTimestamp(sdf.format(new Date(startTimestamp))); + + consumer.subscribe(MessageCodeEnum.TR_AGENT_UP.getCode(), "*"); + + consumer.registerMessageListener(new MessageListenerConcurrently() { + @Override + public ConsumeConcurrentlyStatus consumeMessage( + List list, + ConsumeConcurrentlyContext context) { + + if (!shouldContinue.get()) { + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + } + + try { + for (MessageExt messageExt : list) { + long bornTimestamp = messageExt.getBornTimestamp(); + int queueId = messageExt.getQueueId(); + + // 记录当前队列的offset + queueCurrentOffsets.put(queueId, messageExt.getQueueOffset()); + + // 1. 时间太早，跳过 + if (bornTimestamp < startTimestamp) { + skippedCount.incrementAndGet(); + if (skippedCount.get() % 1000 != 0) { + continue; + } + } + + // 2. 超过结束时间，标记队列完成 + if (bornTimestamp > endTimestamp) { + log.info("队列 {} 已超过结束时间，标记为完成", queueId); + completedQueues.add(queueId); + + // 检查是否所有队列都完成 + if (completedQueues.size() >= 4) { // 假设有4个队列 + log.info("所有队列均已超过结束时间，第一阶段完成"); + shouldContinue.set(false); + } + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + } + + String body = new String(messageExt.getBody(), StandardCharsets.UTF_8); + DeviceMessage message = JSON.parseObject(body, DeviceMessage.class); + String dataType = message.getDataType(); + String clientId = message.getClientId(); + // 只处理"网络上报重试"类型的消息 + if(MsgEnum.网络上报重试.getValue().equals(dataType)){ + + receivedCount.incrementAndGet(); + + // 只缓存消息，不处理 + cacheMessageOnly(message, messageExt, bornTimestamp, sdf); + + // 定期打印进度 + if (receivedCount.get() % 100 == 0) { + printPhase1Progress(); + } + } + } + return ConsumeConcurrentlyStatus.CONSUME_SUCCESS; + } catch (Exception e) { + log.error("重消费消息异常", e); + return ConsumeConcurrentlyStatus.RECONSUME_LATER; + } + } + }); + + // 启动消费者 + consumer.start(); + log.info("时间范围重消费者启动成功，消费者组: {}", consumerGroup); + + // 预热：查找大致的起始位置 + findApproximateStartOffset(consumer, startTimestamp); + + // 监控第一阶段完成 + waitForPhase1Completion(consumer); + + // 第一阶段完成，关闭消费者 + consumer.shutdown(); + log.info("===== 第一阶段完成，共缓存{}条消息，涉及{}个设备 =====", + receivedCount.get(), messageCache.size()); + + // ================== 第二阶段：开始顺序处理 ================== + log.info("===== 第二阶段：开始顺序处理 ====="); + startPhase2Processing(); + + // 等待第二阶段完成 + waitForPhase2Completion(); + + log.info("✓ 时间范围重消费完成，总计接收消息: {}, 处理消息: {}", + receivedCount.get(), processedCount.get()); + } + + /** + * 第一阶段：只缓存消息，不处理 + */ + private void cacheMessageOnly(DeviceMessage message, MessageExt messageExt, + long bornTimestamp, SimpleDateFormat sdf) { + try { + // 解析data中的时间戳 + long timestamp = extractTimestampFromData(message.getData()); + String clientId = message.getClientId(); + + // 创建消息包装 + MessageWrapper wrapper = new MessageWrapper( + messageExt.getMsgId(), + message, + messageExt, + timestamp, + bornTimestamp + ); + + // 获取或创建该客户端的优先级队列 + PriorityBlockingQueue queue = messageCache.computeIfAbsent( + clientId, + id -> new PriorityBlockingQueue<>() + ); + + // 添加到队列（会自动按时间戳排序） + queue.offer(wrapper); + + if (log.isDebugEnabled() && receivedCount.get() % 20 == 0) { + log.debug("消息已缓存: clientId={}, timestamp={}, bornTime={}, 队列大小={}", + clientId, + timestamp, + sdf.format(new Date(bornTimestamp)), + queue.size()); + } + + } catch (Exception e) { + log.error("缓存消息失败: msgId={}", messageExt.getMsgId(), e); + } + } + + /** + * 从data字段解析时间戳 + */ + private long extractTimestampFromData(String data) { + try { + List interfaces = JsonDataParser.parseJsonData(data, InitialBandwidthTraffic.class); + if(!interfaces.isEmpty()) { + // 时间戳转换 + long timestamp = interfaces.get(0).getTimestamp(); + return timestamp; + } + } catch (Exception e) { + log.warn("解析data时间戳失败，使用默认值", e); + } + return System.currentTimeMillis(); // 解析失败使用当前时间 + } + + /** + * 等待第一阶段（消息缓存）完成 + */ + private void waitForPhase1Completion(DefaultMQPushConsumer consumer) { + log.info("等待第一阶段完成（消息缓存完成）..."); + + int checkCount = 0; + long lastMessageCount = 0; + int noProgressCount = 0; + + while (shouldContinue.get()) { + try { + Thread.sleep(5000); + checkCount++; + + long currentMessageCount = receivedCount.get(); + int totalCached = 0; + for (PriorityBlockingQueue queue : messageCache.values()) { + totalCached += queue.size(); + } + + // 每30秒打印一次进度 + if (checkCount % 6 == 0) { + log.info("第一阶段进度: 已接收={}, 已缓存={}, 设备数={}, 完成队列={}/4", + currentMessageCount, totalCached, messageCache.size(), completedQueues.size()); + } + + // 检查是否有进度 + if (currentMessageCount == lastMessageCount) { + noProgressCount++; + } else { + noProgressCount = 0; + lastMessageCount = currentMessageCount; + } + + // 如果30秒内没有新消息，认为第一阶段完成 + if (noProgressCount >= 6 && completedQueues.size() >= 4) { + log.info("30秒内无新消息且所有队列完成，第一阶段完成"); + shouldContinue.set(false); + break; + } + + // 超时保护：最多等待1小时 + if (checkCount >= 720) { + log.warn("第一阶段超时（1小时），强制结束"); + shouldContinue.set(false); + break; + } + + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + } + + /** + * 第二阶段：开始顺序处理所有缓存的消息 + */ + private void startPhase2Processing() { + isProcessingPhase.set(true); + + int totalMessages = messageCache.values().stream() + .mapToInt(Queue::size) + .sum(); + + log.info("开始顺序处理{}个设备的{}条消息...", + messageCache.size(), totalMessages); + + // 为每个设备启动顺序处理任务 + for (Map.Entry> entry : messageCache.entrySet()) { + String clientId = entry.getKey(); + PriorityBlockingQueue queue = entry.getValue(); + + if (!queue.isEmpty()) { + CompletableFuture future = CompletableFuture.runAsync(() -> { + activeSequentialTasks.incrementAndGet(); + try { + processDeviceMessagesSequentially(clientId, queue); + } finally { + activeSequentialTasks.decrementAndGet(); + } + }, sequentialProcessor); + + processingFutures.put(clientId, future); + } + } + + log.info("已启动{}个设备的顺序处理任务", processingFutures.size()); + } + + /** + * 按顺序处理单个设备的所有消息 + */ + private void processDeviceMessagesSequentially(String clientId, + PriorityBlockingQueue queue) { + String threadName = Thread.currentThread().getName(); + Thread.currentThread().setName("SeqProcessor-" + clientId.substring(0, Math.min(8, clientId.length()))); + + try { + int queueSize = queue.size(); + log.info("开始顺序处理设备{}的{}条消息", clientId, queueSize); + + int deviceProcessed = 0; + MessageWrapper lastMessage = null; + long startTime = System.currentTimeMillis(); + + while (!queue.isEmpty()) { + MessageWrapper wrapper = queue.poll(); + if (wrapper != null) { + try { + // 验证时间顺序（可选） + if (lastMessage != null && wrapper.getTimestamp() < lastMessage.getTimestamp()) { + log.warn("设备{}消息时间顺序异常: 前一条timestamp={}, 当前timestamp={}", + clientId, lastMessage.getTimestamp(), wrapper.getTimestamp()); + } + + // 处理消息 + if (messageHandler != null) { + messageHandler.handleMessage(wrapper.getMessage()); + } + + deviceProcessed++; + processedCount.incrementAndGet(); + lastMessage = wrapper; + + // 每处理20条日志一次 + if (deviceProcessed % 20 == 0) { + log.debug("设备{}已处理{}条消息，剩余{}条", + clientId, deviceProcessed, queue.size()); + } + + } catch (Exception e) { + log.error("处理设备{}消息失败: msgId={}", clientId, wrapper.getMsgId(), e); + // 根据业务决定是否重试或跳过 + // 这里选择跳过，记录错误 + } + } + } + + long timeCost = System.currentTimeMillis() - startTime; + log.info("设备{}处理完成: 共处理{}条消息，耗时{}ms", + clientId, deviceProcessed, timeCost); + + } finally { + Thread.currentThread().setName(threadName); + } + } + + /** + * 等待第二阶段处理完成 + */ + private void waitForPhase2Completion() { + log.info("等待所有设备顺序处理完成..."); + + CompletableFuture allFutures = CompletableFuture.allOf( + processingFutures.values().toArray(new CompletableFuture[0]) + ); + + try { + // 等待所有任务完成 + allFutures.get(2, TimeUnit.HOURS); // 最多等待2小时 + + log.info("✓ 所有设备顺序处理完成"); + + } catch (TimeoutException e) { + log.warn("第二阶段处理超时（2小时），强制结束"); + // 可以记录未完成的任务 + List unfinishedDevices = new ArrayList<>(); + for (Map.Entry> entry : processingFutures.entrySet()) { + if (!entry.getValue().isDone()) { + unfinishedDevices.add(entry.getKey()); + } + } + log.warn("以下设备未完成处理: {}", unfinishedDevices); + + } catch (Exception e) { + log.error("等待第二阶段处理异常", e); + } finally { + shutdownResources(); + } + + // 最后统计 + int remainingMessages = messageCache.values().stream() + .mapToInt(Queue::size) + .sum(); + + if (remainingMessages > 0) { + log.warn("仍有{}条消息未处理", remainingMessages); + } else { + log.info("所有消息处理完成，清理缓存"); + messageCache.clear(); + } + } + /** + * 处理完成后关闭资源 - 添加状态重置 + */ + private void shutdownResources() { + log.info("重消费完成，正在关闭资源..."); + + shouldContinue.set(false); + isProcessingPhase.set(false); + isRunning.set(false); + + // 清理缓存 + 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("✓ 重消费者资源已完全释放"); + } + /** + * 查找大致的起始offset + */ + private void findApproximateStartOffset(DefaultMQPushConsumer consumer, long startTimestamp) throws Exception { + SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss"); + log.info("查找起始offset，目标时间: {}", sdf.format(new Date(startTimestamp))); + + try { + // 获取主题队列 + Set queues = consumer.fetchSubscribeMessageQueues(MessageCodeEnum.TR_AGENT_UP.getCode()); + + if (queues == null || queues.isEmpty()) { + log.info("无队列信息，跳过offset查找"); + return; + } + + log.info("找到{}个队列，开始设置起始offset...", queues.size()); + + for (MessageQueue queue : queues) { + try { + // 搜索指定时间戳的offset + long offset = consumer.searchOffset(queue, startTimestamp); + + if (offset >= 0) { + // 获取队列的offset范围 + long minOffset = consumer.minOffset(queue); + long maxOffset = consumer.maxOffset(queue); + + // 确保offset在有效范围内 + if (offset < minOffset) offset = minOffset; + if (offset > maxOffset) offset = maxOffset; + + queueCurrentOffsets.put(queue.getQueueId(), offset); + 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); + } + } catch (Exception e) { + log.warn("队列{}设置offset失败: {}", queue.getQueueId(), e.getMessage()); + } + } + + log.info("起始offset设置完成"); + + } catch (Exception e) { + log.warn("查找起始offset失败，将使用默认时间戳方式: {}", e.getMessage()); + } + } + + /** + * 打印第一阶段进度 + */ + private void printPhase1Progress() { + int totalCached = 0; + for (PriorityBlockingQueue queue : messageCache.values()) { + totalCached += queue.size(); + } + + log.info("第一阶段进度: 已接收={}, 已缓存={}, 设备数={}, 完成队列={}/4", + receivedCount.get(), totalCached, messageCache.size(), completedQueues.size()); + } + + /** + * 打印第二阶段进度 + */ + private void printPhase2Progress() { + int totalRemaining = 0; + int activeTasks = 0; + + for (Map.Entry> entry : messageCache.entrySet()) { + totalRemaining += entry.getValue().size(); + CompletableFuture future = processingFutures.get(entry.getKey()); + if (future != null && !future.isDone()) { + activeTasks++; + } + } + + log.info("第二阶段进度: 已处理={}, 剩余={}, 活跃任务={}", + processedCount.get(), totalRemaining, activeTasks); + } +} \ No newline at end of file diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/HistoryTrafficController.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/HistoryTrafficController.java new file mode 100644 index 0000000..c8c6d2e --- /dev/null +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/HistoryTrafficController.java @@ -0,0 +1,29 @@ +package com.tongran.rocketmq.controller; + +import com.tongran.common.security.annotation.RequiresPermissions; +import com.tongran.rocketmq.consumer.TimeRangeReConsumer; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.web.bind.annotation.GetMapping; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +@RestController +@RequestMapping("/historyTraffic") +@RequiresPermissions("rocketmq:history") +public class HistoryTrafficController { + + @Autowired + private TimeRangeReConsumer timeRangeReConsumer; + /** + * 处理历史流量数据 + * @param timeRange 时间范围,分割 2026-02-03 00:00:00,2026-02-03 12:00:00,172.16.15.52:9876 + */ + @GetMapping("/processHistoryData") + public void processHistoryData(String timeRange){ + String[] timeRangeArr = timeRange.split(","); + String startTime = timeRangeArr[0]; + String endTime = timeRangeArr[1]; + String namesrvAddr = timeRangeArr[2]; + timeRangeReConsumer.startTimeRangeConsumer(startTime, endTime, namesrvAddr); + } +} 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 8f7cd8e..5fc5a7c 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 @@ -115,8 +115,6 @@ public class MessageHandler { @Autowired private IRmAlarmLogService rmAlarmLogService; @Autowired - private IRmAlarmPushConfigService rmAlarmPushConfigService; - @Autowired private IRmFrpcConfigManageService rmFrpcConfigManageService; @Autowired private IRmPppoeConfigSubService rmPppoeConfigSubService; @@ -740,9 +738,9 @@ public class MessageHandler { // 尝试获取锁 locked = lock.tryLock(0, 20, TimeUnit.SECONDS); if (locked) { - log.info("设备{}获取锁成功，开始处理消息", clientId); +// log.info("设备{}获取锁成功，开始处理消息", clientId); processNetRecoverMessageInternal(message); - log.info("设备{}消息处理完成", clientId); +// log.info("设备{}消息处理完成", clientId); } else { log.warn("设备{}获取锁失败，消息处理繁忙", clientId); throw new RuntimeException("设备处理繁忙，请重试"); @@ -755,7 +753,7 @@ public class MessageHandler { if (locked && lock.isHeldByCurrentThread()) { try { lock.unlock(); - log.debug("设备{}锁已释放", clientId); +// log.debug("设备{}锁已释放", clientId); } catch (IllegalMonitorStateException e) { log.warn("设备{}锁释放异常，可能已自动超时", clientId); } @@ -775,10 +773,6 @@ public class MessageHandler { boolean lasttrafficFlag = interfaces.get(0).isLastTrafficFlag(); // 时间戳存储到redis storeTimestampToRedis(clientId, timestamp); - if(lasttrafficFlag){ - // 把redis中存储的时间戳提取出来，删除redis中的时间戳 - processExitsTraffic(clientId); - } long millis = timestamp * 1000; Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss",createTime); @@ -790,13 +784,6 @@ public class MessageHandler { countQuery.setStartTime(timeStr); int exitsCount = initialBandwidthTrafficService.countByClientIdAndTime(countQuery); - // 即使数据已存在，也需要处理临时表逻辑，确保后续计算正确 - if(exitsCount > 0){ - // 不直接返回，继续执行临时表处理逻辑 - // 但跳过最终的数据入库操作 - System.out.println("数据已存在，跳过入库操作，但继续处理临时表逻辑"); - } - // 创建比timestamp少5分钟的时间 long fiveMinutesEarlier = millis - (5 * 60 * 1000); // 减去5分钟的毫秒数 Date fiveMinutesEarlierDate = new Date(fiveMinutesEarlier / 1000 * 1000); // 同样去除毫秒 @@ -943,6 +930,10 @@ public class MessageHandler { // 数据已存在时，只更新临时表，确保后续计算正确 initialBandwidthTrafficTempService.batchInsertServerRecoverTemp(interfaces); } + if(lasttrafficFlag){ + // 把redis中存储的时间戳提取出来，删除redis中的时间戳 + processExitsTraffic(clientId); + } } else { throw new RuntimeException("NET流量data数据为空"); } diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHistoryDataHandler.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHistoryDataHandler.java new file mode 100644 index 0000000..73e89d8 --- /dev/null +++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHistoryDataHandler.java @@ -0,0 +1,291 @@ +package com.tongran.rocketmq.handler; + +import com.tongran.common.core.constant.SecurityConstants; +import com.tongran.common.core.enums.MsgEnum; +import com.tongran.common.core.utils.DateUtils; +import com.tongran.rocketmq.domain.DeviceMessage; +import com.tongran.rocketmq.domain.InitialBandwidthTraffic; +import com.tongran.rocketmq.domain.InitialBandwidthTrafficTemp; +import com.tongran.rocketmq.service.IInitialBandwidthTrafficService; +import com.tongran.rocketmq.service.IInitialBandwidthTrafficTempService; +import com.tongran.rocketmq.utils.DataProcessUtil; +import com.tongran.rocketmq.utils.JsonDataParser; +import com.tongran.system.api.RemoteRevenueConfigService; +import com.tongran.system.api.domain.EpsInitialTrafficDataRemote; +import lombok.extern.slf4j.Slf4j; +import org.redisson.api.RLock; +import org.redisson.api.RedissonClient; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.scheduling.annotation.EnableScheduling; +import org.springframework.stereotype.Component; + +import javax.annotation.PostConstruct; +import java.math.BigDecimal; +import java.math.RoundingMode; +import java.util.Date; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.concurrent.TimeUnit; +import java.util.function.Consumer; +import java.util.function.Function; +import java.util.stream.Collectors; + +/** + * 设备消息处理器 + */ +@Slf4j +@Component +@EnableScheduling +public class MessageHistoryDataHandler { + + private final Map> messageHandlers = new HashMap<>(); + @Autowired + private IInitialBandwidthTrafficService initialBandwidthTrafficService; + @Autowired + private RemoteRevenueConfigService remoteRevenueConfigService; + @Autowired + private DataProcessUtil dataProcessUtil; + @Autowired + private RedissonClient redissonClient; + @Autowired + private IInitialBandwidthTrafficTempService initialBandwidthTrafficTempService; + /** + * 注册消息处理器 + */ + private void registerHandler(String dataType, Consumer handler) { + messageHandlers.put(dataType, handler); + } + + /** + * 处理设备消息（对外暴露的主方法） + */ + public void handleMessage(DeviceMessage message) { + String dataType = message.getDataType(); + Consumer handler = messageHandlers.get(dataType); + + if (handler != null) { + handler.accept(message); + } else { + log.warn("未知数据类型：{}", dataType); + } + } + /** + * 初始化处理器映射 + */ + @PostConstruct + public void init() { + registerHandler(MsgEnum.网络上报重试.getValue(), this::handleNetRecoverMessage); + } + private void handleNetRecoverMessage(DeviceMessage message) { + String clientId = message.getClientId(); + String lockKey = "traffic:recover:" + clientId; + RLock lock = redissonClient.getLock(lockKey); + boolean locked = false; + + try { + // 尝试获取锁 + locked = lock.tryLock(0, 20, TimeUnit.SECONDS); + if (locked) { + processNetRecoverMessageInternal(message); + } else { + log.warn("设备{}获取锁失败，消息处理繁忙", clientId); + throw new RuntimeException("设备处理繁忙，请重试"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + log.error("处理消息时线程被中断", e); + } finally { + // 只有成功获取锁的线程才需要释放 + if (locked && lock.isHeldByCurrentThread()) { + try { + lock.unlock(); + } catch (IllegalMonitorStateException e) { + log.warn("设备{}锁释放异常，可能已自动超时", clientId); + } + } + } + } + /** + * 网络重试流量数据入库 + * @param message + */ + private void processNetRecoverMessageInternal(DeviceMessage message) { + List interfaces = JsonDataParser.parseJsonData(message.getData(), InitialBandwidthTraffic.class); + if(!interfaces.isEmpty()){ + String clientId = message.getClientId(); + // 时间戳转换 + long timestamp = interfaces.get(0).getTimestamp(); + long millis = timestamp * 1000; + Date createTime = new Date(millis / 1000 * 1000); // 去除毫秒 + String timeStr = DateUtils.parseDateToStr("yyyy-MM-dd HH:mm:ss",createTime); + + // 判断数据库中是否已存在数据 + InitialBandwidthTraffic countQuery = new InitialBandwidthTraffic(); + countQuery.setClientId(clientId); + countQuery.setCreateTime(createTime); + countQuery.setStartTime(timeStr); + int exitsCount = initialBandwidthTrafficService.countByClientIdAndTime(countQuery); + + // 创建比timestamp少5分钟的时间 + long fiveMinutesEarlier = millis - (5 * 60 * 1000); // 减去5分钟的毫秒数 + Date fiveMinutesEarlierDate = new Date(fiveMinutesEarlier / 1000 * 1000); // 同样去除毫秒 + + // 查询临时表信息，计算实际流量值 + InitialBandwidthTrafficTemp temp = new InitialBandwidthTrafficTemp(); + temp.setCreateTime(fiveMinutesEarlierDate); + temp.setClientId(clientId); + List tempList = initialBandwidthTrafficTempService.selectInitialBandwidthTrafficRecoverTempList(temp); + + if(!tempList.isEmpty()){ + // 1. 构建快速查找的Map，使用MAC地址+网卡名称作为唯一键 + Map tempMap = tempList.stream() + .collect(Collectors.toMap( + tempItem -> generateKey(tempItem.getMac(), tempItem.getName()), + Function.identity(), + (existing, replacement) -> existing + )); + + // 2. 预计算除数（避免重复创建对象） + BigDecimal divisor = new BigDecimal(300); + + interfaces.forEach(iface -> { + iface.setClientId(clientId); + iface.setCreateTime(createTime); + + // 设置总流量（转换为比特） + iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed())); + iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed())); + iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed())); + iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed())); + iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed())); + iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed())); + // 首次采集，速率设为null + iface.setInSpeed(null); + iface.setOutSpeed(null); + iface.setIpv4InSpeed(null); + iface.setIpv4OutSpeed(null); + iface.setIpv6InSpeed(null); + iface.setIpv6OutSpeed(null); + + // 使用MAC地址+网卡名称作为查找键 + String key = generateKey(iface.getMac(), iface.getName()); + InitialBandwidthTrafficTemp tempInfo = tempMap.get(key); + if (tempInfo != null) { + // 计算总流入速率 + if (iface.getTotalInSpeed() != null && tempInfo.getTotalInSpeed() != null) { + BigDecimal nowInSpeed = new BigDecimal(iface.getTotalInSpeed()); + BigDecimal tempInSpeed = new BigDecimal(tempInfo.getTotalInSpeed()); + BigDecimal inDiff = nowInSpeed.subtract(tempInSpeed); + if (inDiff.compareTo(BigDecimal.ZERO) >= 0) { + iface.setInSpeed(inDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); + } + } + // 计算总流出速率 + if (iface.getTotalOutSpeed() != null && tempInfo.getTotalOutSpeed() != null) { + BigDecimal nowOutSpeed = new BigDecimal(iface.getTotalOutSpeed()); + BigDecimal tempOutSpeed = new BigDecimal(tempInfo.getTotalOutSpeed()); + BigDecimal outDiff = nowOutSpeed.subtract(tempOutSpeed); + if (outDiff.compareTo(BigDecimal.ZERO) >= 0) { + iface.setOutSpeed(outDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); + } + } + // 计算IPv4流入速率 + if (iface.getTotalIpv4InSpeed() != null && tempInfo.getTotalIpv4InSpeed() != null) { + BigDecimal nowIpv4In = new BigDecimal(iface.getTotalIpv4InSpeed()); + BigDecimal tempIpv4In = new BigDecimal(tempInfo.getTotalIpv4InSpeed()); + BigDecimal ipv4InDiff = nowIpv4In.subtract(tempIpv4In); + if (ipv4InDiff.compareTo(BigDecimal.ZERO) >= 0) { + iface.setIpv4InSpeed(ipv4InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); + } + } + // 计算IPv4流出速率 + if (iface.getTotalIpv4OutSpeed() != null && tempInfo.getTotalIpv4OutSpeed() != null) { + BigDecimal nowIpv4Out = new BigDecimal(iface.getTotalIpv4OutSpeed()); + BigDecimal tempIpv4Out = new BigDecimal(tempInfo.getTotalIpv4OutSpeed()); + BigDecimal ipv4OutDiff = nowIpv4Out.subtract(tempIpv4Out); + if (ipv4OutDiff.compareTo(BigDecimal.ZERO) >= 0) { + iface.setIpv4OutSpeed(ipv4OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); + } + } + // 计算IPv6流入速率 + if (iface.getTotalIpv6InSpeed() != null && tempInfo.getTotalIpv6InSpeed() != null) { + BigDecimal nowIpv6In = new BigDecimal(iface.getTotalIpv6InSpeed()); + BigDecimal tempIpv6In = new BigDecimal(tempInfo.getTotalIpv6InSpeed()); + BigDecimal ipv6InDiff = nowIpv6In.subtract(tempIpv6In); + if (ipv6InDiff.compareTo(BigDecimal.ZERO) >= 0) { + iface.setIpv6InSpeed(ipv6InDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); + } + } + // 计算IPv6流出速率 + if (iface.getTotalIpv6OutSpeed() != null && tempInfo.getTotalIpv6OutSpeed() != null) { + BigDecimal nowIpv6Out = new BigDecimal(iface.getTotalIpv6OutSpeed()); + BigDecimal tempIpv6Out = new BigDecimal(tempInfo.getTotalIpv6OutSpeed()); + BigDecimal ipv6OutDiff = nowIpv6Out.subtract(tempIpv6Out); + if (ipv6OutDiff.compareTo(BigDecimal.ZERO) >= 0) { + iface.setIpv6OutSpeed(ipv6OutDiff.divide(divisor, 0, RoundingMode.HALF_UP).toString()); + } + } + } + }); + // 清空临时表对应server信息 + InitialBandwidthTrafficTemp delQuery = new InitialBandwidthTrafficTemp(); + delQuery.setClientId(clientId); + delQuery.setCreateTime(fiveMinutesEarlierDate); + initialBandwidthTrafficTempService.deleteTempMsgByClientIdAndTime(delQuery); + } else { + interfaces.forEach(iface -> { + iface.setClientId(clientId); + iface.setCreateTime(createTime); + // 设置总流量（转换为比特） + iface.setTotalOutSpeed(dataProcessUtil.bytesToBits(iface.getOutSpeed())); + iface.setTotalInSpeed(dataProcessUtil.bytesToBits(iface.getInSpeed())); + iface.setTotalIpv4OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv4OutSpeed())); + iface.setTotalIpv4InSpeed(dataProcessUtil.bytesToBits(iface.getIpv4InSpeed())); + iface.setTotalIpv6OutSpeed(dataProcessUtil.bytesToBits(iface.getIpv6OutSpeed())); + iface.setTotalIpv6InSpeed(dataProcessUtil.bytesToBits(iface.getIpv6InSpeed())); + + // 首次采集，速率设为null + iface.setInSpeed(null); + iface.setOutSpeed(null); + iface.setIpv4InSpeed(null); + iface.setIpv4OutSpeed(null); + iface.setIpv6InSpeed(null); + iface.setIpv6OutSpeed(null); + }); + } + + // 只有在数据不存在时才执行入库操作 + if (exitsCount == 0) { + InitialBandwidthTraffic data = new InitialBandwidthTraffic(); + // 批量入库集合 + data.setList(interfaces); + // 临时表 用来计算流量速率 + initialBandwidthTrafficTempService.batchInsertServerRecoverTemp(interfaces); + // 初始流量数据入库 + initialBandwidthTrafficService.batchInsertRecoverTraffic(data); + EpsInitialTrafficDataRemote epsInitialTrafficDataRemote = new EpsInitialTrafficDataRemote(); + epsInitialTrafficDataRemote.setStartTime(timeStr); + epsInitialTrafficDataRemote.setEndTime(timeStr); + // 复制到业务初始库 + remoteRevenueConfigService.autoSaveServiceRecoverTrafficData(epsInitialTrafficDataRemote, SecurityConstants.INNER); + } else { + // 数据已存在时，只更新临时表，确保后续计算正确 + initialBandwidthTrafficTempService.batchInsertServerRecoverTemp(interfaces); + } + } else { + throw new RuntimeException("NET流量data数据为空"); + } + } + + private String generateKey(String mac, String name) { + if (mac == null) { + mac = ""; + } + if (name == null) { + name = ""; + } + return mac + "|" + name; + } + +} \ No newline at end of file