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

初始流量数据库读取mq入库

Browse Source
This commit is contained in:
gaoyutao
2025-08-21 18:29:22 +08:00
parent 80dbba70fe
commit 34049d1c78
30 changed files with 912 additions and 446 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/config/ConsumerConfig.java
+5 -3
View File
@@ -1,9 +1,8 @@
package com.ruoyi.rocketmq.config;
import com.ruoyi.rocketmq.enums.MessageCodeEnum;
import com.ruoyi.rocketmq.consumer.RocketMsgListener;
import com.ruoyi.rocketmq.enums.MessageTopic;
import com.ruoyi.rocketmq.model.ConsumerMode;
import com.ruoyi.rocketmq.consumer.RocketMsgListener;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
@@ -26,6 +25,9 @@ public class ConsumerConfig {
@Autowired
private ConsumerMode consumerMode;
@Autowired
private RocketMsgListener rocketMsgListener;
@Bean
public DefaultMQPushConsumer getRocketMQConsumer() {
//构建客户端连接
@@ -34,7 +36,7 @@ public class ConsumerConfig {
consumer.setNamesrvAddr(consumerMode.getNamesrvAddr());
consumer.setConsumeThreadMin(consumerMode.getConsumeThreadMin());
consumer.setConsumeThreadMax(consumerMode.getConsumeThreadMax());
consumer.registerMessageListener(new RocketMsgListener());
consumer.registerMessageListener(rocketMsgListener);
/**
* 1. CONSUME_FROM_LAST_OFFSET：第一次启动从队列最后位置消费，后续再启动接着上次消费的进度开始消费
* 2. CONSUME_FROM_FIRST_OFFSET：第一次启动从队列初始位置消费，后续再启动接着上次消费的进度开始消费
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/consumer/RocketMsgListener.java
+49 -6
View File
@@ -1,12 +1,21 @@
package com.ruoyi.rocketmq.consumer;
import com.alibaba.fastjson.JSON;
import com.ruoyi.common.core.constant.SecurityConstants;
import com.ruoyi.common.core.utils.DateUtils;
import com.ruoyi.rocketmq.domain.DeviceMessage;
import com.ruoyi.rocketmq.domain.InitialBandwidthTraffic;
import com.ruoyi.rocketmq.enums.MessageCodeEnum;
import com.ruoyi.rocketmq.producer.ConsumeException;
import com.ruoyi.rocketmq.service.IInitialBandwidthTrafficService;
import com.ruoyi.system.api.RemoteUserService;
import com.ruoyi.system.api.domain.EpsInitialTrafficDataRemote;
import lombok.extern.slf4j.Slf4j;
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.message.MessageExt;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
@@ -20,6 +29,17 @@ import java.util.List;
@Component
public class RocketMsgListener implements MessageListenerConcurrently {
private final IInitialBandwidthTrafficService initialBandwidthTrafficService;
private final RemoteUserService remoteUserService;
@Autowired
public RocketMsgListener(IInitialBandwidthTrafficService initialBandwidthTrafficService,
RemoteUserService remoteUserService) {
this.initialBandwidthTrafficService = initialBandwidthTrafficService;
this.remoteUserService = remoteUserService;
}
/**
* 消费消息
* @param list msgs.size() >= 1
@@ -50,13 +70,19 @@ public class RocketMsgListener implements MessageListenerConcurrently {
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//根据业务返回是否正常
}
// 根据不同的topic处理不同的业务 这里以订单消息为例子
if (MessageCodeEnum.ORDER_MESSAGE_TOPIC.getCode().equals(topic)) {
if (MessageCodeEnum.ORDER_TIMEOUT_TAG.getCode().equals(tags)) {
//处理你的业务
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//业务处理成功
} else {
log.info("未匹配到Tag【{}】" + tags);
if (MessageCodeEnum.AGENT_MESSAGE_TOPIC.getCode().equals(topic)) {
// 拿到信息
DeviceMessage message = JSON.parseObject(body, DeviceMessage.class);
switch (message.getDataType()){
case "NET":
handleNetMessage(message);
break;
default:
log.warn("未知数据类型：{}",message.getDataType());
}
//处理你的业务
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;//业务处理成功
}
}
}
@@ -84,4 +110,21 @@ public class RocketMsgListener implements MessageListenerConcurrently {
}
return ConsumeConcurrentlyStatus.RECONSUME_LATER;
}
private void handleNetMessage(DeviceMessage message) {
List<InitialBandwidthTraffic> interfaces = JSON.parseArray(message.getData(), InitialBandwidthTraffic.class);
InitialBandwidthTraffic data = new InitialBandwidthTraffic();
interfaces.forEach(iface -> {
iface.setClientId(message.getClientId());
});
// 批量入库集合
data.setList(interfaces);
// 初始流量数据入库
initialBandwidthTrafficService.batchInsert(data);
EpsInitialTrafficDataRemote epsInitialTrafficDataRemote = new EpsInitialTrafficDataRemote();
epsInitialTrafficDataRemote.setCreateTime(DateUtils.getNowDate());
// 复制到业务初始库
remoteUserService.autoSaveServiceTrafficData(epsInitialTrafficDataRemote, SecurityConstants.FROM_SOURCE);
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/domain/DeviceMessage.java
+13
View File
@@ -0,0 +1,13 @@
package com.ruoyi.rocketmq.domain;
import lombok.Data;
import java.time.LocalDateTime;
@Data
public class DeviceMessage {
private String clientId;
private String dataType;
private String data;
private LocalDateTime receiveTime = LocalDateTime.now();
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/domain/InitialBandwidthTraffic.java
+22 -131
View File
@@ -1,177 +1,68 @@
package com.ruoyi.rocketmq.domain;
import java.math.BigDecimal;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import com.ruoyi.common.core.annotation.Excel;
import com.ruoyi.common.core.web.domain.BaseEntity;
import lombok.Data;
import java.math.BigDecimal;
import java.util.List;
/**
* 初始带宽流量对象 initial_bandwidth_traffic
*
*
* @author gyt
* @date 2025-08-20
*/
@Data
public class InitialBandwidthTraffic extends BaseEntity
{
private static final long serialVersionUID = 1L;
/** 唯一标识ID */
private Long id;
/** 表名 */
private String tableName;
/** 接口名称 */
@Excel(name = "接口名称")
private String interfaceName;
private String name;
/** MAC地址 */
@Excel(name = "MAC地址")
private String macAddress;
private String mac;
/** 运行状态 */
@Excel(name = "运行状态")
private String operationStatus;
private String status;
/** 接口类型 */
@Excel(name = "接口类型")
private String interfaceType;
private String type;
/** IPv4地址 */
@Excel(name = "IPv4地址")
private String ipv4Address;
private String ipV4;
/** 入站丢包率(%) */
@Excel(name = "入站丢包率(%)")
private BigDecimal inboundPacketLoss;
private BigDecimal inDropped;
/** 出站丢包率(%) */
@Excel(name = "出站丢包率(%)")
private BigDecimal outboundPacketLoss;
private BigDecimal outDropped;
/** 接收带宽(Mbps) */
@Excel(name = "接收带宽(Mbps)")
private BigDecimal receiveBandwidth;
private String inSpeed;
/** 发送带宽(Mbps) */
@Excel(name = "发送带宽(Mbps)")
private BigDecimal sendBandwidth;
private String outSpeed;
public void setId(Long id)
{
this.id = id;
}
/** 设备唯一标识 */
@Excel(name = "设备唯一标识")
private String clientId;
/** 初始带宽流量集合 */
private List<InitialBandwidthTraffic> list;
public Long getId()
{
return id;
}
public void setInterfaceName(String interfaceName)
{
this.interfaceName = interfaceName;
}
public String getInterfaceName()
{
return interfaceName;
}
public void setMacAddress(String macAddress)
{
this.macAddress = macAddress;
}
public String getMacAddress()
{
return macAddress;
}
public void setOperationStatus(String operationStatus)
{
this.operationStatus = operationStatus;
}
public String getOperationStatus()
{
return operationStatus;
}
public void setInterfaceType(String interfaceType)
{
this.interfaceType = interfaceType;
}
public String getInterfaceType()
{
return interfaceType;
}
public void setIpv4Address(String ipv4Address)
{
this.ipv4Address = ipv4Address;
}
public String getIpv4Address()
{
return ipv4Address;
}
public void setInboundPacketLoss(BigDecimal inboundPacketLoss)
{
this.inboundPacketLoss = inboundPacketLoss;
}
public BigDecimal getInboundPacketLoss()
{
return inboundPacketLoss;
}
public void setOutboundPacketLoss(BigDecimal outboundPacketLoss)
{
this.outboundPacketLoss = outboundPacketLoss;
}
public BigDecimal getOutboundPacketLoss()
{
return outboundPacketLoss;
}
public void setReceiveBandwidth(BigDecimal receiveBandwidth)
{
this.receiveBandwidth = receiveBandwidth;
}
public BigDecimal getReceiveBandwidth()
{
return receiveBandwidth;
}
public void setSendBandwidth(BigDecimal sendBandwidth)
{
this.sendBandwidth = sendBandwidth;
}
public BigDecimal getSendBandwidth()
{
return sendBandwidth;
}
@Override
public String toString() {
return new ToStringBuilder(this,ToStringStyle.MULTI_LINE_STYLE)
.append("id", getId())
.append("interfaceName", getInterfaceName())
.append("macAddress", getMacAddress())
.append("operationStatus", getOperationStatus())
.append("interfaceType", getInterfaceType())
.append("ipv4Address", getIpv4Address())
.append("inboundPacketLoss", getInboundPacketLoss())
.append("outboundPacketLoss", getOutboundPacketLoss())
.append("receiveBandwidth", getReceiveBandwidth())
.append("sendBandwidth", getSendBandwidth())
.append("createTime", getCreateTime())
.append("updateTime", getUpdateTime())
.append("createBy", getCreateBy())
.append("updateBy", getUpdateBy())
.toString();
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/domain/NetInterfaceInfo.java
+22
View File
@@ -0,0 +1,22 @@
package com.ruoyi.rocketmq.domain;
import com.alibaba.fastjson.JSON;
import lombok.Data;
import java.util.List;
@Data
public class NetInterfaceInfo {
private String name;
private String type;
private String status;
private String ipV4;
private String mac;
private String inSpeed;
private String outSpeed;
private int inDropped;
private int outDropped;
public static List<NetInterfaceInfo> parseNetData(String jsonData) {
return JSON.parseArray(jsonData, NetInterfaceInfo.class);
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/enums/MessageCodeEnum.java
+6
View File
@@ -10,6 +10,11 @@ import lombok.Getter;
@Getter
public enum MessageCodeEnum {
/**
* agent数据采集的信息
*/
AGENT_MESSAGE_TOPIC("agent_up","agent数据采集的信息topic"),
/**
* 系统消息
*/
@@ -44,6 +49,7 @@ public enum MessageCodeEnum {
*/
ORDER_TIMEOUT_TAG("order_timeout_tag","订单超时处理");
private final String code;
private final String msg;
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/enums/MessageTopic.java
+2 -6
View File
@@ -12,12 +12,8 @@ public class MessageTopic {
//在这里添加topic 用于批量订阅
public List<String> RocketMQTopicList(){
List<String> getTopicLists=new ArrayList<>();
//系统消息
getTopicLists.add("system-message");
//用户消息
getTopicLists.add("user-message");
//订单消息
getTopicLists.add("order-message");
// agent采集消息
getTopicLists.add("agent_up");
return getTopicLists;
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/mapper/InitialBandwidthTrafficMapper.java
+14 -1
View File
@@ -1,8 +1,9 @@
package com.ruoyi.rocketmq.mapper;
import java.util.List;
import com.ruoyi.rocketmq.domain.InitialBandwidthTraffic;
import java.util.List;
/**
* 初始带宽流量Mapper接口
*
@@ -58,4 +59,16 @@ public interface InitialBandwidthTrafficMapper
* @return 结果
*/
public int deleteInitialBandwidthTrafficByIds(Long[] ids);
/**
* 按时间分表插入流量数据
* @param data
* @return
*/
int insert(InitialBandwidthTraffic data);
/**
* 批量插入数据
* @param data 流量数据实体类
*/
int batchInsert(InitialBandwidthTraffic data);
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/service/IInitialBandwidthTrafficService.java
+14 -1
View File
@@ -1,8 +1,9 @@
package com.ruoyi.rocketmq.service;
import java.util.List;
import com.ruoyi.rocketmq.domain.InitialBandwidthTraffic;
import java.util.List;
/**
* 初始带宽流量Service接口
*
@@ -58,4 +59,16 @@ public interface IInitialBandwidthTrafficService
* @return 结果
*/
public int deleteInitialBandwidthTrafficById(Long id);
/**
* 保存单条流量数据
* @param data 流量数据
*/
void save(InitialBandwidthTraffic data);
/**
* 保存单条流量数据
* @param data 流量数据
*/
void batchInsert(InitialBandwidthTraffic data);
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/service/impl/InitialBandwidthTrafficServiceImpl.java
+86 -4
View File
@@ -1,12 +1,23 @@
package com.ruoyi.rocketmq.service.impl;
import java.util.List;
import com.ruoyi.common.core.utils.DateUtils;
import com.ruoyi.rocketmq.domain.InitialBandwidthTraffic;
import com.ruoyi.rocketmq.mapper.InitialBandwidthTrafficMapper;
import com.ruoyi.rocketmq.service.IInitialBandwidthTrafficService;
import com.ruoyi.rocketmq.utils.TableRouterUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import com.ruoyi.rocketmq.mapper.InitialBandwidthTrafficMapper;
import com.ruoyi.rocketmq.domain.InitialBandwidthTraffic;
import com.ruoyi.rocketmq.service.IInitialBandwidthTrafficService;
import org.springframework.transaction.annotation.Transactional;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.Date;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
/**
* 初始带宽流量Service业务层处理
@@ -15,6 +26,7 @@ import com.ruoyi.rocketmq.service.IInitialBandwidthTrafficService;
* @date 2025-08-20
*/
@Service
@Slf4j
public class InitialBandwidthTrafficServiceImpl implements IInitialBandwidthTrafficService
{
@Autowired
@@ -51,6 +63,7 @@ public class InitialBandwidthTrafficServiceImpl implements IInitialBandwidthTraf
* @return 结果
*/
@Override
@Transactional(rollbackFor = Exception.class)
public int insertInitialBandwidthTraffic(InitialBandwidthTraffic initialBandwidthTraffic)
{
initialBandwidthTraffic.setCreateTime(DateUtils.getNowDate());
@@ -93,4 +106,73 @@ public class InitialBandwidthTrafficServiceImpl implements IInitialBandwidthTraf
{
return initialBandwidthTrafficMapper.deleteInitialBandwidthTrafficById(id);
}
/**
* 保存单条数据到对应分表
* @param data 流量数据
*/
@Override
@Transactional
public void save(InitialBandwidthTraffic data) {
if (data.getCreateTime() == null) {
data.setCreateTime(DateUtils.getNowDate());
}
Date date = data.getCreateTime();
LocalDateTime createTime = date.toInstant()
.atZone(ZoneId.systemDefault())
.toLocalDateTime();
data.setTableName(TableRouterUtil.getTableName(createTime));
initialBandwidthTrafficMapper.insert(data);
}
/**
* 保存多条数据到对应分表
* @param initialBandwidthTraffic 流量数据
*/
@Override
@Transactional
public void batchInsert(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.batchInsert(data);
} catch (Exception e) {
log.error("表{}插入失败", tableName, e);
throw new RuntimeException("批量插入失败", e);
}
});
}
}
ruoyi-rocketmq/src/main/java/com/ruoyi/rocketmq/utils/TableRouterUtil.java
+85
View File
@@ -0,0 +1,85 @@
package com.ruoyi.rocketmq.utils;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.LinkedHashSet;
import java.util.Set;
public class TableRouterUtil {
// 表名前缀
private static final String TABLE_PREFIX = "eps_initial_traffic";
// 表名前缀
private static final String TABLE_PREFIX_INITIAL = "initial_bandwidth_traffic";
// 日期格式
private static final DateTimeFormatter YEAR_MONTH_FORMAT =
DateTimeFormatter.ofPattern("yyyy_MM");
/**
* 根据创建时间获取表名
* @param createTime 记录创建时间
* @return 对应的分表名称
* @throws IllegalArgumentException 如果createTime为null
*
* 示例：
* 2023-08-05 14:30:00 → eps_initial_traffic_2023_08_1_10
* 2023-08-15 09:15:00 → eps_initial_traffic_2023_08_11_20
* 2023-08-25 18:45:00 → eps_initial_traffic_2023_08_21_31
*/
public static String getTableName(LocalDateTime createTime) {
if (createTime == null) {
throw new IllegalArgumentException("创建时间不能为null");
}
String yearMonth = createTime.format(YEAR_MONTH_FORMAT);
int day = createTime.getDayOfMonth();
return String.format("%s_%s_%s",
TABLE_PREFIX_INITIAL,
yearMonth,
getDayRange(day));
}
/**
* 获取时间范围内涉及的所有表名
* @param startTime 开始时间（包含）
* @param endTime 结束时间（包含）
* @return 按时间顺序排列的表名集合
*/
public static Set<String> getTableNamesBetween(LocalDateTime startTime, LocalDateTime endTime) {
validateTimeRange(startTime, endTime);
Set<String> tableNames = new LinkedHashSet<>();
LocalDateTime current = startTime.withHour(0).withMinute(0).withSecond(0);
while (!current.isAfter(endTime)) {
tableNames.add(getTableName(current));
current = current.plusDays(1);
}
return tableNames;
}
// 获取日期区间
private static String getDayRange(int day) {
if (day < 1 || day > 31) {
throw new IllegalArgumentException("日期必须在1-31之间");
}
if (day <= 10) return "1_10";
if (day <= 20) return "11_20";
return "21_31";
}
// 验证时间范围
private static void validateTimeRange(LocalDateTime start, LocalDateTime end) {
if (start == null || end == null) {
throw new IllegalArgumentException("时间范围参数不能为null");
}
if (start.isAfter(end)) {
throw new IllegalArgumentException("开始时间不能晚于结束时间");
}
}
}