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

TCP数据包优化

Browse Source
This commit is contained in:
qiminbao
2025-09-23 09:45:00 +08:00
parent 9e75926e77
commit 86415bf526
1 changed files with 36 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/tongran/agent/client/netty/handler/AgentDecoderHandler.java
+36 -22
View File
@@ -50,6 +50,7 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
AssertLog.info("<<[up1]:IP:{},[up-content]==>{}", sessionManager.client(ctx),messages); AssertLog.info("<<[up1]:IP:{},[up-content]==>{}", sessionManager.client(ctx),messages);
String content = ""; String content = "";
StringBuilder sb = new StringBuilder();
boolean isClear = false; boolean isClear = false;
String tempMsg = lruCache.get(sessionManager.client(ctx)); String tempMsg = lruCache.get(sessionManager.client(ctx));
tempMsg = tempMsg == null ? "" : tempMsg; tempMsg = tempMsg == null ? "" : tempMsg;
@@ -66,11 +67,13 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
if(endsWith){ if(endsWith){
lruCache.remove(sessionManager.client(ctx)); lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0; tmpMsgSize = 0;
content = arr_msg[0]+"@tong-ran"; sb.append(arr_msg[0]+"@tong-ran");
// content = arr_msg[0]+"@tong-ran";
}else{ }else{
lruCache.remove(sessionManager.client(ctx)); lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0; tmpMsgSize = 0;
content = arr_msg[0]; sb.append(arr_msg[0]);
// content = arr_msg[0];
lruCache.put(sessionManager.client(ctx), content, DateUnit.SECOND.getMillis() * 3000); lruCache.put(sessionManager.client(ctx), content, DateUnit.SECOND.getMillis() * 3000);
} }
} }
@@ -79,15 +82,23 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
if(!endsWith){ if(!endsWith){
lruCache.remove(sessionManager.client(ctx)); lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0; tmpMsgSize = 0;
content = arr_msg[0]+"@tong-ran"; for (int i = 0; i < arr_msg.length; i++) {
lruCache.put(sessionManager.client(ctx), arr_msg[1], DateUnit.SECOND.getMillis() * 3000); if(i == arr_msg.length -1){
lruCache.put(sessionManager.client(ctx), arr_msg[i], DateUnit.SECOND.getMillis() * 5000);
}else{
sb.append(arr_msg[i].replaceAll("agent-server:","")+"@tong-ran");
}
}
}else{ }else{
lruCache.remove(sessionManager.client(ctx)); lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0; tmpMsgSize = 0;
content = String.join("@tong-ran", arr_msg); sb.append(String.join("@tong-ran", arr_msg));
// content = String.join("@tong-ran", arr_msg);
} }
} }
} }
content = sb.toString();
if (tmpMsgSize > 0) { if (tmpMsgSize > 0) {
content = tempMsg + messages; content = tempMsg + messages;
endsWith = content.endsWith("@tong-ran"); endsWith = content.endsWith("@tong-ran");
@@ -95,33 +106,36 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
isClear = true; isClear = true;
}else{ }else{
lruCache.remove(sessionManager.client(ctx)); lruCache.remove(sessionManager.client(ctx));
tmpMsgSize = 0;
lruCache.put(sessionManager.client(ctx), content, DateUnit.SECOND.getMillis() * 3000); lruCache.put(sessionManager.client(ctx), content, DateUnit.SECOND.getMillis() * 3000);
} }
} }
AssertLog.info("<<[up2]:IP:{},[up-content]==>{}", sessionManager.client(ctx),content); AssertLog.info("<<[up2]:IP:{},[up-content]==>{}", sessionManager.client(ctx),content);
endsWith = content.endsWith("@tong-ran"); endsWith = content.endsWith("@tong-ran");
//判断是否是整包 //判断是否是整包
if(endsWith){ if(endsWith){
content = content.replaceAll("agent-server:","");
String[] arr = content.split("@tong-ran"); String[] arr = content.split("@tong-ran");
for (String message : arr) { try {
JSONObject jsonObject = JSONObject.parseObject(message); for (String message : arr) {
String clientId = jsonObject.getString("clientId"); JSONObject jsonObject = JSONObject.parseObject(message);
String dataType = jsonObject.getString("dataType"); String clientId = jsonObject.getString("clientId");
String data = jsonObject.getString("data"); String dataType = jsonObject.getString("dataType");
AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value); String data = jsonObject.getString("data");
if (ObjectUtil.isNotEmpty(msgHandler)) { AgentHandler msgHandler = agentDispatcherManager.getHandler(dataType + "&" + AgentDispatcher.VersionEnum.V1.value);
UpMsgResponse response = msgHandler.upHandle(data, clientId ); if (ObjectUtil.isNotEmpty(msgHandler)) {
AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", response.getClientId(), dataType, message); UpMsgResponse response = msgHandler.upHandle(data, clientId );
if(Objects.nonNull(response)){ AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}", response.getClientId(), dataType, message);
Message agentMessage = Message.builder().build(); if(Objects.nonNull(response)){
agentMessage.setClientId(response.getClientId()); Message agentMessage = Message.builder().build();
agentMessage.setDataType(response.getDataType()); agentMessage.setClientId(response.getClientId());
agentMessage.setData(response.getContent()); agentMessage.setDataType(response.getDataType());
ctx.fireChannelRead(agentMessage);//传递到下一个handler agentMessage.setData(response.getContent());
ctx.fireChannelRead(agentMessage);//传递到下一个handler
}
} }
} }
}catch (Exception e){
AssertLog.error("=====channelRead:{}=====" + e.getMessage());
} }
} }
if (isClear) { if (isClear) {