Netty ByteToMessageDecoder 没有接收到不同 tcp 数据包中发送的消息

Question

自定义 ByteToMessageEncoder 不接收在同一 tcp 连接中但在不同 tcp 消息中发送的字节。

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

面临的问题是，在正确解码标头并知道正文长度后，相同的解码器期望正文进入相同的 ByteBuf（即来自 byteChannel 的一个 fireChannelRead 事件）。

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

使用 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

查看超类ByteToMessageDecoder 很明显，子类应该通过它读取的字节数或解码的消息数来传达解码进度。我认为我继承此代码的人错过了这一点。

这个实现通过了一些初步测试：

public class MessageDecoder extends ByteToMessageDecoder {

    private static final Logger logger = LoggerFactory.getLogger(MessageDecoder.class);

    private ByteBuf headBuf = Unpooled.buffer(MessageHeader.SIZE);

    private MessageHeader header = null;

    private ByteBuf bodyBuf;

    private int bodylength = 0;

    private int messageBytes = 0;

    private final static int STATE_READ_HEADER = 1, STATE_READ_BODY = 2;

    private int state = STATE_READ_HEADER;

    @Override
    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 {
        int readBytes = 0;
        logstate(in);
        switch (state) {
        case STATE_READ_HEADER:
            if (in.readableBytes() <= MessageHeader.SIZE - messageBytes) {
                readBytes = in.readableBytes();
            } else {
                readBytes = MessageHeader.SIZE - messageBytes;
            }
            headBuf.writeBytes(in, readBytes);
            messageBytes += readBytes;
            if (messageBytes == MessageHeader.SIZE) {
                state = STATE_READ_BODY;
                header = MessageHeader.decode(headBuf, new ProcessingTracker(MessagePart.HEADER));
                bodylength = header.getMessageType().getMessageBodySize();
                bodyBuf = Unpooled.buffer(bodylength);
            }
            break;
        case STATE_READ_BODY:
            if (in.readableBytes() <= bodylength - (messageBytes - MessageHeader.SIZE)) {
                readBytes = in.readableBytes();
            } else {
                readBytes = bodylength - (messageBytes - MessageHeader.SIZE);
            }
            bodyBuf.writeBytes(in, readBytes);
            messageBytes += readBytes;
            if (messageBytes == MessageHeader.SIZE + bodylength) {
                state = STATE_READ_HEADER;
                Message message = Message.decode(header, bodyBuf, 0);
                reset();
                return message;
            }
            break;
        }
        return null;
    }
}

Answer 2

您的代码中存在多个问题...

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

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