优化agent本地存储流量数据处理

2026-01-29 12:00:05 +08:00
parent 29a30b1f52
commit 1dab9b1cf4
15 changed files with 465 additions and 89 deletions
@@ -721,23 +721,31 @@ public class MessageHandler {
            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);
+
+            // 即使数据已存在，也需要处理临时表逻辑，确保后续计算正确
            if(exitsCount > 0){
-                return;
+                // 不直接返回，继续执行临时表处理逻辑
+                // 但跳过最终的数据入库操作
+                System.out.println("数据已存在，跳过入库操作，但继续处理临时表逻辑");
            }
+
            // 创建比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<InitialBandwidthTrafficTemp> tempList = initialBandwidthTrafficTempService.selectInitialBandwidthTrafficRecoverTempList(temp);
+
            if(!tempList.isEmpty()){
                // 1. 构建快速查找的Map，使用MAC地址+网卡名称作为唯一键
                Map<String, InitialBandwidthTrafficTemp> tempMap = tempList.stream()
@@ -834,7 +842,7 @@ public class MessageHandler {
                delQuery.setClientId(clientId);
                delQuery.setCreateTime(fiveMinutesEarlierDate);
                initialBandwidthTrafficTempService.deleteTempMsgByClientIdAndTime(delQuery);
-            }else{
+            } else {
                interfaces.forEach(iface -> {
                    iface.setClientId(clientId);
                    iface.setCreateTime(createTime);
@@ -855,19 +863,26 @@ public class MessageHandler {
                    iface.setIpv6OutSpeed(null);
                });
            }
-            InitialBandwidthTraffic data = new InitialBandwidthTraffic();
-            // 批量入库集合
-            data.setList(interfaces);
-            // 临时表 用来计算流量速率
-            initialBandwidthTrafficTempService.batchInsertServerRecoverTemp(interfaces);
-            // 初始流量数据入库
-            initialBandwidthTrafficService.batchInsert(data);
-            EpsInitialTrafficDataRemote epsInitialTrafficDataRemote = new EpsInitialTrafficDataRemote();
-            epsInitialTrafficDataRemote.setStartTime(timeStr);
-            epsInitialTrafficDataRemote.setEndTime(timeStr);
-            // 复制到业务初始库
-            remoteRevenueConfigService.autoSaveServiceTrafficData(epsInitialTrafficDataRemote, SecurityConstants.INNER);
-        }else{
+
+            // 只有在数据不存在时才执行入库操作
+            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数据为空");
        }
    }
@@ -971,77 +986,77 @@ public class MessageHandler {
        long timestamp = disks.get(0).getTimestamp();
        long millis = timestamp * 1000;
        Date createTime = new Date(millis / 1000 * 1000);
-        disks.forEach(disk -> {
-            disk.setClientId(clientId);
-            disk.setCreateTime(createTime);
-        });
-        // 关键：每个clientId有自己独立的key
-//        String diskCountKey = DISK_COUNT_PREFIX + clientId;
-//
-//        // 1. 给这个客户端的所有磁盘次数+1
-//        Map<Object, Object> diskCountMap = redisTemplate.opsForHash().entries(diskCountKey);
-//        for (Map.Entry<Object, Object> entry : diskCountMap.entrySet()) {
-//            String diskName = (String) entry.getKey();
-//            String countStr = (String) entry.getValue();
-//
-//            try {
-//                int count = Integer.parseInt(countStr) + 1;
-//                redisTemplate.opsForHash().put(diskCountKey, diskName, String.valueOf(count));
-//            } catch (NumberFormatException e) {
-//                redisTemplate.opsForHash().put(diskCountKey, diskName, "1");
-//            }
-//        }
-//
-//        // 2. 处理本次上报的磁盘
-//        Set<String> reportedDisks = new HashSet<>();
 //        disks.forEach(disk -> {
-//            String diskName = disk.getName();
-//            reportedDisks.add(diskName);
-//
-//            // 本次上报的磁盘，次数重置为0
-//            redisTemplate.opsForHash().put(diskCountKey, diskName, "0");
-//
 //            disk.setClientId(clientId);
 //            disk.setCreateTime(createTime);
 //        });
-//
-//        // 3. 检查次数≥3的磁盘
-//        diskCountMap = redisTemplate.opsForHash().entries(diskCountKey);
-//        List<String> disksToRemove = new ArrayList<>();
-//
-//        for (Map.Entry<Object, Object> entry : diskCountMap.entrySet()) {
-//            String diskName = (String) entry.getKey();
-//            String countStr = (String) entry.getValue();
-//
-//            try {
-//                int count = Integer.parseInt(countStr);
-//                // 如果次数≥3且本次没上报
-//                if (count >= 3 && !reportedDisks.contains(diskName)) {
-//                    AllDiskName allDiskName = new AllDiskName();
-//                    allDiskName.setStatus(0);
-//                    allDiskName.setClientId(clientId);
-//                    allDiskName.setName(diskName);
-//                    allDiskNameService.updateAllDiskName(allDiskName);
-//                    disksToRemove.add(diskName);
-//                    // 磁盘缺失，触发告警
-//                    RmAlarmLog rmAlarmLog = new RmAlarmLog();
-//                    rmAlarmLog.setClientId(clientId);
-//                    rmAlarmLog.setAlarmTime(DateUtils.getNowDate());
-//                    rmAlarmLog.setAlarmType(AlarmTypeEnum.磁盘缺失.getCode());
-//                    rmAlarmLog.setAlarmContent("服务器" + clientId + "磁盘缺失，磁盘名称:" + diskName);
-//                    rmAlarmLogService.insertRmAlarmLog(rmAlarmLog);
-//                    sendAlarmPushUtil.sendAlarmPush(rmAlarmLog, AlarmTypeEnum.磁盘缺失.getMsg());
-//
-//                }
-//            } catch (NumberFormatException e) {
-//                disksToRemove.add(diskName);
-//            }
-//        }
-//
-//        // 4. 删除已处理的磁盘记录
-//        if (!disksToRemove.isEmpty()) {
-//            redisTemplate.opsForHash().delete(diskCountKey, disksToRemove.toArray());
-//        }
+        // 关键：每个clientId有自己独立的key
+        String diskCountKey = DISK_COUNT_PREFIX + clientId;
+
+        // 1. 给这个客户端的所有磁盘次数+1
+        Map<Object, Object> diskCountMap = redisTemplate.opsForHash().entries(diskCountKey);
+        for (Map.Entry<Object, Object> entry : diskCountMap.entrySet()) {
+            String diskName = (String) entry.getKey();
+            String countStr = (String) entry.getValue();
+
+            try {
+                int count = Integer.parseInt(countStr) + 1;
+                redisTemplate.opsForHash().put(diskCountKey, diskName, String.valueOf(count));
+            } catch (NumberFormatException e) {
+                redisTemplate.opsForHash().put(diskCountKey, diskName, "1");
+            }
+        }
+
+        // 2. 处理本次上报的磁盘
+        Set<String> reportedDisks = new HashSet<>();
+        disks.forEach(disk -> {
+            String diskName = disk.getName();
+            reportedDisks.add(diskName);
+
+            // 本次上报的磁盘，次数重置为0
+            redisTemplate.opsForHash().put(diskCountKey, diskName, "0");
+
+            disk.setClientId(clientId);
+            disk.setCreateTime(createTime);
+        });
+
+        // 3. 检查次数≥3的磁盘
+        diskCountMap = redisTemplate.opsForHash().entries(diskCountKey);
+        List<String> disksToRemove = new ArrayList<>();
+
+        for (Map.Entry<Object, Object> entry : diskCountMap.entrySet()) {
+            String diskName = (String) entry.getKey();
+            String countStr = (String) entry.getValue();
+
+            try {
+                int count = Integer.parseInt(countStr);
+                // 如果次数≥3且本次没上报
+                if (count >= 3 && !reportedDisks.contains(diskName)) {
+                    AllDiskName allDiskName = new AllDiskName();
+                    allDiskName.setStatus(0);
+                    allDiskName.setClientId(clientId);
+                    allDiskName.setName(diskName);
+                    allDiskNameService.updateAllDiskName(allDiskName);
+                    disksToRemove.add(diskName);
+                    // 磁盘缺失，触发告警
+                    RmAlarmLog rmAlarmLog = new RmAlarmLog();
+                    rmAlarmLog.setClientId(clientId);
+                    rmAlarmLog.setAlarmTime(DateUtils.getNowDate());
+                    rmAlarmLog.setAlarmType(AlarmTypeEnum.磁盘缺失.getCode());
+                    rmAlarmLog.setAlarmContent("服务器" + clientId + "磁盘缺失，磁盘名称:" + diskName);
+                    rmAlarmLogService.insertRmAlarmLog(rmAlarmLog);
+                    sendAlarmPushUtil.sendAlarmPush(rmAlarmLog, AlarmTypeEnum.磁盘缺失.getMsg());
+
+                }
+            } catch (NumberFormatException e) {
+                disksToRemove.add(diskName);
+            }
+        }
+
+        // 4. 删除已处理的磁盘记录
+        if (!disksToRemove.isEmpty()) {
+            redisTemplate.opsForHash().delete(diskCountKey, disksToRemove.toArray());
+        }

        // 5. 数据入库
        initialDiskInfoService.batchInsertInitialDiskInfo(disks, createTime);
@@ -1961,7 +1976,7 @@ public class MessageHandler {
     * 查询IP地址归属地信息，返回运营商和省份
     */
    private Map<String, String> queryIpLocation(String ip) {
-        String apiUrl = "http://172.16.15.51:10000/?ip=" + ip;
+        String apiUrl = "http://172.16.15.103:10000/?ip=" + ip;
        CloseableHttpClient httpClient = HttpClients.createDefault();
        HttpGet httpGet = new HttpGet(apiUrl);

@@ -71,6 +71,7 @@ public interface InitialBandwidthTrafficMapper
     * @param data 流量数据实体类
     */
    int batchInsert(InitialBandwidthTraffic data);
+    int batchInsertRecoverTraffic(InitialBandwidthTraffic data);
    /**
     * 网络接口基础信息
     * @param initialBandwidthTraffic
@@ -71,6 +71,7 @@ public interface IInitialBandwidthTrafficService
     * @param data 流量数据
     */
    void batchInsert(InitialBandwidthTraffic data);
+    void batchInsertRecoverTraffic(InitialBandwidthTraffic data);
    /**
     * 网络接口基础信息
     * @param initialBandwidthTraffic
@@ -183,6 +183,56 @@ public class InitialBandwidthTrafficServiceImpl implements IInitialBandwidthTraf
            }
        });
    }
+    /**
+     * 保存多条数据到对应分表
+     * @param initialBandwidthTraffic 流量数据
+     */
+    @Override
+    @Transactional(rollbackFor = Exception.class, isolation = Isolation.READ_COMMITTED)
+    public void batchInsertRecoverTraffic(InitialBandwidthTraffic initialBandwidthTraffic) {
+        if (initialBandwidthTraffic == null) {
+            return;
+        }
+        List<InitialBandwidthTraffic> dataList = initialBandwidthTraffic.getList();
+        if (dataList.isEmpty()){
+            return;
+        }
+        // 按表名分组批量插入
+        Map<String, List<InitialBandwidthTraffic>> groupedData = dataList.stream()
+                .map(data -> {
+                    try {
+                        InitialBandwidthTraffic processed = new InitialBandwidthTraffic();
+                        BeanUtils.copyProperties(data,processed);
+                        if (data.getCreateTime() == null) {
+                            data.setCreateTime(DateUtils.getNowDate());
+                        }
+                        LocalDateTime createTime = data.getCreateTime().toInstant()
+                                .atZone(ZoneId.systemDefault())
+                                .toLocalDateTime();
+                        processed.setTableName(TableRouterUtil.getTableName(createTime));
+                        return processed;
+                    } catch (Exception e){
+                        log.error("数据处理失败",e.getMessage());
+                        return null;
+                    }
+                }).collect(Collectors.groupingBy(
+                        InitialBandwidthTraffic::getTableName,
+                        LinkedHashMap::new, // 保持插入顺序
+                        Collectors.toList()));
+
+        groupedData.forEach((tableName, list) -> {
+            try {
+                InitialBandwidthTraffic data = new InitialBandwidthTraffic();
+                BeanUtils.copyProperties(initialBandwidthTraffic,data);
+                data.setTableName(tableName);
+                data.setList(list);
+                initialBandwidthTrafficMapper.batchInsertRecoverTraffic(data);
+            } catch (Exception e) {
+                log.error("表{}插入失败", tableName, e);
+                throw new RuntimeException("批量插入失败", e);
+            }
+        });
+    }
    /**
     * 网络接口基础信息
     * @param initialBandwidthTraffic