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

开发frpc接口、增加网卡绑定触发逻辑。

Browse Source
This commit is contained in:
gaoyutao
2025-12-23 18:55:27 +08:00
parent 1af9d3fbc5
commit 5eda46802e
9 changed files with 120 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/RmFrpcConfigManageController.java
-5
View File
@@ -47,7 +47,6 @@ public class RmFrpcConfigManageController extends BaseController
/**
* 导出FRPC配置管理列表
*/
@RequiresPermissions("rocketmq:frpcConfigManage:export")
@Log(title = "FRPC配置管理", businessType = BusinessType.EXPORT)
@PostMapping("/export")
public void export(HttpServletResponse response, @RequestBody RmFrpcConfigManage rmFrpcConfigManage)
@@ -61,7 +60,6 @@ public class RmFrpcConfigManageController extends BaseController
/**
* 获取FRPC配置管理详细信息
*/
@RequiresPermissions("rocketmq:frpcConfigManage:query")
@GetMapping(value = "/{id}")
public AjaxResult getInfo(@PathVariable("id") Long id)
{
@@ -71,7 +69,6 @@ public class RmFrpcConfigManageController extends BaseController
/**
* 新增FRPC配置管理
*/
@RequiresPermissions("rocketmq:frpcConfigManage:add")
@Log(title = "FRPC配置管理", businessType = BusinessType.INSERT)
@PostMapping
public AjaxResult add(@RequestBody RmFrpcConfigManage rmFrpcConfigManage)
@@ -82,7 +79,6 @@ public class RmFrpcConfigManageController extends BaseController
/**
* 修改FRPC配置管理
*/
@RequiresPermissions("rocketmq:frpcConfigManage:edit")
@Log(title = "FRPC配置管理", businessType = BusinessType.UPDATE)
@PutMapping
public AjaxResult edit(@RequestBody RmFrpcConfigManage rmFrpcConfigManage)
@@ -93,7 +89,6 @@ public class RmFrpcConfigManageController extends BaseController
/**
* 删除FRPC配置管理
*/
@RequiresPermissions("rocketmq:frpcConfigManage:remove")
@Log(title = "FRPC配置管理", businessType = BusinessType.DELETE)
@DeleteMapping("/{ids}")
public AjaxResult remove(@PathVariable Long[] ids)
tongran-rocketmq/src/main/java/com/tongran/rocketmq/controller/RmFrpcPortMappingController.java
+1 -1
View File
@@ -20,6 +20,7 @@ import java.util.List;
*/
@RestController
@RequestMapping("/frpcPortMapping")
@RequiresPermissions("rocketmq:frpcConfigManage")
public class RmFrpcPortMappingController extends BaseController
{
@Autowired
@@ -28,7 +29,6 @@ public class RmFrpcPortMappingController extends BaseController
/**
* 查询FRPC端口映射配置列表
*/
@RequiresPermissions("rocketmq:frpcPortMapping:list")
@PostMapping("/list")
public AjaxResult list(@RequestBody RmFrpcPortMapping rmFrpcPortMapping)
{
tongran-rocketmq/src/main/java/com/tongran/rocketmq/domain/vo/FrpRspVo.java
+1 -1
View File
@@ -9,7 +9,7 @@ import java.time.Instant;
@JsonIgnoreProperties(ignoreUnknown = true)
public class FrpRspVo {
/**
* 状态码，0、失败；1、成功
* 状态码，0、失败；1、成功；2、端口冲突；3、frpc状态上报；
*/
private Integer resCode;
/**
tongran-rocketmq/src/main/java/com/tongran/rocketmq/handler/MessageHandler.java
+65 -2
View File
@@ -53,6 +53,9 @@ public class MessageHandler {
String HEARTBEAT_RECOVERY_COUNT_PREFIX = "heartbeat:recovery:count:";
String HEARTBEAT_COUNT_PREFIX = "heartbeat:count:";
// Redis key定义
private static final String USED_PORTS_KEY = "frpc:used_ports";
private static final long HEARTBEAT_TIMEOUT = 30000; // 3分钟超时
@@ -149,12 +152,44 @@ public class MessageHandler {
RmFrpcConfigManage configManage = new RmFrpcConfigManage();
configManage.setFrpcRemotePort(rsp.getRemotePort());
rmFrpcConfigManageService.addFrpConnect(configManage);
} else if(rsp.getResCode() == 3){
// frpc状态上报
RmFrpcConfigManage frpcData = new RmFrpcConfigManage();
frpcData.setClientId(message.getClientId());
frpcData.setFrpcName(rsp.getName());
frpcData.setFrpcStatus(rsp.getIsRunning());
frpcData.setFrpcLocalPort(rsp.getLocalPort());
frpcData.setFrpcLocalIp(rsp.getLocalIP());
frpcData.setFrpcCreateTime(createTime);
frpcData.setFrpcRemotePort(rsp.getRemotePort());
frpcData.setFrpcServerAddr(rsp.getServerAddr());
frpcData.setFrpcServerPort(rsp.getServerPort());
rmFrpcConfigManageService.insertRmFrpcConfigManage(frpcData);
} else{
if(rsp.getRemotePort() != null){
// 生成失败 ，释放端口资源
removeUsedPort(Integer.parseInt(rsp.getRemotePort()));
}
log.error("生成连接失败：{}",rsp.getResMsg());
}
}
}
/**
* 添加端口到已用集合
*/
public boolean addUsedPort(int port) {
Long result = redisTemplate.opsForSet().add(USED_PORTS_KEY, String.valueOf(port));
return result != null && result > 0;
}
/**
* 从已用集合中移除端口
*/
public boolean removeUsedPort(int port) {
Long result = redisTemplate.opsForSet().remove(USED_PORTS_KEY, String.valueOf(port));
return result != null && result > 0;
}
/**
* 保存网卡信息
* @param message
@@ -1070,6 +1105,20 @@ public class MessageHandler {
}
boolean isSingleInterface = networkInfoList.size() == 1;
boolean isMultiInterface = networkInfoList.size() > 1;
// 查询数据库中当前的网卡数量
RmNetworkInterface countQuery = new RmNetworkInterface();
countQuery.setClientId(clientId);
countQuery.setNewFlag(1);
List<RmNetworkInterface> currentInterfaces = rmNetworkInterfaceService.selectRmNetworkInterfaceList(countQuery);
int currentInterfaceCount = currentInterfaces.size();
// 判断网卡数量是否发生变化
boolean interfaceCountChanged = false;
if (!isRegister && currentInterfaceCount != networkInfoList.size()) {
interfaceCountChanged = true;
}
for (NetworkInfo networkInfo : networkInfoList) {
List<NetworkInfo> childList = networkInfo.getSubInterfaces();
@@ -1101,13 +1150,20 @@ public class MessageHandler {
rmNetworkInterfaceChildService.insertRmNetworkInterfaceChild(insertChild);
}
}
// 如果网卡数量有变动，需要更新网卡绑定状态
if (!isRegister && interfaceCountChanged) {
RmResourceRegistrationRemote updateParam = new RmResourceRegistrationRemote();
updateParam.setClientId(clientId);
updateParam.setMultiPublicIpStatus("0");
remoteRevenueConfigService.updateStatusByResource(updateParam, SecurityConstants.INNER);
}
} else {
// 更新网卡信息
RmNetworkInterface oldInterfaceMsg = exits.get(0);
cleanChildOldRecords(clientId, networkInfo.getMac());
// 判断是否需要创建新记录
boolean needCreateNew = !StringUtils.equals(networkInfo.getName(), oldInterfaceMsg.getInterfaceName())
|| !StringUtils.equals(networkInfo.getGateway(), oldInterfaceMsg.getGateway());
boolean needCreateNew = (!StringUtils.equals(networkInfo.getName(), oldInterfaceMsg.getInterfaceName())
|| !StringUtils.equals(networkInfo.getGateway(), oldInterfaceMsg.getGateway())) && StringUtils.equals(networkInfo.getMac(), oldInterfaceMsg.getMacAddress());
if(childList != null && !childList.isEmpty()){
for (NetworkInfo info : childList) {
RmNetworkInterfaceChild insertChild = new RmNetworkInterfaceChild();
@@ -1130,6 +1186,13 @@ public class MessageHandler {
RmNetworkInterface insertData = new RmNetworkInterface();
setNetworkInterfaceData(insertData, networkInfo, clientId);
rmNetworkInterfaceService.insertRmNetworkInterface(insertData);
// 如果网卡名称有变化，需要更新网卡绑定状态
if(!isRegister && !StringUtils.equals(networkInfo.getName(), oldInterfaceMsg.getInterfaceName())){
RmResourceRegistrationRemote updateParam = new RmResourceRegistrationRemote();
updateParam.setClientId(clientId);
updateParam.setMultiPublicIpStatus("0");
remoteRevenueConfigService.updateStatusByResource(updateParam, SecurityConstants.INNER);
}
} else {
// 更新现有记录
updateNetworkInterface(networkInfo, clientId, oldInterfaceMsg, isSingleInterface, isRegister);
tongran-rocketmq/src/main/java/com/tongran/rocketmq/service/impl/RmFrpcConfigManageServiceImpl.java
+34 -3
View File
@@ -159,10 +159,12 @@ public class RmFrpcConfigManageServiceImpl implements IRmFrpcConfigManageService
@Override
public int addFrpConnect(RmFrpcConfigManage rmFrpcConfigManage) {
rmFrpcConfigManage.setFrpcConnectionStatus(1L);
int rows = rmFrpcConfigManageMapper.insertRmFrpcConfigManage(rmFrpcConfigManage);
// 查询frps配置
List<RmFrpcPortMapping> portMappingList = rmFrpcPortMappingMapper.selectRmFrpcPortMappingList(new RmFrpcPortMapping());
Integer startRemotePort = rmFrpcConfigManage.getStartRemotePort();
Integer endRemotePort = rmFrpcConfigManage.getEndRemotePort();
Integer startRemotePort = null;
Integer endRemotePort = null;
String serverPort = null;
String serverAddr = null;
if(portMappingList != null && !portMappingList.isEmpty()){
@@ -181,7 +183,7 @@ public class RmFrpcConfigManageServiceImpl implements IRmFrpcConfigManageService
addUsedPort(Integer.parseInt(rmFrpcConfigManage.getFrpcRemotePort()));
}
// 获取一个可用的随机端口
int availablePort = getRandomAvailablePort(startRemotePort, endRemotePort);
int availablePort = getRandomAvailablePort(clientId, startRemotePort, endRemotePort);
if (availablePort != -1) {
// 将端口标记为已使用
@@ -243,6 +245,35 @@ public class RmFrpcConfigManageServiceImpl implements IRmFrpcConfigManageService
return -1; // 没有可用端口
}
public int getRandomAvailablePort(String clientId, int startPort, int endPort) {
// 1. 先清理这个客户端可能存在的旧端口
cleanupClientOldPorts(clientId);
// 2. 获取新端口
int newPort = getRandomAvailablePort(startPort, endPort);
// 3. 记录客户端与端口的关联
if (newPort != -1) {
redisTemplate.opsForHash().put(
"client:port:mapping",
clientId,
String.valueOf(newPort)
);
}
return newPort;
}
/**
* 清理客户端旧端口
*/
private void cleanupClientOldPorts(String clientId) {
Object oldPortObj = redisTemplate.opsForHash().get("client:port:mapping", clientId);
if (oldPortObj != null) {
int oldPort = Integer.parseInt(oldPortObj.toString());
removeUsedPort(oldPort);
}
}
/**
* 判断端口是否可用（只检查Redis中是否存在）