Java 开发 - 粘包处理器 - 基于消息头 + 消息体（使用 ByteBuffer）（魔数验证、长度验证）

weixin_52173250

445人浏览 · 2025-11-15 23:15:37

weixin_52173250 · 2025-11-15 23:15:37 发布

一、原始版本

DynamicLengthPacketByteBufferHandler.java

public class DynamicLengthPacketByteBufferHandler {

    private ByteBuffer buffer = ByteBuffer.allocate(0);
    private int parseFlag = PARSE_FLAG_HEADER;
    public static final int HEADER_LENGTH = 4;
    private int bodyLength;

    private static final int PARSE_FLAG_HEADER = 0;
    private static final int PARSE_FLAG_BODY = 1;

    public List<byte[]> handleData(byte[] data) {

        // 合并缓冲区
        ByteBuffer newBuffer = ByteBuffer.allocate(buffer.remaining() + data.length);
        newBuffer.put(buffer);
        newBuffer.put(data);
        newBuffer.flip(); // 切换到读模式
        buffer = newBuffer;

        List<byte[]> completeResults = new ArrayList<>();

        while (buffer.remaining() > 0) {
            if (parseFlag == PARSE_FLAG_HEADER) {

                // 处理消息头

                if (buffer.remaining() < HEADER_LENGTH) {
                    break;
                }
                bodyLength = buffer.getInt();
                parseFlag = PARSE_FLAG_BODY;
            }

            // 处理消息体

            if (buffer.remaining() < bodyLength) {

                // 回退位置，消息头已经读取，但是消息体不完整
                buffer.position(buffer.position() - HEADER_LENGTH);
                parseFlag = PARSE_FLAG_HEADER;
                break;
            }

            byte[] completeResult = new byte[bodyLength];
            buffer.get(completeResult);
            completeResults.add(completeResult);
            parseFlag = PARSE_FLAG_HEADER;
        }

        // 保存剩余数据
        if (buffer.remaining() > 0) {
            ByteBuffer remaining = ByteBuffer.allocate(buffer.remaining());
            remaining.put(buffer);
            remaining.flip();
            buffer = remaining;
        }

        return completeResults;
    }

    public void clear() {
        buffer = ByteBuffer.allocate(0);
        parseFlag = PARSE_FLAG_HEADER;
    }

    public static byte[] buildPacket(String data) {
        byte[] body = data.getBytes();
        ByteBuffer packet = ByteBuffer.allocate(4 + body.length);
        packet.putInt(body.length);
        packet.put(body);
        return packet.array();
    }
}

原始版本对异常长度完全没有处理能力，异常的长度包括错误的长度值、过长的长度值

错误的长度值：直接采用读取到的任何长度值，一旦解析到错误长度就卡住，无法继续处理后续数据
过长的长度值：例如，如果读取到过长的长度值，会尝试分配过大内存，导致内存溢出

二、原始版本测试用例

遇到错误的长度值

byte[] someData1 = DynamicLengthPacketByteBufferHandler.buildPacket("Hello");
byte[] someData2 = DynamicLengthPacketByteBufferHandler.buildPacket("World");
byte[] someData3 = DynamicLengthPacketByteBufferHandler.buildPacket("Java");

someData1[0] = (byte) 0x00;
someData1[1] = (byte) 0x00;
someData1[2] = (byte) 0x00;
someData1[3] = (byte) 0x07;

DynamicLengthPacketByteBufferHandler handler = new DynamicLengthPacketByteBufferHandler();

List<byte[]> results = handler.handleData(someData1);

System.out.println("第 1 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData2);

System.out.println("第 2 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData3);

System.out.println("第 3 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

# 输出结果

第 1 次解析结果数量: 0
第 2 次解析结果数量: 1
解析内容: Hello  
第 3 次解析结果数量: 0

遇到过长的长度值

byte[] someData1 = DynamicLengthPacketByteBufferHandler.buildPacket("Hello");
byte[] someData2 = DynamicLengthPacketByteBufferHandler.buildPacket("World");
byte[] someData3 = DynamicLengthPacketByteBufferHandler.buildPacket("Java");

someData1[0] = (byte) 0x0F;
someData1[1] = (byte) 0xFF;
someData1[2] = (byte) 0xFF;
someData1[3] = (byte) 0xFF;

DynamicLengthPacketByteBufferHandler handler = new DynamicLengthPacketByteBufferHandler();

List<byte[]> results = handler.handleData(someData1);

System.out.println("第 1 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData2);

System.out.println("第 2 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData3);

System.out.println("第 3 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

# 输出结果

第 1 次解析结果数量: 0
第 2 次解析结果数量: 0
第 3 次解析结果数量: 0

byte[] someData1 = DynamicLengthPacketByteBufferHandler.buildPacket("Hello");
byte[] someData2 = DynamicLengthPacketByteBufferHandler.buildPacket("World");
byte[] someData3 = DynamicLengthPacketByteBufferHandler.buildPacket("Java");

someData1[0] = (byte) 0xFF;
someData1[1] = (byte) 0xFF;
someData1[2] = (byte) 0xFF;
someData1[3] = (byte) 0xFF;

DynamicLengthPacketByteBufferHandler handler = new DynamicLengthPacketByteBufferHandler();

List<byte[]> results = handler.handleData(someData1);

System.out.println("第 1 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData2);

System.out.println("第 2 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData3);

System.out.println("第 3 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

# 输出结果

Exception in thread "main" java.lang.NegativeArraySizeException: -1

三、优化版本

DynamicLengthPacketByteBufferHandler.java

public class DynamicLengthPacketByteBufferHandler {

    private ByteBuffer buffer = ByteBuffer.allocate(0);

    public static final int HEADER_LENGTH = 8;

    private static final int MAGIC_NUMBER = 0x12345678;

    private int bodyLength;
    private static final int MAX_BODY_LENGTH = 20 * 1024 * 1024;
    private static final int MIN_BODY_LENGTH = 0;

    private int parseFlag = PARSE_FLAG_HEADER;
    private static final int PARSE_FLAG_HEADER = 0;
    private static final int PARSE_FLAG_BODY = 1;

    public List<byte[]> handleData(byte[] data) {
        ByteBuffer newBuffer = ByteBuffer.allocate(buffer.remaining() + data.length);
        newBuffer.put(buffer);
        newBuffer.put(data);
        newBuffer.flip(); // 切换到读模式
        buffer = newBuffer;

        List<byte[]> completeResults = new ArrayList<>();

        while (buffer.remaining() > 0) {
            if (parseFlag == PARSE_FLAG_HEADER) {

                // 处理消息头

                if (buffer.remaining() < HEADER_LENGTH) {
                    break;
                }

                buffer.mark();

                int magic = buffer.getInt();
                if (magic != MAGIC_NUMBER) {
                    buffer.reset();
                    buffer.get();
                    continue;
                }
                bodyLength = buffer.getInt();
                if (bodyLength < MIN_BODY_LENGTH || bodyLength > MAX_BODY_LENGTH) {
                    buffer.reset();
                    buffer.get();
                    continue;
                }

                parseFlag = PARSE_FLAG_BODY;
            }

            // 处理消息体

            if (buffer.remaining() < bodyLength) {

                // 回退位置，消息头已经读取，但是消息体不完整
                buffer.reset();
                parseFlag = PARSE_FLAG_HEADER;
                break;
            }

            byte[] completeResult = new byte[bodyLength];
            buffer.get(completeResult);
            completeResults.add(completeResult);
            parseFlag = PARSE_FLAG_HEADER;
        }

        // 保存剩余数据
        if (buffer.remaining() > 0) {
            ByteBuffer remaining = ByteBuffer.allocate(buffer.remaining());
            remaining.put(buffer);
            remaining.flip();
            buffer = remaining;
        }

        return completeResults;
    }

    public void clear() {
        buffer = ByteBuffer.allocate(0);
        parseFlag = PARSE_FLAG_HEADER;
    }

    public static byte[] buildPacket(String data) {
        byte[] body = data.getBytes();
        ByteBuffer packet = ByteBuffer.allocate(HEADER_LENGTH + body.length);

        // 设置魔数
        packet.putInt(MAGIC_NUMBER);

        // 设置消息头
        packet.putInt(body.length);

        // 设置消息体
        packet.put(body);

        return packet.array();
    }
}

优化版本通过魔数验证，只有正确的协议头才会解析长度，过滤掉垃圾数据
同时设置 MAX_BODY_LENGTH = 20MB，拒绝处理过大的包

四、优化版本测试用例

遇到错误的长度值

byte[] someData1 = DynamicLengthPacketByteBufferHandler.buildPacket("Hello");
byte[] someData2 = DynamicLengthPacketByteBufferHandler.buildPacket("World");
byte[] someData3 = DynamicLengthPacketByteBufferHandler.buildPacket("Java");

someData1[0] = (byte) 0x00;
someData1[1] = (byte) 0x00;
someData1[2] = (byte) 0x00;
someData1[3] = (byte) 0x07;

DynamicLengthPacketByteBufferHandler handler = new DynamicLengthPacketByteBufferHandler();

List<byte[]> results = handler.handleData(someData1);

System.out.println("第 1 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData2);

System.out.println("第 2 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData3);

System.out.println("第 3 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

# 输出结果

第 1 次解析结果数量: 0
第 2 次解析结果数量: 1
解析内容: World
第 3 次解析结果数量: 1
解析内容: Java

遇到过长的长度值

byte[] someData1 = DynamicLengthPacketByteBufferHandler.buildPacket("Hello");
byte[] someData2 = DynamicLengthPacketByteBufferHandler.buildPacket("World");
byte[] someData3 = DynamicLengthPacketByteBufferHandler.buildPacket("Java");

someData1[0] = (byte) 0x0F;
someData1[1] = (byte) 0xFF;
someData1[2] = (byte) 0xFF;
someData1[3] = (byte) 0xFF;

DynamicLengthPacketByteBufferHandler handler = new DynamicLengthPacketByteBufferHandler();

List<byte[]> results = handler.handleData(someData1);

System.out.println("第 1 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData2);

System.out.println("第 2 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData3);

System.out.println("第 3 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

# 输出结果

第 1 次解析结果数量: 0
第 2 次解析结果数量: 1
解析内容: World
第 3 次解析结果数量: 1
解析内容: Java

byte[] someData1 = DynamicLengthPacketByteBufferHandler.buildPacket("Hello");
byte[] someData2 = DynamicLengthPacketByteBufferHandler.buildPacket("World");
byte[] someData3 = DynamicLengthPacketByteBufferHandler.buildPacket("Java");

someData1[0] = (byte) 0xFF;
someData1[1] = (byte) 0xFF;
someData1[2] = (byte) 0xFF;
someData1[3] = (byte) 0xFF;

DynamicLengthPacketByteBufferHandler handler = new DynamicLengthPacketByteBufferHandler();

List<byte[]> results = handler.handleData(someData1);

System.out.println("第 1 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData2);

System.out.println("第 2 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

results = handler.handleData(someData3);

System.out.println("第 3 次解析结果数量: " + results.size());
for (byte[] result : results) {
    System.out.println("解析内容: " + new String(result));
}

# 输出结果

第 1 次解析结果数量: 0
第 2 次解析结果数量: 1
解析内容: World
第 3 次解析结果数量: 1
解析内容: Java

AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

别让你的 Agent 一口吃成胖子：AI Agent 任务拆分的工程实践指南

AI Agent技术社区

AI Agent律师行业智能体的实践步骤

AI Agent技术社区

MCP + 企业IM架构：AI Agent调用OA、ERP、MES时，身份和权限如何传递？

<think>我们只需要根据内容生成摘要，不超过150字。先理解文章核心：MCP+企业IM架构，AI Agent调用OA/ERP/MES时身份和权限传递。要点：IM从消息入口变任务入口；MCP解决连接，但不解决权限；需要五层身份映射；权限分层（可见性、调用、数据范围、确认）；建议组合MCP+API网关。摘要需简洁。参考生成： "本文探讨MCP与企业IM集成中AI Agent调用业务系统时的身份权