Netty ByteToMessageDecoder无法接收以不同tcp数据包发送的消息

Question

自定义ByteToMessageEncoder不接收在相同tcp连接中发送的字节，而是接收在不同tcp消息中发送的字节。

我被指派解决一个问题，一个几年前的系统开始出现不良行为。在我看来，有一个tcp服务器是由其他一些开发人员用netty编写的，它接收带有静态长度头和可变长度主体的二进制消息。正文长度由a头字段定义，该字段告诉消息类型。我们维护一个消息类型及其长度的地图。

所面临的问题是，在正确解码头部并知道主体长度之后，相同的解码器希望主体出现在相同的ByteBuf中（即来自byteChannel的一个fireChannelRead事件）。

然而，有时缓冲区中没有足够的东西，解码器就放弃了。但是下次调用decode-method时，正文字节会出现，并被错误地解释为头部，从而使解码器不同步。

使用netty来组装消息的正确方法是什么，这些消息的字节可能会以较小的块插入？

以下是当前解码器的基本知识。

    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        Message message = decode(ctx, in); 
        if (message != null) {
            out.add(message);
        }
    }

    protected Message decode(ChannelHandlerContext ctx, ByteBuf in) throws Exception {
        if (in.readableBytes() < MessageHeader.SIZE) {
            return null;
        }
        ByteBuf headerBytes = in.readBytes(MessageHeader.SIZE);
        MessageHeader header = MessageHeader.decode(headerBytes, new ProcessingTracker(MessagePart.HEADER));
        if (header == null) {
            ctx.disconnect().sync();
            logger.debug("Disconnected from channel");
            return null;
        }
        int bodySize = header.getMessageType().getMessageBodySize();
        if (!waitingForBytes(in, bodySize, READ_TRY_TIMES)) {
            ctx.disconnect().sync();
            logger.debug("Disconnected from channel");
            return null;
        }
        ByteBuf messageBytes = in.readBytes(bodySize);
        messageBytes.resetReaderIndex();
        Message message = Message.decode(header, messageBytes, 0);
        return message;
    }


    public boolean waitingForBytes(ByteBuf in, int bodySize, int counter) {
        if (counter == 0) {
            logger.warn("Didn't get enough bytes of message body in MessagDecoder. Giving up and disconnecting from remote peer.");
            return false;
        }
        logger.debug(String.format("Readable bytes in buffer %d, expected %d", in.readableBytes(), bodySize));

        if (in.readableBytes() < bodySize) {

            try {
                Thread.sleep(20L);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            return waitingForBytes(in, bodySize, counter - 1);
        } else {
            return true;
        }
    }

Answer 1

你的代码有多个问题...

首先，不允许在代码中调用sync()，因为这样将使EventLoop“死锁”。

其次，您不能在这里使用WaitingForBytes，因为它基本上会使EventLoop上的所有其他IO过时，这意味着您将永远不会继续执行任何IO。在像Netty这样的a框架中，千万不要阻塞EventLoop线程，因为这基本上会导致一切都过时，没有任何进展。