指令下发优化

pom.xml

@@ -92,6 +92,15 @@
                 <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>
 

src/main/java/com/tongran/agent/server/controller/TcpController.java

@@ -1,6 +1,6 @@
 package com.tongran.agent.server.controller;
 
-import com.alibaba.fastjson2.JSON;
+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;
@@ -15,17 +15,13 @@ public class TcpController {
     @Resource
     private TcpService tcpService;
 
-    @PostMapping("/create")
-    public void create(@Validated @RequestBody Message dto) {
-        tcpService.create(dto.getClientId());
-    }
-
     @PostMapping("/send")
     public void sendMessage(@Validated @RequestBody Message dto) {
-        String json =  JSON.toJSONString(dto);
+        JSONObject jsonObject = new JSONObject();
-//        ScriptPolicy policy = JSON.parseObject(dto.getData(), ScriptPolicy.class);
+        jsonObject.put("clientId", dto.getClientId());
-//        System.out.println(policy.getFiles().get(0).getFileData());
+        jsonObject.put("dataType", dto.getDataType());
-        tcpService.sendMessage(json);
+        jsonObject.put("data", dto.getData());
+        tcpService.sendMessage(jsonObject.toString());
     }
 
 

src/main/java/com/tongran/agent/server/controller/TcpService.java

@@ -10,6 +10,8 @@ import com.tongran.agent.server.netty.annotation.AgentDispatcher;
 import com.tongran.agent.server.netty.basics.AgentDispatcherManager;
 import com.tongran.agent.server.netty.basics.AgentHandler;
 import com.tongran.agent.server.netty.model.Message;
+import com.tongran.agent.server.rockermq.AgentProducer;
+import com.tongran.agent.server.rockermq.model.MqMsg;
 import com.tongran.agent.server.utils.AssertLog;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;
@@ -29,6 +31,9 @@ public class TcpService {
     @Resource
     private MultiTargetNettyClient client;
 
+    @Resource
+    private AgentProducer agentProducer;
+
     public TcpService() {
         this.sessionManager = SessionManager.getInstance();
     }
@@ -60,67 +65,58 @@ public class TcpService {
         AssertLog.info("consumer==> received down message: {}", message);
         try {
             Message dto = JSON.parseObject(message, Message.class);
-            if(Objects.nonNull(dto)){
+            if(StringUtils.equals(dto.getDataType(), MsgEnum.注册.getValue())){
-                if(StringUtils.equals(dto.getDataType(), MsgEnum.注册.getValue())){
+                JSONObject jsonObject = JSONObject.parseObject(dto.getData());
-                    JSONObject jsonObject = JSONObject.parseObject(dto.getData());
+                String clientIp = "";
-                    String clientIp = "";
+                int clientPort = 0;
-                    int port = 0;
+                String switchBoard = "";
-                    long timestamp = 0;
+                long timestamp = 0;
-                    if(jsonObject.containsKey("clientIp")){
+                if(jsonObject.containsKey("clientIp")){
-                        clientIp = jsonObject.getString("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);
                     }
-                    if(jsonObject.containsKey("port")){
+                    Message msg = Message.builder().clientId(dto.getClientId()).dataType(MsgEnum.注册.getValue()).data(object.toString()).build();
-                        port = jsonObject.getInteger("port");
+                    // 将对象转为 JSON 字符串
-                    }
+                    client.sendMessages(dto.getClientId(), msg);
-                    if(jsonObject.containsKey("timestamp")){
+                }
-                        timestamp = jsonObject.getLongValue("timestamp");
+            } else {
-                    }
+                AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value);
-                    if(StringUtils.isBlank(clientIp) || port == 0){
+                if (ObjectUtil.isNotEmpty(msgHandler)) {
-                        //返回提醒ip 端口 空
+                    String msg = msgHandler.downHandle(dto.getData(), dto.getClientId());
-                        return;
+                    Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build();
-                    }
+                    if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) {
-                    boolean success = client.createConnection(dto.getClientId(), clientIp, port, 5);
+                        sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage);
-                    if(success){
-                        String community = "";
-                        String ip = "";
-                        int switchPort = 0;
-                        String oids = "";
-                        String params = "";
-                        if(jsonObject.containsKey("switchBoard")){
-                            String switchBoard = jsonObject.getString("switchBoard");
-                            jsonObject = JSONObject.parseObject(switchBoard);
-                            community = jsonObject.getString("community");
-                            ip = jsonObject.getString("ip");
-                            switchPort = jsonObject.getInteger("port");
-                            oids = jsonObject.getString("oids");
-                            params = jsonObject.getString("params");
-                        }
-                        //连接成功
-                        // 连接建立后可以发送登录认证消息
-                        JSONObject object = new JSONObject();
-                        object.put("clientId",dto.getClientId());
-                        object.put("timestamp",timestamp);
-                        JSONObject switchBoard = new JSONObject();
-                        switchBoard.put("community", community);
-                        switchBoard.put("ip", ip);
-                        switchBoard.put("port", switchPort);
-                        switchBoard.put("oids", oids);
-                        switchBoard.put("params", params);
-                        object.put("switchBoard",switchBoard.toString());
-                        Message msg = Message.builder().clientId(dto.getClientId()).dataType(MsgEnum.注册.getValue()).data(object.toString()).build();
-                        // 将对象转为 JSON 字符串
-                        client.sendMessages(dto.getClientId(), msg);
-                    }else{
-                        //连接失败
-                    }
-                } else {
-                    AgentHandler msgHandler = agentDispatcherManager.getHandler(dto.getDataType() + "&" + AgentDispatcher.VersionEnum.V1.value);
-                    if (ObjectUtil.isNotEmpty(msgHandler)) {
-                        String msg = msgHandler.downHandle(dto.getData(), dto.getClientId());
-                        Message chargeMessage = Message.builder().clientId(dto.getClientId()).dataType(dto.getDataType()).data(msg).build();
-                        if (Objects.nonNull(sessionManager.getSessionById(dto.getClientId()))) {
-                            sessionManager.writeAndFlush(sessionManager.getSessionById(dto.getClientId()).getChannel(), chargeMessage);
-                        }
                     }
                 }
             }

src/main/java/com/tongran/agent/server/core/enums/MsgEnum.java

@@ -22,6 +22,8 @@ public enum MsgEnum {
 
     心跳上报("HEARTBEAT"),
 
+    心跳上报应答("HEARTBEAT_RSP"),
+
     CPU上报("CPU"),
 
     磁盘上报("DISK"),

src/main/java/com/tongran/agent/server/netty/enpoint/AgentEndpoint.java

@@ -81,7 +81,9 @@ public class AgentEndpoint {
             jsonObject.put("data", data);
             MqMsg mqMsg = MqMsg.builder().topic(agentMqConfig.getAgentTopic()).content(jsonObject.toString()).build();
             agentProducer.asyncSend(mqMsg);
-            return null;
+            JSONObject json = new JSONObject();
+            json.put("resCode",1);
+            return UpMsgResponse.builder().clientId(clientId).dataType(MsgEnum.心跳上报应答.getValue()).content(json.toString()).build();
         }
     }
 

src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java

@@ -121,8 +121,8 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
 					if(Objects.nonNull(agentDispatcherManager)){
 						AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
 						if (ObjectUtil.isNotEmpty(msgHandler)) {
-							UpMsgResponse response = msgHandler.upHandle(data, clientId );
 							AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", clientId, dataType, data);
+							UpMsgResponse response = msgHandler.upHandle(data, clientId );
 							if(Objects.nonNull(response)){
 								Message agentMessage = Message.builder().build();
 								agentMessage.setClientId(clientId);

src/main/resources/application-dev.yml

@@ -1,7 +1,7 @@
 server:
-  port: 9018
+  port: 7012
   servlet:
-    context-path: /agent-server
+    context-path: /tr-agent-server
   
 # 接口文档配置  
 knife4j:
@@ -12,9 +12,13 @@ knife4j:
 logging:
   file:
 #    path: /usr/local/tongran/logs
-    path: /data/agent-server/logs
+    path: /data/tr-agent-server/logs
 #    path: D:/job/agent-logs/logs
 
+rocketmq:
+  name-server: 172.16.15.103:9876
+  enabled: true
+
 netty:
   server:
 #    host: 172.16.15.103

src/main/resources/application.yml

@@ -5,7 +5,7 @@ spring:
     pathmatch:
       matching-strategy: ant_path_matcher
   application:
-    name: agent-server
+    name: tr-agent-server
     version: 1.0
   web:
     resources: