netty解码器之ReplayingDecoder

亦凉 2022-12-31 11:27 204阅读 0赞

# 解码器之ReplayingDecoder #

ReplayingDecoder继承了ByteToMessageDecoder，是一个解码器，是ByteToMessageDecoder的一个特殊变体。

## ReplayingDecoder和ByteToMessageDecoder的区别 ##

ReplayingDecoder和ByteToMessageDecoder最大的不同在于，ReplayingDecoder允许让你实现decode()方法，就像已经接收到所有所需的字节，而不用去检查所需字节的可用性。

假设现在自定义协议传递ByteBuf格式如下：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]

协议传递的数据需要封装为如下格式：

import lombok.Data;
    
    @Data
    public class CustomProtocol { 
    
        private int length;
        private byte[] content;
    }

### 使用ByteToMessageDecoder处理粘包半包问题 ###

package com.morris.netty.code.replayingdecoder;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ByteToMessageDecoder;
    
    import java.util.List;
    
    public class CustomProtocolDecoder extends ByteToMessageDecoder { 
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 
    
            if(in.readableBytes() < 4) {  // length is ready?
                return;
            }
    
            in.markReaderIndex();
    
            int length = in.readInt();
    
            if(in.readableBytes() < length) {  // content is ready？
                in.resetReaderIndex();
                return;
            }
    
            CustomProtocol customProtocol = new CustomProtocol();
            customProtocol.setLength(length);
            byte[] bytes = new byte[length];
            in.readBytes(bytes);
            customProtocol.setContent(bytes);
    
            out.add(customProtocol);
        }
    }

### 使用ReplayingDecoder处理粘包半包问题 ###

package com.morris.netty.code.replayingdecoder;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ReplayingDecoder;
    
    import java.util.List;
    
    public class IntegerHeaderFrameDecoder extends ReplayingDecoder<Void> { 
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception { 
    
            int length = in.readInt();
    
            CustomProtocol customProtocol = new CustomProtocol();
            customProtocol.setLength(length);
            byte[] bytes = new byte[length];
            in.readBytes(bytes);
            customProtocol.setContent(bytes);
    
            out.add(customProtocol);
        }
    }

## ReplayingDecoder底层如何实现的 ##

ReplayingDecoder通过一个专门的ByteBuf实现，当缓冲区中没有足够的数据时，会抛出一个某种类型的Error。在上面的IntegerHeaderFrameDecoder中，当调用in.readInt()时，只需要假设缓冲区有4个或更多的字节。如果，缓冲区中确实有4个字节，会返回你所期望的integer header。否则，会抛出这个Error。ReplayingDecoder会捕获这个Error，然后重置缓冲区的readerindex到缓冲区的开始位置，并且当更多的数据到达缓冲区时，会再次调用decode()方法。

需要注意一些限制：

*  buffer的部分操作，如readBytes(ByteBuffer dst)、retain()、release()等方法会直接抛出异常。
 *  如果网络缓慢而且消息格式复杂，性能会变差，在这种情况，你的解码器不得不解码同一部分的信息。
 *  解码一条信息，decode()方法可以被多次调用，如下代码不能正常工作：

package com.morris.netty.code.replayingdecoder;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ReplayingDecoder;
    
    import java.util.LinkedList;
    import java.util.List;
    import java.util.Queue;
    
    public class MyDecoder extends ReplayingDecoder<Void> { 
    
        private final Queue<Integer> values = new LinkedList<>();
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) throws Exception { 
            // A message contains 2 integers.
            values.offer(buf.readInt());
            values.offer(buf.readInt());
    
            // This assertion will fail intermittently since values.offer()
            // can be called more than two times!
            assert values.size() == 2;
            out.add(values.poll() + values.poll());
        }
    }

正确的做法如下：

package com.morris.netty.code.replayingdecoder;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ReplayingDecoder;
    
    import java.util.LinkedList;
    import java.util.List;
    import java.util.Queue;
    
    public class MyDecoder extends ReplayingDecoder<Void> { 
    
        private final Queue<Integer> values = new LinkedList<>();
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) throws Exception { 
    
            values.clear(); // 没有这一行就会有问题
    
            // A message contains 2 integers.
            values.offer(buf.readInt());
            values.offer(buf.readInt());
    
            // This assertion will fail intermittently since values.offer()
            // can be called more than two times!
            assert values.size() == 2;
            out.add(values.poll() + values.poll());
        }
    }

## 提升ReplayingDecoder的性能 ##

复杂解码器的性能可以使用checkpoint()方法得到显著的提高，checkpoint()方法更新缓冲区的“初始”位置，ReplayingDecoder将回滚缓冲区的readerIndex到最后调用checkpoint()方法的位置。

### 带枚举参数的checkpoint() ###

管理解码器状态的最简单方法是创建表示解码器当前状态的Enum类型，并在状态改变时调用checkpoint(T)方法，你可以有许多你想要的状态取决于你想解码的信息的复杂性:

package com.morris.netty.code.replayingdecoder;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ReplayingDecoder;
    
    import java.util.List;
    
    public class IntegerHeaderFrameDecoder2 extends ReplayingDecoder<MyDecoderState> { 
    
        private int length;
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) throws Exception { 
            switch (state()) { 
                case READ_LENGTH:
                    length = buf.readInt();
                    checkpoint(MyDecoderState.READ_CONTENT);
                case READ_CONTENT:
                    ByteBuf frame = buf.readBytes(length);
                    checkpoint(MyDecoderState.READ_LENGTH);
                    out.add(frame);
                    break;
                default:
                    throw new Error("Shouldn't reach here.");
            }
        }
    }

### 无参的checkpoint() ###

也可以自己管理缓冲器的状态，不适用枚举类型。

package com.morris.netty.code.replayingdecoder;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ReplayingDecoder;
    
    import java.util.List;
    
    public class IntegerHeaderFrameDecoder3 extends ReplayingDecoder<MyDecoderState> { 
    
        private boolean readLength; // 通过这个字段来标记当前的状态
    
        private int length;
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf buf, List<Object> out) throws Exception { 
    
            if (!readLength) { 
                length = buf.readInt();
                readLength = true;
                checkpoint();
            } else { 
                ByteBuf frame = buf.readBytes(length);
                checkpoint(MyDecoderState.READ_LENGTH);
                out.add(frame);
                checkpoint();
            }
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMjI4MTI4NDk_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/19fca92674374ef5a54fe4bd469e495a.png