netty ChannelInboundHandlerAdapter将帧切成约1500个字节

Question

我已经实现了一个服务器应用程序，该应用程序使用netty框架使用ChannelInblundHandlerAdapter读取传入的字节。

如标题所示，我的问题是，我不定期地从客户端获取内容，我认为是在〜1.500字节后被削减。例如：在这种情况下，我应该收到一个大的JSON数组。因为它被剪切，所以我无法解析它。

我尝试在使用消息之前使用管道中的附加ByteToMessageDecoder通道对消息进行解码。但这不能解决问题。我在JSON中没有定界符，因此我可以检查并再次将两个（或多个）部分粘在一起。

这是我的管道配置：

        ServerBootstrap b = new ServerBootstrap();
        b.group(bossGroup, workerGroup)
                .channel(NioServerSocketChannel.class)
                .childHandler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new IdleStateHandler(45,0,0));
                        ch.pipeline().addLast(new MyByteToMessageDecoder());
                        ch.pipeline().addLast(new GatewayCommunicationHandler());
                    }
                })
                .option(ChannelOption.SO_BACKLOG, 128)
                .option(ChannelOption.SO_RCVBUF, 8192)
                .childOption(ChannelOption.RCVBUF_ALLOCATOR, new FixedRecvByteBufAllocator(8192))
                .childOption(ChannelOption.SO_KEEPALIVE, true);

        initRestServer();

        // Bind and start to accept incoming connections.
        ChannelFuture f = b.bind(Config.gatewayPort).sync();
        f.channel().closeFuture().sync();

多数民众赞成在我的ByteToMessageDecoder：（我知道这是一团糟，但我不知道如何处理我的情况）

public class MyByteToMessageDecoder extends ByteToMessageDecoder {

@Override
protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) {
    byte[] receivedBytes = new byte[in.readableBytes()];
    in.getBytes(in.readerIndex(), receivedBytes);


    if (receivedBytes[in.readableBytes()-1] != (byte) 0) {
        out.add(receivedBytes);
        return;
    }

    int lenForOutBytes = 0;
    for (Object o : out) {
        byte[] bytes = (byte[]) o;
        lenForOutBytes += bytes.length;
    }

    byte[] outBytes = new byte[lenForOutBytes];

    for (Object o : out) {
        byte[] bytes = (byte[]) o;

        if (out.size() == 1) {
            outBytes = (byte[]) out.get(0);
        }
        else {
            int i = 0;

            for (int j = 0; j < bytes.length; j++) {
                outBytes[i + j] = bytes[j];
            }
            i += bytes.length;
        }
    }

    ctx.fireChannelRead(outBytes);
    in.resetReaderIndex();
}
...

还有其他人有这样的问题吗？

感谢您的回覆

布鲁·乔

Answer 1

我已经看到这个问题经常发生，所以我的目标范围比平常大了一点

发生此问题的原因是TCP是基于流的，而不是基于数据包的。

这基本上是发生的：

1. [客户端]想要发送10k字节的数据
2. [客户端]将数据发送到TCP层
3. [客户端] TCP层拆分数据包，它知道最大数据包大小为1500（这是几乎所有网络都使用的默认MTU ）
4. [客户端]客户端向服务器发送包含40字节作为报头和1460字节作为数据的数据包
5. [服务器] Netty接收第一个数据包，并直接调用您的函数，第一个数据包包含1460个字节的数据
6. [服务器]在您的功能需要处理剩余数据时（初始数据-1260）

所以解决这个问题，有多种方法

带有以下长度的邮件：

虽然这通常是解决数据包最简单的方法，但在同时处理大小消息时效率也最低。这也需要更改协议。

基本思想是在发送数据包之前预先设置长度，这样可以正确拆分邮件

好处

• 无需遍历数据即可滤除字符或阻止禁用字符
• 如果您的网络中有中继系统，则不必为消息边界进行任何硬解析

缺点

• 必须知道消息的长度，在长消息中，这会占用大量内存

怎么样？

如果您使用标准的整数字段，这很简单，因为Netty为此内置了类：

• LengthFieldBasedFrameDecoder
• LengthFieldPrepender

在管道中按以下方式使用它

// int maxFrameLength, int lengthFieldOffset, int lengthFieldLength, int lengthAdjustment, int initialBytesToStrip
pipeline.addLast(new LengthFieldBasedFrameDecoder(1024 * 4, 0, 2, 0, 2));
// int lengthFieldLength, int lengthAdjustment
pipeline.addLast(new LengthFieldPrepender(2, 0));

这基本上像以下这样对数据包进行了构架：

您发送：

DATA: 12B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 65 6c 6c 6f 20 57 6f 72 6c 64 21             |Hello World!    |
+--------+-------------------------------------------------+----------------+

LengthFieldPrepender 将此转换为：

DATA: 14B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 00 0c 48 65 6c 6c 6f 20 57 6f 72 6c 64 21       |..Hello World!  |
+--------+-------------------------------------------------+----------------+

然后，当您收到消息时，将其LengthFieldBasedFrameDecoder解码为：

DATA: 12B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 65 6c 6c 6f 20 57 6f 72 6c 64 21             |Hello World!    |
+--------+-------------------------------------------------+----------------+

在简单的分隔符上分割消息

某些协议采用不同的方法，而不是在固定长度上进行拆分，而是在定界符上进行拆分。快速查看方法是Java中的字符串以a结束，"文本文件中的行以换行结束，自然文本中的段落以双换行结束。

好处

• 如果您知道某些数据不包含字符，则相对容易生成，例如JSON通常不包含空格，因此用空格分隔消息很容易。
• 无需状态即可通过脚本语言轻松实现

缺点

• 与框架字符的冲突可能会使邮件大小膨胀
• 长度事先未知，因此要么在代码中设置硬编码限制，要么继续读取直到内存不足或数据结束
• 即使您对数据包不感兴趣，也需要阅读每个字符

怎么样？

从Netty发送消息时，您需要在消息本身中手动添加定界符，接收时可以使用DelimiterBasedFrameDecoder将输入流解码为消息。

管道示例：

在管道中按以下方式使用它

// int maxFrameLength, ByteBuf... delimiters
pipeline.addLast(1024 * 4, DelimiterBasedFrameDecoder(Delimiters.lineDelimiter()));

发送消息时，您需要手动添加定界符：

DATA: 14B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 65 6c 6c 6f 20 57 6f 72 6c 64 21 0d 0a       |Hello World!..  |
+--------+-------------------------------------------------+----------------+

接收消息时，DelimiterBasedFrameDecoder为您将消息转换为帧：

DATA: 12B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 48 65 6c 6c 6f 20 57 6f 72 6c 64 21             |Hello World!    |
+--------+-------------------------------------------------+----------------+

拆分复杂的业务定界符

并非所有取景都是容易的，如果避免的话，某些解决方案实际上是最好的，但是有时候，您确实需要做一些肮脏的工作。

好处

• 可以虚拟处理所有现有数据结构
• 无需修改协议

缺点

• 通常您必须检查每个字节
• 代码可能很难遵循
• 快速解决方案可能会因输入错误而给出怪异的错误

这分为2类：

• 基于现有的解码器
• 模式检测

基于现有的解码器

使用这些解决方案，您基本上可以使用其他框架中的现有解码器来解析数据包，并检测其处理失败。

GSON和的示例ReplayingDecoder：

public class GSONDecoder
    extends ReplayingDecoder<Void> {

    Gson gson = new GsonBuilder().create();

    protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) 
        throws Exception {

        out.add(gson.fromJson(new ByteBufInputStream(buf, false), Object.class));
    }
}

模式检测

如果要使用模式检测方法，则需要了解协议。让我们为JSON创建一个模式检测解码器。

基于JSON的结构，让我们做以下假设：

1. JSON是基于匹配的双{和}，和[和]
2. {和}之间的匹配对应忽略"
3. " 当以 \
4. 一个\如果由前缀应该被忽略\，从解析时，从左向右

基于这些属性，让我们ByteToMessageDecoder基于以下假设进行：

public static class JSONDecoder extends ByteToMessageDecoder {

    // Notice, this class is designed for JSON without a charset definition at the start, adding this is hard as we basicly have to call differend
    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        in.markReaderIndex();

        int fromIndex = in.readerIndex();

        int unclosedCurlyBracketsSeen = 0;
        boolean inQuotedSection = false;
        boolean nonWhitespaceSeen = false;
        boolean slashSeen = false;

        while (in.isReadable()) {
            boolean newSlashSeenState = false;
            byte character = in.readByte();
            if (character == '{' && !inQuotedSection) {
                unclosedCurlyBracketsSeen++;
            }
            if (character == '}' && !inQuotedSection) {
                unclosedCurlyBracketsSeen--;
            }
            if (character == '[' && !inQuotedSection) {
                unclosedCurlyBracketsSeen++;
            }
            if (character == ']' && !inQuotedSection) {
                unclosedCurlyBracketsSeen--;
            }
            if (character == '"' && !slashSeen) {
                inQuotedSection = !inQuotedSection;
            }
            if (character == '\\' && !slashSeen) {
                newSlashSeenState = true;
            }
            if (!Character.isWhitespace(character)) {
                nonWhitespaceSeen = true;
            }
            slashSeen = newSlashSeenState;
            if(unclosedCurlyBracketsSeen == 0 && nonWhitespaceSeen) {
                int targetIndex = in.readerIndex();
                out.add(in.slice(fromIndex, targetIndex - fromIndex).retain());
                return;
            }
        }

        // End of stream reached, but our JSON is not complete, reset our progress!
        in.resetReaderIndex();
    }

}

接收消息时，它是这样工作的：

DATA: 35B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 7b 22 68 69 21 22 2c 22 53 74 72 69 6e 67 3a 20 |{"hi!","String: |
|00000010| 5c 22 48 69 5c 22 22 7d 20 20 7b 22 73 6c 61 73 |\"Hi\""}  {"slas|
|00000020| 68 22 3a                                        |h":             |
+--------+-------------------------------------------------+----------------+

DATA: 34B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 22 5c 5c 22 7d 7b 22 4e 65 73 74 65 64 3a 22 3a |"\\"}{"Nested:":|
|00000010| 7b 22 64 65 65 70 65 72 22 3a 7b 22 6f 6b 22 7d |{"deeper":{"ok"}|
|00000020| 7d 7d                                           |}}              |
+--------+-------------------------------------------------+----------------+

如您所见，我们收到了2条消息，其中1条甚至被分割为2个“虚拟TCP”数据包，由我们的“ JSON解码器”将其转换为以下ByteBuf数据包：

DATA: 24B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 7b 22 68 69 21 22 2c 22 53 74 72 69 6e 67 3a 20 |{"hi!","String: |
|00000010| 5c 22 48 69 5c 22 22 7d                         |\"Hi\""}        |
+--------+-------------------------------------------------+----------------+

DATA: 16B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 20 20 7b 22 73 6c 61 73 68 22 3a 22 5c 5c 22 7d |  {"slash":"\\"}|
+--------+-------------------------------------------------+----------------+

DATA: 29B
         +-------------------------------------------------+
         |  0  1  2  3  4  5  6  7  8  9  a  b  c  d  e  f |
+--------+-------------------------------------------------+----------------+
|00000000| 7b 22 4e 65 73 74 65 64 3a 22 3a 7b 22 64 65 65 |{"Nested:":{"dee|
|00000010| 70 65 72 22 3a 7b 22 6f 6b 22 7d 7d 7d          |per":{"ok"}}}   |
+--------+-------------------------------------------------+----------------+