diff --git a/src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java b/src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java index e9efb0c..9051c4d 100644 --- a/src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java +++ b/src/main/java/com/tongran/agent/server/netty/handler/AgentDecoderHandler.java @@ -43,7 +43,7 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter { AgentDispatcherManager agentDispatcherManager; // 用来临时保留没有处理过的请求报文 - private final Cache lruCache = CacheUtil.newLRUCache(5000); + private final Cache 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 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); }