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

初始化

Browse Source
This commit is contained in:
qiminbao
2025-09-10 15:41:46 +08:00
commit afecffc5db
42 changed files with 2949 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pom.xml
+98
View File
@@ -0,0 +1,98 @@
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.5.6</version>
<relativePath/> <!-- lookup parent from repository -->
</parent>
<groupId>com.tongran.agent</groupId>
<artifactId>tr-agent-server</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>tr-agent-server</name>
<description>tr-agent-server</description>
<properties>
<java.version>1.8</java.version>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-aop</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-validation</artifactId>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<dependency>
<groupId>io.swagger.core.v3</groupId>
<artifactId>swagger-annotations</artifactId>
<version>2.2.19</version> <!-- 请检查并使用最新版本 -->
</dependency>
<dependency>
<groupId>cn.hutool</groupId>
<artifactId>hutool-all</artifactId>
<version>5.7.11</version>
</dependency>
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
</dependency>
<dependency>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
<version>4.1.86.Final</version>
</dependency>
<dependency>
<groupId>com.alibaba.fastjson2</groupId>
<artifactId>fastjson2</artifactId>
<version>2.0.31</version>
</dependency>
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.2.3</version>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</project>
src/main/java/com/tongran/agent/server/TrAgentServerApplication.java
+13
View File
@@ -0,0 +1,13 @@
package com.tongran.agent.server;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
@SpringBootApplication
public class TrAgentServerApplication {
public static void main(String[] args) {
SpringApplication.run(TrAgentServerApplication.class, args);
}
}
src/main/java/com/tongran/agent/server/controller/TcpController.java
+33
View File
@@ -0,0 +1,33 @@
package com.tongran.agent.server.controller;
import com.alibaba.fastjson2.JSON;
import com.tongran.agent.server.netty.model.Message;
import org.springframework.validation.annotation.Validated;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RestController;
import javax.annotation.Resource;
@RestController
public class TcpController {
@Resource
private TcpService tcpService;
@PostMapping("/create")
public void create(@Validated @RequestBody Message dto) {
tcpService.create(dto.getClientId());
}
@PostMapping("/send")
public void sendMessage(@Validated @RequestBody Message dto) {
String json = JSON.toJSONString(dto);
// ScriptPolicy policy = JSON.parseObject(dto.getData(), ScriptPolicy.class);
// System.out.println(policy.getFiles().get(0).getFileData());
tcpService.sendMessage(json);
}
}
src/main/java/com/tongran/agent/server/controller/TcpService.java
+133
View File
@@ -0,0 +1,133 @@
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.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
public class TcpService {
protected final SessionManager sessionManager;
@Resource
private AgentDispatcherManager agentDispatcherManager;
@Resource
private MultiTargetNettyClient client;
public TcpService() {
this.sessionManager = SessionManager.getInstance();
}
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);
}
}
public void sendMessage(String message){
AssertLog.info("consumer==> received down message: {}", message);
try {
Message dto = JSON.parseObject(message, Message.class);
if(Objects.nonNull(dto)){
if(StringUtils.equals(dto.getDataType(), MsgEnum.注册.getValue())){
JSONObject jsonObject = JSONObject.parseObject(dto.getData());
String clientIp = "";
int port = 0;
long timestamp = 0;
if(jsonObject.containsKey("clientIp")){
clientIp = jsonObject.getString("clientIp");
}
if(jsonObject.containsKey("port")){
port = jsonObject.getInteger("port");
}
if(jsonObject.containsKey("timestamp")){
timestamp = jsonObject.getLongValue("timestamp");
}
if(StringUtils.isBlank(clientIp) || port == 0){
//返回提醒ip 端口 空
return;
}
boolean success = client.createConnection(dto.getClientId(), clientIp, port, 5);
if(success){
String community = "";
String ip = "";
int switchPort = 0;
String oids = "";
String params = "";
if(jsonObject.containsKey("switchBoard")){
String switchBoard = jsonObject.getString("switchBoard");
jsonObject = JSONObject.parseObject(switchBoard);
community = jsonObject.getString("community");
ip = jsonObject.getString("ip");
switchPort = jsonObject.getInteger("port");
oids = jsonObject.getString("oids");
params = jsonObject.getString("params");
}
//连接成功
// 连接建立后可以发送登录认证消息
JSONObject object = new JSONObject();
object.put("clientId",dto.getClientId());
object.put("timestamp",timestamp);
JSONObject switchBoard = new JSONObject();
switchBoard.put("community", community);
switchBoard.put("ip", ip);
switchBoard.put("port", switchPort);
switchBoard.put("oids", oids);
switchBoard.put("params", params);
object.put("switchBoard",switchBoard.toString());
Message msg = Message.builder().clientId(dto.getClientId()).dataType(MsgEnum.注册.getValue()).data(object.toString()).build();
// 将对象转为 JSON 字符串
client.sendMessages(dto.getClientId(), msg);
}else{
//连接失败
}
} 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/config/GlobalConfig.java
+17
View File
@@ -0,0 +1,17 @@
package com.tongran.agent.server.core.config;
/**
* 全局配置类
*/
public class GlobalConfig {
/**
* 交换机
*/
public static String SWITCH_COMMUNITY; //团名
public static String SWITCH_IP; //交换机IP
public static int SWITCH_PORT; //交换机端口
public static String[] SWITCH_OID; //交换机OID
}
src/main/java/com/tongran/agent/server/core/enums/MsgEnum.java
+73
View File
@@ -0,0 +1,73 @@
package com.tongran.agent.server.core.enums;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.NoArgsConstructor;
/**
* 控制码
*/
@Getter
@AllArgsConstructor
@NoArgsConstructor
public enum MsgEnum {
注册("REGISTER"),
注册应答("REGISTER_RSP"),
断开("DISCONNECT"),
断开应答("DISCONNECT_RSP"),
心跳上报("HEARTBEAT"),
CPU上报("CPU"),
磁盘上报("DISK"),
容器上报("DOCKER"),
内存上报("MEMORY"),
网络上报("NET"),
挂载上报("POINT"),
系统其他上报("OTHER_SYSTEM"),
交换机上报("SWITCHBOARD"),
开启系统采集("SYSTEM_COLLECT_START"),
开启系统采集应答("SYSTEM_COLLECT_START_RSP"),
关闭系统采集("SYSTEM_COLLECT_STOP"),
关闭系统采集应答("SYSTEM_COLLECT_STOP_RSP"),
开启交换机采集("SWITCH_COLLECT_START"),
开启交换机采集应答("SWITCH_COLLECT_START_RSP"),
关闭交换机采集("SWITCH_COLLECT_STOP"),
关闭交换机采集应答("SWITCH_COLLECT_STOP_RSP"),
告警设置("ALARM_SET"),
告警设置应答("ALARM_SET_RSP"),
执行脚本策略("SCRIPT_POLICY"),
执行脚本策略应答("SCRIPT_POLICY_RSP"),
Agent版本更新("AGENT_VERSION_UPDATE"),
Agent版本更新应答("AGENT_VERSION_UPDATE_RSP");
private String value;
}
src/main/java/com/tongran/agent/server/core/session/Session.java
+85
View File
@@ -0,0 +1,85 @@
package com.tongran.agent.server.core.session;
import io.netty.channel.ChannelHandlerContext;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.Accessors;
import java.io.Serializable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
/**
* SESSION
*
*/
@Data
@Builder
@Accessors(chain = true)
@AllArgsConstructor
@NoArgsConstructor
public class Session implements Serializable {
private static final long serialVersionUID = 1L;
/**
* sessionId
*/
private String sessionId;
/**
* 客户端唯一标识
*/
private String clientId;
/**
* Seesion额外参数
*/
private ConcurrentHashMap<String,Object> extraMap;
/**
* 是否鉴权
*/
@Builder.Default
private boolean isAuthenticated = false;
@Builder.Default
private long createTime = System.currentTimeMillis();
@Builder.Default
private long updateTime = System.currentTimeMillis();
/**
* 消息渠道
*/
private ChannelHandlerContext channel;
@Builder.Default
private AtomicInteger serialNo = new AtomicInteger(0);
public static String buildSessionId(ChannelHandlerContext channel) {
return channel.channel().id().asLongText();
}
public static Session buildSession(ChannelHandlerContext channel, String clientId) {
return Session.builder().channel(channel).sessionId(buildSessionId(channel)).clientId(clientId).build();
}
/**
* 自生成流水号
*
* @return
*/
public int nextSerialNo() {
int current;
int next;
do {
current = serialNo.get();
next = current > 0xffff ? 0 : current;
} while (!serialNo.compareAndSet(current, next + 1));
return next;
}
}
src/main/java/com/tongran/agent/server/core/session/SessionManager.java
+137
View File
@@ -0,0 +1,137 @@
package com.tongran.agent.server.core.session;
import cn.hutool.cache.Cache;
import cn.hutool.cache.CacheUtil;
import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.util.ObjectUtil;
import com.tongran.agent.server.exception.ServerException;
import com.tongran.agent.server.utils.AssertLog;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.util.Attribute;
import io.netty.util.AttributeKey;
import lombok.Data;
import org.apache.commons.lang3.StringUtils;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Data
public class SessionManager {
private static volatile SessionManager instance = null;
// clientId-Session
private final Map<String, Session> sessionMap = new ConcurrentHashMap<>();
private final Cache<String, Object> sessionCache = CacheUtil.newTimedCache(10 * 60 * 1000);
public static SessionManager getInstance() {
if (instance == null) {
synchronized (SessionManager.class) {
if (instance == null) {
instance = new SessionManager();
}
}
}
return instance;
}
public synchronized void put(String clientId, Session session) {
if (StringUtils.isNotBlank(session.getClientId())) {
sessionMap.put(session.getClientId(), session);
}
}
public synchronized void remove(String clientId) {
if (StringUtils.isNotBlank(clientId)) {
sessionMap.remove(clientId);
}
}
public synchronized void remove(Session session) {
if (session != null && StringUtils.isNotBlank(session.getClientId())) {
sessionMap.remove(session.getClientId(), session);
}
}
public synchronized void remove(Channel channel) {
remove(getSessionByChannel(channel));
}
public Session getSessionById(String clientId) {
return sessionMap.get(clientId);
}
public Session getSessionByChannel(Channel channel) {
Session session = new Session();
sessionMap.values().forEach(s -> {
if (s.getChannel().channel() == channel) {
BeanUtil.copyProperties(s, session, true);
}
});
return session;
}
public boolean containsSession(String clientId) {
return sessionMap.containsKey(clientId);
}
public boolean containsSession(Session session) {
return sessionMap.containsValue(session);
}
public void setSessionCache(String clientId, Object value) {
sessionCache.put(clientId, value);
}
public Object getSessionCache(String clientId) {
return sessionCache.get(clientId);
}
public String client(ChannelHandlerContext ctx) {
Channel channel = ctx.channel();
Session session = this.getSessionByChannel(channel);
if (ObjectUtil.isNotNull(session) && StringUtils.isNotBlank(session.getClientId())) {
return channel.remoteAddress().toString() + "/" + session.getClientId();
}
return channel.remoteAddress().toString();
}
/**
* 根据channel生成流水号
*
* @param channel
* @return
*/
public short getSerialNumber(Channel channel, AttributeKey<Short> serialNumber) {
Attribute<Short> flowIdAttr = channel.attr(serialNumber);
Short flowId = flowIdAttr.get();
if (flowId == null) {
flowId = 0;
} else {
flowId++;
}
flowIdAttr.set(flowId);
return flowId;
}
public void writeAndFlush(String clientId, Object msg) {
Session session = this.getSessionById(clientId);
if (ObjectUtil.isNotNull(session)) {
this.writeAndFlush(session.getChannel(), msg);
} else {
throw new ServerException(500, "终端:" + clientId + "离线或者不存在");
}
}
public void writeAndFlush(ChannelHandlerContext ctx, Object msg) {
ctx.writeAndFlush(msg).addListener(future -> {
if (!future.isSuccess()) {
AssertLog.error("消息发送失败:{}", future.cause());
}
});
}
}
src/main/java/com/tongran/agent/server/core/vo/RegisterVO.java
+63
View File
@@ -0,0 +1,63 @@
package com.tongran.agent.server.core.vo;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class RegisterVO {
//客户端ID（对应平台唯一SN）
private String clientId;
//客户端IP
private String clientIp;
//客户端端口
private int clientPort;
//交换机配置
private SwitchBoard switchBoard;
//时间戳
private long timestamp;
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public static class SwitchBoard{
//交换机团名
private String community;
//交换机IP
private String ip;
//交换机端口
private int port;
//交换机网络端口发现OID
private String netOID;
//交换机光模块发现OID
private String moduleOID;
//交换机MPU发现OID
private String mpuOID;
//交换机电源发现OID
private String pwrOID;
//交换机风扇发现OID
private String fanOID;
//交换机系统其他OID
private String otherOID;
}
}
src/main/java/com/tongran/agent/server/exception/ServerException.java
+27
View File
@@ -0,0 +1,27 @@
package com.tongran.agent.server.exception;
import cn.hutool.core.util.StrUtil;
import com.tongran.agent.server.exception.base.BaseException;
import com.tongran.agent.server.exception.code.ErrorCode;
public class ServerException extends BaseException {
private static final long serialVersionUID = -519977195881081739L;
public ServerException(ErrorCode code) {
super(code.getCode(), code.getMsg());
}
public ServerException(Integer code, String msg) {
super(code, msg);
}
public ServerException(String msg) {
super(500, msg);
}
public ServerException(String msg, Object... arguments) {
super(500, StrUtil.format(msg, arguments));
}
}
src/main/java/com/tongran/agent/server/exception/base/BaseException.java
+42
View File
@@ -0,0 +1,42 @@
package com.tongran.agent.server.exception.base;
import com.tongran.agent.server.exception.code.ErrorCode;
public class BaseException extends RuntimeException implements ErrorCode {
private static final long serialVersionUID = 1966249840643379123L;
private Integer code;
private String msg;
public BaseException() {
}
public BaseException(String msg, Throwable cause) {
super(msg, cause);
}
public BaseException(Integer code, String msg) {
super(msg);
this.code = code;
this.msg = msg;
}
public BaseException(Integer code, String msg, Throwable cause) {
super(msg, cause);
this.code = code;
this.msg = msg;
}
@Override
public Integer getCode() {
return code;
}
@Override
public String getMsg() {
return msg;
}
}
src/main/java/com/tongran/agent/server/exception/code/ErrorCode.java
+16
View File
@@ -0,0 +1,16 @@
package com.tongran.agent.server.exception.code;
import com.tongran.agent.server.utils.R;
public interface ErrorCode {
Integer getCode();
String getMsg();
default R<?> toResult() {
return R.error(getMsg(), getCode());
}
}
src/main/java/com/tongran/agent/server/exception/code/GlobalErrorCode.java
+20
View File
@@ -0,0 +1,20 @@
package com.tongran.agent.server.exception.code;
import lombok.AllArgsConstructor;
import lombok.Getter;
@Getter
@AllArgsConstructor
public enum GlobalErrorCode implements ErrorCode {
SUCCESS(200, "成功"),
NOT_FOUND(404, "未找到相关资源"),
ERROR(500, "系统错误");
private final Integer code;
private final String msg;
}
src/main/java/com/tongran/agent/server/netty/BaseNettyServer.java
+117
View File
@@ -0,0 +1,117 @@
package com.tongran.agent.server.netty;
import com.tongran.agent.server.netty.config.BaseNettyConfig;
import com.tongran.agent.server.utils.AssertLog;
import io.netty.bootstrap.AbstractBootstrap;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioChannelOption;
import io.netty.channel.socket.nio.NioDatagramChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.ResourceLeakDetector;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.Future;
/**
* 基础Netty服务
*/
public abstract class BaseNettyServer {
protected boolean isRunning;
protected BaseNettyConfig config;
protected EventLoopGroup bossGroup;
protected EventLoopGroup workerGroup;
protected EventExecutorGroup businessGroup;
protected BaseNettyServer(BaseNettyConfig config) {
this.config = config;
}
protected AbstractBootstrap<?, ?> initializeTcp() {
bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY));
workerGroup = new NioEventLoopGroup(config.workerCore, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY));
if (config.businessCore > 0) {
businessGroup = new DefaultEventExecutorGroup(config.businessCore);
}
ServerBootstrap serverBootstrap = new ServerBootstrap();
serverBootstrap.group(bossGroup, workerGroup)
.channel(NioServerSocketChannel.class)
.childOption(ChannelOption.SO_REUSEADDR, true)
.option(ChannelOption.SO_BACKLOG, 1024)
.childOption(NioChannelOption.TCP_NODELAY, true)
.childHandler(config.hander);
//内存泄漏检测 开发推荐PARANOID 线上SIMPLE
ResourceLeakDetector.setLevel(ResourceLeakDetector.Level.SIMPLE);
return serverBootstrap;
}
protected AbstractBootstrap<?, ?> initializeUdp() {
bossGroup = new NioEventLoopGroup(1, new DefaultThreadFactory(config.name, Thread.MAX_PRIORITY));
if (config.businessCore > 0) {
businessGroup = new DefaultEventExecutorGroup(config.businessCore);
}
return new Bootstrap()
.group(bossGroup).channel(NioDatagramChannel.class)
.option(NioChannelOption.SO_REUSEADDR, true)
.option(NioChannelOption.SO_RCVBUF, 1024 * 1024 * 50)
.handler(config.hander);
}
public synchronized boolean start() {
if (!config.enable) {
return false;
}
if (isRunning) {
AssertLog.info("======{}已经启动,port:{}======", config.name, config.port);
return isRunning;
}
AbstractBootstrap<?, ?> bootstrap = config.isTcp ? initializeTcp() : initializeUdp();
ChannelFuture future = bootstrap.bind(config.port).awaitUninterruptibly();
future.channel().closeFuture().addListener(f -> {
if (isRunning) {
stop();
}
});
if (future.cause() != null) {
AssertLog.error("===启动失败===", future.cause());
}
if (isRunning = future.isSuccess()) {
AssertLog.info("\n\n\t\t\t\t\t\t\t\t======{}启动成功,port:{}======\n", config.name, config.port);
}
return isRunning;
}
public synchronized void stop() {
if (!config.enable) {
return;
}
isRunning = false;
try {
Future<?> future = this.workerGroup.shutdownGracefully().await();
if (!future.isSuccess()) {
AssertLog.error("workerGroup 无法正常停止:{}", future.cause());
}
future = this.bossGroup.shutdownGracefully().await();
if (!future.isSuccess()) {
AssertLog.error("bossGroup 无法正常停止:{}", future.cause());
}
future = this.businessGroup.shutdownGracefully().await();
if (!future.isSuccess()) {
AssertLog.error("businessGroup 无法正常停止:{}", future.cause());
}
} catch (InterruptedException e) {
e.printStackTrace();
}
AssertLog.info("\n\n\t\t\t\t\t\t\t\t======{} 已经停止,port:{}======\n", config.name, config.port);
}
}
src/main/java/com/tongran/agent/server/netty/MultiTargetNettyClient.java
+301
View File
@@ -0,0 +1,301 @@
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/annotation/AgentDispatcher.java
+47
View File
@@ -0,0 +1,47 @@
package com.tongran.agent.server.netty.annotation;
import com.tongran.agent.server.core.enums.MsgEnum;
import lombok.AllArgsConstructor;
import lombok.Getter;
import org.springframework.stereotype.Component;
import java.lang.annotation.*;
/**
* 指明类为Agent消息
*/
@Component
@Documented
@Target({ElementType.TYPE})
@Retention(RetentionPolicy.RUNTIME)
public @interface AgentDispatcher {
@Getter
@AllArgsConstructor
enum VersionEnum {
V1("2026");
public final String value;
}
/**
* 消息ID
*
* @return
*/
MsgEnum msgId();
/**
* 消息版本 默认版本2025
*
* @return
*/
VersionEnum version() default VersionEnum.V1;
/**
* 描述
*
* @return
*/
String desc() default "";
}
src/main/java/com/tongran/agent/server/netty/basics/AgentDispatcherManager.java
+48
View File
@@ -0,0 +1,48 @@
package com.tongran.agent.server.netty.basics;
import com.tongran.agent.server.netty.annotation.AgentDispatcher;
import org.springframework.beans.BeansException;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
@Component
@Order(1)
public class AgentDispatcherManager implements ApplicationContextAware {
/**
* 所有实现的包处理器
*/
private static Map<String, AgentHandler> MSG_HANDLER_MAP;
/**
* 唤醒时 初始化 packHandlerMap
*/
@Override
public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
// 仅一次性初始化完成
if (MSG_HANDLER_MAP == null) {
MSG_HANDLER_MAP = new ConcurrentHashMap<>();
Map<String, Object> handlers = applicationContext.getBeansWithAnnotation(AgentDispatcher.class);
if (!CollectionUtils.isEmpty(handlers)) {
handlers.values().forEach(tempHandler -> {
boolean result = tempHandler.getClass().isAnnotationPresent(AgentDispatcher.class);
if (result) {
AgentDispatcher annotation = tempHandler.getClass().getAnnotation(AgentDispatcher.class);
MSG_HANDLER_MAP.put(annotation.msgId().getValue() + "&" + annotation.version().value, (AgentHandler) tempHandler);
}
});
}
}
}
public AgentHandler getHandler(String msgIdVersion) {
return MSG_HANDLER_MAP == null ? null : MSG_HANDLER_MAP.get(msgIdVersion);
}
}
src/main/java/com/tongran/agent/server/netty/basics/AgentHandler.java
+21
View File
@@ -0,0 +1,21 @@
package com.tongran.agent.server.netty.basics;
import com.tongran.agent.server.netty.model.UpMsgResponse;
public interface AgentHandler {
/**
* 处理终端传入的消息, 然后进行返回
*
* @return 需要发送给终端的消息
*/
default UpMsgResponse upHandle(String data, String clientId) {
return null;
}
default String downHandle(String data, String clientId) {
return "";
}
}
src/main/java/com/tongran/agent/server/netty/config/BaseNettyConfig.java
+58
View File
@@ -0,0 +1,58 @@
package com.tongran.agent.server.netty.config;
import com.tongran.agent.server.core.session.SessionManager;
import io.netty.channel.ChannelHandler;
import io.netty.util.NettyRuntime;
import lombok.Data;
import java.io.Serializable;
/**
* 基础Netty配置
*/
@Data
public class BaseNettyConfig implements Serializable {
private static final long serialVersionUID = -2736718499972710895L;
/**
* 是否开启
*/
public Boolean enable = false;
/**
* 是否TCP
*/
public Boolean isTcp = true;
/**
* 服务名称
*/
public String name = "Netty";
/**
* 端口号
*/
public Integer port = 11111;
public Integer workerCore = NettyRuntime.availableProcessors() * 2;
public Integer businessCore = Math.max(1, NettyRuntime.availableProcessors() >> 1);
public Integer readerIdleTime = 240;
public Integer writerIdleTime = 0;
public Integer allIdleTime = 0;
/**
* netty Session管理
*/
public SessionManager sessionManager;
/**
* netty Channel处理列表
*/
public ChannelHandler hander;
}
src/main/java/com/tongran/agent/server/netty/config/ConnectionConfig.java
+39
View File
@@ -0,0 +1,39 @@
package com.tongran.agent.server.netty.config;
import java.util.Objects;
/**
* 连接配置信息
*/
public class ConnectionConfig {
private final String connectionKey;
private final String host;
private final int port;
private final int timeout;
public ConnectionConfig(String connectionKey, String host, int port, int timeout) {
this.connectionKey = connectionKey;
this.host = host;
this.port = port;
this.timeout = timeout;
}
// Getter方法
public String getConnectionKey() { return connectionKey; }
public String getHost() { return host; }
public int getPort() { return port; }
public int getTimeout() { return timeout; }
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
ConnectionConfig that = (ConnectionConfig) o;
return Objects.equals(connectionKey, that.connectionKey);
}
@Override
public int hashCode() {
return Objects.hash(connectionKey);
}
}
src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java
+313
View File
@@ -0,0 +1,313 @@
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;
import com.tongran.agent.server.rockermq.AgentProducer;
import com.tongran.agent.server.rockermq.config.AgentMqConfig;
import com.tongran.agent.server.rockermq.model.MqMsg;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@Component
public class AgentEndpoint {
@Resource
private MultiTargetNettyClient client;
@Resource
private AgentMqConfig agentMqConfig;
@Resource
private AgentProducer agentProducer;
@AgentDispatcher(msgId = MsgEnum.注册应答)
public class RegisterRspHandler 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 DisconnectHandler implements AgentHandler {
@Override
public String downHandle(String data, String clientId) {
return data;
}
}
@AgentDispatcher(msgId = MsgEnum.断开应答)
public class DisconnectRspHandler 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 = JSONObject.parseObject(data);
int resCode = 0;
if(json.containsKey("resCode")){
resCode = json.getInteger("resCode");
}
if(resCode == 1){
client.closeConnection(clientId);
}
return UpMsgResponse.builder().clientId(clientId).build();
}
}
@AgentDispatcher(msgId = MsgEnum.心跳上报)
public class HeartBeatHandler 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.CPU上报)
public class CpuHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = new JSONObject();
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.CPU上报.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 DiskHandler 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 DockerHandler 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 MemoryHandler 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 NetHandler 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 PointHandler 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 SystemHandler 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 SwitchBoardHandler 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 SystemCollectStartHandler implements AgentHandler {
@Override
public String downHandle(String data, String clientId) {
return data;
}
}
@AgentDispatcher(msgId = MsgEnum.开启系统采集应答)
public class SystemCollectStartRspHandler 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 SwitchCollectStartHandler implements AgentHandler {
@Override
public String downHandle(String data, String clientId) {
return data;
}
}
@AgentDispatcher(msgId = MsgEnum.开启交换机采集应答)
public class SwitchCollectStartRspHandler 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 AlarmSetHandler implements AgentHandler {
@Override
public String downHandle(String data, String clientId) {
return data;
}
}
@AgentDispatcher(msgId = MsgEnum.告警设置应答)
public class AlarmSetRspHandler 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 ScriptPolicyHandler implements AgentHandler {
@Override
public String downHandle(String data, String clientId) {
return data;
}
}
@AgentDispatcher(msgId = MsgEnum.执行脚本策略应答)
public class ScriptPolicyRspHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = JSONObject.parseObject(data);
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.Agent版本更新)
public class AgentVersionUpdateHandler implements AgentHandler {
@Override
public String downHandle(String data, String clientId) {
return data;
}
}
@AgentDispatcher(msgId = MsgEnum.Agent版本更新应答)
public class AgentVersionUpdateRspHandler implements AgentHandler {
@Override
public UpMsgResponse upHandle(String data, String clientId) {
JSONObject jsonObject = JSONObject.parseObject(data);
jsonObject.put("clientId", clientId);
jsonObject.put("dataType", MsgEnum.Agent版本更新应答.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
+162
View File
@@ -0,0 +1,162 @@
package com.tongran.agent.server.netty.handler;
import cn.hutool.cache.Cache;
import cn.hutool.cache.CacheUtil;
import cn.hutool.core.date.DateUnit;
import cn.hutool.core.util.ObjectUtil;
import com.alibaba.fastjson2.JSONObject;
import com.tongran.agent.server.core.session.SessionManager;
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.netty.model.UpMsgResponse;
import com.tongran.agent.server.utils.AssertLog;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.util.AttributeKey;
import io.netty.util.CharsetUtil;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Objects;
@Component
@ChannelHandler.Sharable
public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
// 连接标识属性键
private static final AttributeKey<String> CONNECTION_KEY = AttributeKey.valueOf("connectionKey");
protected final SessionManager sessionManager;
public AgentDecoderHandler() {
this.sessionManager = SessionManager.getInstance();
}
@Resource
AgentDispatcherManager agentDispatcherManager;
// 用来临时保留没有处理过的请求报文
private final Cache<String, String> lruCache = 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);
String content = "";
StringBuilder sb = new StringBuilder();
boolean isClear = false;
String tempMsg = lruCache.get(sessionManager.client(ctx));
tempMsg = tempMsg == null ? "" : tempMsg;
int tmpMsgSize = tempMsg.length();
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), sb.toString(), DateUnit.SECOND.getMillis() * 5000);
}
}
//判定是否粘包
if(arr_msg.length > 1){
if(!endsWith){
lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0;
sb.append(arr_msg[0]+"@tong-ran");
lruCache.put(sessionManager.client(ctx), arr_msg[1], DateUnit.SECOND.getMillis() * 5000);
}else{
lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0;
sb.append(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);
}
}
AssertLog.info("<<[up2]:IP:{},[up-content]==>{}", sessionManager.client(ctx),content);
endsWith = content.endsWith("@tong-ran");
//判断是否是整包
if(endsWith){
String[] arr = content.split("@tong-ran");
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)) {
UpMsgResponse response = msgHandler.upHandle(data, clientId );
AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", clientId, dataType, data);
if(Objects.nonNull(response)){
Message agentMessage = Message.builder().build();
agentMessage.setClientId(clientId);
agentMessage.setDataType(response.getDataType());
agentMessage.setData(response.getContent());
ctx.fireChannelRead(agentMessage);//传递到下一个handler
}
}
}
}
}
if (isClear) {
lruCache.remove(sessionManager.client(ctx));
}
byteBuf.release(); // 释放 ByteBuf 资源（重要！）
} else {
System.out.println("Unexpected message type: " + msg.getClass());
}
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
String connectionKey = ctx.channel().attr(CONNECTION_KEY).get();
System.err.println("连接 " + connectionKey + " 发生异常: " + cause.getMessage());
ctx.close();
}
public byte[] subByte(byte[] b, int off, int length) {
byte[] bytes = new byte[length];
System.arraycopy(b, off, bytes, 0, length);
return bytes;
}
}
src/main/java/com/tongran/agent/server/netty/handler/AgentDispatcherHandler.java
+54
View File
@@ -0,0 +1,54 @@
package com.tongran.agent.server.netty.handler;
import com.tongran.agent.server.core.session.Session;
import com.tongran.agent.server.core.session.SessionManager;
import com.tongran.agent.server.netty.model.Message;
import com.tongran.agent.server.utils.AssertLog;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
@Component
@ChannelHandler.Sharable
public class AgentDispatcherHandler extends SimpleChannelInboundHandler<Message> {
protected final SessionManager sessionManager;
public AgentDispatcherHandler() {
this.sessionManager = SessionManager.getInstance();
}
@Override
protected void channelRead0(ChannelHandlerContext ctx, Message msg) {
// AssertLog.info(">>>>>>[要处理的终端数据] {}", msg.toString());
String clientId = msg.getClientId();
if (StringUtils.isBlank(clientId)) {
AssertLog.error("<<<<<<错误的信息from:{}", ctx.channel().remoteAddress());
return;
}
Session session = Session.buildSession(ctx, clientId);
sessionManager.put(clientId, session);
if (StringUtils.isNotBlank(msg.getData())) {
// AssertLog.info(">>>>>>[getSessionById的终端数据] {}", sessionManager.getSessionById(clientId));
sessionManager.writeAndFlush(sessionManager.getSessionById(clientId).getChannel(), msg);
}
}
// @Override
// public void channelActive(ChannelHandlerContext ctx) {
// AssertLog.info("与服务器建立连接: {}", ctx.channel().remoteAddress());
// // 连接建立后可以发送登录认证消息
// long timestamp = System.currentTimeMillis();
// String clientId = "clientId001";
// JSONObject object = new JSONObject();
// object.put("uuid",clientId);
// object.put("timestamp",timestamp);
// Message message = Message.builder().clientId(clientId).dataType("LOGIN").data(object.toString()).build();
// // 将对象转为 JSON 字符串
// String json = JSON.toJSONString(message);
// System.out.println("连接建立成功，发送登录认证="+json);
// ctx.writeAndFlush(message);
// }
}
src/main/java/com/tongran/agent/server/netty/handler/AgentEncoderHandler.java
+49
View File
@@ -0,0 +1,49 @@
package com.tongran.agent.server.netty.handler;
import com.alibaba.fastjson2.JSON;
import com.tongran.agent.server.netty.model.Message;
import com.tongran.agent.server.core.session.SessionManager;
import com.tongran.agent.server.utils.AssertLog;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelOutboundHandlerAdapter;
import io.netty.channel.ChannelPromise;
import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Component;
import java.nio.charset.StandardCharsets;
@Component
@ChannelHandler.Sharable
public class AgentEncoderHandler extends ChannelOutboundHandlerAdapter {
protected final SessionManager sessionManager;
public AgentEncoderHandler() {
this.sessionManager = SessionManager.getInstance();
}
/**
* 消息解码器
*/
@Override
public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
if (msg instanceof Message) {
Message entity = (Message) msg;
if (StringUtils.isBlank(entity.getData())) {
AssertLog.error(">>[down]:IP:{},errorContent:{}", sessionManager.client(ctx), msg);
return;
}
AssertLog.info(">>[down]:IP:{},[content]==>{}", sessionManager.client(ctx), entity.getData());
// String json = JSON.toJSONString(entity);
String json = "agent-server:"+JSON.toJSONString(entity)+"@tong-ran";
byte[] bytes = json.getBytes(StandardCharsets.UTF_8); // 显式指定 UTF-8
// byte[] bytes = EscapeUtil.hexStringToByteArray(entity.getContent());
ByteBuf buf = Unpooled.wrappedBuffer(bytes);
ctx.write(buf, promise);
}
}
}
src/main/java/com/tongran/agent/server/netty/model/Message.java
+37
View File
@@ -0,0 +1,37 @@
package com.tongran.agent.server.netty.model;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
/**
* @author egrias
*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class Message implements Serializable {
private static final long serialVersionUID = -1267013167162440610L;
/**
* clientId
*/
private String clientId;
/**
* 数据类型：LOGIN、HEARTBEAT、CPU、MEMORY、SYSTEM、POINT、NET、DISK、DOCKER、SWITCHBOARD
*/
private String dataType;
/**
* 发送内容
*/
private String data;
}
src/main/java/com/tongran/agent/server/netty/model/UpMsgResponse.java
+22
View File
@@ -0,0 +1,22 @@
package com.tongran.agent.server.netty.model;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.experimental.SuperBuilder;
@Data
@SuperBuilder
@AllArgsConstructor
@NoArgsConstructor
public class UpMsgResponse {
private String clientId;
/**
* type
*/
private String dataType;
private String content;
}
src/main/java/com/tongran/agent/server/rockermq/AgentProducer.java
+54
View File
@@ -0,0 +1,54 @@
package com.tongran.agent.server.rockermq;
import com.tongran.agent.server.rockermq.model.MqMsg;
import com.tongran.agent.server.utils.AssertLog;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
@Component
public class AgentProducer implements RocketMqService {
@Resource
private RocketMQTemplate rocketMQTemplate;
@Override
public void send(MqMsg msg) {
AssertLog.info("send发送消息:==>{}", msg);
rocketMQTemplate.send(msg.getTopic(), MessageBuilder.withPayload(msg.getContent()).build());
}
@Override
public void asyncSend(MqMsg msg) {
AssertLog.info("asyncSend发送消息:==>{}", msg);
rocketMQTemplate.asyncSend(msg.getTopic(), msg.getContent(), new SendCallback() {
@Override
public void onSuccess(SendResult sendResult) {
// AssertLog.info("asyncSend发送消息成功:==>{}", sendResult.getSendStatus());
}
@Override
public void onException(Throwable throwable) {
AssertLog.error("asyncSend发送消息失败:==>{}", throwable.getMessage());
}
});
}
@Override
public void syncSendOrderly(MqMsg msg) {
AssertLog.info("syncSendOrderly发送消息:==>{}", msg);
rocketMQTemplate.sendOneWay(msg.getTopic(), msg.getContent());
}
@Override
public void delayedSendOrderly(MqMsg msg) {
AssertLog.info("delayedSendOrderly发送消息:==>{}", msg);
rocketMQTemplate.syncSend(msg.getTopic(), MessageBuilder.withPayload(msg.getContent()).build(), 30000, msg.getLevel());
}
}
src/main/java/com/tongran/agent/server/rockermq/RocketMqService.java
+45
View File
@@ -0,0 +1,45 @@
package com.tongran.agent.server.rockermq;
import com.tongran.agent.server.rockermq.model.MqMsg;
/**
* Rocket MQ 对应服务封装
* @author BAO
*
*/
public interface RocketMqService {
/**
* 同步发送消息<br/>
* <p>
* 当发送的消息很重要是，且对响应时间不敏感的时候采用sync方式;
*
* @param msg 发送消息实体类
*/
void send(MqMsg msg);
/**
* 异步发送消息，异步返回消息结果 当发送的消息很重要，且对响应时间非常敏感的时候采用async方式；
*
* @param msg 发送消息实体类
*/
void asyncSend(MqMsg msg);
/**
* 直接发送发送消息，不关心返回结果，容易消息丢失，适合日志收集、不精确统计等消息发送; 发送的消息不重要时，采用one-way方式，以提高吞吐量；
*
* @param msg 发送消息实体类
*/
void syncSendOrderly(MqMsg msg);
/**
* 发送延时消息，不关心返回结果<br/>
* <p>
* 当发送的消息不重要时，采用one-way方式，以提高吞吐量；
*
* @param msg 发送消息实体类
*/
default void delayedSendOrderly(MqMsg msg){};
}
src/main/java/com/tongran/agent/server/rockermq/config/AgentMqConfig.java
+14
View File
@@ -0,0 +1,14 @@
package com.tongran.agent.server.rockermq.config;
import lombok.Data;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Configuration;
@Data
@Configuration
@ConfigurationProperties(prefix = "rocketmq.producer")
public class AgentMqConfig {
private String agentTopic;
}
src/main/java/com/tongran/agent/server/rockermq/listener/AgentConsumerDownListener.java
+114
View File
@@ -0,0 +1,114 @@
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;
import org.springframework.stereotype.Component;
import javax.annotation.Resource;
import java.util.Objects;
@Component
@ConditionalOnProperty(name = "rocketmq.enabled", havingValue = "true")
@RocketMQMessageListener(topic = "agent_down", consumerGroup = "agent_down_group")
public class AgentConsumerDownListener implements RocketMQListener<String> {
protected final SessionManager sessionManager;
@Resource
private AgentDispatcherManager agentDispatcherManager;
@Resource
private MultiTargetNettyClient client;
@Resource
private AgentMqConfig agentMqConfig;
@Resource
private AgentProducer agentProducer;
public AgentConsumerDownListener() {
this.sessionManager = SessionManager.getInstance();
}
@Override
public void onMessage(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(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)){
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/rockermq/model/MqMsg.java
+38
View File
@@ -0,0 +1,38 @@
package com.tongran.agent.server.rockermq.model;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
/**
* RocketMQ消息发送对象
*
* @author BAO
*/
@Data
@Builder
@AllArgsConstructor
@NoArgsConstructor
public class MqMsg implements Serializable {
private static final long serialVersionUID = 4164379745748817325L;
/**
* 消息topic
*/
private String topic;
/**
* 消息content
*/
private Object content;
/**
* 消息延时等级
*/
private Integer level;
}
src/main/java/com/tongran/agent/server/service/AgentService.java
+10
View File
@@ -0,0 +1,10 @@
package com.tongran.agent.server.service;
import com.tongran.agent.server.core.vo.RegisterVO;
public interface AgentService {
boolean createConn(String clientId, String host, int port, int timeout);
void register(RegisterVO registerVO);
}
src/main/java/com/tongran/agent/server/service/impl/AgentServiceImpl.java
+57
View File
@@ -0,0 +1,57 @@
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;
@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;
}
@Override
public void register(RegisterVO registerVO) {
// 发送注册
JSONObject object = new JSONObject();
object.put("clientId", registerVO.getClientId());
object.put("timestamp", registerVO.getTimestamp());
JSONObject switchBoard = new JSONObject();
switchBoard.put("community", registerVO.getSwitchBoard().getCommunity());
switchBoard.put("ip", registerVO.getSwitchBoard().getIp());
switchBoard.put("port", registerVO.getSwitchBoard().getPort());
switchBoard.put("oid", String.join(",", registerVO.getSwitchBoard().getFanOID()));
object.put("switchBoard", switchBoard.toString());
Message message = Message.builder().clientId(registerVO.getClientId()).dataType("REGISTER").data(object.toString()).build();
// 将对象转为 JSON 字符串
String json = JSON.toJSONString(message);
System.out.println("发送注册消息="+json);
client.sendMessages(registerVO.getClientId(), message);
}
}
src/main/java/com/tongran/agent/server/utils/AssertLog.java
+73
View File
@@ -0,0 +1,73 @@
package com.tongran.agent.server.utils;
import lombok.extern.slf4j.Slf4j;
/**
* 日志断言类
*
* @author BAO
*
*/
@Slf4j
public class AssertLog {
/**
* 打印Info 日志
*
* @param format
* @param arguments
*/
public static void info(String format, Object... arguments) {
if (log.isInfoEnabled()) {
log.info(format, arguments);
}
}
/**
* 打印Debug 日志
*
* @param format
* @param arguments
*/
public static void debug(String format, Object... arguments) {
if (log.isDebugEnabled()) {
log.debug(format, arguments);
}
}
/**
* 打印Error 日志
*
* @param format
* @param arguments
*/
public static void error(String format, Object... arguments) {
if (log.isErrorEnabled()) {
log.error(format, arguments);
}
}
/**
* 打印Trace 日志
*
* @param format
* @param arguments
*/
public static void trace(String format, Object... arguments) {
if (log.isTraceEnabled()) {
log.trace(format, arguments);
}
}
/**
* 打印Warn 日志
*
* @param format
* @param arguments
*/
public static void warn(String format, Object... arguments) {
if (log.isWarnEnabled()) {
log.warn(format, arguments);
}
}
}
src/main/java/com/tongran/agent/server/utils/ClientExample.java
+46
View File
@@ -0,0 +1,46 @@
package com.tongran.agent.server.utils;
import com.tongran.agent.server.netty.config.ConnectionConfig;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
public class ClientExample {
public static void main(String[] args) {
EnhancedConnectionManager manager = new EnhancedConnectionManager();
try {
// 配置多个连接
List<ConnectionConfig> configs = Arrays.asList(
// new ConnectionConfig("db_server", "192.168.1.101", 3306, 5),
// new ConnectionConfig("redis_server", "192.168.1.102", 6379, 3),
// new ConnectionConfig("api_server", "api.example.com", 8080, 10),
new ConnectionConfig("server1", "127.0.0.1", 6610, 5)
);
// 批量创建连接
Map<String, Boolean> results = manager.createConnections(configs);
System.out.println("连接创建结果: " + results);
// 等待连接建立
Thread.sleep(3000);
// 根据消息类型路由
manager.routeMessageByType("TYPE_A", "Database query");
// manager.routeMessageByType("TYPE_B", "Cache operation");
// manager.routeMessageByType("UNKNOWN", "Broadcast message");
// 检查连接状态
Map<String, Boolean> status = manager.getConnectionStatus();
System.out.println("连接状态: " + status);
Thread.sleep(5000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
manager.shutdown();
}
}
}
src/main/java/com/tongran/agent/server/utils/EnhancedConnectionManager.java
+82
View File
@@ -0,0 +1,82 @@
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;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
/**
* 增强版连接管理器
*/
public class EnhancedConnectionManager {
private final MultiTargetNettyClient client;
private final Map<String, ConnectionConfig> connectionConfigs;
public EnhancedConnectionManager() {
this.client = new MultiTargetNettyClient();
this.connectionConfigs = new ConcurrentHashMap<>();
}
/**
* 批量创建连接
*/
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);
});
return results;
}
/**
* 根据消息类型路由到不同连接
*/
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));
}
}
/**
* 获取所有连接状态
*/
public Map<String, Boolean> getConnectionStatus() {
Map<String, Boolean> status = new HashMap<>();
connectionConfigs.forEach((key, config) -> {
// 这里可以添加更详细的状态检查
status.put(key, client.sendMessage(key, "PING"));
});
return status;
}
public void shutdown() {
client.shutdown();
}
}
src/main/java/com/tongran/agent/server/utils/R.java
+76
View File
@@ -0,0 +1,76 @@
package com.tongran.agent.server.utils;
import com.tongran.agent.server.exception.code.ErrorCode;
import com.tongran.agent.server.exception.code.GlobalErrorCode;
import io.swagger.v3.oas.annotations.media.Schema;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
import java.io.Serializable;
/**
* @author BAO
*/
@Data
@Builder
@NoArgsConstructor
@AllArgsConstructor
@Schema(description = "接口交互统一数据返回标准")
public class R<T> implements Serializable {
private static final long serialVersionUID = 1L;
@Schema(description = "返回代码")
private Integer code;
@Schema(description = "消息描述")
private String msg;
@Schema(description = "结果对象")
private T data;
public static <T> R<T> success(String msg, T t) {
R<T> r = new R<>();
r.setData(t);
r.setMsg(msg);
r.setCode(GlobalErrorCode.SUCCESS.getCode());
return r;
}
public static <T> R<T> success(T t) {
return R.success(GlobalErrorCode.SUCCESS.getMsg(), t);
}
public static <T> R<T> success() {
return R.success(null);
}
public static <T> R<T> error(String msg, Integer code) {
R<T> r = new R<>();
r.setMsg(msg);
r.setCode(code);
return r;
}
public static <T> R<T> error(Integer code, String msg) {
R<T> r = new R<>();
r.setMsg(msg);
r.setCode(code);
return r;
}
public static <T> R<T> error(ErrorCode err) {
return R.error(err.getMsg(), err.getCode());
}
public static <T> R<T> error() {
return R.error(GlobalErrorCode.ERROR.getMsg(), GlobalErrorCode.ERROR.getCode());
}
public static <T> R<T> error(String msg) {
return R.error(msg, GlobalErrorCode.ERROR.getCode());
}
}
src/main/resources/application-dev.yml
+28
View File
@@ -0,0 +1,28 @@
server:
port: 9018
servlet:
context-path: /agent-server
# 接口文档配置
knife4j:
enable: true
production: false # 开启屏蔽文档资源
# 日志配置
logging:
file:
# path: /usr/local/tongran/logs
path: /data/agent-server/logs
# path: D:/job/agent-logs/logs
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 # 最大重连延迟(毫秒)
src/main/resources/application.yml
+43
View File
@@ -0,0 +1,43 @@
spring:
profiles:
active: dev
mvc:
pathmatch:
matching-strategy: ant_path_matcher
application:
name: agent-server
version: 1.0
web:
resources:
static-locations: classpath*:/META-INF/resources/
logging:
config: classpath:logback/logback-${spring.profiles.active}.xml
# springdoc-openapi项目配置
knife4j:
setting:
enable-footer-custom: true
footer-custom-content: Apache License 2.0
springdoc:
config:
title: AGENT服务
description: AGENT服务接口文档
contact: SERVER
email:
version: ${spring.application.version}
group-configs:
- group: 'AGENT服务'
paths-to-match: '/**'
packages-to-scan: com.tongran.agent.server
rocketmq:
producer:
group: tongran-group
topic: default_topic
agent-group: tongran_agent_up_group
agent-topic: tongran_agent_up
consumer:
group: tongran-group
topic: default_topic
agent-group: tongran_agent_down_group
agent-topic: tongran_agent_down
src/main/resources/logback/logback-dev.xml
+116
View File
@@ -0,0 +1,116 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60000" debug="false">
<springProperty scope="context" name="appName" source="spring.application.name" defaultValue="default"/>
<springProperty scope="context" name="logPath" source="logging.file.path" defaultValue="/logs/${appName}"/>
<contextName>${appName}</contextName>
<!--定义日志文件的存储地址 勿在 LogBack 的配置中使用相对路径 -->
<property name="LOG_HOME" value="${logPath}/${appName}"/>
<!-- 控制台输出 -->
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
<pattern>[requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n</pattern>
</encoder>
</appender>
<!-- 按照每天生成日志文件, debug级别 -->
<appender name="DEBUG" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 日志日常打印文件 -->
<file>${LOG_HOME}/debug.log</file>
<!-- 日志滚动规则 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!--日志滚动文件名命令规则 -->
<FileNamePattern>${LOG_HOME}/debug.%d{yyyy-MM-dd}.%i.AssertLog.zip</FileNamePattern>
<!--日志文件保留天数 -->
<MaxHistory>30</MaxHistory>
<!-- 日志保存总量 -->
<totalSizeCap>10GB</totalSizeCap>
<!-- 日志切分规则 -->
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- 文件切分压缩的大小阈值 -->
<maxFileSize>10MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<!-- 日志输出的样式 -->
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
<pattern>[requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n</pattern>
</encoder>
</appender>
<!-- 按照每天生成日志文件, info级别 -->
<appender name="INFO" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 精确过滤, 只保存info级别 -->
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<level>INFO</level>
<onMatch>ACCEPT</onMatch>
<onMismatch>DENY</onMismatch>
</filter>
<!-- 日志日常打印文件 -->
<file>${LOG_HOME}/info.log</file>
<!-- 日志滚动规则 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 日志滚动文件名命令规则 -->
<FileNamePattern>${LOG_HOME}/info.%d{yyyy-MM-dd}.%i.AssertLog.zip</FileNamePattern>
<!--日志文件保留天数 -->
<MaxHistory>30</MaxHistory>
<!-- 日志保存总量 -->
<totalSizeCap>10GB</totalSizeCap>
<!-- 日志切分规则 -->
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- 文件切分压缩的大小阈值 -->
<maxFileSize>10MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<!-- 日志输出的样式 -->
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
<pattern>[requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n</pattern>
</encoder>
</appender>
<!-- 按照固定大小生成日志文件, error级别 -->
<appender name="ERROR" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 精确过滤, 只保存error级别 -->
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<level>ERROR</level>
<onMatch>ACCEPT</onMatch>
<onMismatch>DENY</onMismatch>
</filter>
<file>${LOG_HOME}/error.log</file>
<!-- 日志滚动规则 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 日志滚动文件名命令规则 -->
<FileNamePattern>${LOG_HOME}/error.%d{yyyy-MM-dd}.%i.AssertLog.zip</FileNamePattern>
<!--日志文件保留天数 -->
<MaxHistory>30</MaxHistory>
<!-- 日志保存总量 -->
<totalSizeCap>10GB</totalSizeCap>
<!-- 日志切分规则 -->
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- 文件切分压缩的大小阈值 -->
<maxFileSize>10MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<!-- 日志输出的样式 -->
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
<pattern>[requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n</pattern>
</encoder>
</appender>
<logger name="com.alibaba.nacos.client" level="OFF"></logger>
<logger name="com.netflix" level="OFF"></logger>
<logger name="RocketmqClient" additivity="false">
<level value="warn" />
<appender-ref ref="ERROR"/>
</logger>
<!-- 日志输出级别 -->
<root level="info">
<appender-ref ref="STDOUT"/>
<appender-ref ref="INFO"/>
<appender-ref ref="ERROR"/>
<appender-ref ref="DEBUG"/>
</root>
</configuration>
src/main/resources/logback/logback-prod.xml
+115
View File
@@ -0,0 +1,115 @@
<?xml version="1.0" encoding="UTF-8"?>
<configuration scan="true" scanPeriod="60000" debug="false">
<springProperty scope="context" name="appName" source="spring.application.name" defaultValue="default"/>
<springProperty scope="context" name="logPath" source="logging.file.path" defaultValue="/logs/${appName}"/>
<contextName>${appName}</contextName>
<!--定义日志文件的存储地址 勿在 LogBack 的配置中使用相对路径 -->
<property name="LOG_HOME" value="${logPath}/${appName}"/>
<!-- 控制台输出 -->
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
<pattern>[requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n</pattern>
</encoder>
</appender>
<!-- 按照每天生成日志文件, debug级别 -->
<appender name="DEBUG" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 日志日常打印文件 -->
<file>${LOG_HOME}/debug.log</file>
<!-- 日志滚动规则 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!--日志滚动文件名命令规则 -->
<FileNamePattern>${LOG_HOME}/debug.%d{yyyy-MM-dd}.%i.AssertLog.zip</FileNamePattern>
<!--日志文件保留天数 -->
<MaxHistory>30</MaxHistory>
<!-- 日志保存总量 -->
<totalSizeCap>10GB</totalSizeCap>
<!-- 日志切分规则 -->
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- 文件切分压缩的大小阈值 -->
<maxFileSize>10MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<!-- 日志输出的样式 -->
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
<pattern>[requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n</pattern>
</encoder>
</appender>
<!-- 按照每天生成日志文件, info级别 -->
<appender name="INFO" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 精确过滤, 只保存info级别 -->
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<level>INFO</level>
<onMatch>ACCEPT</onMatch>
<onMismatch>DENY</onMismatch>
</filter>
<!-- 日志日常打印文件 -->
<file>${LOG_HOME}/info.log</file>
<!-- 日志滚动规则 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 日志滚动文件名命令规则 -->
<FileNamePattern>${LOG_HOME}/info.%d{yyyy-MM-dd}.%i.AssertLog.zip</FileNamePattern>
<!--日志文件保留天数 -->
<MaxHistory>30</MaxHistory>
<!-- 日志保存总量 -->
<totalSizeCap>10GB</totalSizeCap>
<!-- 日志切分规则 -->
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- 文件切分压缩的大小阈值 -->
<maxFileSize>100MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<!-- 日志输出的样式 -->
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
<pattern>[requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n</pattern>
</encoder>
</appender>
<!-- 按照固定大小生成日志文件, error级别 -->
<appender name="ERROR" class="ch.qos.logback.core.rolling.RollingFileAppender">
<!-- 精确过滤, 只保存error级别 -->
<filter class="ch.qos.logback.classic.filter.LevelFilter">
<level>ERROR</level>
<onMatch>ACCEPT</onMatch>
<onMismatch>DENY</onMismatch>
</filter>
<file>${LOG_HOME}/error.log</file>
<!-- 日志滚动规则 -->
<rollingPolicy class="ch.qos.logback.core.rolling.TimeBasedRollingPolicy">
<!-- 日志滚动文件名命令规则 -->
<FileNamePattern>${LOG_HOME}/error.%d{yyyy-MM-dd}.%i.AssertLog.zip</FileNamePattern>
<!--日志文件保留天数 -->
<MaxHistory>30</MaxHistory>
<!-- 日志保存总量 -->
<totalSizeCap>10GB</totalSizeCap>
<!-- 日志切分规则 -->
<timeBasedFileNamingAndTriggeringPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedFNATP">
<!-- 文件切分压缩的大小阈值 -->
<maxFileSize>10MB</maxFileSize>
</timeBasedFileNamingAndTriggeringPolicy>
</rollingPolicy>
<!-- 日志输出的样式 -->
<encoder class="ch.qos.logback.classic.encoder.PatternLayoutEncoder">
<!--格式化输出：%d表示日期，%thread表示线程名，%-5level：级别从左显示5个字符宽度%msg：日志消息，%n是换行符 -->
<pattern>[requestId:%X{requestId}] [time:%d{yyyy-MM-dd HH:mm:ss.SSS}] [thread:%thread] [level:%-5level] [appName:${appName}] msg- %msg%n</pattern>
</encoder>
</appender>
<logger name="com.alibaba.nacos.client" level="OFF"></logger>
<logger name="com.netflix" level="OFF"></logger>
<logger name="RocketmqClient" additivity="false">
<level value="warn" />
<appender-ref ref="ERROR"/>
</logger>
<!-- 日志输出级别 -->
<root level="info">
<appender-ref ref="STDOUT"/>
<appender-ref ref="INFO"/>
<appender-ref ref="ERROR"/>
</root>
</configuration>
src/test/java/com/tongran/agent/server/TrAgentServerApplicationTests.java
+13
View File
@@ -0,0 +1,13 @@
package com.tongran.agent.server;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
@SpringBootTest
class TrAgentServerApplicationTests {
@Test
void contextLoads() {
}
}