定时任务流量上报
This commit is contained in:
lys
parent
07d74769df
commit
a850899638
|
|
@ -17,18 +17,27 @@ dataSources:
|
|||
rules:
|
||||
- !SHARDING
|
||||
tables: # 数据分片规则配置
|
||||
# 表名
|
||||
net_flow_record:
|
||||
# 数据源 + 表名
|
||||
actualDataNodes: net_flow_record.net_flow_record_${['00','01','02','03','04','05','06','07','08','09','10','11','12','13','14','15','16','17','18','19','20','21','22','23']}
|
||||
# 节点流量记录表（按天分表）
|
||||
net_node_flow_record:
|
||||
actualDataNodes: net_flow_record.net_node_flow_record_${['20260323']}
|
||||
tableStrategy:
|
||||
standard:
|
||||
shardingColumn: create_time
|
||||
shardingAlgorithmName: hourly_sharding
|
||||
shardingAlgorithmName: daily_sharding
|
||||
|
||||
# 用户流量记录表（按天分表）
|
||||
net_user_flow_record:
|
||||
actualDataNodes: net_flow_record.net_user_flow_record_${['20260323']}
|
||||
tableStrategy:
|
||||
standard:
|
||||
shardingColumn: create_time
|
||||
shardingAlgorithmName: daily_sharding
|
||||
|
||||
shardingAlgorithms:
|
||||
|
||||
hourly_sharding:
|
||||
# 新增: 按天分表算法
|
||||
daily_sharding:
|
||||
type: CLASS_BASED
|
||||
props:
|
||||
strategy: standard
|
||||
algorithmClassName: org.dromara.common.mybatis.sharding.HourlyShardingAlgorithm
|
||||
algorithmClassName: org.dromara.common.mybatis.sharding.DailyShardingAlgorithm
|
||||
|
|
|
|||
|
|
@ -0,0 +1,314 @@
|
|||
package org.dromara.job.snailjob;
|
||||
|
||||
import com.aizuda.snailjob.client.job.core.annotation.JobExecutor;
|
||||
import com.aizuda.snailjob.client.job.core.dto.JobArgs;
|
||||
import com.aizuda.snailjob.common.core.util.JsonUtil;
|
||||
import com.aizuda.snailjob.common.log.SnailJobLog;
|
||||
import com.aizuda.snailjob.model.dto.ExecuteResult;
|
||||
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
||||
import lombok.Data;
|
||||
import lombok.RequiredArgsConstructor;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.dromara.common.core.enums.NetNodeSoftTypeEnum;
|
||||
import org.dromara.common.mybatis.utils.IdGeneratorUtil;
|
||||
import org.dromara.common.tenant.helper.TenantHelper;
|
||||
import org.dromara.net.domain.NetNode;
|
||||
import org.dromara.net.domain.NetNodeConfig;
|
||||
import org.dromara.net.domain.NetNodeFlowRecord;
|
||||
import org.dromara.net.domain.NetUserFlowRecord;
|
||||
import org.dromara.net.dto.NodeUserStatsRequest;
|
||||
import org.dromara.net.dto.NodeUserStatsResponse;
|
||||
import org.dromara.net.mapper.NetNodeConfigMapper;
|
||||
import org.dromara.net.mapper.NetNodeFlowRecordMapper;
|
||||
import org.dromara.net.mapper.NetNodeMapper;
|
||||
import org.dromara.net.mapper.NetUserFlowRecordMapper;
|
||||
import org.dromara.net.util.RemnaNodeHttpClient;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.Date;
|
||||
import java.util.HashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
|
||||
/**
|
||||
* Remna节点用户流量统计定时任务
|
||||
* <p>
|
||||
* 功能说明：
|
||||
* 1. 查询开启的或状态正常的REMNAWAVE节点
|
||||
* 2. 调用节点接口 /node/stats/get-users-stats 获取用户流量信息（reset=true 重置统计）
|
||||
* 3. 将流量信息写入 net_node_flow_record 和 net_user_flow_record 表
|
||||
* </p>
|
||||
*
|
||||
* @author lys
|
||||
* @date 2025-03-23
|
||||
*/
|
||||
@Slf4j
|
||||
@Component
|
||||
@RequiredArgsConstructor
|
||||
@JobExecutor(name = "remnaNodeUserStatsJob")
|
||||
public class RemnaNodeUserStatsJob {
|
||||
|
||||
private final NetNodeMapper netNodeMapper;
|
||||
private final NetNodeConfigMapper netNodeConfigMapper;
|
||||
private final NetNodeFlowRecordMapper netNodeFlowRecordMapper;
|
||||
private final NetUserFlowRecordMapper netUserFlowRecordMapper;
|
||||
|
||||
/**
|
||||
* 请求超时时间（毫秒）
|
||||
*/
|
||||
private static final int REQUEST_TIMEOUT = 10000;
|
||||
|
||||
/**
|
||||
* 用户流量统计接口路径
|
||||
*/
|
||||
private static final String USER_STATS_PATH = "/node/stats/get-users-stats";
|
||||
|
||||
/**
|
||||
* 节点状态超时阈值（5分钟，与NodeStatusFilterHandler保持一致）
|
||||
*/
|
||||
private static final long NODE_STATUS_TIMEOUT_MS = 5 * 60 * 1000L;
|
||||
|
||||
public ExecuteResult jobExecute(JobArgs jobArgs) {
|
||||
SnailJobLog.LOCAL.info("remnaNodeUserStatsJob 开始执行. jobArgs:{}", JsonUtil.toJsonString(jobArgs));
|
||||
SnailJobLog.REMOTE.info("remnaNodeUserStatsJob 开始执行. jobArgs:{}", JsonUtil.toJsonString(jobArgs));
|
||||
|
||||
// 1. 查询所有可用的REMNAWAVE节点
|
||||
List<NetNode> nodes = queryAvailableNodes();
|
||||
|
||||
SnailJobLog.LOCAL.info("remnaNodeUserStatsJob 找到 {} 个可用RemnaWave节点", nodes.size());
|
||||
SnailJobLog.REMOTE.info("remnaNodeUserStatsJob 找到 {} 个可用RemnaWave节点", nodes.size());
|
||||
|
||||
if (nodes.isEmpty()) {
|
||||
return ExecuteResult.success("没有可用的RemnaWave节点");
|
||||
}
|
||||
|
||||
// 2. 用于汇总用户级别的流量（跨节点汇总）
|
||||
Map<Long, UserFlowAggregate> userFlowMap = new HashMap<>();
|
||||
int successCount = 0;
|
||||
int failCount = 0;
|
||||
Date recordTime = new Date();
|
||||
|
||||
// 3. 遍历节点获取流量数据
|
||||
for (NetNode node : nodes) {
|
||||
try {
|
||||
// 获取节点配置
|
||||
NetNodeConfig config = TenantHelper.ignore(() -> {
|
||||
LambdaQueryWrapper<NetNodeConfig> configQueryWrapper = new LambdaQueryWrapper<>();
|
||||
configQueryWrapper.eq(NetNodeConfig::getNodeId, node.getId());
|
||||
return netNodeConfigMapper.selectOne(configQueryWrapper);
|
||||
});
|
||||
|
||||
if (config == null) {
|
||||
SnailJobLog.LOCAL.warn("remnaNodeUserStatsJob 节点 {} ({}) 配置不存在，跳过",
|
||||
node.getId(), node.getNodeName());
|
||||
SnailJobLog.REMOTE.warn("remnaNodeUserStatsJob 节点 {} ({}) 配置不存在，跳过",
|
||||
node.getId(), node.getNodeName());
|
||||
failCount++;
|
||||
continue;
|
||||
}
|
||||
|
||||
// 构建请求URL
|
||||
String nodeUrl = RemnaNodeHttpClient.buildNodeUrl(node.getNodeIp(), node.getServerPort());
|
||||
String statsUrl = nodeUrl + USER_STATS_PATH;
|
||||
|
||||
// 构建请求体（reset=true 重置节点统计）
|
||||
NodeUserStatsRequest request = new NodeUserStatsRequest(true);
|
||||
|
||||
// 调用接口获取用户流量统计
|
||||
NodeUserStatsResponse statsResponse = RemnaNodeHttpClient.postAndParse(
|
||||
config, statsUrl, request, NodeUserStatsResponse.class, REQUEST_TIMEOUT);
|
||||
|
||||
if (statsResponse != null && statsResponse.getResponse() != null
|
||||
&& statsResponse.getResponse().getUsers() != null) {
|
||||
|
||||
List<NodeUserStatsResponse.UserFlowData> users = statsResponse.getResponse().getUsers();
|
||||
|
||||
// 计算节点级别的流量汇总
|
||||
long nodeUpTotal = 0L;
|
||||
long nodeDownTotal = 0L;
|
||||
|
||||
for (NodeUserStatsResponse.UserFlowData userData : users) {
|
||||
Long uplink = userData.getUplink() != null ? userData.getUplink() : 0L;
|
||||
Long downlink = userData.getDownlink() != null ? userData.getDownlink() : 0L;
|
||||
|
||||
nodeUpTotal += uplink;
|
||||
nodeDownTotal += downlink;
|
||||
|
||||
// 解析用户ID（username 就是用户ID 的字符串形式）
|
||||
Long userId = parseUserId(userData.getUsername());
|
||||
if (userId != null) {
|
||||
// 汇总用户流量（跨节点累加）
|
||||
userFlowMap.compute(userId, (k, v) -> {
|
||||
if (v == null) {
|
||||
return new UserFlowAggregate(uplink, downlink);
|
||||
} else {
|
||||
v.addUplink(uplink);
|
||||
v.addDownlink(downlink);
|
||||
return v;
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// 保存节点流量记录
|
||||
saveNodeFlowRecord(node.getId(), nodeUpTotal, nodeDownTotal, recordTime);
|
||||
|
||||
successCount++;
|
||||
SnailJobLog.LOCAL.info("remnaNodeUserStatsJob 节点 {} ({}) 流量采集成功, 用户数={}, 上传={}, 下载={}",
|
||||
node.getId(), node.getNodeName(), users.size(), nodeUpTotal, nodeDownTotal);
|
||||
} else {
|
||||
failCount++;
|
||||
SnailJobLog.LOCAL.warn("remnaNodeUserStatsJob 节点 {} ({}) 流量采集失败：响应为空",
|
||||
node.getId(), node.getNodeName());
|
||||
SnailJobLog.REMOTE.warn("remnaNodeUserStatsJob 节点 {} ({}) 流量采集失败：响应为空",
|
||||
node.getId(), node.getNodeName());
|
||||
}
|
||||
} catch (Exception e) {
|
||||
failCount++;
|
||||
SnailJobLog.LOCAL.error("remnaNodeUserStatsJob 节点 {} ({}) 流量采集异常: {}",
|
||||
node.getId(), node.getNodeName(), e.getMessage());
|
||||
SnailJobLog.REMOTE.error("remnaNodeUserStatsJob 节点 {} ({}) 流量采集异常: {}",
|
||||
node.getId(), node.getNodeName(), e.getMessage());
|
||||
}
|
||||
}
|
||||
|
||||
// 4. 保存用户级别的流量汇总记录
|
||||
int userRecordCount = saveUserFlowRecords(userFlowMap, recordTime);
|
||||
|
||||
String result = String.format("执行完成, 节点成功: %d, 节点失败: %d, 用户记录数: %d",
|
||||
successCount, failCount, userRecordCount);
|
||||
SnailJobLog.LOCAL.info("remnaNodeUserStatsJob {}", result);
|
||||
SnailJobLog.REMOTE.info("remnaNodeUserStatsJob {}", result);
|
||||
|
||||
return ExecuteResult.success(result);
|
||||
}
|
||||
|
||||
/**
|
||||
* 查询可用的REMNAWAVE节点
|
||||
* 条件：
|
||||
* 1. nodeSoftType = REMNAWAVE
|
||||
* 2. enableFlag = 1 (开启状态)
|
||||
* 3. reportStatusTime 在超时阈值内（状态正常）
|
||||
*
|
||||
* @return 可用节点列表
|
||||
*/
|
||||
private List<NetNode> queryAvailableNodes() {
|
||||
return TenantHelper.ignore(() -> {
|
||||
LambdaQueryWrapper<NetNode> queryWrapper = new LambdaQueryWrapper<>();
|
||||
queryWrapper.eq(NetNode::getNodeSoftType, NetNodeSoftTypeEnum.REMNAWAVE.getNodeValue())
|
||||
.eq(NetNode::getEnableFlag, 1)
|
||||
.isNotNull(NetNode::getReportStatusTime);
|
||||
|
||||
List<NetNode> nodes = netNodeMapper.selectList(queryWrapper);
|
||||
|
||||
// 过滤掉状态超时的节点
|
||||
Date timeoutThreshold = new Date(System.currentTimeMillis() - NODE_STATUS_TIMEOUT_MS);
|
||||
nodes.removeIf(node ->
|
||||
node.getReportStatusTime() != null &&
|
||||
node.getReportStatusTime().before(timeoutThreshold)
|
||||
);
|
||||
|
||||
return nodes;
|
||||
});
|
||||
}
|
||||
|
||||
/**
|
||||
* 解析用户ID
|
||||
* username 是用户ID的字符串形式
|
||||
*
|
||||
* @param username 用户名字符串
|
||||
* @return 用户ID，解析失败返回null
|
||||
*/
|
||||
private Long parseUserId(String username) {
|
||||
if (username == null || username.isEmpty()) {
|
||||
return null;
|
||||
}
|
||||
try {
|
||||
return Long.parseLong(username);
|
||||
} catch (NumberFormatException e) {
|
||||
SnailJobLog.LOCAL.warn("remnaNodeUserStatsJob 解析用户ID失败: {}", username);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* 保存节点流量记录
|
||||
*
|
||||
* @param nodeId 节点ID
|
||||
* @param upNum 上传流量
|
||||
* @param downNum 下载流量
|
||||
* @param recordTime 记录时间
|
||||
*/
|
||||
private void saveNodeFlowRecord(Long nodeId, Long upNum, Long downNum, Date recordTime) {
|
||||
NetNodeFlowRecord record = new NetNodeFlowRecord();
|
||||
record.setId(IdGeneratorUtil.nextLongId());
|
||||
record.setNodeId(nodeId);
|
||||
record.setUpNum(upNum);
|
||||
record.setDownNum(downNum);
|
||||
record.setTotalNum(upNum + downNum);
|
||||
record.setCreateTime(recordTime);
|
||||
|
||||
netNodeFlowRecordMapper.insert(record);
|
||||
}
|
||||
|
||||
/**
|
||||
* 批量保存用户流量记录
|
||||
*
|
||||
* @param userFlowMap 用户流量汇总Map
|
||||
* @param recordTime 记录时间
|
||||
* @return 保存的记录数
|
||||
*/
|
||||
private int saveUserFlowRecords(Map<Long, UserFlowAggregate> userFlowMap, Date recordTime) {
|
||||
if (userFlowMap.isEmpty()) {
|
||||
return 0;
|
||||
}
|
||||
|
||||
List<NetUserFlowRecord> records = new ArrayList<>();
|
||||
for (Map.Entry<Long, UserFlowAggregate> entry : userFlowMap.entrySet()) {
|
||||
Long userId = entry.getKey();
|
||||
UserFlowAggregate flow = entry.getValue();
|
||||
|
||||
// 只记录有流量的用户
|
||||
if (flow.getUplink() > 0 || flow.getDownlink() > 0) {
|
||||
NetUserFlowRecord record = new NetUserFlowRecord();
|
||||
record.setId(IdGeneratorUtil.nextLongId());
|
||||
record.setUserId(userId);
|
||||
record.setUpNum(flow.getUplink());
|
||||
record.setDownNum(flow.getDownlink());
|
||||
record.setTotalNum(flow.getUplink() + flow.getDownlink());
|
||||
record.setCreateTime(recordTime);
|
||||
records.add(record);
|
||||
}
|
||||
}
|
||||
|
||||
// 批量插入
|
||||
if (!records.isEmpty()) {
|
||||
netUserFlowRecordMapper.insert(records, 1000);
|
||||
}
|
||||
|
||||
return records.size();
|
||||
}
|
||||
|
||||
/**
|
||||
* 用户流量汇总内部类
|
||||
*/
|
||||
@Data
|
||||
private static class UserFlowAggregate {
|
||||
private Long uplink;
|
||||
private Long downlink;
|
||||
|
||||
public UserFlowAggregate(Long uplink, Long downlink) {
|
||||
this.uplink = uplink;
|
||||
this.downlink = downlink;
|
||||
}
|
||||
|
||||
public void addUplink(Long value) {
|
||||
this.uplink += value;
|
||||
}
|
||||
|
||||
public void addDownlink(Long value) {
|
||||
this.downlink += value;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -121,7 +121,7 @@ public class NetNodeConfigServiceImpl implements INetNodeConfigService {
|
|||
private XrayConfig.Inbound buildUserInbound(NetNodeVo node, NetNodeConfigVo config) {
|
||||
XrayConfig.Inbound inbound = new XrayConfig.Inbound();
|
||||
inbound.setTag(node.getProtocolType() + "-inbound");
|
||||
inbound.setPort(node.getServerPort());
|
||||
inbound.setPort(node.getConnectionPort());
|
||||
inbound.setProtocol(node.getProtocolType().toLowerCase());
|
||||
|
||||
// 设置协议配置
|
||||
|
|
|
|||
Loading…
Reference in New Issue