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

1、优化agent本地存储流量文件删除逻辑。

Browse Source 
2、开发tcpdump探测策略下发接口。
3、开发tcpdump探测接口。
4、开发出省流量计算统计接口。
This commit is contained in:
gaoyutao
2026-01-22 18:00:17 +08:00
parent fe0225012f
commit 493ecba0ba
9 changed files with 144 additions and 204 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/RmOutboundTrafficStatisticsController.java
+8 -66
View File
@@ -1,18 +1,16 @@
package com.tongran.rocketmq.controller;
import com.tongran.common.core.utils.poi.ExcelUtil;
import com.tongran.common.core.web.controller.BaseController;
import com.tongran.common.core.web.domain.AjaxResult;
import com.tongran.common.core.web.page.TableDataInfo;
import com.tongran.common.log.annotation.Log;
import com.tongran.common.log.enums.BusinessType;
import com.tongran.common.security.annotation.RequiresPermissions;
import com.tongran.rocketmq.domain.RmOutboundTrafficStatistics;
import com.tongran.rocketmq.service.IRmOutboundTrafficStatisticsService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.web.bind.annotation.*;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import javax.servlet.http.HttpServletResponse;
import java.util.List;
/**
@@ -23,6 +21,7 @@ import java.util.List;
*/
@RestController
@RequestMapping("/rmOutboundTrafficStatistics")
@RequiresPermissions("rocketmq:traffic")
public class RmOutboundTrafficStatisticsController extends BaseController
{
@Autowired
@@ -31,68 +30,11 @@ public class RmOutboundTrafficStatisticsController extends BaseController
/**
* 查询出省流量统计列表
*/
@RequiresPermissions("rocketmq:rmOutboundTrafficStatistics:list")
@GetMapping("/list")
public TableDataInfo list(RmOutboundTrafficStatistics rmOutboundTrafficStatistics)
{
startPage();
List<RmOutboundTrafficStatistics> list = rmOutboundTrafficStatisticsService.selectRmOutboundTrafficStatisticsList(rmOutboundTrafficStatistics);
return getDataTable(list);
}
/**
* 导出出省流量统计列表
*/
@RequiresPermissions("rocketmq:rmOutboundTrafficStatistics:export")
@Log(title = "出省流量统计", businessType = BusinessType.EXPORT)
@PostMapping("/export")
public void export(HttpServletResponse response, RmOutboundTrafficStatistics rmOutboundTrafficStatistics)
@PostMapping("/list")
public AjaxResult list(@RequestBody RmOutboundTrafficStatistics rmOutboundTrafficStatistics)
{
List<RmOutboundTrafficStatistics> list = rmOutboundTrafficStatisticsService.selectRmOutboundTrafficStatisticsList(rmOutboundTrafficStatistics);
ExcelUtil<RmOutboundTrafficStatistics> util = new ExcelUtil<RmOutboundTrafficStatistics>(RmOutboundTrafficStatistics.class);
util.exportExcel(response, list, "出省流量统计数据");
return success(list);
}
/**
* 获取出省流量统计详细信息
*/
@RequiresPermissions("rocketmq:rmOutboundTrafficStatistics:query")
@GetMapping(value = "/{id}")
public AjaxResult getInfo(@PathVariable("id") Long id)
{
return success(rmOutboundTrafficStatisticsService.selectRmOutboundTrafficStatisticsById(id));
}
/**
* 新增出省流量统计
*/
@RequiresPermissions("rocketmq:rmOutboundTrafficStatistics:add")
@Log(title = "出省流量统计", businessType = BusinessType.INSERT)
@PostMapping
public AjaxResult add(@RequestBody RmOutboundTrafficStatistics rmOutboundTrafficStatistics)
{
return toAjax(rmOutboundTrafficStatisticsService.insertRmOutboundTrafficStatistics(rmOutboundTrafficStatistics));
}
/**
* 修改出省流量统计
*/
@RequiresPermissions("rocketmq:rmOutboundTrafficStatistics:edit")
@Log(title = "出省流量统计", businessType = BusinessType.UPDATE)
@PutMapping
public AjaxResult edit(@RequestBody RmOutboundTrafficStatistics rmOutboundTrafficStatistics)
{
return toAjax(rmOutboundTrafficStatisticsService.updateRmOutboundTrafficStatistics(rmOutboundTrafficStatistics));
}
/**
* 删除出省流量统计
*/
@RequiresPermissions("rocketmq:rmOutboundTrafficStatistics:remove")
@Log(title = "出省流量统计", businessType = BusinessType.DELETE)
@DeleteMapping("/{ids}")
public AjaxResult remove(@PathVariable Long[] ids)
{
return toAjax(rmOutboundTrafficStatisticsService.deleteRmOutboundTrafficStatisticsByIds(ids));
}
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/DeviceMessage.java
-2
View File
@@ -9,6 +9,4 @@ public class DeviceMessage {
private String clientId;
private String dataType;
private String data;
/** 最后一次标志 */
private boolean lastTrafficFlag;
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/InitialBandwidthTraffic.java
+2
View File
@@ -109,4 +109,6 @@ public class InitialBandwidthTraffic extends BaseEntity
private Double pingDropped;
/** 多个clientId */
private Map<String, List<String>> clientInterfaces;
/** 是否最后一次 */
private boolean lastTrafficFlag;
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/RmOutboundTrafficStatistics.java
+2 -44
View File
@@ -1,9 +1,8 @@
package com.tongran.rocketmq.domain;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import com.tongran.common.core.annotation.Excel;
import com.tongran.common.core.web.domain.BaseEntity;
import lombok.Data;
/**
* 出省流量统计对象 rm_outbound_traffic_statistics
@@ -11,6 +10,7 @@ import com.tongran.common.core.web.domain.BaseEntity;
* @author tongran
* @date 2026-01-21
*/
@Data
public class RmOutboundTrafficStatistics extends BaseEntity
{
private static final long serialVersionUID = 1L;
@@ -26,46 +26,4 @@ public class RmOutboundTrafficStatistics extends BaseEntity
@Excel(name = "出省流量统计详情")
private String description;
public void setId(Long id)
{
this.id = id;
}
public Long getId()
{
return id;
}
public void setClientId(String clientId)
{
this.clientId = clientId;
}
public String getClientId()
{
return clientId;
}
public void setDescription(String description)
{
this.description = description;
}
public String getDescription()
{
return description;
}
@Override
public String toString() {
return new ToStringBuilder(this,ToStringStyle.MULTI_LINE_STYLE)
.append("id", getId())
.append("clientId", getClientId())
.append("description", getDescription())
.append("createTime", getCreateTime())
.append("updateTime", getUpdateTime())
.append("createBy", getCreateBy())
.append("updateBy", getUpdateBy())
.toString();
}
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/RmTcpdumpConfig.java
+4 -68
View File
@@ -1,9 +1,8 @@
package com.tongran.rocketmq.domain;
import org.apache.commons.lang3.builder.ToStringBuilder;
import org.apache.commons.lang3.builder.ToStringStyle;
import com.tongran.common.core.annotation.Excel;
import com.tongran.common.core.web.domain.BaseEntity;
import lombok.Data;
/**
* tcpdump探测策略对象 rm_tcpdump_config
@@ -11,6 +10,7 @@ import com.tongran.common.core.web.domain.BaseEntity;
* @author tongran
* @date 2026-01-21
*/
@Data
public class RmTcpdumpConfig extends BaseEntity
{
private static final long serialVersionUID = 1L;
@@ -30,72 +30,8 @@ public class RmTcpdumpConfig extends BaseEntity
@Excel(name = "探测频率")
private String frequency;
/** 探测时间列表(多个时间用;分隔) */
@Excel(name = "探测时间列表(多个时间用;分隔)")
/** 探测时间列表(多个时间用,分隔) */
@Excel(name = "探测时间列表(多个时间用,分隔)")
private String detectTimes;
public void setId(Long id)
{
this.id = id;
}
public Long getId()
{
return id;
}
public void setClientId(String clientId)
{
this.clientId = clientId;
}
public String getClientId()
{
return clientId;
}
public void setDetectFlag(Integer detectFlag)
{
this.detectFlag = detectFlag;
}
public Integer getDetectFlag()
{
return detectFlag;
}
public void setFrequency(String frequency)
{
this.frequency = frequency;
}
public String getFrequency()
{
return frequency;
}
public void setDetectTimes(String detectTimes)
{
this.detectTimes = detectTimes;
}
public String getDetectTimes()
{
return detectTimes;
}
@Override
public String toString() {
return new ToStringBuilder(this,ToStringStyle.MULTI_LINE_STYLE)
.append("id", getId())
.append("clientId", getClientId())
.append("detectFlag", getDetectFlag())
.append("frequency", getFrequency())
.append("detectTimes", getDetectTimes())
.append("createTime", getCreateTime())
.append("updateTime", getUpdateTime())
.append("createBy", getCreateBy())
.append("updateBy", getUpdateBy())
.toString();
}
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/TcpdumpVo.java
+1 -1
View File
@@ -7,5 +7,5 @@ public class TcpdumpVo {
/** ip地址 */
private String ip;
/** 出现次数 */
private Integer count;
private Double count;
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java
+114 -20
View File
@@ -44,6 +44,7 @@ import javax.annotation.PostConstruct;
import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
@@ -126,6 +127,8 @@ public class MessageHandler {
private ProducerMode producerMode;
@Autowired
private RedissonClient redissonClient;
@Autowired
private IRmOutboundTrafficStatisticsService rmOutboundTrafficStatisticsService;
/**
@@ -158,26 +161,116 @@ public class MessageHandler {
private void handleTcpdumpResultMessage(DeviceMessage message) {
List<TcpdumpVo> tcpdumpVoList = JsonDataParser.parseJsonData(message.getData(), TcpdumpVo.class);
if(tcpdumpVoList != null && !tcpdumpVoList.isEmpty()){
for (TcpdumpVo tcpdumpVo : tcpdumpVoList) {
String ip = tcpdumpVo.getIp();
Integer count = tcpdumpVo.getCount();
String type = checkIPVersion(ip);
// 查询IP地址归属地信息
Map<String, String> locationInfo = queryIpLocation(ip);
if (locationInfo != null) {
String operator = locationInfo.get("operator"); // 运营商
String province = locationInfo.get("province"); // 省份
System.out.println("IP: " + ip +
", 运营商: " + operator +
", 省份: " + province +
", 抓包数量: " + count);
// 设置到对象中（如果TcpdumpVo有这些字段）
// tcpdumpVo.setOperator(operator);
// tcpdumpVo.setProvince(province);
String localProvince = "";
// 根据clientId查询省份信息
RmNetworkInterface networkInterfaceQuery = new RmNetworkInterface();
networkInterfaceQuery.setClientId(message.getClientId());
networkInterfaceQuery.setNewFlag(1);
List<RmNetworkInterface> networkInfoList = rmNetworkInterfaceService.selectRmNetworkInterfaceList(networkInterfaceQuery);
if(networkInfoList != null && !networkInfoList.isEmpty()){
for (RmNetworkInterface rmNetworkInterface : networkInfoList) {
if(rmNetworkInterface.getProvince() != null){
localProvince = rmNetworkInterface.getProvince();
break;
}
}
}
// 计算总数量
Double totalCount = 0.0;
for (TcpdumpVo tcpdumpVo : tcpdumpVoList) {
totalCount += tcpdumpVo.getCount();
}
// 分别统计IPv4和IPv6
Map<String, List<TcpdumpVo>> ipTypeMap = new HashMap<>();
ipTypeMap.put("IPv4", new ArrayList<>());
ipTypeMap.put("IPv6", new ArrayList<>());
// 按类型分类
for (TcpdumpVo tcpdumpVo : tcpdumpVoList) {
String ip = tcpdumpVo.getIp();
String type = checkIPVersion(ip);
if ("IPv4".equals(type)) {
ipTypeMap.get("IPv4").add(tcpdumpVo);
} else if ("IPv6".equals(type)) {
ipTypeMap.get("IPv6").add(tcpdumpVo);
}
}
// 构建content内容 - 根据图片格式
StringBuilder content = new StringBuilder();
// 添加时间戳 - 根据图片格式
content.append("时间：").append(new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date())).append("\n");
// 处理IPv4统计
if (!ipTypeMap.get("IPv4").isEmpty()) {
content.append("##### V4 统计 ##### ##\n");
// 按省份运营商分组统计（过滤本地省份）
Map<String, Double> v4StatMap = new HashMap<>();
for (TcpdumpVo tcpdumpVo : ipTypeMap.get("IPv4")) {
Map<String, String> locationInfo = queryIpLocation(tcpdumpVo.getIp());
if (locationInfo != null) {
String operator = locationInfo.get("operator");
String province = locationInfo.get("province");
// 过滤本地省份和空值
if (province != null && operator != null && !localProvince.equals(province)) {
String key = province + operator; // 根据图片格式，省份和运营商之间没有空格
double currentCount = v4StatMap.getOrDefault(key, 0.0);
v4StatMap.put(key, currentCount + tcpdumpVo.getCount());
}
}
}
// 计算百分比并排序
List<Map.Entry<String, Double>> sortedV4List = new ArrayList<>(v4StatMap.entrySet());
sortedV4List.sort((a, b) -> b.getValue().compareTo(a.getValue())); // 降序排序
for (Map.Entry<String, Double> entry : sortedV4List) {
BigDecimal percentage = new BigDecimal(entry.getValue() / totalCount * 100)
.setScale(2, RoundingMode.HALF_UP);
content.append(entry.getKey()).append(": ").append(percentage).append("%\n");
}
}
// 处理IPv6统计
if (!ipTypeMap.get("IPv6").isEmpty()) {
content.append("##### V6 统计 ##### ##\n");
// 按省份运营商分组统计（过滤本地省份）
Map<String, Double> v6StatMap = new HashMap<>();
for (TcpdumpVo tcpdumpVo : ipTypeMap.get("IPv6")) {
Map<String, String> locationInfo = queryIpLocation(tcpdumpVo.getIp());
if (locationInfo != null) {
String operator = locationInfo.get("operator");
String province = locationInfo.get("province");
// 过滤本地省份和空值
if (province != null && operator != null && !localProvince.equals(province)) {
String key = province + operator; // 根据图片格式，省份和运营商之间没有空格
double currentCount = v6StatMap.getOrDefault(key, 0.0);
v6StatMap.put(key, currentCount + tcpdumpVo.getCount());
}
}
}
// 计算百分比并排序
List<Map.Entry<String, Double>> sortedV6List = new ArrayList<>(v6StatMap.entrySet());
sortedV6List.sort((a, b) -> b.getValue().compareTo(a.getValue())); // 降序排序
for (Map.Entry<String, Double> entry : sortedV6List) {
BigDecimal percentage = new BigDecimal(entry.getValue() / totalCount * 100)
.setScale(2, RoundingMode.HALF_UP);
content.append(entry.getKey()).append(": ").append(percentage).append("%\n");
}
}
RmOutboundTrafficStatistics insertData = new RmOutboundTrafficStatistics();
insertData.setClientId(message.getClientId());
insertData.setDescription(content.toString());
rmOutboundTrafficStatisticsService.insertRmOutboundTrafficStatistics(insertData);
}
}
@@ -608,7 +701,7 @@ public class MessageHandler {
}
}
/**
* 网络流量数据入库
* 网络重试流量数据入库
* @param message
*/
private void processNetRecoverMessageInternal(DeviceMessage message) {
@@ -617,9 +710,10 @@ public class MessageHandler {
String clientId = message.getClientId();
// 时间戳转换
long timestamp = interfaces.get(0).getTimestamp();
boolean lasttrafficFlag = interfaces.get(0).isLastTrafficFlag();
// 时间戳存储到redis
storeTimestampToRedis(clientId, timestamp);
if(message.isLastTrafficFlag()){
if(lasttrafficFlag){
// 把redis中存储的时间戳提取出来，删除redis中的时间戳
processExitsTraffic(clientId);
}