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

1、优化tcpdump解析速度。

Browse Source 
2、补充tcpdump晚8点默认策略下发。
3、服务器管理列表增加出省流量占比列。
This commit is contained in:
gaoyutao
2026-02-04 17:41:00 +08:00
parent f8e0ab1b2a
commit 5dd9004081
12 changed files with 183 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/RmTcpdumpConfigController.java
+18 -1
View File
@@ -4,6 +4,7 @@ import com.tongran.common.core.web.controller.BaseController;
import com.tongran.common.core.web.domain.AjaxResult;
import com.tongran.common.log.annotation.Log;
import com.tongran.common.log.enums.BusinessType;
import com.tongran.common.security.annotation.InnerAuth;
import com.tongran.common.security.annotation.RequiresPermissions;
import com.tongran.rocketmq.domain.RmTcpdumpConfig;
import com.tongran.rocketmq.service.IRmTcpdumpConfigService;
@@ -55,5 +56,21 @@ public class RmTcpdumpConfigController extends BaseController
{
return toAjax(rmTcpdumpConfigService.insertRmTcpdumpConfig(rmTcpdumpConfig));
}
/**
* 批量新增tcpdump探测策略
*/
@PostMapping("/batchInsertRmTcpdumpConfig")
public AjaxResult batchInsertRmTcpdumpConfig(@RequestBody RmTcpdumpConfig rmTcpdumpConfig)
{
return toAjax(rmTcpdumpConfigService.batchInsertRmTcpdumpConfig(rmTcpdumpConfig));
}
/**
* 新增tcpdump探测策略
*/
@InnerAuth
@PostMapping("/innerInsertRmTcpdumpConfig")
public AjaxResult innerInsertRmTcpdumpConfig(@RequestBody RmTcpdumpConfig rmTcpdumpConfig)
{
return toAjax(rmTcpdumpConfigService.insertRmTcpdumpConfig(rmTcpdumpConfig));
}
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/RmTcpdumpConfig.java
+2
View File
@@ -34,4 +34,6 @@ public class RmTcpdumpConfig extends BaseEntity
@Excel(name = "探测时间列表(多个时间用,分隔)")
private String detectTimes;
private String clientIds;
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java
+43 -30
View File
@@ -21,11 +21,11 @@ import com.tongran.rocketmq.utils.SendAlarmPushUtil;
import com.tongran.system.api.RemoteRevenueConfigService;
import com.tongran.system.api.domain.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.impl.conn.PoolingHttpClientConnectionManager;
import org.apache.http.util.EntityUtils;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
@@ -40,13 +40,11 @@ import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
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.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.*;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -58,7 +56,23 @@ import java.util.stream.Collectors;
@Component
@EnableScheduling
public class MessageHandler {
// 全局HTTP连接池
private static final PoolingHttpClientConnectionManager connectionManager =
new PoolingHttpClientConnectionManager();
private static final CloseableHttpClient httpClient;
static {
connectionManager.setMaxTotal(100); // 最大连接数
connectionManager.setDefaultMaxPerRoute(50); // 每个路由最大连接数
httpClient = HttpClients.custom()
.setConnectionManager(connectionManager)
.build();
}
// 使用自定义线程池限制并发数
private static final ExecutorService DEVICE_PROCESS_POOL =
Executors.newFixedThreadPool(40); // 限制40个并发
private static final ExecutorService IP_QUERY_POOL =
Executors.newFixedThreadPool(100);
private final Map<String, Consumer<DeviceMessage>> messageHandlers = new HashMap<>();
// 硬盘上报状态
String DISK_COUNT_PREFIX = "disk:count:";
@@ -161,6 +175,10 @@ public class MessageHandler {
}
private void handleTcpdumpResultMessage(DeviceMessage message) {
// 立即提交到专属线程池
DEVICE_PROCESS_POOL.execute(() -> processMessage(message));
}
private void processMessage(DeviceMessage message){
List<TcpdumpVo> tcpdumpVoList = JsonDataParser.parseJsonData(message.getData(), TcpdumpVo.class);
if(tcpdumpVoList != null && !tcpdumpVoList.isEmpty()){
String localProvince = "";
@@ -196,9 +214,15 @@ public class MessageHandler {
// 在计算百分比之前，先并行查询所有IP归属地
Map<String, Map<String, String>> ipLocationCache = new ConcurrentHashMap<>();
tcpdumpVoList.parallelStream().forEach(vo -> {
ipLocationCache.computeIfAbsent(vo.getIp(), ip -> queryIpLocation(ip));
});
List<CompletableFuture<Void>> futures = new ArrayList<>();
for (TcpdumpVo vo : tcpdumpVoList) {
CompletableFuture<Void> future = CompletableFuture.runAsync(() -> {
ipLocationCache.computeIfAbsent(vo.getIp(), ip -> queryIpLocation(ip));
}, IP_QUERY_POOL);
futures.add(future);
}
// 等待所有查询完成
CompletableFuture.allOf(futures.toArray(new CompletableFuture[0])).join();
// 计算总数量
Double totalCount = 0.0;
@@ -319,6 +343,11 @@ public class MessageHandler {
}
// 在更新后的位置插入IPv6总计
content.insert(insertPosition, "总占比: " + totalRate + "%\n");
// 添加总占比
RmResourceRegistrationRemote updateData = new RmResourceRegistrationRemote();
updateData.setClientId(message.getClientId());
updateData.setOutboundRate(totalRate);
remoteRevenueConfigService.innerUpdateRegist(updateData, SecurityConstants.INNER);
RmOutboundTrafficStatistics insertData = new RmOutboundTrafficStatistics();
insertData.setClientId(message.getClientId());
insertData.setDescription(content.toString());
@@ -1368,7 +1397,7 @@ public class MessageHandler {
long currentTime = System.currentTimeMillis();
log.debug("开始心跳状态检查，当前时间: {}", currentTime);
// 获取所有客户端时间键
// 获取所有客户端时间键0
Set<String> timeKeys = redisTemplate.keys(HEARTBEAT_TIME_PREFIX + "*");
if (timeKeys == null) {
log.debug("未找到任何心跳时间键");
@@ -2033,35 +2062,19 @@ public class MessageHandler {
*/
private Map<String, String> queryIpLocation(String ip) {
String apiUrl = "http://172.16.15.103:10000/?ip=" + ip;
CloseableHttpClient httpClient = HttpClients.createDefault();
HttpGet httpGet = new HttpGet(apiUrl);
try {
CloseableHttpResponse response = httpClient.execute(httpGet);
try (CloseableHttpResponse response = httpClient.execute(httpGet)) {
if (response.getStatusLine().getStatusCode() == 200) {
HttpEntity entity = response.getEntity();
if (entity != null) {
String jsonResponse = EntityUtils.toString(entity, "UTF-8");
// 使用Jackson解析JSON
ObjectMapper mapper = new ObjectMapper();
Map<String, Object> resultMap = mapper.readValue(jsonResponse, new TypeReference<Map<String, Object>>() {});
return extractLocationInfo(resultMap);
}
} else {
System.err.println("获取IP归属地信息失败，HTTP状态码: " + response.getStatusLine().getStatusCode());
String jsonResponse = EntityUtils.toString(response.getEntity(), "UTF-8");
ObjectMapper mapper = new ObjectMapper();
Map<String, Object> resultMap = mapper.readValue(jsonResponse,
new TypeReference<Map<String, Object>>() {});
return extractLocationInfo(resultMap);
}
} catch (Exception e) {
System.err.println("查询IP归属地时发生异常: " + e.getMessage());
e.printStackTrace();
} finally {
try {
httpClient.close();
} catch (IOException e) {
e.printStackTrace();
}
}
return null;
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/IRmTcpdumpConfigService.java
+2
View File
@@ -61,4 +61,6 @@ public interface IRmTcpdumpConfigService
public int deleteRmTcpdumpConfigById(Long id);
RmTcpdumpConfig selectRmTcpdumpConfigByClientId(String clientId);
int batchInsertRmTcpdumpConfig(RmTcpdumpConfig rmTcpdumpConfig);
}
tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/RmTcpdumpConfigServiceImpl.java
+33
View File
@@ -129,4 +129,37 @@ public class RmTcpdumpConfigServiceImpl implements IRmTcpdumpConfigService
public RmTcpdumpConfig selectRmTcpdumpConfigByClientId(String clientId) {
return rmTcpdumpConfigMapper.selectRmTcpdumpConfigByClientId(clientId);
}
@Override
public int batchInsertRmTcpdumpConfig(RmTcpdumpConfig rmTcpdumpConfig)
{
String clientIds = rmTcpdumpConfig.getClientIds();
String[] clientIdArr = clientIds.split(",");
for (String clientId : clientIdArr) {
rmTcpdumpConfig.setClientId(clientId);
rmTcpdumpConfig.setCreateTime(DateUtils.getNowDate());
rmTcpdumpConfigMapper.insertRmTcpdumpConfig(rmTcpdumpConfig);
// 下发消息给agent
PolicyTypeVo policyTypeVo = new PolicyTypeVo();
RmTcpdumpVo rmTcpdumpVo = new RmTcpdumpVo();
rmTcpdumpVo.setDetectFlag(rmTcpdumpConfig.getDetectFlag());
rmTcpdumpVo.setFrequency(rmTcpdumpConfig.getFrequency());
rmTcpdumpVo.setDetectTimes(rmTcpdumpConfig.getDetectTimes()==null?"":rmTcpdumpConfig.getDetectTimes());
policyTypeVo.setTcpdumpTimes(JSONObject.toJSONString(rmTcpdumpVo));
MessageProducer messageProducer = new MessageProducer();
String configJson = JSONObject.toJSONString(policyTypeVo);
DeviceMessage message = new DeviceMessage();
message.setClientId(rmTcpdumpConfig.getClientId());
message.setData(configJson);
message.setDataType(MsgEnum.获取最新策略应答.getValue());
messageProducer.sendAsyncProducerMessage(
producerMode.getAgentTopic(),
"",
"",
JSONObject.toJSONString(message)
);
}
return 1;
}
}