mtr-agent-server v1.0.0初始化
This commit is contained in:
gaoyutao
@@ -0,0 +1,107 @@
|
||||
<?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>
|
||||
<!-- 打包跳过测试 -->
|
||||
<plugin>
|
||||
<groupId>org.apache.maven.plugins</groupId>
|
||||
<artifactId>maven-surefire-plugin</artifactId>
|
||||
<version>${maven-surefire-plugin.version}</version>
|
||||
<configuration>
|
||||
<skipTests>true</skipTests>
|
||||
</configuration>
|
||||
</plugin>
|
||||
</plugins>
|
||||
</build>
|
||||
|
||||
</project>
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
package com.tongran.agent.server.controller;
|
||||
|
||||
import com.alibaba.fastjson2.JSONObject;
|
||||
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("/send")
|
||||
public void sendMessage(@Validated @RequestBody Message dto) {
|
||||
JSONObject jsonObject = new JSONObject();
|
||||
jsonObject.put("clientId", dto.getClientId());
|
||||
jsonObject.put("dataType", dto.getDataType());
|
||||
jsonObject.put("data", dto.getData());
|
||||
tcpService.sendMessage(jsonObject.toString());
|
||||
}
|
||||
|
||||
|
||||
|
||||
}
|
||||
@@ -0,0 +1,118 @@
|
||||
package com.tongran.agent.server.controller;
|
||||
|
||||
import com.tongran.agent.server.core.session.SessionManager;
|
||||
import com.tongran.agent.server.netty.basics.AgentDispatcherManager;
|
||||
import com.tongran.agent.server.rockermq.AgentProducer;
|
||||
import com.tongran.agent.server.utils.AssertLog;
|
||||
import lombok.extern.slf4j.Slf4j;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
|
||||
@Slf4j
|
||||
@Service
|
||||
public class TcpService {
|
||||
protected final SessionManager sessionManager;
|
||||
|
||||
@Resource
|
||||
private AgentDispatcherManager agentDispatcherManager;
|
||||
|
||||
// @Resource
|
||||
// private MultiTargetNettyClient client;
|
||||
|
||||
@Resource
|
||||
private AgentProducer agentProducer;
|
||||
|
||||
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(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());
|
||||
// }
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -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
|
||||
|
||||
|
||||
}
|
||||
@@ -0,0 +1,86 @@
|
||||
package com.tongran.agent.server.core.enums;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Getter;
|
||||
import lombok.NoArgsConstructor;
|
||||
|
||||
/**
|
||||
* 控制码
|
||||
*/
|
||||
@Getter
|
||||
@AllArgsConstructor
|
||||
@NoArgsConstructor
|
||||
public enum MsgEnum {
|
||||
建立连接("CONNECT"),
|
||||
|
||||
建立连接应答("CONNECT_RSP"),
|
||||
|
||||
注册("REGISTER"),
|
||||
|
||||
注册应答("REGISTER_RSP"),
|
||||
|
||||
获取最新策略("GET_POLICY"),
|
||||
|
||||
获取最新策略应答("GET_POLICY_RSP"),
|
||||
|
||||
断开("DISCONNECT"),
|
||||
|
||||
断开应答("DISCONNECT_RSP"),
|
||||
|
||||
心跳上报("HEARTBEAT"),
|
||||
|
||||
心跳上报应答("HEARTBEAT_RSP"),
|
||||
|
||||
CPU上报("CPU"),
|
||||
|
||||
磁盘上报("DISK"),
|
||||
|
||||
容器上报("DOCKER"),
|
||||
|
||||
内存上报("MEMORY"),
|
||||
|
||||
网络上报("NET"),
|
||||
|
||||
挂载上报("POINT"),
|
||||
|
||||
系统其他上报("OTHER_SYSTEM"),
|
||||
|
||||
交换机上报("SWITCHBOARD"),
|
||||
|
||||
采集上报应答("COLLECT_RSP"),
|
||||
|
||||
开启或更新系统采集("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"),
|
||||
|
||||
多网IP探测上报("NETWORK_DETECT");
|
||||
|
||||
private String value;
|
||||
|
||||
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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));
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
}
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -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 "";
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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 "";
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,420 @@
|
||||
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.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 AgentMqConfig agentMqConfig;
|
||||
|
||||
@Resource
|
||||
private AgentProducer agentProducer;
|
||||
|
||||
@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("data", data);
|
||||
MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
|
||||
agentProducer.asyncSend(mqMsg);
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
@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);
|
||||
JSONObject json = new JSONObject();
|
||||
json.put("resCode",1);
|
||||
return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.心跳上报应答.getValue()).content(json.toString()).build();
|
||||
}
|
||||
}
|
||||
|
||||
@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);
|
||||
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 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);
|
||||
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 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);
|
||||
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 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);
|
||||
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 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);
|
||||
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
|
||||
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 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);
|
||||
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 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);
|
||||
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 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 SystemCollectClosetHandler implements AgentHandler {
|
||||
@Override
|
||||
public String downHandle(String data, String clientId) {
|
||||
return data;
|
||||
}
|
||||
}
|
||||
|
||||
@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 SwitchCollectClosetHandler implements AgentHandler {
|
||||
@Override
|
||||
public String downHandle(String data, String clientId) {
|
||||
return data;
|
||||
}
|
||||
}
|
||||
|
||||
@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 = new JSONObject();
|
||||
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);
|
||||
JSONObject json = JSONObject.parseObject(data);
|
||||
int resCode = 0;
|
||||
if(json.containsKey("resCode")){
|
||||
resCode = json.getInteger("resCode");
|
||||
}
|
||||
// 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;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
}
|
||||
@@ -0,0 +1,174 @@
|
||||
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), content, DateUnit.SECOND.getMillis() * 5000);
|
||||
}
|
||||
}
|
||||
//判定是否粘包
|
||||
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 = ms;
|
||||
}
|
||||
}
|
||||
}
|
||||
// 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){
|
||||
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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
}catch (Exception e){
|
||||
AssertLog.error("=====channelRead:{}=====" + e.getMessage());
|
||||
}
|
||||
}
|
||||
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;
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
// }
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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();// 暂未实现
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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){};
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
}
|
||||
+51
@@ -0,0 +1,51 @@
|
||||
package com.tongran.agent.server.rockermq.listener;
|
||||
|
||||
import cn.hutool.core.util.ObjectUtil;
|
||||
import com.alibaba.fastjson2.JSON;
|
||||
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.utils.AssertLog;
|
||||
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 = "tr_agent_down", consumerGroup = "tr_agent_down_group")
|
||||
public class AgentConsumerDownListener implements RocketMQListener<String> {
|
||||
|
||||
protected final SessionManager sessionManager;
|
||||
|
||||
@Resource
|
||||
private AgentDispatcherManager agentDispatcherManager;
|
||||
|
||||
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);
|
||||
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());
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
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.model.Message;
|
||||
import com.tongran.agent.server.service.AgentService;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
@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;
|
||||
return false;
|
||||
}
|
||||
|
||||
@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);
|
||||
}
|
||||
}
|
||||
@@ -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);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,81 @@
|
||||
package com.tongran.agent.server.utils;
|
||||
|
||||
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();
|
||||
// }
|
||||
}
|
||||
@@ -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());
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,29 @@
|
||||
server:
|
||||
port: -1
|
||||
servlet:
|
||||
context-path: /tr-agent-server
|
||||
|
||||
# 接口文档配置
|
||||
knife4j:
|
||||
enable: true
|
||||
production: false # 开启屏蔽文档资源
|
||||
|
||||
# 日志配置
|
||||
logging:
|
||||
file:
|
||||
# path: /usr/local/tongran/logs
|
||||
path: /usr/local/tongran_server/logs
|
||||
# path: D:/job/agent-logs/logs
|
||||
|
||||
rocketmq:
|
||||
name-server: 172.16.15.103:9876
|
||||
enabled: true
|
||||
|
||||
tcp:
|
||||
netty:
|
||||
charge:
|
||||
enable: true
|
||||
name: AGENT-SERVER-服务
|
||||
port: 6620
|
||||
readerIdleTime: 300
|
||||
|
||||
@@ -0,0 +1,43 @@
|
||||
spring:
|
||||
profiles:
|
||||
active: dev
|
||||
mvc:
|
||||
pathmatch:
|
||||
matching-strategy: ant_path_matcher
|
||||
application:
|
||||
name: tr-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: tr_agent_up_group
|
||||
agent-topic: tr_agent_up
|
||||
consumer:
|
||||
group: tongran-group
|
||||
topic: default_topic
|
||||
agent-group: tr_agent_down_group
|
||||
agent-topic: tr_agent_down
|
||||
@@ -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>
|
||||
@@ -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>
|
||||
Reference in New Issue
Block a user