liuyupeng/saas-houduan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

saas平台联调agent存储业务流量。

Browse Source 
优化ip存储，未探测到，保留上次结果。
This commit is contained in:
gaoyutao
2026-03-10 17:50:48 +08:00
parent d4902a488d
commit cffcb2efd3
11 changed files with 318 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java
+158 -10
View File
@@ -128,6 +128,10 @@ public class MessageHandler {
@Autowired
private IInitialBandwidthTrafficTempService initialBandwidthTrafficTempService;
@Autowired
private IInitialNetBusinessTrafficTempService initialNetBusinessTrafficTempService;
@Autowired
private IInitialNetBusinessTrafficService iInitialNetBusinessTrafficService;
@Autowired
private IRmAlarmLogService rmAlarmLogService;
@Autowired
private IRmFrpcConfigManageService rmFrpcConfigManageService;
@@ -168,6 +172,7 @@ public class MessageHandler {
registerHandler(MsgEnum.内存上报.getValue(), this::handleMemoryMessage);
registerHandler(MsgEnum.网络上报.getValue(), this::handleNetMessage);
registerHandler(MsgEnum.网络上报重试.getValue(), this::handleNetRecoverMessage);
registerHandler(MsgEnum.业务网络上报.getValue(), this::handleBusinessNetMessage);
registerHandler(MsgEnum.挂载上报.getValue(), this::handleMountPointMessage);
registerHandler(MsgEnum.系统其他上报.getValue(), this::handleOtherSystemMessage);
registerHandler(MsgEnum.心跳上报.getValue(), this::handleHeartbeatMessage);
@@ -179,6 +184,149 @@ public class MessageHandler {
registerHandler(MsgEnum.内存详情上报.getValue(), this::handleMemoryDetailsMessage);
}
private void handleBusinessNetMessage(DeviceMessage message) {
List<InitialNetBusinessTraffic> interfaces = JsonDataParser.parseJsonData(message.getData(), InitialNetBusinessTraffic.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);
// 创建比timestamp少5分钟的时间
long fiveMinutesEarlier = millis - (5 * 60 * 1000); // 减去5分钟的毫秒数
Date fiveMinutesEarlierDate = new Date(fiveMinutesEarlier / 1000 * 1000); // 同样去除毫秒
// 查询临时表信息，计算实际流量值
InitialNetBusinessTrafficTemp temp = new InitialNetBusinessTrafficTemp();
temp.setCreateTime(fiveMinutesEarlierDate);
temp.setClientId(clientId);
List<InitialNetBusinessTrafficTemp> tempList = initialNetBusinessTrafficTempService.selectInitialNetBusinessTrafficTempList(temp);
if(!tempList.isEmpty()){
// 1. 构建快速查找的Map，使用MAC地址+网卡名称作为唯一键
Map<String, InitialNetBusinessTrafficTemp> 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());
InitialNetBusinessTrafficTemp 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信息
initialNetBusinessTrafficTempService.deleteBusinessTempMsgByClientId(clientId);
}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);
});
}
InitialNetBusinessTraffic data = new InitialNetBusinessTraffic();
// 批量入库集合
data.setList(interfaces);
// 临时表 用来计算流量速率
initialNetBusinessTrafficTempService.batchInsertBusinessTemp(interfaces);
// 初始流量数据入库
iInitialNetBusinessTrafficService.batchInsertBusinessTraffic(data);
}else{
throw new RuntimeException("业务NET流量data数据为空");
}
}
private void handleMemoryDetailsMessage(DeviceMessage message) {
List<MemoryVo> memoryVoList = JsonDataParser.parseJsonData(message.getData(), MemoryVo.class);
String clientId = message.getClientId();
@@ -1413,8 +1561,8 @@ public class MessageHandler {
String prevTime = redisTemplate.opsForValue().get(timeKey);
String prevHeartbeatCount = redisTemplate.opsForValue().get(heartbeatCountKey);
Boolean prevAlertStatus = redisTemplate.hasKey(alertKey);
log.debug("客户端ID: {} 处理前状态 - status: {}, time: {}, heartbeatCount: {}, hasAlert: {}",
clientId, prevStatus, prevTime, prevHeartbeatCount, prevAlertStatus);
// log.debug("客户端ID: {} 处理前状态 - status: {}, time: {}, heartbeatCount: {}, hasAlert: {}",
// clientId, prevStatus, prevTime, prevHeartbeatCount, prevAlertStatus);
// 原子递增心跳计数（线程安全）
Long newHeartbeatCount = redisTemplate.opsForValue().increment(heartbeatCountKey);
@@ -1431,7 +1579,7 @@ public class MessageHandler {
}
});
log.debug("客户端ID: {} 心跳处理完成，当前心跳次数: {}", clientId, newHeartbeatCount);
// log.debug("客户端ID: {} 心跳处理完成，当前心跳次数: {}", clientId, newHeartbeatCount);
// 检查是否之前有告警状态（心跳恢复检测）
if (Boolean.TRUE.equals(redisTemplate.hasKey(alertKey))) {
@@ -1463,7 +1611,7 @@ public class MessageHandler {
// 只有达到3次心跳才执行数据库操作
if (newHeartbeatCount >= 3) {
log.debug("客户端ID: {} 达到{}次心跳，开始执行数据库操作", clientId, newHeartbeatCount);
// log.debug("客户端ID: {} 达到{}次心跳，开始执行数据库操作", clientId, newHeartbeatCount);
// 添加逻辑节点标识
RmResourceRegistrationRemote updateData = new RmResourceRegistrationRemote();
@@ -1518,8 +1666,8 @@ public class MessageHandler {
String currentTime = redisTemplate.opsForValue().get(timeKey);
String currentHeartbeatCount = redisTemplate.opsForValue().get(heartbeatCountKey);
Boolean currentAlertStatus = redisTemplate.hasKey(alertKey);
log.debug("客户端ID: {} 处理后状态 - status: {}, time: {}, heartbeatCount: {}, hasAlert: {}",
clientId, currentStatus, currentTime, currentHeartbeatCount, currentAlertStatus);
// log.debug("客户端ID: {} 处理后状态 - status: {}, time: {}, heartbeatCount: {}, hasAlert: {}",
// clientId, currentStatus, currentTime, currentHeartbeatCount, currentAlertStatus);
} catch (Exception e) {
log.error("处理心跳消息异常, clientId: {}", clientId, e);
@@ -2079,7 +2227,7 @@ public class MessageHandler {
boolean needUpdate = false;
// 逐个字段比较是否需要更新
if (!StringUtils.equals(networkInfo.getCity(), oldInterfaceMsg.getCity())) {
if (networkInfo.getCity() != null && !StringUtils.equals(networkInfo.getCity(), oldInterfaceMsg.getCity())) {
updateData.setCity(networkInfo.getCity());
needUpdate = true;
}
@@ -2091,15 +2239,15 @@ public class MessageHandler {
updateData.setIpv6Address(networkInfo.getIpv6());
needUpdate = true;
}
if (!StringUtils.equals(networkInfo.getProvince(), oldInterfaceMsg.getProvince())) {
if (networkInfo.getProvince() != null && !StringUtils.equals(networkInfo.getProvince(), oldInterfaceMsg.getProvince())) {
updateData.setProvince(networkInfo.getProvince());
needUpdate = true;
}
if (!StringUtils.equals(networkInfo.getPublicIp(), oldInterfaceMsg.getPublicIp())) {
if (networkInfo.getPublicIp() != null && !StringUtils.equals(networkInfo.getPublicIp(), oldInterfaceMsg.getPublicIp())) {
updateData.setPublicIp(networkInfo.getPublicIp());
needUpdate = true;
}
if (!StringUtils.equals(networkInfo.getCarrier(), oldInterfaceMsg.getIsp())) {
if (networkInfo.getCarrier() != null && !StringUtils.equals(networkInfo.getCarrier(), oldInterfaceMsg.getIsp())) {
updateData.setIsp(networkInfo.getCarrier());
needUpdate = true;
}