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

增加流量重试入库功能

Browse Source
This commit is contained in:
gaoyutao
2026-01-20 18:04:23 +08:00
parent 14a750a736
commit 31cf6dbb04
14 changed files with 397 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tongran-rocketmq/src/main/java/com/tongran/rocketmq/config/ConsumerConfig.java
+1
View File
@@ -54,6 +54,7 @@ public class ConsumerConfig {
consumer.setVipChannelEnabled(false);
consumer.setConsumeMessageBatchMaxSize(consumerMode.getConsumeMessageBatchMaxSize());
consumer.setSuspendCurrentQueueTimeMillis(5);
try {
/**
* 订阅topic，可以对指定消息进行过滤，例如："TopicTest","tagl||tag2||tag3",*或null表示topic所有消息
tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/RocketMsgListener.java
+13 -2
View File
@@ -1,6 +1,7 @@
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.enums.MessageCodeEnum;
import com.tongran.rocketmq.handler.DeviceMessageHandler;
@@ -56,8 +57,18 @@ public class RocketMsgListener implements MessageListenerConcurrently {
int reConsume = messageExt.getReconsumeTimes();
// 消息已经重试了3次，如果不需要再次消费，则返回成功
if (reConsume == 3) {
// TODO 补偿信息
log.error("消息消费三次失败，消息内容：{}", body);
// 拿到信息
DeviceMessage message = JSON.parseObject(body, DeviceMessage.class);
if(!message.getDataType().equals(MsgEnum.网络上报重试.getValue())){
// TODO 补偿信息
log.error("消息消费三次失败，消息内容：{}", body);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//根据业务返回是否正常
}
}
// 流量数据重试
if (reConsume == 6) {
// 补偿信息
log.error("流量数据重试消息消费6次失败，消息内容：{}", body);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//根据业务返回是否正常
}
if(MessageCodeEnum.TONGRAN_AGENT_UP.getCode().equals(topic)){
tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/DeviceMessage.java
+4
View File
@@ -1,10 +1,14 @@
package com.tongran.rocketmq.domain;
import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import lombok.Data;
@Data
@JsonIgnoreProperties(ignoreUnknown = true)
public class DeviceMessage {
private String clientId;
private String dataType;
private String data;
/** 最后一次标志 */
private boolean lastTrafficFlag;
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/PolicyTypeVo.java
+2
View File
@@ -25,6 +25,8 @@ public class PolicyTypeVo {
private String diskIopsTest;
/** 卸载磁盘分区 */
private String umountDisk;
/** 需要删除的时间戳 */
private String trafficUsedTimestamp;
/** 时间戳 */
private Long timestamp = Instant.now().getEpochSecond();
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java
+266
View File
@@ -11,6 +11,8 @@ import com.tongran.rocketmq.domain.*;
import com.tongran.rocketmq.domain.vo.*;
import com.tongran.rocketmq.enums.AlarmTypeEnum;
import com.tongran.rocketmq.enums.PushMethodEnum;
import com.tongran.rocketmq.model.ProducerMode;
import com.tongran.rocketmq.producer.MessageProducer;
import com.tongran.rocketmq.service.*;
import com.tongran.rocketmq.utils.DataProcessUtil;
import com.tongran.rocketmq.utils.JsonDataParser;
@@ -18,6 +20,8 @@ import com.tongran.rocketmq.utils.WeChatWorkBot;
import com.tongran.system.api.RemoteRevenueConfigService;
import com.tongran.system.api.domain.*;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
@@ -32,6 +36,7 @@ import javax.annotation.PostConstruct;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -108,6 +113,10 @@ public class MessageHandler {
private IRmPppoeConfigSubService rmPppoeConfigSubService;
@Autowired
private IAllDiskNameService allDiskNameService;
@Autowired
private ProducerMode producerMode;
@Autowired
private RedissonClient redissonClient;
/**
@@ -126,6 +135,7 @@ public class MessageHandler {
registerHandler(MsgEnum.容器上报.getValue(), this::handleDockerMessage);
registerHandler(MsgEnum.内存上报.getValue(), this::handleMemoryMessage);
registerHandler(MsgEnum.网络上报.getValue(), this::handleNetMessage);
registerHandler(MsgEnum.网络上报重试.getValue(), this::handleNetRecoverMessage);
registerHandler(MsgEnum.挂载上报.getValue(), this::handleMountPointMessage);
registerHandler(MsgEnum.系统其他上报.getValue(), this::handleOtherSystemMessage);
registerHandler(MsgEnum.心跳上报.getValue(), this::handleHeartbeatMessage);
@@ -505,6 +515,262 @@ public class MessageHandler {
throw new RuntimeException("NET流量data数据为空");
}
}
/**
* 存储时间戳到Redis
* @param clientId 客户端ID
* @param timestamp 时间戳
*/
private void storeTimestampToRedis(String clientId, long timestamp) {
try {
String redisKey = "net_traffic_timestamps:" + clientId;
String existingTimestamps = redisTemplate.opsForValue().get(redisKey);
if (existingTimestamps != null && !existingTimestamps.isEmpty()) {
// 已存在时间戳，追加新的时间戳
String newTimestamps = existingTimestamps + "," + timestamp;
redisTemplate.opsForValue().set(redisKey, newTimestamps);
} else {
// 首次存储
redisTemplate.opsForValue().set(redisKey, String.valueOf(timestamp));
}
} catch (Exception e) {
// Redis操作异常处理，记录日志但不中断主流程
log.error("存储时间戳到Redis失败，clientId: {}, timestamp: {}", clientId, timestamp, e);
}
}
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) {
log.info("设备{}获取锁成功，开始处理消息", clientId);
processNetRecoverMessageInternal(message);
log.info("设备{}消息处理完成", clientId);
} else {
log.warn("设备{}获取锁失败，消息处理繁忙", clientId);
throw new RuntimeException("设备处理繁忙，请重试");
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
log.error("处理消息时线程被中断", e);
} finally {
// 只有成功获取锁的线程才需要释放
if (locked && lock.isHeldByCurrentThread()) {
try {
lock.unlock();
log.debug("设备{}锁已释放", clientId);
} catch (IllegalMonitorStateException e) {
log.warn("设备{}锁释放异常，可能已自动超时", clientId);
}
}
}
}
/**
* 网络流量数据入库
* @param message
*/
private void processNetRecoverMessageInternal(DeviceMessage message) {
List<InitialBandwidthTraffic> interfaces = JsonDataParser.parseJsonData(message.getData(), InitialBandwidthTraffic.class);
if(!interfaces.isEmpty()){
String clientId = message.getClientId();
// 时间戳转换
long timestamp = interfaces.get(0).getTimestamp();
// 时间戳存储到redis
storeTimestampToRedis(clientId, timestamp);
if(message.isLastTrafficFlag()){
// 把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);
// 判断数据库中是否已存在数据
InitialBandwidthTraffic countQuery = new InitialBandwidthTraffic();
countQuery.setClientId(clientId);
countQuery.setCreateTime(createTime);
countQuery.setStartTime(timeStr);
int exitsCount = initialBandwidthTrafficService.countByClientIdAndTime(countQuery);
if(exitsCount > 0){
return;
}
// 创建比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()
.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);
});
}
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{
throw new RuntimeException("NET流量data数据为空");
}
}
private void processExitsTraffic(String clientId) {
try {
String redisKey = "net_traffic_timestamps:" + clientId;
String storedTimestamps = redisTemplate.opsForValue().get(redisKey);
if (storedTimestamps != null && !storedTimestamps.isEmpty()) {
// 删除Redis中的时间戳数据
redisTemplate.delete(redisKey);
// 构建磁盘下发信息
PolicyTypeVo policyTypeVo = new PolicyTypeVo();
policyTypeVo.setTrafficUsedTimestamp(storedTimestamps);
MessageProducer messageProducer = new MessageProducer();
String configJson = JSONObject.toJSONString(policyTypeVo);
DeviceMessage message = new DeviceMessage();
message.setClientId(clientId);
message.setData(configJson);
message.setDataType(MsgEnum.获取最新策略应答.getValue());
messageProducer.sendAsyncProducerMessage(
producerMode.getAgentTopic(),
"",
"",
JSONObject.toJSONString(message)
);
}
} catch (Exception e) {
log.error("提取并删除Redis时间戳失败，clientId: {}", clientId, e);
}
}
private String generateKey(String mac, String name) {
if (mac == null) {
mac = "";
tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialBandwidthTrafficMapper.java
+2
View File
@@ -83,4 +83,6 @@ public interface InitialBandwidthTrafficMapper
* @return
*/
List<InitialBandwidthTraffic> getNetTrafficList(InitialBandwidthTraffic condition);
int countByClientIdAndTime(InitialBandwidthTraffic countQuery);
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialBandwidthTrafficTempMapper.java
+6
View File
@@ -70,4 +70,10 @@ public interface InitialBandwidthTrafficTempMapper
* @return
*/
int batchInsertServerTemp(@Param("list") List<InitialBandwidthTraffic> interfaces);
int batchInsertServerRecoverTemp(@Param("list") List<InitialBandwidthTraffic> interfaces);
int deleteTempMsgByClientIdAndTime(InitialBandwidthTrafficTemp initialBandwidthTrafficTemp);
List<InitialBandwidthTrafficTemp> selectInitialBandwidthTrafficRecoverTempList(InitialBandwidthTrafficTemp initialBandwidthTrafficTemp);
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialBandwidthTrafficService.java
+2
View File
@@ -121,4 +121,6 @@ public interface IInitialBandwidthTrafficService
Map<String, Object> serverDetails(InitialBandwidthTraffic initialBandwidthTraffic);
Map<String, Object> pingDroppedEcharts(InitialBandwidthTraffic initialBandwidthTraffic);
int countByClientIdAndTime(InitialBandwidthTraffic countQuery);
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialBandwidthTrafficTempService.java
+9
View File
@@ -29,6 +29,8 @@ public interface IInitialBandwidthTrafficTempService
*/
public List<InitialBandwidthTrafficTemp> selectInitialBandwidthTrafficTempList(InitialBandwidthTrafficTemp initialBandwidthTrafficTemp);
public List<InitialBandwidthTrafficTemp> selectInitialBandwidthTrafficRecoverTempList(InitialBandwidthTrafficTemp initialBandwidthTrafficTemp);
/**
* 新增初始带宽流量临时表
*
@@ -67,6 +69,7 @@ public interface IInitialBandwidthTrafficTempService
* @return
*/
int deleteTempMsgByClientId(String clientId);
int deleteTempMsgByClientIdAndTime(InitialBandwidthTrafficTemp initialBandwidthTrafficTemp);
/**
* 批量新增临时表信息
@@ -74,4 +77,10 @@ public interface IInitialBandwidthTrafficTempService
* @return
*/
int batchInsertServerTemp(List<InitialBandwidthTraffic> interfaces);
/**
* 批量新增重试临时表信息
* @param interfaces
* @return
*/
int batchInsertServerRecoverTemp(List<InitialBandwidthTraffic> interfaces);
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialBandwidthTrafficServiceImpl.java
+7
View File
@@ -649,6 +649,13 @@ public class InitialBandwidthTrafficServiceImpl implements IInitialBandwidthTraf
);
}
@Override
public int countByClientIdAndTime(InitialBandwidthTraffic countQuery) {
String tableName = TableSubUtil.getTableName(countQuery.getCreateTime(), "initial_bandwidth_traffic");
countQuery.setTableName(tableName);
return initialBandwidthTrafficMapper.countByClientIdAndTime(countQuery);
}
/**
* 构建支持子网卡的批量echarts图表数据
* @param initialBandwidthTraffic
tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialBandwidthTrafficTempServiceImpl.java
+20 -1
View File
@@ -45,7 +45,17 @@ public class InitialBandwidthTrafficTempServiceImpl implements IInitialBandwidth
{
return initialBandwidthTrafficTempMapper.selectInitialBandwidthTrafficTempList(initialBandwidthTrafficTemp);
}
/**
* 查询初始带宽流量临时表列表
*
* @param initialBandwidthTrafficTemp 初始带宽流量临时表
* @return 初始带宽流量临时表
*/
@Override
public List<InitialBandwidthTrafficTemp> selectInitialBandwidthTrafficRecoverTempList(InitialBandwidthTrafficTemp initialBandwidthTrafficTemp)
{
return initialBandwidthTrafficTempMapper.selectInitialBandwidthTrafficRecoverTempList(initialBandwidthTrafficTemp);
}
/**
* 新增初始带宽流量临时表
*
@@ -101,8 +111,17 @@ public class InitialBandwidthTrafficTempServiceImpl implements IInitialBandwidth
return initialBandwidthTrafficTempMapper.deleteTempMsgByClientId(clientId);
}
@Override
public int deleteTempMsgByClientIdAndTime(InitialBandwidthTrafficTemp initialBandwidthTrafficTemp) {
return initialBandwidthTrafficTempMapper.deleteTempMsgByClientIdAndTime(initialBandwidthTrafficTemp);
}
@Override
public int batchInsertServerTemp(List<InitialBandwidthTraffic> interfaces) {
return initialBandwidthTrafficTempMapper.batchInsertServerTemp(interfaces);
}
@Override
public int batchInsertServerRecoverTemp(List<InitialBandwidthTraffic> interfaces) {
return initialBandwidthTrafficTempMapper.batchInsertServerRecoverTemp(interfaces);
}
}