如何让我的bytebuf将我的全部信息发送给Netty

Question

我每个人。

我问你是因为我在Netty的解码器中的ByteBuf有问题。

我想解码我的消息谁到达一个服务器，但ByteBuf不工作，因为我会。

问题是ByteBuf没有获取消息的所有字节。

我解释说，我有一条长度为1221字节的消息（这是一个示例），但缓冲区大小只有64字节。

当我试图阅读，缓冲区与我的长度和我有这样的错误：

io.netty.handler.codec.DecoderException: java.lang.IndexOutOfBoundsException: readerIndex(117) + length(101) exceeds writerIndex(192): PooledUnsafeDirectByteBuf(ridx: 117, widx: 192, cap: 192)
at io.netty.handler.codec.ByteToMessageDecoder.callDecode(ByteToMessageDecoder.java:470) ~[netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.handler.codec.ByteToMessageDecoder.channelRead(ByteToMessageDecoder.java:276) ~[netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:352) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1408) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:374) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:360) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:930) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:697) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:632) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:549) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:511) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:918) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.util.internal.ThreadExecutorMap$2.run(ThreadExecutorMap.java:74) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30) [netty-all-4.1.37.Final.jar:4.1.37.Final]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_181]

我认为Netty没有时间阅读所有内容并只发送部分消息，但我现在不知道是否可以配置Netty，因为他必须等到消息全部到达。

如果有人能帮我，我很感激

为了获得更多帮助，我给您提供了解码器的代码

    int length = buffer.readInt();
    int messageType = buffer.readInt();


        Supplier<AbstractMessage> supplier = SUPPLIERS.get(messageType);
        if (supplier == null) {
            LOGGER.debug("This message type isn't supported: {}", messageType);
        } else {
            ByteBuf data = buffer.readBytes(length);
            if (messageType != 6) {
                AbstractMessage message = supplier.get();
                message.read(data, version);
                list.add(message);
                LOGGER.debug("{}", message);
            }
        }
    }

}

消息格式如下：MessageLength in 4字节（int）MessageType in 4字节（int）Data in n字节（MessageLength size）

我给你我用来解释这里的文档。

Answer 1

非常感谢。我做了一些改变，但你的代码激励了我，解决了我的问题。我想我收到了碎片信息，但我并没有太多的知识。

如果这篇文章能帮助别人，我把下面的代码更正

if (length == 0) {
        length = buffer.readInt();
        messageType = buffer.readInt();
    }

    if (buffer.writerIndex() < length + buffer.readerIndex()) {
        // ensure we have enough data so we can also read the message type and the whole message body
        return;
    }

    if (!(messageType == MessageType.HEARTBEAT_REQ.getValue() || messageType == MessageType.HEARTBEAT_CONF.getValue())) {
        LOGGER.debug("The message type is : {}", messageType);
    }

    Supplier<AbstractMessage> supplier = SUPPLIERS.get(messageType);
    if (supplier == null) {
        LOGGER.debug("This message type isn't supported: {}", messageType);
        buffer.skipBytes(length);
    } else {
        ByteBuf data = buffer.readSlice(length);
        AbstractMessage message = supplier.get();
        message.read(data, version);
        list.add(message);
        if (messageType != 6) {
            LOGGER.debug("{}", message);
        }
        length = 0;
        messageType = 0;
    }

请注意，length和messageType变量现在在解码器中是静态的

private static int length = 0;
private static int messageType = 0;

Answer 2

您需要通过扩展ByteToMessageDecoder和缓冲区来编写自己的解码器，直到您收到所有内容。由于这是TCP，您可能会以碎片的方式接收字节，因此您需要自己重新组装它。

像这样的方法应该会奏效：

class MyDecoder extends ByteToMessageDecoder {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf input, List<Object> out) {
        if (input.readableBytes() < 4) {
            // we need to have at least 4 bytes to read to be able to get the message length
            return;
        }
        int length = input.getInt(input.readerIndex());
        if (input.readableBytes() < 8 + length) {
            // ensure we have enough data so we can also read the message type and the whole message body
            return;
        }
        // skip the length now
        input.skipBytes(4);

        int messageType = input.readInt();

        Supplier<AbstractMessage> supplier = SUPPLIERS.get(messageType);
        if (supplier == null) {
            LOGGER.debug("This message type isn't supported: {}", messageType);
            input.skip(length);
        } else {
            if (messageType != 6) {
                ByteBuf data = buffer.readSlice(length);
                AbstractMessage message = supplier.get();
                message.read(data, version);
                list.add(message);
                LOGGER.debug("{}", message);
            }
        }
    }
}