netty ChannelInboundHandlerAdapter 在 ~1500 字节处削减帧

netty ChannelInboundHandlerAdapter cuts frame at ~1500 bytes

我已经实现了一个服务器应用程序,它使用 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();
}
...

有没有人遇到这样的问题。

感谢您的回复

乔兄弟

I have seen that this issue happens frequently, so I'm purposely a bit broader than I usually do

出现此问题是因为 TCP 是基于流的,而不是基于数据包的。

这基本上会发生:

  1. [客户端] 想要发送 10k 字节的数据
  2. [client]发送数据到TCP层
  3. [客户端] TCP 层拆分数据包,它知道最大数据包大小为 1500(这是默认 MTU 几乎所有 网络使用)
  4. [client] 客户端向服务器发送数据包,包头40字节,数据1460字节
  5. [server] Netty收到第一包数据,直接调用你函数,第一包包含1460字节数据
  6. [服务器]在您的函数需要处理剩余数据(初始数据 - 1260)的时间内

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

在消息前加上长度:

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

基本思想是在发送数据包之前添加长度,这样可以正确拆分消息

优势

  • 无需遍历数据来过滤掉字符,或阻止禁用字符
  • 如果您的网络中有中继系统,他们不必对消息边界进行任何硬解析

缺点

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

怎么办?

如果你使用一个标准的整数字段,这真的很简单,因为 Netty 在 类 中为此构建:

这在您的管道中按以下方式使用

// 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 中的字符串以 " 结尾,文本文件中的行以换行符结尾,自然文本中的段落以双换行符结尾。

优势

  • 如果您知道某个数据不包含字符,则相对容易生成,例如 JSON 通常不包含 space,因此用 space 分隔消息很简单。
  • 易于通过脚本语言实现,因为不需要状态

缺点

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

怎么办?

从 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 and 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. " 前面加上 \
  5. 时应该被忽略
  6. A \ 如果前面有 \,当从左到右解析时 \ 应该被忽略

基于这些属性,让我们根据这些假设做一个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 个 "virtual TCP" 数据包之间分段,这被我们的 "JSON decoder" 转换为以下 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"}}}   |
+--------+-------------------------------------------------+----------------+