diff --git a/tongran-rocketmq/pom.xml b/tongran-rocketmq/pom.xml
index b3202cf..d1ba322 100644
--- a/tongran-rocketmq/pom.xml
+++ b/tongran-rocketmq/pom.xml
@@ -121,6 +121,12 @@
tea-util
0.2.21
+
+
+ org.redisson
+ redisson-spring-boot-starter
+ 3.27.2
+
diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/config/ConsumerConfig.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/config/ConsumerConfig.java
index c6971ea..f7c09c8 100644
--- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/config/ConsumerConfig.java
+++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/config/ConsumerConfig.java
@@ -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所有消息
diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/RocketMsgListener.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/RocketMsgListener.java
index d14fb99..5a118ac 100644
--- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/RocketMsgListener.java
+++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/consumer/RocketMsgListener.java
@@ -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)){
diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/DeviceMessage.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/DeviceMessage.java
index 909d989..3807875 100644
--- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/DeviceMessage.java
+++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/DeviceMessage.java
@@ -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;
}
\ No newline at end of file
diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/PolicyTypeVo.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/PolicyTypeVo.java
index 6389281..9f837f1 100644
--- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/PolicyTypeVo.java
+++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/PolicyTypeVo.java
@@ -25,6 +25,8 @@ public class PolicyTypeVo {
private String diskIopsTest;
/** 卸载磁盘分区 */
private String umountDisk;
+ /** 需要删除的时间戳 */
+ private String trafficUsedTimestamp;
/** 时间戳 */
private Long timestamp = Instant.now().getEpochSecond();
}
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 c98fff3..740f3ad 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
@@ -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 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 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);
+ });
+ }
+ 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 = "";
diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialBandwidthTrafficMapper.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialBandwidthTrafficMapper.java
index f3ad380..de95e72 100644
--- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialBandwidthTrafficMapper.java
+++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialBandwidthTrafficMapper.java
@@ -83,4 +83,6 @@ public interface InitialBandwidthTrafficMapper
* @return
*/
List getNetTrafficList(InitialBandwidthTraffic condition);
+
+ int countByClientIdAndTime(InitialBandwidthTraffic countQuery);
}
diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialBandwidthTrafficTempMapper.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialBandwidthTrafficTempMapper.java
index e7f020c..0fcfd3e 100644
--- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialBandwidthTrafficTempMapper.java
+++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/mapper/InitialBandwidthTrafficTempMapper.java
@@ -70,4 +70,10 @@ public interface InitialBandwidthTrafficTempMapper
* @return
*/
int batchInsertServerTemp(@Param("list") List interfaces);
+
+ int batchInsertServerRecoverTemp(@Param("list") List interfaces);
+
+ int deleteTempMsgByClientIdAndTime(InitialBandwidthTrafficTemp initialBandwidthTrafficTemp);
+
+ List selectInitialBandwidthTrafficRecoverTempList(InitialBandwidthTrafficTemp initialBandwidthTrafficTemp);
}
diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialBandwidthTrafficService.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialBandwidthTrafficService.java
index dfa68b5..18440c6 100644
--- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialBandwidthTrafficService.java
+++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialBandwidthTrafficService.java
@@ -121,4 +121,6 @@ public interface IInitialBandwidthTrafficService
Map serverDetails(InitialBandwidthTraffic initialBandwidthTraffic);
Map pingDroppedEcharts(InitialBandwidthTraffic initialBandwidthTraffic);
+
+ int countByClientIdAndTime(InitialBandwidthTraffic countQuery);
}
diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialBandwidthTrafficTempService.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialBandwidthTrafficTempService.java
index 746276e..afb0cec 100644
--- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialBandwidthTrafficTempService.java
+++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IInitialBandwidthTrafficTempService.java
@@ -29,6 +29,8 @@ public interface IInitialBandwidthTrafficTempService
*/
public List selectInitialBandwidthTrafficTempList(InitialBandwidthTrafficTemp initialBandwidthTrafficTemp);
+ public List 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 interfaces);
+ /**
+ * 批量新增重试临时表信息
+ * @param interfaces
+ * @return
+ */
+ int batchInsertServerRecoverTemp(List interfaces);
}
diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialBandwidthTrafficServiceImpl.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialBandwidthTrafficServiceImpl.java
index 07c1ca7..0b730d2 100644
--- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialBandwidthTrafficServiceImpl.java
+++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialBandwidthTrafficServiceImpl.java
@@ -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
diff --git a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialBandwidthTrafficTempServiceImpl.java b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialBandwidthTrafficTempServiceImpl.java
index d8c0203..3445017 100644
--- a/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialBandwidthTrafficTempServiceImpl.java
+++ b/tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/InitialBandwidthTrafficTempServiceImpl.java
@@ -45,7 +45,17 @@ public class InitialBandwidthTrafficTempServiceImpl implements IInitialBandwidth
{
return initialBandwidthTrafficTempMapper.selectInitialBandwidthTrafficTempList(initialBandwidthTrafficTemp);
}
-
+ /**
+ * 查询初始带宽流量临时表列表
+ *
+ * @param initialBandwidthTrafficTemp 初始带宽流量临时表
+ * @return 初始带宽流量临时表
+ */
+ @Override
+ public List 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 interfaces) {
return initialBandwidthTrafficTempMapper.batchInsertServerTemp(interfaces);
}
+ @Override
+ public int batchInsertServerRecoverTemp(List interfaces) {
+ return initialBandwidthTrafficTempMapper.batchInsertServerRecoverTemp(interfaces);
+ }
}
diff --git a/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficMapper.xml b/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficMapper.xml
index c786a40..c32088e 100644
--- a/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficMapper.xml
+++ b/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficMapper.xml
@@ -223,5 +223,12 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
and create_time <= #{endTime}
-
+
\ No newline at end of file
diff --git a/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficTempMapper.xml b/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficTempMapper.xml
index 1eb7587..9030ab4 100644
--- a/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficTempMapper.xml
+++ b/tongran-rocketmq/src/main/resources/mapper/rocketmq/InitialBandwidthTrafficTempMapper.xml
@@ -59,7 +59,30 @@ PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN"
and create_time = #{createTime}
-
+