Agent/tr-agent-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

merge into: Agent/tr-agent-server:main
Branches Tags
Agent/tr-agent-server:main Agent/tr-agent-server:v1.1
...
pull from: Agent/tr-agent-server:v1.1
Branches Tags
Agent/tr-agent-server:v1.1 Agent/tr-agent-server:main

15 Commits
main ... v1.1

Author SHA1 Message Date
gaoyutao 81181869da 区分配置文件 2026-03-30 17:59:48 +08:00
gaoyutao b52f56e2e9 新增业务流量上报 2026-03-10 17:51:35 +08:00
gaoyutao 7bd91aa14f 优化tcp解码 2026-03-03 17:46:11 +08:00
gaoyutao 5244f1d09b 流量补全消息改为顺序上报 2026-02-28 18:00:56 +08:00
gaoyutao 816f833ff5 增加采集内存详情方法 2026-02-25 17:54:48 +08:00
gaoyutao 9b395b2be7 定位并解决中文乱码问题 2026-01-27 19:09:21 +08:00
gaoyutao d579f85eee 增加网络上报重试，和tcpdump结果上报 2026-01-22 17:58:31 +08:00
gaoyutao f9b9f06390 增加磁盘IOPS上报方法 2026-01-15 18:15:37 +08:00
gaoyutao 75c5db6f91 增加pppoe状态上报 2025-12-30 18:21:52 +08:00
gaoyutao 8950c0ce3c 增加nett传输日志 2025-12-26 18:31:27 +08:00
gaoyutao 5fd0e913ae 增加frpc响应处理 2025-12-22 18:54:47 +08:00
qiminbao 2017bfe437 数据格式优化 2025-10-30 22:06:09 +08:00
qiminbao c8980142d8 增加多网IP探测上报 2025-10-28 14:26:41 +08:00
qiminbao a26981a92e v1.1初始化 2025-10-27 10:26:29 +08:00
qiminbao eb53489e90 v1.1初始化 2025-10-21 14:21:54 +08:00
18 changed files with 747 additions and 698 deletions
Expand all files Collapse all files
.gitignore
+46
View File
@@ -0,0 +1,46 @@
######################################################################
# Build Tools
.gradle
/build/
!gradle/wrapper/gradle-wrapper.jar
target/
!.mvn/wrapper/maven-wrapper.jar
######################################################################
# IDE
### STS ###
.apt_generated
.classpath
.factorypath
.project
.settings
.springBeans
### IntelliJ IDEA ###
.idea
*.iws
*.iml
*.ipr
### JRebel ###
rebel.xml
### NetBeans ###
nbproject/private/
build/*
nbbuild/
dist/
nbdist/
.nb-gradle/
######################################################################
# Others
*.log
*.xml.versionsBackup
*.swp
!*/build/*.java
!*/build/*.html
!*/build/*.xml
src/main/java/com/tongran/agent/server/controller/TcpService.java
+80 -91
View File
@@ -1,24 +1,13 @@
package com.tongran.agent.server.controller;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.tongran.agent.server.core.enums.MsgEnum;
import com.tongran.agent.server.core.session.SessionManager;
import com.tongran.agent.server.netty.MultiTargetNettyClient;
import com.tongran.agent.server.netty.annotation.AgentDispatcher;
import com.tongran.agent.server.netty.basics.AgentDispatcherManager;
import com.tongran.agent.server.netty.basics.AgentHandler;
import com.tongran.agent.server.netty.model.Message;
import com.tongran.agent.server.rockermq.AgentProducer;
import com.tongran.agent.server.rockermq.model.MqMsg;
import com.tongran.agent.server.utils.AssertLog;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
import java.util.Objects;
@Slf4j
@Service
@@ -28,8 +17,8 @@ public class TcpService {
@Resource
private AgentDispatcherManager agentDispatcherManager;
@Resource
private MultiTargetNettyClient client;
// @Resource
// private MultiTargetNettyClient client;
@Resource
private AgentProducer agentProducer;
@@ -41,88 +30,88 @@ public class TcpService {
public void create(String clientId){
boolean success1 = false;
// 动态创建多个不同IP和端口的连接
if(StringUtils.equals(clientId, "clientId007")){
success1 = client.createConnection(clientId, "127.0.0.1", 6610, 5);
}else{
success1 = client.createConnection(clientId, "127.0.0.1", 6710, 5);
}
System.out.println("当前活跃连接数: " + client.getActiveConnections());
if(success1){
// 连接建立后可以发送登录认证消息
long timestamp = System.currentTimeMillis();
JSONObject object = new JSONObject();
object.put("uuid",clientId);
object.put("timestamp",timestamp);
Message message = Message.builder().clientId(clientId).dataType("CREATE").data(object.toString()).build();
// 将对象转为 JSON 字符串
String json = JSON.toJSONString(message);
System.out.println("连接建立成功，发送登录认证="+json);
client.sendMessages(clientId, message);
}
// if(StringUtils.equals(clientId, "clientId007")){
// success1 = client.createConnection(clientId, "127.0.0.1", 6610, 5);
// }else{
// success1 = client.createConnection(clientId, "127.0.0.1", 6710, 5);
// }
// System.out.println("当前活跃连接数: " + client.getActiveConnections());
// if(success1){
// // 连接建立后可以发送登录认证消息
// long timestamp = System.currentTimeMillis();
// JSONObject object = new JSONObject();
// object.put("uuid",clientId);
// object.put("timestamp",timestamp);
// Message message = Message.builder().clientId(clientId).dataType("CREATE").data(object.toString()).build();
// // 将对象转为 JSON 字符串
// String json = JSON.toJSONString(message);
// System.out.println("连接建立成功，发送登录认证="+json);
// client.sendMessages(clientId, message);
// }
}
public void sendMessage(String message){
AssertLog.info("consumer==> received down message: {}", message);
try {
Message dto = JSON.parseObject(message, Message.class);
if(StringUtils.equals(dto.getDataType(), MsgEnum.注册.getValue())){
JSONObject jsonObject = JSONObject.parseObject(dto.getData());
String clientIp = "";
int clientPort = 0;
String switchBoard = "";
long timestamp = 0;
if(jsonObject.containsKey("clientIp")){
clientIp = jsonObject.getString("clientIp");
}
if(jsonObject.containsKey("clientPort")){
clientPort = jsonObject.getInteger("clientPort");
}
if(jsonObject.containsKey("switchBoard")){
switchBoard = jsonObject.getString("switchBoard");
}
if(jsonObject.containsKey("timestamp")){
timestamp = jsonObject.getLongValue("timestamp");
}
if(StringUtils.isBlank(clientIp) || clientPort == 0){
//返回提醒ip 端口 空
JSONObject json = new JSONObject();
json.put("clientId",dto.getClientId());
json.put("dataType",MsgEnum.注册应答.getValue());
JSONObject dataJson = new JSONObject();
dataJson.put("resCode",0);
dataJson.put("resMag","客户端IP、端口不能为空");
dataJson.put("timestamp", System.currentTimeMillis());
json.put("data",dataJson.toString());
MqMsg mqMsg = MqMsg.builder().topic("tongran_agent_down").content(json.toString()).build();
agentProducer.asyncSend(mqMsg);
return;
}
boolean success = client.createConnection(dto.getClientId(), clientIp, clientPort, 5);
if(success){
//连接成功
// 连接建立后可以发送登录认证消息
JSONObject object = new JSONObject();
object.put("timestamp", timestamp);
if(StringUtils.isNotBlank(switchBoard)){
object.put("switchBoard",switchBoard);
}
Message msg = Message.builder().clientId(dto.getClientId()).dataType(MsgEnum.注册.getValue()).data(object.toString()).build();
// 将对象转为 JSON 字符串
client.sendMessages(dto.getClientId(), msg);
}
} else {
AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value);
if (ObjectUtil.isNotEmpty(msgHandler)) {
String msg = msgHandler.downHandle(dto.getData(), dto.getClientId());
Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build();
if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) {
sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage);
}
}
}
} catch (Exception e) {
AssertLog.error("=====rocketmq-exception{}=====" + e.getMessage());
}
// try {
// Message dto = JSON.parseObject(message, Message.class);
// if(StringUtils.equals(dto.getDataType(), MsgEnum.注册.getValue())){
// JSONObject jsonObject = JSONObject.parseObject(dto.getData());
// String clientIp = "";
// int clientPort = 0;
// String switchBoard = "";
// long timestamp = 0;
// if(jsonObject.containsKey("clientIp")){
// clientIp = jsonObject.getString("clientIp");
// }
// if(jsonObject.containsKey("clientPort")){
// clientPort = jsonObject.getInteger("clientPort");
// }
// if(jsonObject.containsKey("switchBoard")){
// switchBoard = jsonObject.getString("switchBoard");
// }
// if(jsonObject.containsKey("timestamp")){
// timestamp = jsonObject.getLongValue("timestamp");
// }
// if(StringUtils.isBlank(clientIp) || clientPort == 0){
// //返回提醒ip 端口 空
// JSONObject json = new JSONObject();
// json.put("clientId",dto.getClientId());
// json.put("dataType",MsgEnum.注册应答.getValue());
// JSONObject dataJson = new JSONObject();
// dataJson.put("resCode",0);
// dataJson.put("resMag","客户端IP、端口不能为空");
// dataJson.put("timestamp", System.currentTimeMillis());
// json.put("data",dataJson.toString());
// MqMsg mqMsg = MqMsg.builder().topic("tongran_agent_down").content(json.toString()).build();
// agentProducer.asyncSend(mqMsg);
// return;
// }
// boolean success = client.createConnection(dto.getClientId(), clientIp, clientPort, 5);
// if(success){
// //连接成功
// // 连接建立后可以发送登录认证消息
// JSONObject object = new JSONObject();
// object.put("timestamp", timestamp);
// if(StringUtils.isNotBlank(switchBoard)){
// object.put("switchBoard",switchBoard);
// }
// Message msg = Message.builder().clientId(dto.getClientId()).dataType(MsgEnum.注册.getValue()).data(object.toString()).build();
// // 将对象转为 JSON 字符串
// client.sendMessages(dto.getClientId(), msg);
// }
// } else {
// AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value);
// if (ObjectUtil.isNotEmpty(msgHandler)) {
// String msg = msgHandler.downHandle(dto.getData(), dto.getClientId());
// Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build();
// if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) {
// sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage);
// }
// }
// }
// } catch (Exception e) {
// AssertLog.error("=====rocketmq-exception{}=====" + e.getMessage());
// }
}
src/main/java/com/tongran/agent/server/core/enums/MsgEnum.java
+24 -1
View File
@@ -11,11 +11,18 @@ import lombok.NoArgsConstructor;
@AllArgsConstructor
@NoArgsConstructor
public enum MsgEnum {
建立连接("CONNECT"),
建立连接应答("CONNECT_RSP"),
注册("REGISTER"),
注册应答("REGISTER_RSP"),
获取最新策略("GET_POLICY"),
获取最新策略应答("GET_POLICY_RSP"),
断开("DISCONNECT"),
断开应答("DISCONNECT_RSP"),
@@ -34,6 +41,10 @@ public enum MsgEnum {
网络上报("NET"),
网络上报重试("NET_RECOVER"),
业务网络上报("BUSINESS_NET"),
挂载上报("POINT"),
系统其他上报("OTHER_SYSTEM"),
@@ -68,7 +79,19 @@ public enum MsgEnum {
Agent版本更新("AGENT_VERSION_UPDATE"),
Agent版本更新应答("AGENT_VERSION_UPDATE_RSP");
Agent版本更新应答("AGENT_VERSION_UPDATE_RSP"),
修改frp配置文件应答("UPDATE_FRP_RSP"),
macvlan状态上报("MACVLAN_STATUS_RSP"),
内存详情上报("MEMORY_DETAILS"),
iops结果上报("IOPS_RESULT"),
tcpdump结果上报("TCPDUMP_RESULT"),
多网IP探测上报("NETWORK_DETECT");
private String value;
src/main/java/com/tongran/agent/server/netty/AgentNettyConfig.java
+22
View File
@@ -0,0 +1,22 @@
package com.tongran.agent.server.netty;
import com.tongran.agent.server.netty.config.BaseNettyConfig;
import lombok.Data;
import lombok.EqualsAndHashCode;
import lombok.ToString;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
/**
* Netty配置
*/
@Data
@Configuration
@ToString(callSuper = true)
@EqualsAndHashCode(callSuper = true)
@ConfigurationProperties(prefix = "tcp.netty.charge")
public class AgentNettyConfig extends BaseNettyConfig {
private static final long serialVersionUID = -4284214721383107291L;
}
src/main/java/com/tongran/agent/server/netty/AgentNettyServer.java
+56
View File
@@ -0,0 +1,56 @@
package com.tongran.agent.server.netty;
import com.tongran.agent.server.netty.handler.AgentDecoderHandler;
import com.tongran.agent.server.netty.handler.AgentDispatcherHandler;
import com.tongran.agent.server.netty.handler.AgentEncoderHandler;
import com.tongran.agent.server.netty.handler.TCPListenHandler;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.timeout.IdleStateHandler;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import javax.annotation.Resource;
@Configuration
public class AgentNettyServer extends BaseNettyServer {
@Resource
private AgentNettyConfig config;
@Resource
private TCPListenHandler tcpListenHandler;
@Resource
private AgentDecoderHandler decoderHandler;
@Resource
private AgentEncoderHandler encoderHandler;
@Resource
private AgentDispatcherHandler dispatcherHandler;
protected AgentNettyServer(AgentNettyConfig config) {
super(config);
}
@Bean(name = "NettyCharge", initMethod = "start", destroyMethod = "stop")
BaseNettyServer run() {
config.setHander(new ChannelInitializer<NioSocketChannel>() {
@Override
public void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new IdleStateHandler(config.readerIdleTime, config.writerIdleTime, config.allIdleTime));// 心跳
ch.pipeline().addLast(tcpListenHandler);// 监听器
//入栈
ch.pipeline().addLast(decoderHandler);//解码器
//出栈
ch.pipeline().addLast(encoderHandler);//加码器
// 业务分发是入站最后一个处理器, 出站第一个处理器, 位置要放在最后
ch.pipeline().addLast(businessGroup, dispatcherHandler);//业务分发 这里使用业务线程组
}
});
return new AgentNettyServer(config);
}
}
src/main/java/com/tongran/agent/server/netty/MultiTargetNettyClient.java
-301
View File
@@ -1,301 +0,0 @@
package com.tongran.agent.server.netty;
import com.tongran.agent.server.netty.handler.AgentDecoderHandler;
import com.tongran.agent.server.netty.handler.AgentDispatcherHandler;
import com.tongran.agent.server.netty.handler.AgentEncoderHandler;
import com.tongran.agent.server.netty.model.Message;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.AttributeKey;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
@Component
public class MultiTargetNettyClient {
// 连接标识属性键
private static final AttributeKey<String> CONNECTION_KEY = AttributeKey.valueOf("connectionKey");
// 连接管理器
private final Map<String, Channel> connectionMap = new ConcurrentHashMap<>();
private final Map<Channel, String> reverseConnectionMap = new ConcurrentHashMap<>();
private EventLoopGroup workerGroup;
@Resource
private AgentDecoderHandler decoderHandler;
@Resource
private AgentEncoderHandler encoderHandler;
@Resource
private AgentDispatcherHandler dispatcherHandler;
/**
* 动态创建TCP连接
* @param connectionKey 连接唯一标识（格式：ip:port 或自定义）
* @param host 目标主机
* @param port 目标端口
* @param timeout 超时时间（秒）
* @return 是否创建成功
*/
public boolean createConnection(String connectionKey, String host, int port, int timeout) {
if (workerGroup == null) {
workerGroup = new NioEventLoopGroup();
}
if (connectionMap.containsKey(connectionKey)) {
System.out.println("连接已存在: " + connectionKey);
return true;
}
Bootstrap bootstrap = new Bootstrap();
bootstrap.group(workerGroup)
.channel(NioSocketChannel.class)
.option(ChannelOption.TCP_NODELAY, true)
.option(ChannelOption.CONNECT_TIMEOUT_MILLIS, timeout * 1000)
.handler(new ChannelInitializer<SocketChannel>() {
@Override
protected void initChannel(SocketChannel ch) {
ChannelPipeline pipeline = ch.pipeline();
// 添加编解码器
pipeline.addLast(decoderHandler);
pipeline.addLast(encoderHandler);
// 添加心跳机制
// pipeline.addLast(new IdleStateHandler(0, 30, 0, TimeUnit.SECONDS));
// 添加自定义处理器
pipeline.addLast(dispatcherHandler);
}
});
try {
CountDownLatch latch = new CountDownLatch(1);
final boolean[] success = {false};
bootstrap.connect(host, port).addListener((ChannelFuture future) -> {
if (future.isSuccess()) {
Channel channel = future.channel();
connectionMap.put(connectionKey, channel);
reverseConnectionMap.put(channel, connectionKey);
success[0] = true;
System.out.println("连接建立成功: " + connectionKey + " -> " + host + ":" + port);
} else {
System.err.println("连接建立失败: " + connectionKey + " - " + future.cause().getMessage());
success[0] = false;
}
latch.countDown();
});
// 等待连接结果
return latch.await(timeout, TimeUnit.SECONDS) && success[0];
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
/**
* 向指定连接发送消息
* @param connectionKey 连接标识
* @param message 消息内容
* @return 是否发送成功
*/
public boolean sendMessage(String connectionKey, String message) {
Channel channel = connectionMap.get(connectionKey);
if (channel == null) {
System.err.println("连接不存在: " + connectionKey);
return false;
}
if (!channel.isActive()) {
System.err.println("连接已断开: " + connectionKey);
removeConnection(connectionKey);
return false;
}
try {
ChannelFuture future = channel.writeAndFlush(message).sync();
System.out.println("消息发送成功到: " + connectionKey + " - " + message);
return future.isSuccess();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
public boolean sendMessages(String connectionKey, Message message) {
Channel channel = connectionMap.get(connectionKey);
if (channel == null) {
System.err.println("连接不存在: " + connectionKey);
return false;
}
if (!channel.isActive()) {
System.err.println("连接已断开: " + connectionKey);
removeConnection(connectionKey);
return false;
}
try {
ChannelFuture future = channel.writeAndFlush(message).sync();
System.out.println("消息发送成功到: " + connectionKey + " - " + message);
return future.isSuccess();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
return false;
}
}
/**
* 批量发送消息到不同连接
* @param messages Map<连接标识, 消息内容>
*/
public void sendMessages(Map<String, String> messages) {
messages.forEach((connectionKey, message) -> {
new Thread(() -> {
if (sendMessage(connectionKey, message)) {
System.out.println("批量发送成功: " + connectionKey);
} else {
System.err.println("批量发送失败: " + connectionKey);
}
}).start();
});
}
/**
* 关闭指定连接
* @param connectionKey 连接标识
*/
public void closeConnection(String connectionKey) {
Channel channel = connectionMap.get(connectionKey);
if (channel != null) {
channel.close();
removeConnection(connectionKey);
System.out.println("连接已关闭: " + connectionKey);
}
}
/**
* 移除连接记录
* @param connectionKey 连接标识
*/
private void removeConnection(String connectionKey) {
Channel channel = connectionMap.remove(connectionKey);
if (channel != null) {
reverseConnectionMap.remove(channel);
}
}
/**
* 关闭所有连接
*/
public void shutdown() {
// 关闭所有连接
connectionMap.forEach((key, channel) -> {
if (channel.isActive()) {
channel.close();
}
});
connectionMap.clear();
reverseConnectionMap.clear();
// 关闭EventLoopGroup
if (workerGroup != null) {
workerGroup.shutdownGracefully();
}
System.out.println("所有连接已关闭");
}
/**
* 获取连接状态
* @return 当前活跃连接数
*/
public int getActiveConnections() {
return (int) connectionMap.values().stream()
.filter(Channel::isActive)
.count();
}
// /**
// * 客户端处理器
// */
// private class ClientHandler extends SimpleChannelInboundHandler<String> {
// @Override
// protected void channelRead0(ChannelHandlerContext ctx, String msg) {
// String connectionKey = ctx.channel().attr(CONNECTION_KEY).get();
// System.out.println("收到来自 " + connectionKey + " 的消息: " + msg);
// }
//
// @Override
// public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
// String connectionKey = ctx.channel().attr(CONNECTION_KEY).get();
// System.err.println("连接 " + connectionKey + " 发生异常: " + cause.getMessage());
// ctx.close();
// }
//
// @Override
// public void channelInactive(ChannelHandlerContext ctx) {
// String connectionKey = ctx.channel().attr(CONNECTION_KEY).get();
// System.out.println("连接断开: " + connectionKey);
// removeConnection(connectionKey);
// }
// }
/**
* 使用示例
*/
public static void main(String[] args) throws InterruptedException {
MultiTargetNettyClient client = new MultiTargetNettyClient();
try {
// 动态创建多个不同IP和端口的连接
boolean success1 = client.createConnection("server1", "127.0.0.1", 8080, 5);
boolean success2 = client.createConnection("server2", "192.168.1.100", 8081, 5);
boolean success3 = client.createConnection("server3", "10.0.0.50", 8082, 5);
// 等待连接建立
Thread.sleep(2000);
if (success1) {
client.sendMessage("server1", "Hello Server 1 from 8080");
}
if (success2) {
client.sendMessage("server2", "Hello Server 2 from 8081");
}
if (success3) {
client.sendMessage("server3", "Hello Server 3 from 8082");
}
// // 批量发送示例
// Map<String, String> batchMessages = Map.of(
// "server1", "Batch message to server 1",
// "server2", "Batch message to server 2",
// "server3", "Batch message to server 3"
// );
//
// client.sendMessages(batchMessages);
// 保持运行一段时间
Thread.sleep(5000);
System.out.println("当前活跃连接数: " + client.getActiveConnections());
} finally {
client.shutdown();
}
}
public Channel get(String connectionKey){
Channel channel = connectionMap.get(connectionKey);
return channel;
}
}
src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java
+172 -17
View File
@@ -2,7 +2,6 @@ package com.tongran.agent.server.netty.enpoint;
import com.alibaba.fastjson2.JSONObject;
import com.tongran.agent.server.core.enums.MsgEnum;
import com.tongran.agent.server.netty.MultiTargetNettyClient;
import com.tongran.agent.server.netty.annotation.AgentDispatcher;
import com.tongran.agent.server.netty.basics.AgentHandler;
import com.tongran.agent.server.netty.model.UpMsgResponse;
@@ -15,8 +14,6 @@ import javax.annotation.Resource;
@Component
public class AgentEndpoint {
@Resource
private MultiTargetNettyClient client;
@Resource
private AgentMqConfig agentMqConfig;
@@ -24,20 +21,66 @@ public class AgentEndpoint {
@Resource
private AgentProducer agentProducer;
@AgentDispatcher(msgId = MsgEnum.注册应答)
public class RegisterRspHandler implements AgentHandler {
@AgentDispatcher(msgId = MsgEnum.建立连接)
public class ConnectHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject json = new JSONObject();
json.put("resCode",1);
return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.建立连接应答.getValue())
.content(json.toString()).build();
}
}
@AgentDispatcher(msgId = MsgEnum.建立连接应答)
public class ConnectRspHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
return null;
}
}
@AgentDispatcher(msgId = MsgEnum.注册)
public class RegisterHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.注册应答.getValue());
jsonObject.put("dataType", MsgEnum.注册.getValue());
jsonObject.put("data", data);
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
agentProducer.asyncSend(mqMsg);
JSONObject json = new JSONObject();
json.put("resCode",1);
return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.注册应答.getValue())
.content(json.toString()).build();
return null;
}
}
@AgentDispatcher(msgId = MsgEnum.注册应答)
public class RegisterRspHandler implements AgentHandler {
@Override
public String downHandle(String data, String clientId) {
return data;
}
}
@AgentDispatcher(msgId = MsgEnum.获取最新策略)
public class GetPolicyHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.获取最新策略.getValue());
jsonObject.put("data", data);
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
agentProducer.asyncSend(mqMsg);
return null;
}
}
@AgentDispatcher(msgId = MsgEnum.获取最新策略应答)
public class GetPolicyRspHandler implements AgentHandler {
@Override
public String downHandle(String data, String clientId) {
return data;
}
}
@@ -64,9 +107,9 @@ public class AgentEndpoint {
if(json.containsKey("resCode")){
resCode = json.getInteger("resCode");
}
if(resCode == 1){
client.closeConnection(clientId);
}
// if(resCode == 1){
// client.closeConnection(clientId);
// }
return UpMsgResponse.builder().clientId(clientId).build();
}
}
@@ -171,7 +214,39 @@ public class AgentEndpoint {
.content(json.toString()).build();
}
}
@AgentDispatcher(msgId = MsgEnum.网络上报重试)
public class NetRecoverHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.网络上报重试.getValue());
jsonObject.put("data", data);
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
agentProducer.sendOrderly(mqMsg, clientId);
JSONObject json = new JSONObject();
json.put("resCode",1);
return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.采集上报应答.getValue())
.content(json.toString()).build();
}
}
@AgentDispatcher(msgId = MsgEnum.业务网络上报)
public class BusinessNetHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.业务网络上报.getValue());
jsonObject.put("data", data);
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
agentProducer.asyncSend(mqMsg);
JSONObject json = new JSONObject();
json.put("resCode",1);
return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.采集上报应答.getValue())
.content(json.toString()).build();
}
}
@AgentDispatcher(msgId = MsgEnum.挂载上报)
public class PointHandler implements AgentHandler {
@Override
@@ -350,14 +425,94 @@ public class AgentEndpoint {
if(json.containsKey("resCode")){
resCode = json.getInteger("resCode");
}
if(resCode == 1){
client.closeConnection(clientId);
}
// if(resCode == 1){
// client.closeConnection(clientId);
// }
return null;
}
}
@AgentDispatcher(msgId = MsgEnum.多网IP探测上报)
public class DetectNetworkHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.多网IP探测上报.getValue());
jsonObject.put("data", data);
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
agentProducer.asyncSend(mqMsg);
return null;
}
}
@AgentDispatcher(msgId = MsgEnum.修改frp配置文件应答)
public class UpdateFrpHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.修改frp配置文件应答.getValue());
jsonObject.put("data", data);
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
agentProducer.asyncSend(mqMsg);
return null;
}
}
@AgentDispatcher(msgId = MsgEnum.macvlan状态上报)
public class upMacvlanStatusHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.macvlan状态上报.getValue());
jsonObject.put("data", data);
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
agentProducer.asyncSend(mqMsg);
return null;
}
}
@AgentDispatcher(msgId = MsgEnum.iops结果上报)
public class iopsResultHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.iops结果上报.getValue());
jsonObject.put("data", data);
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
agentProducer.asyncSend(mqMsg);
return null;
}
}
@AgentDispatcher(msgId = MsgEnum.tcpdump结果上报)
public class tcpdumpResultHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.tcpdump结果上报.getValue());
jsonObject.put("data", data);
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
agentProducer.asyncSend(mqMsg);
return null;
}
}
@AgentDispatcher(msgId = MsgEnum.内存详情上报)
public class MemoryDetailsHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.内存详情上报.getValue());
jsonObject.put("data", data);
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
agentProducer.asyncSend(mqMsg);
return null;
}
}
}
src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java
+154 -91
View File
@@ -21,6 +21,9 @@ import io.netty.util.CharsetUtil;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
@Component
@@ -40,121 +43,181 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
AgentDispatcherManager agentDispatcherManager;
// 用来临时保留没有处理过的请求报文
private final Cache<String, String> lruCache = CacheUtil.newLRUCache(5000);
private final Cache<String, byte[]> byteCache = CacheUtil.newLRUCache(5000);
/**
* 消息解码器
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
// 检查是否为 ByteBuf（未解码的原始数据）
if (msg instanceof ByteBuf) {
ByteBuf byteBuf = (ByteBuf) msg;
String messages = byteBuf.toString(CharsetUtil.UTF_8); // 指定字符集解码
AssertLog.info("<<[up1]:[up-content]==>{}", messages);
boolean released = false; // 标记是否已释放
try {
// 使用字节数组读取
byte[] bytes = new byte[byteBuf.readableBytes()];
byteBuf.readBytes(bytes);
String content = "";
StringBuilder sb = new StringBuilder();
boolean isClear = false;
String tempMsg = lruCache.get(sessionManager.client(ctx));
tempMsg = tempMsg == null ? "" : tempMsg;
int tmpMsgSize = tempMsg.length();
String clientKey = sessionManager.client(ctx);
boolean startsWith = messages.startsWith("agent-client:");
boolean endsWith = messages.endsWith("@tong-ran");
String ms = messages.replaceAll("agent-client:","");
String[] arr_msg = ms.split("@tong-ran");
if(startsWith){
//判定是否整包
if(arr_msg.length == 1){
//判定是否拆包
if(endsWith){
lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0;
// sb.append(arr_msg[0]+"@tong-ran");
content = arr_msg[0]+"@tong-ran";
}else{
lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0;
// sb.append(arr_msg[0]);
content = arr_msg[0];
lruCache.put(sessionManager.client(ctx), content, DateUnit.SECOND.getMillis() * 5000);
// 1. 获取缓存的字节数组
byte[] cachedBytes = byteCache.get(clientKey);
// 2. 合并字节数组
byte[] mergedBytes;
if (cachedBytes != null && cachedBytes.length > 0) {
mergedBytes = new byte[cachedBytes.length + bytes.length];
System.arraycopy(cachedBytes, 0, mergedBytes, 0, cachedBytes.length);
System.arraycopy(bytes, 0, mergedBytes, cachedBytes.length, bytes.length);
// 合并后清除旧缓存
byteCache.remove(clientKey);
} else {
mergedBytes = bytes;
}
// 3. 尝试解码
String rawMessage = new String(mergedBytes, StandardCharsets.UTF_8);
// 4. 检查是否有乱码
boolean hasGarbledChars = false;
for (int i = 0; i < rawMessage.length(); i++) {
if (rawMessage.charAt(i) == '\ufffd') {
hasGarbledChars = true;
break;
}
}
//判定是否粘包
if(arr_msg.length > 1){
if(!endsWith){
lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0;
for (int i = 0; i < arr_msg.length; i++) {
if(i == arr_msg.length -1){
lruCache.put(sessionManager.client(ctx), arr_msg[i], DateUnit.SECOND.getMillis() * 5000);
}else{
content += arr_msg[i].replaceAll("agent-client:","")+"@tong-ran";
// sb.append(arr_msg[i].replaceAll("agent-client:","")+"@tong-ran");
}
}
}else{
lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0;
// sb.append(String.join("@tong-ran", arr_msg));
content = String.join("@tong-ran", arr_msg);
}
}
}
// content = sb.toString();
if (tmpMsgSize > 0) {
content = tempMsg + messages;
endsWith = content.endsWith("@tong-ran");
if(endsWith){
isClear = true;
}else{
lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0;
lruCache.put(sessionManager.client(ctx), content, DateUnit.SECOND.getMillis() * 5000);
if (hasGarbledChars) {
// 有乱码，缓存字节数组等待后续数据
byteCache.put(clientKey, mergedBytes, DateUnit.SECOND.getMillis() * 5000);
// 打印调试信息
int garbledIndex = rawMessage.indexOf('\ufffd');
AssertLog.info("<<[up1]:[bytes-cached]==>长度: {}, 乱码位置: {}, 预览: {}",
mergedBytes.length, garbledIndex, rawMessage);
byteBuf.release();
released = true;
return; // 跳过处理
}
}
AssertLog.info("<<[up2]:IP:{},[up-content]==>{}", sessionManager.client(ctx),content);
endsWith = content.endsWith("@tong-ran");
//判断是否是整包
if(endsWith){
content = content.replaceAll("agent-client:","");
String[] arr = content.split("@tong-ran");
try {
for (String message : arr) {
JSONObject jsonObject = JSONObject.parseObject(message);
String clientId = jsonObject.getString("clientId");
String dataType = jsonObject.getString("dataType");
String data = jsonObject.getString("data");
if(Objects.nonNull(agentDispatcherManager)){
AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
if (ObjectUtil.isNotEmpty(msgHandler)) {
AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", clientId, dataType, data);
UpMsgResponse response = msgHandler.upHandle(data, clientId );
if(Objects.nonNull(response)){
Message agentMessage = Message.builder().build();
agentMessage.setClientId(clientId);
agentMessage.setDataType(response.getDataType());
agentMessage.setData(response.getContent());
ctx.fireChannelRead(agentMessage);//传递到下一个handler
}
// 5. 没有乱码，处理完整消息
AssertLog.info("<<[up1]:[raw-content]==>{}", rawMessage);
// 6. 处理消息分割（原有逻辑）
// 这里不需要从缓存获取，因为已经在字节层面处理过了
String fullMessage = rawMessage;
// 7. 使用更精确的分割逻辑
List<String> completeMessages = new ArrayList<>();
StringBuilder currentMsg = new StringBuilder();
// 按消息边界分割
String[] parts = fullMessage.split("(?=agent-client:)");
for (String part : parts) {
if (part.isEmpty()) continue;
currentMsg.append(part);
String current = currentMsg.toString();
// 检查是否包含完整消息边界
if (current.contains("@tong-ran")) {
String[] messages = current.split("@tong-ran", -1); // 使用limit=-1保留空字符串
// 前面的都是完整消息
for (int i = 0; i < messages.length - 1; i++) {
String completeMsg = messages[i];
if (!completeMsg.isEmpty()) {
completeMessages.add(completeMsg);
}
}
// 最后一部分可能是未完成的消息
String lastPart = messages[messages.length - 1];
currentMsg = new StringBuilder(lastPart);
}
}catch (Exception e){
AssertLog.error("=====channelRead:{}=====" + e.getMessage());
}
// 8. 处理缓存逻辑
if (currentMsg.length() > 0) {
// 有未完成的消息，需要缓存
// 注意：这里缓存的是字符串，不是字节数组
String remainingMsg = currentMsg.toString();
if (remainingMsg.length() > 0) {
// 将未完成的字符串转为字节数组缓存
byte[] remainingBytes = remainingMsg.getBytes(StandardCharsets.UTF_8);
byteCache.put(clientKey, remainingBytes, DateUnit.SECOND.getMillis() * 5000);
}
} else {
// 所有消息都处理完成，清除缓存
byteCache.remove(clientKey);
}
// 9. 处理完整的消息
for (String message : completeMessages) {
processCompleteMessage(ctx, message);
}
} catch (Exception e) {
AssertLog.error("=====channelRead error:{}=====", e.getMessage(), e);
} finally {
if(!released && byteBuf.refCnt() > 0){
byteBuf.release(); // 确保资源释放
}
}
if (isClear) {
lruCache.remove(sessionManager.client(ctx));
}
byteBuf.release(); // 释放 ByteBuf 资源（重要！）
} else {
System.out.println("Unexpected message type: " + msg.getClass());
}
}
private void processCompleteMessage(ChannelHandlerContext ctx, String message) {
try {
// 清理消息格式
String cleanMessage = message.replaceAll("^agent-client:", "").trim();
if (cleanMessage.isEmpty()) return;
// 截断过长的日志
AssertLog.info("<<[up2]:IP:{},[clean-content]==>{}...",
sessionManager.client(ctx), cleanMessage);
// 解析JSON
JSONObject jsonObject = JSONObject.parseObject(cleanMessage);
String clientId = jsonObject.getString("clientId");
String dataType = jsonObject.getString("dataType");
String data = jsonObject.getString("data");
if (Objects.nonNull(agentDispatcherManager)) {
AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
if (ObjectUtil.isNotEmpty(msgHandler)) {
AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}",
clientId, dataType, data);
UpMsgResponse response = msgHandler.upHandle(data, clientId);
if (Objects.nonNull(response)) {
Message agentMessage = Message.builder()
.clientId(clientId)
.dataType(response.getDataType())
.data(response.getContent())
.build();
ctx.fireChannelRead(agentMessage); // 传递到下一个handler
}
}
}
} catch (Exception e) {
// 如果解析失败，可能是消息不完整，尝试缓存字节
try {
String clientKey = sessionManager.client(ctx);
// 将失败的消息转为字节数组缓存
byte[] failedBytes = message.getBytes(StandardCharsets.UTF_8);
byteCache.put(clientKey, failedBytes, DateUnit.SECOND.getMillis() * 5000);
AssertLog.warn("<<[parse-failed-cached]: 消息解析失败，已缓存字节，长度: {}", failedBytes.length);
} catch (Exception cacheEx) {
AssertLog.error("=====缓存失败消息时出错:{}=====", cacheEx.getMessage(), cacheEx);
}
AssertLog.error("=====processCompleteMessage error, message: {}, error: {}=====",
message, e.getMessage(), e);
}
}
@Override
src/main/java/com/tongran/agent/server/netty/handler/TCPListenHandler.java
+66
View File
@@ -0,0 +1,66 @@
package com.tongran.agent.server.netty.handler;
import com.tongran.agent.server.core.session.SessionManager;
import com.tongran.agent.server.service.AgentService;
import com.tongran.agent.server.utils.AssertLog;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.io.IOException;
/**
* @Description: TCP消息适配器
*/
@Component
@ChannelHandler.Sharable
public class TCPListenHandler extends ChannelInboundHandlerAdapter {
private final SessionManager sessionManager;
@Resource
private AgentService agentService;
public TCPListenHandler() {
this.sessionManager = SessionManager.getInstance();
}
@Override
public void channelActive(ChannelHandlerContext ctx) {
AssertLog.info("<<<<<<[终端连接]{}", ctx.channel().remoteAddress());
}
@Override
public void channelInactive(ChannelHandlerContext ctx) {
AssertLog.info(">>>>>>[断开连接]{}", sessionManager.client(ctx));
sessionManager.remove(ctx.channel());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable e) {
if (e instanceof IOException) {
AssertLog.info("<<<<<<[终端断开连接]{} {}", sessionManager.client(ctx), e.getMessage());
} else {
AssertLog.info(">>>>>>[消息处理异常]" + sessionManager.client(ctx), e);
}
}
@Override
public void userEventTriggered(ChannelHandlerContext ctx, Object evt) {
if (evt instanceof IdleStateEvent) {
IdleStateEvent event = (IdleStateEvent) evt;
IdleState state = event.state();
if (state == IdleState.READER_IDLE || state == IdleState.WRITER_IDLE || state == IdleState.ALL_IDLE) {
AssertLog.warn(">>>>>>[终端心跳超时]{} {}", state, sessionManager.client(ctx));
sessionManager.remove(ctx.channel());
ctx.close();
}
}
}
}
src/main/java/com/tongran/agent/server/netty/handler/UDPListenHandler.java
+24
View File
@@ -0,0 +1,24 @@
package com.tongran.agent.server.netty.handler;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.socket.DatagramPacket;
import org.springframework.stereotype.Component;
/**
* @Description: UDP消息适配器
*/
@Component
@ChannelHandler.Sharable
public class UDPListenHandler extends ChannelInboundHandlerAdapter {
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) {
DatagramPacket packet = (DatagramPacket) msg;
ByteBuf buf = packet.content();
buf.clear();// 暂未实现
}
}
src/main/java/com/tongran/agent/server/rockermq/AgentProducer.java
+15
View File
@@ -51,4 +51,19 @@ public class AgentProducer implements RocketMqService {
rocketMQTemplate.syncSend(msg.getTopic(), MessageBuilder.withPayload(msg.getContent()).build(), 30000, msg.getLevel());
}
// 新增顺序发送
@Override
public void sendOrderly(MqMsg mqMsg, String shardingKey) {
try {
AssertLog.info("sendOrderly发送消息:==>{}", mqMsg);
rocketMQTemplate.syncSendOrderly(
mqMsg.getTopic(),
MessageBuilder.withPayload(mqMsg.getContent()).build(),
shardingKey
);
} catch (Exception e) {
AssertLog.error("顺序发送消息失败", e);
}
}
}
src/main/java/com/tongran/agent/server/rockermq/RocketMqService.java
+7
View File
@@ -41,5 +41,12 @@ public interface RocketMqService {
* @param msg 发送消息实体类
*/
default void delayedSendOrderly(MqMsg msg){};
/**
* 发送顺序消息
* @param mqMsg
* @param shardingKey
*/
public void sendOrderly(MqMsg mqMsg, String shardingKey);
}
src/main/java/com/tongran/agent/server/rockermq/listener/AgentConsumerDownListener.java
+9 -122
View File
@@ -2,19 +2,13 @@ package com.tongran.agent.server.rockermq.listener;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.tongran.agent.server.core.enums.MsgEnum;
import com.tongran.agent.server.core.session.SessionManager;
import com.tongran.agent.server.netty.MultiTargetNettyClient;
import com.tongran.agent.server.netty.annotation.AgentDispatcher;
import com.tongran.agent.server.netty.basics.AgentDispatcherManager;
import com.tongran.agent.server.netty.basics.AgentHandler;
import com.tongran.agent.server.netty.model.Message;
import com.tongran.agent.server.rockermq.AgentProducer;
import com.tongran.agent.server.rockermq.config.AgentMqConfig;
import com.tongran.agent.server.rockermq.model.MqMsg;
import com.tongran.agent.server.utils.AssertLog;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
@@ -25,7 +19,7 @@ import java.util.Objects;
@Component
@ConditionalOnProperty(name = "rocketmq.enabled", havingValue = "true")
@RocketMQMessageListener(topic = "tongran_agent_down", consumerGroup = "tongran_agent_down_group")
@RocketMQMessageListener(topic = "tr_agent_down", consumerGroup = "tr_agent_down_group")
public class AgentConsumerDownListener implements RocketMQListener<String> {
protected final SessionManager sessionManager;
@@ -33,15 +27,6 @@ public class AgentConsumerDownListener implements RocketMQListener<String> {
@Resource
private AgentDispatcherManager agentDispatcherManager;
@Resource
private MultiTargetNettyClient client;
@Resource
private AgentMqConfig agentMqConfig;
@Resource
private AgentProducer agentProducer;
public AgentConsumerDownListener() {
this.sessionManager = SessionManager.getInstance();
}
@@ -51,112 +36,14 @@ public class AgentConsumerDownListener implements RocketMQListener<String> {
AssertLog.info("consumer==> received down message: {}", message);
try {
Message dto = JSON.parseObject(message, Message.class);
if(StringUtils.equals(dto.getDataType(), MsgEnum.注册.getValue())){
JSONObject jsonObject = JSONObject.parseObject(dto.getData());
String clientIp = "";
int clientPort = 0;
String switchBoard = "";
long timestamp = 0;
if(jsonObject.containsKey("clientIp")){
clientIp = jsonObject.getString("clientIp");
}
if(jsonObject.containsKey("clientPort")){
clientPort = jsonObject.getInteger("clientPort");
}
if(jsonObject.containsKey("switchBoard")){
switchBoard = jsonObject.getString("switchBoard");
}
if(jsonObject.containsKey("timestamp")){
timestamp = jsonObject.getLongValue("timestamp");
}
if(StringUtils.isBlank(clientIp) || clientPort == 0){
//返回提醒ip 端口 空
JSONObject json = new JSONObject();
json.put("clientId",dto.getClientId());
json.put("dataType",MsgEnum.注册应答.getValue());
JSONObject dataJson = new JSONObject();
dataJson.put("resCode",0);
dataJson.put("resMag","客户端IP、端口不能为空");
dataJson.put("timestamp", System.currentTimeMillis());
json.put("data",dataJson.toString());
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(json.toString()).build();
agentProducer.asyncSend(mqMsg);
return;
}
boolean success = client.createConnection(dto.getClientId(), clientIp, clientPort, 5);
if(success){
//连接成功
// 连接建立后可以发送登录认证消息
JSONObject object = new JSONObject();
object.put("timestamp", timestamp);
if(StringUtils.isNotBlank(switchBoard)){
String community = "";
String switchIp = "";
int switchPort = 0;
String netOID = "";
String moduleOID = "";
String mpuOID = "";
String pwrOID = "";
String fanOID = "";
String otherOID = "";
String filters = "";
JSONObject json = JSONObject.parseObject(switchBoard);
if(json.containsKey("community")){
community = json.getString("community");
}
if(json.containsKey("ip")){
switchIp = json.getString("ip");
}
if(json.containsKey("port")){
switchPort = json.getInteger("port");
}
if(json.containsKey("netOID")){
netOID = json.getString("netOID");
}
if(json.containsKey("moduleOID")){
moduleOID = json.getString("moduleOID");
}
if(json.containsKey("mpuOID")){
mpuOID = json.getString("mpuOID");
}
if(json.containsKey("pwrOID")){
pwrOID = json.getString("pwrOID");
}
if(json.containsKey("fanOID")){
fanOID = json.getString("fanOID");
}
if(json.containsKey("otherOID")){
otherOID = json.getString("otherOID");
}
if(json.containsKey("filters")){
filters = json.getString("filters");
}
JSONObject switchBoardJson = new JSONObject();
switchBoardJson.put("community",community);
switchBoardJson.put("ip",switchIp);
switchBoardJson.put("port",switchPort);
switchBoardJson.put("netOID",netOID);
switchBoardJson.put("moduleOID",moduleOID);
switchBoardJson.put("mpuOID",mpuOID);
switchBoardJson.put("pwrOID",pwrOID);
switchBoardJson.put("fanOID",fanOID);
switchBoardJson.put("otherOID",otherOID);
switchBoardJson.put("filters",filters);
object.put("switchBoard",switchBoardJson.toString());
AssertLog.info("交换机配置信息={}",switchBoardJson.toString());
}
Message msg = Message.builder().clientId(dto.getClientId()).dataType(MsgEnum.注册.getValue()).data(object.toString()).build();
// 将对象转为 JSON 字符串
client.sendMessages(dto.getClientId(), msg);
}
} else {
AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value);
if (ObjectUtil.isNotEmpty(msgHandler)) {
String msg = msgHandler.downHandle(dto.getData(), dto.getClientId());
Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build();
if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) {
sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage);
}
AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value);
if (ObjectUtil.isNotEmpty(msgHandler)) {
String msg = msgHandler.downHandle(dto.getData(), dto.getClientId());
Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build();
AssertLog.info("连接是否存在：{},clientId:{},session:{}",
Objects.nonNull(sessionManager.getSessionById(dto.getClientId())), dto.getClientId(), sessionManager.getSessionById(dto.getClientId()));
if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) {
sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage);
}
}
} catch (Exception e) {
src/main/java/com/tongran/agent/server/service/impl/AgentServiceImpl.java
+18 -20
View File
@@ -3,37 +3,35 @@ package com.tongran.agent.server.service.impl;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.tongran.agent.server.core.vo.RegisterVO;
import com.tongran.agent.server.netty.MultiTargetNettyClient;
import com.tongran.agent.server.netty.model.Message;
import com.tongran.agent.server.service.AgentService;
import org.springframework.stereotype.Service;
import javax.annotation.Resource;
@Service
public class AgentServiceImpl implements AgentService {
@Resource
private MultiTargetNettyClient client;
// @Resource
// private MultiTargetNettyClient client;
@Override
public boolean createConn(String clientId, String host, int port, int timeout) {
// 动态创建不同IP和端口的连接
boolean success = client.createConnection(clientId, host, port, timeout);
if(success){
// 连接建立后可以发送建立消息
long timestamp = System.currentTimeMillis();
JSONObject object = new JSONObject();
object.put("clientId",clientId);
object.put("timestamp",timestamp);
Message message = Message.builder().clientId(clientId).dataType("CREATE").data(object.toString()).build();
// 将对象转为 JSON 字符串
String json = JSON.toJSONString(message);
System.out.println("连接建立成功，发送建立消息="+json);
client.sendMessages(clientId, message);
}
return success;
// boolean success = client.createConnection(clientId, host, port, timeout);
// if(success){
// // 连接建立后可以发送建立消息
// long timestamp = System.currentTimeMillis();
// JSONObject object = new JSONObject();
// object.put("clientId",clientId);
// object.put("timestamp",timestamp);
// Message message = Message.builder().clientId(clientId).dataType("CREATE").data(object.toString()).build();
// // 将对象转为 JSON 字符串
// String json = JSON.toJSONString(message);
// System.out.println("连接建立成功，发送建立消息="+json);
// client.sendMessages(clientId, message);
// }
// return success;
return false;
}
@Override
@@ -52,6 +50,6 @@ public class AgentServiceImpl implements AgentService {
// 将对象转为 JSON 字符串
String json = JSON.toJSONString(message);
System.out.println("发送注册消息="+json);
client.sendMessages(registerVO.getClientId(), message);
// client.sendMessages(registerVO.getClientId(), message);
}
}
src/main/java/com/tongran/agent/server/utils/ClientExample.java
+1 -1
View File
@@ -40,7 +40,7 @@ public class ClientExample {
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
manager.shutdown();
// manager.shutdown();
}
}
}
src/main/java/com/tongran/agent/server/utils/EnhancedConnectionManager.java
+36 -37
View File
@@ -1,6 +1,5 @@
package com.tongran.agent.server.utils;
import com.tongran.agent.server.netty.MultiTargetNettyClient;
import com.tongran.agent.server.netty.config.ConnectionConfig;
import java.util.HashMap;
@@ -12,11 +11,11 @@ import java.util.concurrent.ConcurrentHashMap;
* 增强版连接管理器
*/
public class EnhancedConnectionManager {
private final MultiTargetNettyClient client;
// private final MultiTargetNettyClient client;
private final Map<String, ConnectionConfig> connectionConfigs;
public EnhancedConnectionManager() {
this.client = new MultiTargetNettyClient();
// this.client = new MultiTargetNettyClient();
this.connectionConfigs = new ConcurrentHashMap<>();
}
@@ -26,18 +25,18 @@ public class EnhancedConnectionManager {
public Map<String, Boolean> createConnections(List<ConnectionConfig> configs) {
Map<String, Boolean> results = new HashMap<>();
configs.forEach(config -> {
boolean success = client.createConnection(
config.getConnectionKey(),
config.getHost(),
config.getPort(),
config.getTimeout()
);
if (success) {
connectionConfigs.put(config.getConnectionKey(), config);
}
results.put(config.getConnectionKey(), success);
});
// configs.forEach(config -> {
// boolean success = client.createConnection(
// config.getConnectionKey(),
// config.getHost(),
// config.getPort(),
// config.getTimeout()
// );
// if (success) {
// connectionConfigs.put(config.getConnectionKey(), config);
// }
// results.put(config.getConnectionKey(), success);
// });
return results;
}
@@ -47,21 +46,21 @@ public class EnhancedConnectionManager {
*/
public void routeMessageByType(String messageType, String message) {
// 这里可以根据业务逻辑决定发送到哪个连接
switch (messageType) {
case "TYPE_A":
client.sendMessage("server1", "[TYPE_A] " + message);
break;
case "TYPE_B":
client.sendMessage("server2", "[TYPE_B] " + message);
break;
case "TYPE_C":
client.sendMessage("server3", "[TYPE_C] " + message);
break;
default:
// 广播到所有连接
connectionConfigs.keySet().forEach(key ->
client.sendMessage(key, "[BROADCAST] " + message));
}
// switch (messageType) {
// case "TYPE_A":
// client.sendMessage("server1", "[TYPE_A] " + message);
// break;
// case "TYPE_B":
// client.sendMessage("server2", "[TYPE_B] " + message);
// break;
// case "TYPE_C":
// client.sendMessage("server3", "[TYPE_C] " + message);
// break;
// default:
// // 广播到所有连接
// connectionConfigs.keySet().forEach(key ->
// client.sendMessage(key, "[BROADCAST] " + message));
// }
}
/**
@@ -69,14 +68,14 @@ public class EnhancedConnectionManager {
*/
public Map<String, Boolean> getConnectionStatus() {
Map<String, Boolean> status = new HashMap<>();
connectionConfigs.forEach((key, config) -> {
// 这里可以添加更详细的状态检查
status.put(key, client.sendMessage(key, "PING"));
});
// connectionConfigs.forEach((key, config) -> {
// // 这里可以添加更详细的状态检查
// status.put(key, client.sendMessage(key, "PING"));
// });
return status;
}
public void shutdown() {
client.shutdown();
}
// public void shutdown() {
// client.shutdown();
// }
}
src/main/resources/application-dev.yml
+13 -13
View File
@@ -1,5 +1,5 @@
server:
port: 7012
port: -1
servlet:
context-path: /tr-agent-server
@@ -12,21 +12,21 @@ knife4j:
logging:
file:
# path: /usr/local/tongran/logs
path: /data/tr-agent-server/logs
path: /usr/local/tongran_server/logs
# path: D:/job/agent-logs/logs
rocketmq:
#测试环境
#name-server: 172.16.15.52:9876
#生产环境
name-server: 172.16.15.103:9876
enabled: true
netty:
server:
# host: 172.16.15.103
host: 127.0.0.1
port: 6610
client:
client-id: client-001
reconnect-interval: 5
maxReconnectAttempts: 10 # 最大重连次数
initialReconnectDelay: 1000 # 初始重连延迟(毫秒)
maxReconnectDelay: 30000 # 最大重连延迟(毫秒)
tcp:
netty:
charge:
enable: true
name: AGENT-SERVER-服务
port: 6620
readerIdleTime: 300
src/main/resources/application.yml
+4 -4
View File
@@ -34,10 +34,10 @@ rocketmq:
producer:
group: tongran-group
topic: default_topic
agent-group: tongran_agent_up_group
agent-topic: tongran_agent_up
agent-group: tr_agent_up_group
agent-topic: tr_agent_up
consumer:
group: tongran-group
topic: default_topic
agent-group: tongran_agent_down_group
agent-topic: tongran_agent_down
agent-group: tr_agent_down_group
agent-topic: tr_agent_down