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

优化tcp解码

Browse Source
This commit is contained in:
gaoyutao
2026-03-03 17:46:11 +08:00
parent 5244f1d09b
commit 7bd91aa14f
1 changed files with 71 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java
+71 -30
View File
@@ -43,7 +43,7 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
AgentDispatcherManager agentDispatcherManager;
// 用来临时保留没有处理过的请求报文
private final Cache<String, String> lruCache = CacheUtil.newLRUCache(5000);
private final Cache<String, byte[]> byteCache = CacheUtil.newLRUCache(5000);
/**
* 消息解码器
@@ -52,44 +52,63 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
public void channelRead(ChannelHandlerContext ctx, Object msg) {
if (msg instanceof ByteBuf) {
ByteBuf byteBuf = (ByteBuf) msg;
boolean released = false; // 标记是否已释放
try {
// 使用字节数组读取，避免字符串解码时丢失信息
// 使用字节数组读取
byte[] bytes = new byte[byteBuf.readableBytes()];
byteBuf.readBytes(bytes);
String rawMessage = new String(bytes, StandardCharsets.UTF_8);
// ==== 精确乱码检测 ====
String clientKey = sessionManager.client(ctx);
// 1. 获取缓存的字节数组
byte[] cachedBytes = byteCache.get(clientKey);
// 2. 合并字节数组
byte[] mergedBytes;
if (cachedBytes != null && cachedBytes.length > 0) {
mergedBytes = new byte[cachedBytes.length + bytes.length];
System.arraycopy(cachedBytes, 0, mergedBytes, 0, cachedBytes.length);
System.arraycopy(bytes, 0, mergedBytes, cachedBytes.length, bytes.length);
// 合并后清除旧缓存
byteCache.remove(clientKey);
} else {
mergedBytes = bytes;
}
// 3. 尝试解码
String rawMessage = new String(mergedBytes, StandardCharsets.UTF_8);
// 4. 检查是否有乱码
boolean hasGarbledChars = false;
// 检查字符串末尾是否有UTF-8替换字符（乱码）
for (int i = Math.max(0, rawMessage.length() - 3); i < rawMessage.length(); i++) {
if (i >= 0 && rawMessage.charAt(i) == '\ufffd') {
for (int i = 0; i < rawMessage.length(); i++) {
if (rawMessage.charAt(i) == '\ufffd') {
hasGarbledChars = true;
break;
}
}
if (hasGarbledChars) {
// 有乱码，说明消息被截断，缓存起来等待后续数据
String clientKey = sessionManager.client(ctx);
String cachedMsg = lruCache.get(clientKey);
cachedMsg = cachedMsg == null ? "" : cachedMsg;
// 有乱码，缓存字节数组等待后续数据
byteCache.put(clientKey, mergedBytes, DateUnit.SECOND.getMillis() * 5000);
lruCache.put(clientKey, cachedMsg + rawMessage, DateUnit.SECOND.getMillis() * 5000);
// 打印调试信息（不改变原有日志格式）
AssertLog.info("<<[up1]:[raw-content]==>{} (truncated)", rawMessage);
// 打印调试信息
int garbledIndex = rawMessage.indexOf('\ufffd');
AssertLog.info("<<[up1]:[bytes-cached]==>长度: {}, 乱码位置: {}, 预览: {}",
mergedBytes.length, garbledIndex, rawMessage);
byteBuf.release();
released = true;
return; // 跳过处理
}
// 5. 没有乱码，处理完整消息
AssertLog.info("<<[up1]:[raw-content]==>{}", rawMessage);
String clientKey = sessionManager.client(ctx);
String cachedMsg = lruCache.get(clientKey);
cachedMsg = cachedMsg == null ? "" : cachedMsg;
// 6. 处理消息分割（原有逻辑）
// 这里不需要从缓存获取，因为已经在字节层面处理过了
String fullMessage = rawMessage;
// 拼接缓存和当前消息
String fullMessage = cachedMsg + rawMessage;
// 使用更精确的分割逻辑
// 7. 使用更精确的分割逻辑
List<String> completeMessages = new ArrayList<>();
StringBuilder currentMsg = new StringBuilder();
@@ -119,16 +138,22 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
}
}
// 处理缓存逻辑
// 8. 处理缓存逻辑
if (currentMsg.length() > 0) {
// 有未完成的消息，存入缓存
lruCache.put(clientKey, currentMsg.toString(), DateUnit.SECOND.getMillis() * 5000);
// 有未完成的消息，需要缓存
// 注意：这里缓存的是字符串，不是字节数组
String remainingMsg = currentMsg.toString();
if (remainingMsg.length() > 0) {
// 将未完成的字符串转为字节数组缓存
byte[] remainingBytes = remainingMsg.getBytes(StandardCharsets.UTF_8);
byteCache.put(clientKey, remainingBytes, DateUnit.SECOND.getMillis() * 5000);
}
} else {
// 所有消息都处理完成，清除缓存
lruCache.remove(clientKey);
byteCache.remove(clientKey);
}
// 处理完整的消息
// 9. 处理完整的消息
for (String message : completeMessages) {
processCompleteMessage(ctx, message);
}
@@ -136,7 +161,9 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
} catch (Exception e) {
AssertLog.error("=====channelRead error:{}=====", e.getMessage(), e);
} finally {
byteBuf.release(); // 确保资源释放
if(!released && byteBuf.refCnt() > 0){
byteBuf.release(); // 确保资源释放
}
}
} else {
System.out.println("Unexpected message type: " + msg.getClass());
@@ -149,7 +176,9 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
String cleanMessage = message.replaceAll("^agent-client:", "").trim();
if (cleanMessage.isEmpty()) return;
AssertLog.info("<<[up2]:IP:{},[clean-content]==>{}", sessionManager.client(ctx), cleanMessage);
// 截断过长的日志
AssertLog.info("<<[up2]:IP:{},[clean-content]==>{}...",
sessionManager.client(ctx), cleanMessage);
// 解析JSON
JSONObject jsonObject = JSONObject.parseObject(cleanMessage);
@@ -160,7 +189,8 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
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);
AssertLog.info("<<[up-after-handle]:clientId:{},type={},[handle-content]={}",
clientId, dataType, data);
UpMsgResponse response = msgHandler.upHandle(data, clientId);
if (Objects.nonNull(response)) {
@@ -174,6 +204,17 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
}
}
} catch (Exception e) {
// 如果解析失败，可能是消息不完整，尝试缓存字节
try {
String clientKey = sessionManager.client(ctx);
// 将失败的消息转为字节数组缓存
byte[] failedBytes = message.getBytes(StandardCharsets.UTF_8);
byteCache.put(clientKey, failedBytes, DateUnit.SECOND.getMillis() * 5000);
AssertLog.warn("<<[parse-failed-cached]: 消息解析失败，已缓存字节，长度: {}", failedBytes.length);
} catch (Exception cacheEx) {
AssertLog.error("=====缓存失败消息时出错:{}=====", cacheEx.getMessage(), cacheEx);
}
AssertLog.error("=====processCompleteMessage error, message: {}, error: {}=====",
message, e.getMessage(), e);
}