定位并解决中文乱码问题
This commit is contained in:
gaoyutao
@@ -21,6 +21,9 @@ import io.netty.util.CharsetUtil;
|
||||
import org.springframework.stereotype.Component;
|
||||
|
||||
import javax.annotation.Resource;
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.Objects;
|
||||
|
||||
@Component
|
||||
@@ -47,114 +50,133 @@ public class AgentDecoderHandler extends ChannelInboundHandlerAdapter {
|
||||
*/
|
||||
@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);
|
||||
try {
|
||||
// 使用字节数组读取，避免字符串解码时丢失信息
|
||||
byte[] bytes = new byte[byteBuf.readableBytes()];
|
||||
byteBuf.readBytes(bytes);
|
||||
String rawMessage = new String(bytes, StandardCharsets.UTF_8);
|
||||
// ==== 精确乱码检测 ====
|
||||
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') {
|
||||
hasGarbledChars = true;
|
||||
break;
|
||||
}
|
||||
}
|
||||
//判定是否粘包
|
||||
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);
|
||||
}
|
||||
if (hasGarbledChars) {
|
||||
// 有乱码，说明消息被截断，缓存起来等待后续数据
|
||||
String clientKey = sessionManager.client(ctx);
|
||||
String cachedMsg = lruCache.get(clientKey);
|
||||
cachedMsg = cachedMsg == null ? "" : cachedMsg;
|
||||
|
||||
}
|
||||
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
|
||||
}
|
||||
lruCache.put(clientKey, cachedMsg + rawMessage, DateUnit.SECOND.getMillis() * 5000);
|
||||
|
||||
// 打印调试信息（不改变原有日志格式）
|
||||
AssertLog.info("<<[up1]:[raw-content]==>{} (truncated)", rawMessage);
|
||||
|
||||
byteBuf.release();
|
||||
return; // 跳过处理
|
||||
}
|
||||
AssertLog.info("<<[up1]:[raw-content]==>{}", rawMessage);
|
||||
|
||||
String clientKey = sessionManager.client(ctx);
|
||||
String cachedMsg = lruCache.get(clientKey);
|
||||
cachedMsg = cachedMsg == null ? "" : cachedMsg;
|
||||
|
||||
// 拼接缓存和当前消息
|
||||
String fullMessage = cachedMsg + rawMessage;
|
||||
|
||||
// 使用更精确的分割逻辑
|
||||
List<String> completeMessages = new ArrayList<>();
|
||||
StringBuilder currentMsg = new StringBuilder();
|
||||
|
||||
// 按消息边界分割
|
||||
String[] parts = fullMessage.split("(?=agent-client:)");
|
||||
for (String part : parts) {
|
||||
if (part.isEmpty()) continue;
|
||||
|
||||
currentMsg.append(part);
|
||||
String current = currentMsg.toString();
|
||||
|
||||
// 检查是否包含完整消息边界
|
||||
if (current.contains("@tong-ran")) {
|
||||
String[] messages = current.split("@tong-ran", -1); // 使用limit=-1保留空字符串
|
||||
|
||||
// 前面的都是完整消息
|
||||
for (int i = 0; i < messages.length - 1; i++) {
|
||||
String completeMsg = messages[i];
|
||||
if (!completeMsg.isEmpty()) {
|
||||
completeMessages.add(completeMsg);
|
||||
}
|
||||
}
|
||||
|
||||
// 最后一部分可能是未完成的消息
|
||||
String lastPart = messages[messages.length - 1];
|
||||
currentMsg = new StringBuilder(lastPart);
|
||||
}
|
||||
}catch (Exception e){
|
||||
AssertLog.error("=====channelRead:{}=====" + e.getMessage());
|
||||
}
|
||||
|
||||
// 处理缓存逻辑
|
||||
if (currentMsg.length() > 0) {
|
||||
// 有未完成的消息，存入缓存
|
||||
lruCache.put(clientKey, currentMsg.toString(), DateUnit.SECOND.getMillis() * 5000);
|
||||
} else {
|
||||
// 所有消息都处理完成，清除缓存
|
||||
lruCache.remove(clientKey);
|
||||
}
|
||||
|
||||
// 处理完整的消息
|
||||
for (String message : completeMessages) {
|
||||
processCompleteMessage(ctx, message);
|
||||
}
|
||||
|
||||
} catch (Exception e) {
|
||||
AssertLog.error("=====channelRead error:{}=====", e.getMessage(), e);
|
||||
} finally {
|
||||
byteBuf.release(); // 确保资源释放
|
||||
}
|
||||
if (isClear) {
|
||||
lruCache.remove(sessionManager.client(ctx));
|
||||
}
|
||||
byteBuf.release(); // 释放 ByteBuf 资源（重要！）
|
||||
} else {
|
||||
System.out.println("Unexpected message type: " + msg.getClass());
|
||||
}
|
||||
}
|
||||
|
||||
private void processCompleteMessage(ChannelHandlerContext ctx, String message) {
|
||||
try {
|
||||
// 清理消息格式
|
||||
String cleanMessage = message.replaceAll("^agent-client:", "").trim();
|
||||
if (cleanMessage.isEmpty()) return;
|
||||
|
||||
AssertLog.info("<<[up2]:IP:{},[clean-content]==>{}", sessionManager.client(ctx), cleanMessage);
|
||||
|
||||
// 解析JSON
|
||||
JSONObject jsonObject = JSONObject.parseObject(cleanMessage);
|
||||
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()
|
||||
.clientId(clientId)
|
||||
.dataType(response.getDataType())
|
||||
.data(response.getContent())
|
||||
.build();
|
||||
ctx.fireChannelRead(agentMessage); // 传递到下一个handler
|
||||
}
|
||||
}
|
||||
}
|
||||
} catch (Exception e) {
|
||||
AssertLog.error("=====processCompleteMessage error, message: {}, error: {}=====",
|
||||
message, e.getMessage(), e);
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
Reference in New Issue
Block a user