Netty ReplayingDecoder 源码分析与特性解读

作者: 嘟嘟碰碰叮叮当当 | 来源:发表于2021-01-12 15:57 被阅读0次

转自：https://blog.csdn.net/wzq6578702/article/details/78826494

在介绍ReplayingDecoder之前想看一下它的用法，构建一个服务端和客户端的模型：

服务端：

public class MyServer {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup bossGroup = new NioEventLoopGroup(1);
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try{
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            serverBootstrap.group(bossGroup,workerGroup).channel(NioServerSocketChannel.class)
                    .childHandler(new MyServerInitializer());
            ChannelFuture channelFuture = serverBootstrap.bind(8899).sync();
            channelFuture.channel().closeFuture().sync();
        }finally{
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
}

服务端initializer：

public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipline = ch.pipeline();
        pipline.addLast(new MyReplayingDecoder());//使用ReplayingDecoder
        pipline.addLast(new MyLongToByteEncoder());
        pipline.addLast(new MyServerHandler());
    }
}

ServerHandler:

public class MyServerHandler extends SimpleChannelInboundHandler<Long> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
        System.out.println(ctx.channel().remoteAddress()+" --> "+msg);
        ctx.writeAndFlush(654321L);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        ctx.close();
    }
}

MyReplayingDecoder:

public class MyReplayingDecoder extends ReplayingDecoder<Void> {

    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
        System.out.println("MyReplayingDecoder decode invoked!");
        //注意没有判断字节数！！！！
        out.add(in.readLong());
    }
}

MyLongToByteEncoder

public class MyLongToByteEncoder extends MessageToByteEncoder<Long> {
    @Override
    protected void encode(ChannelHandlerContext ctx, Long msg, ByteBuf out) throws Exception {
        System.out.println("encode invoked");
        System.out.println(msg);
        out.writeLong(msg);
    }
}

客户端：

public class Myclient {
    public static void main(String[] args) throws InterruptedException {
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(eventLoopGroup).channel(NioSocketChannel.class).handler(new MyClientIniatializer());
            ChannelFuture channelFuture = bootstrap.connect("localhost", 8899).sync();
            channelFuture.channel().writeAndFlush("hello");
            channelFuture.channel().closeFuture().sync();
        } finally {
            eventLoopGroup.shutdownGracefully();
        }
    }
}

客户端Iniatializer：

public class MyClientIniatializer extends ChannelInitializer<SocketChannel> {

    @Override
    protected void initChannel(SocketChannel ch) throws Exception {
        ChannelPipeline pipline = ch.pipeline();

        pipline.addLast(new MyReplayingDecoder());
        pipline.addLast(new MyLongToByteEncoder());
        pipline.addLast(new MyClientHandler());
    }
}

客户端Handler：

public class MyClientHandler extends SimpleChannelInboundHandler<Long> {
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
        System.out.println(ctx.channel().remoteAddress());
        System.out.println("client output "+msg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        ctx.writeAndFlush(123456L);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        super.exceptionCaught(ctx, cause);
        ctx.close();
    }
}

运行服务端，之后运行客户端：
服务端输出结果：

MyReplayingDecoder decode invoked!
/127.0.0.1:4448 --> 123456
encode invoked
654321

客户端输出结果：

encode invoked
123456
MyReplayingDecoder decode invoked!
localhost/127.0.0.1:8899
client output 654321

加了下，数据传输流程分析

/**
 * describe:  1. client channelActive 发送12345
 *            2. client 触发encode方法发送到server端
 *            3. server 触发decode方法
 *            4. server channelRead0
 *            5. server write flush 发送出去
 *            6. server 触发encode方法发送到客户端
 *            7. clinet 触发decode方法
 *            8. clinet 触发channelRead0
 */

ReplayingDecoder java doc

ByteToMessageDecoder一种特殊变体，它可以在阻塞I / O范例中实现非阻塞解码器。
最大的区别ReplayingDecoder和ByteToMessageDecoder是ReplayingDecoder可以
让你实现decode()和decodeLast()方法，就像已经获得所有所需的字节，而不是检查
所需的字节的可用性。

例如，以下ByteToMessageDecoder实现：

   public class IntegerHeaderFrameDecoder extends ByteToMessageDecoder {
  
      @Override
     protected void decode(ChannelHandlerContext ctx,
                             ByteBuf buf, List<Object> out) throws Exception {
  
       if (buf.readableBytes() < 4) {
          return;
       }
  
       buf.markReaderIndex();
       int length = buf.readInt();
  
       if (buf.readableBytes() < length) {
          buf.resetReaderIndex();
          return;
       }
  
       out.add(buf.readBytes(length));
     }
   }

使用ReplayingDecoder了如下简化：

   public class IntegerHeaderFrameDecoder
        extends ReplayingDecoder<Void> {
  
     protected void decode(ChannelHandlerContext ctx,
                             ByteBuf buf) throws Exception {
  
       out.add(buf.readBytes(buf.readInt()));
     }
   }

这是如何运作的？

ReplayingDecoder通过一个专门的ByteBuf实现其抛出一个Error的某些类型的时候有没有在缓冲区足够的数据。在上面的IntegerHeaderFrameDecoder ，您仅假设调用buf.readInt()时缓冲区中将有4个或更多字节。如果缓冲区中确实有4个字节，它将按预期返回整数标头。否则，将引发Error并将控件返回给ReplayingDecoder 。如果ReplayingDecoder捕获到Error ，则它将把缓冲区的readerIndex倒回到“初始”位置（即缓冲区的开头），并在缓冲区中收到更多数据时再次调用readerIndex decode(..)方法。
请注意， ReplayingDecoder始终会抛出相同的缓存Error实例，以避免创建新Error并为每次抛出填充其堆栈跟踪的开销。

局限性

以简单为代价， ReplayingDecoder强制您执行一些限制：
禁止某些缓冲区操作。
如果网络速度慢且消息格式复杂，则性能可能会变差，这与上面的示例不同。在这种情况下，您的解码器可能不得不一遍又一遍地解码消息的相同部分。
您必须记住，可以多次调用decode(..)方法来解码一条消息。例如，以下代码将不起作用：

 public class MyDecoder extends ReplayingDecoder<Void> {
  
     private final Queue<Integer> values = new LinkedList<Integer>();
  
      @Override
     public void decode(.., 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());
     }
   }

正确的实现如下所示，您还可以利用“检查点”功能，下一部分将对此进行详细说明。

 public class MyDecoder extends ReplayingDecoder<Void> {
  
     private final Queue<Integer> values = new LinkedList<Integer>();
  
      @Override
     public void decode(.., ByteBuf buf, List<Object> out) throws Exception {
  
       // Revert the state of the variable that might have been changed
       // since the last partial decode.
       values.clear();
  
       // A message contains 2 integers.
       values.offer(buf.readInt());
       values.offer(buf.readInt());
  
       // Now we know this assertion will never fail.
       assert values.size() == 2;
       out.add(values.poll() + values.poll());
     }
   }

改善表现

幸运的是，使用checkpoint()方法可以显着提高复杂解码器实现的性能。
该checkpoint()这样方法更新缓冲区的“初始”位置ReplayingDecoder倒回readerIndex缓冲剂与在那里你叫的最后一个位置checkpoint()方法。

用Enum调用checkpoint(T)

尽管您可以只使用checkpoint()方法并自己管理解码器的状态，
但是管理解码器状态的最简单方法是
创建一个代表解码器当前状态的Enum类型并调用checkpoint(T)状态改变时的方法。
根据要解码的消息的复杂性，可以根据需要设置任意多个状态：

   public enum MyDecoderState {
     READ_LENGTH,
     READ_CONTENT;
   }
  
   public class IntegerHeaderFrameDecoder
        extends ReplayingDecoder<MyDecoderState> {
  
     private int length;
  
     public IntegerHeaderFrameDecoder() {
       // Set the initial state.
       super(MyDecoderState.READ_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()

管理解码器状态的另一种方法是自己管理。

   public class IntegerHeaderFrameDecoder
        extends ReplayingDecoder<Void> {
  
     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();
       }
  
       if (readLength) {
         ByteBuf frame = buf.readBytes(length);
         readLength = false;
         checkpoint();
         out.add(frame);
       }
     }
   }

用流水线中的另一个解码器替换一个解码器

如果你打算写一个协议复用器，你可能会想更换ReplayingDecoder与另一个（协议检测） ReplayingDecoder ， ByteToMessageDecoder或MessageToMessageDecoder （实际协议解码器）。不能仅通过调用ChannelPipeline.replace(ChannelHandler, String, ChannelHandler)来实现此目的，但是需要一些其他步骤：


   public class FirstDecoder extends ReplayingDecoder<Void> {
  
        @Override
       protected void decode(ChannelHandlerContext ctx,
                               ByteBuf buf, List<Object> out) {
           ...
           // Decode the first message
           Object firstMessage = ...;
  
           // Add the second decoder
           ctx.pipeline().addLast("second", new SecondDecoder());
  
           if (buf.isReadable()) {
               // Hand off the remaining data to the second decoder
               out.add(firstMessage);
               out.add(buf.readBytes(super.actualReadableBytes()));
           } else {
               // Nothing to hand off
               out.add(firstMessage);
           }
           // Remove the first decoder (me)
           ctx.pipeline().remove(this);
       }

几种常见的编解码器:

LineBasedFrameDecoder

解码器，将接收到的ByteBuf在行尾拆分。
"\n"和"\r\n"都被处理。 有关基于分隔符的更通用解码器，请参见DelimiterBasedFrameDecoder

FixedLengthFrameDecoder

解码器，将接收到的ByteBuf为固定的字节数。 例如，如果您收到以下四个分段的数据包：
   +---+----+------+----+
   | A | BC | DEFG | HI |
   +---+----+------+----+
   
FixedLengthFrameDecoder (3)会将它们解码为以下三个具有固定长度的数据包：
   +-----+-----+-----+
   | ABC | DEF | GHI |
   +-----+-----+-----+

DelimiterBasedFrameDecoder

一种解码器， ByteBuf通过一个或多个定界符对接收到的ByteBuf拆分。 这对于解码以分隔符（例如NUL或换行符）结尾的帧特别有用。
预定义的分隔符
为了方便起见， Delimiters定义了常用的定界符。
指定多个定界符
DelimiterBasedFrameDecoder允许您指定多个定界符。 如果在缓冲区中找到多个定界符，它将选择产生`最短帧`的定界符。 例如，如果缓冲区中包含以下数据：
   +--------------+
   | ABC\nDEF\r\n |
   +--------------+
   
DelimiterBasedFrameDecoder （ Delimiters.lineDelimiter() ）将选择'\n'作为第一个定界符并产生两个帧：
   +-----+-----+
   | ABC | DEF |
   +-----+-----+
   
而不是错误地选择'\r\n'作为第一个定界符：
   +----------+
   | ABC\nDEF |
   +----------+

LengthFieldBasedFrameDecoder

解码器按消息中的length字段的值动态拆分接收到的ByteBuf 。 当您解码二进制消息时，此消息特别有用，该二进制消息具有代表消息正文或整个消息长度的整数头字段。
LengthFieldBasedFrameDecoder具有许多配置参数，因此它可以解码带有长度字段的任何消息，这在专有的客户端-服务器协议中经常出现。 以下是一些示例，可让您基本了解哪个选项可以执行什么操作。
2字节长度的字段，偏移量为0，不剥离标题
在此示例中，长度字段的值为12（0x0C） ，代表“ HELLO，WORLD”的长度。 默认情况下，解码器假定length字段表示在length字段之后的字节数。 因此，可以使用简单的参数组合对其进行解码。
   lengthFieldOffset   = 0
   lengthFieldLength   = 2
   lengthAdjustment    = 0
   initialBytesToStrip = 0 (= do not strip header)
  
   BEFORE DECODE (14 bytes)         AFTER DECODE (14 bytes)
   +--------+----------------+      +--------+----------------+
   | Length | Actual Content |----->| Length | Actual Content |
   | 0x000C | "HELLO, WORLD" |      | 0x000C | "HELLO, WORLD" |
   +--------+----------------+      +--------+----------------+
   
2个字节的长度字段，偏移量为0，带头
因为我们可以通过调用ByteBuf.readableBytes()获得内容的长度，所以您可能希望通过指定initialBytesToStrip来剥离长度字段。 在此示例中，我们指定2 ，它与length字段的长度相同，以剥离前两个字节。
   lengthFieldOffset   = 0
   lengthFieldLength   = 2
   lengthAdjustment    = 0
   initialBytesToStrip = 2 (= the length of the Length field)
  
   BEFORE DECODE (14 bytes)         AFTER DECODE (12 bytes)
   +--------+----------------+      +----------------+
   | Length | Actual Content |----->| Actual Content |
   | 0x000C | "HELLO, WORLD" |      | "HELLO, WORLD" |
   +--------+----------------+      +----------------+
   
2个字节的长度字段，偏移量为0，不剥离标题，该长度字段表示整个消息的长度
在大多数情况下，长度字段仅表示消息正文的长度，如前面的示例所示。 但是，在某些协议中，长度字段表示整个消息的长度，包括消息头。 在这种情况下，我们指定一个非零的lengthAdjustment 。 因为此示例消息中的length值总是比主体长度大2 ，所以我们将-2指定为lengthAdjustment以进行补偿。
   lengthFieldOffset   =  0
   lengthFieldLength   =  2
   lengthAdjustment    = -2 (= the length of the Length field)
   initialBytesToStrip =  0
  
   BEFORE DECODE (14 bytes)         AFTER DECODE (14 bytes)
   +--------+----------------+      +--------+----------------+
   | Length | Actual Content |----->| Length | Actual Content |
   | 0x000E | "HELLO, WORLD" |      | 0x000E | "HELLO, WORLD" |
   +--------+----------------+      +--------+----------------+
   
3个字节的长度字段位于5个字节的报头末尾，不剥离报头
以下消息是第一个示例的简单变体。 该消息之前会附加一个额外的标头值。 lengthAdjustment再次为零，因为解码器始终在帧长度计算过程中考虑前置数据的长度。
   lengthFieldOffset   = 2 (= the length of Header 1)
   lengthFieldLength   = 3
   lengthAdjustment    = 0
   initialBytesToStrip = 0
  
   BEFORE DECODE (17 bytes)                      AFTER DECODE (17 bytes)
   +----------+----------+----------------+      +----------+----------+----------------+
   | Header 1 |  Length  | Actual Content |----->| Header 1 |  Length  | Actual Content |
   |  0xCAFE  | 0x00000C | "HELLO, WORLD" |      |  0xCAFE  | 0x00000C | "HELLO, WORLD" |
   +----------+----------+----------------+      +----------+----------+----------------+
   
5字节标题开头的3字节长度字段，不剥离标题
这是一个高级示例，显示了在length字段和消息正文之间存在一个额外的标头的情况。 您必须指定一个正的lengthAdjustment，以便解码器将额外的标头计入帧长度计算中。
   lengthFieldOffset   = 0
   lengthFieldLength   = 3
   lengthAdjustment    = 2 (= the length of Header 1)
   initialBytesToStrip = 0
  
   BEFORE DECODE (17 bytes)                      AFTER DECODE (17 bytes)
   +----------+----------+----------------+      +----------+----------+----------------+
   |  Length  | Header 1 | Actual Content |----->|  Length  | Header 1 | Actual Content |
   | 0x00000C |  0xCAFE  | "HELLO, WORLD" |      | 0x00000C |  0xCAFE  | "HELLO, WORLD" |
   +----------+----------+----------------+      +----------+----------+----------------+
   
2个字节的长度字段在4字节标题的中间偏移量1处，除去第一个标题字段和长度字段
这是以上所有示例的组合。 在length字段之前有前置报头，在length字段之后有多余的报头。 前置标头会影响lengthFieldOffset ，额外标头会影响lengthAdjustment 。 我们还指定了一个非零的initialBytesToStrip来从帧中剥离长度字段和前置标头。 如果不想剥离前置标头，则可以将initialBytesToSkip指定为0 。
   lengthFieldOffset   = 1 (= the length of HDR1)
   lengthFieldLength   = 2
   lengthAdjustment    = 1 (= the length of HDR2)
   initialBytesToStrip = 3 (= the length of HDR1 + LEN)
  
   BEFORE DECODE (16 bytes)                       AFTER DECODE (13 bytes)
   +------+--------+------+----------------+      +------+----------------+
   | HDR1 | Length | HDR2 | Actual Content |----->| HDR2 | Actual Content |
   | 0xCA | 0x000C | 0xFE | "HELLO, WORLD" |      | 0xFE | "HELLO, WORLD" |
   +------+--------+------+----------------+      +------+----------------+
   
2个字节的长度字段位于4字节标题的中间偏移量1处，去掉第一个标题字段和length字段，length字段代表整个消息的长度
让我们再对前面的示例进行一些修改。 与前一个示例的唯一区别是，长度字段表示整个消息的长度，而不是消息主体，就像第三个示例一样。 我们必须将HDR1的长度和Length计入lengthAdjustment中。 请注意，我们不需要考虑HDR2的长度，因为length字段已经包含了整个标头长度。
   lengthFieldOffset   =  1
   lengthFieldLength   =  2
   lengthAdjustment    = -3 (= the length of HDR1 + LEN, negative)
   initialBytesToStrip =  3
  
   BEFORE DECODE (16 bytes)                       AFTER DECODE (13 bytes)
   +------+--------+------+----------------+      +------+----------------+
   | HDR1 | Length | HDR2 | Actual Content |----->| HDR2 | Actual Content |
   | 0xCA | 0x0010 | 0xFE | "HELLO, WORLD" |      | 0xFE | "HELLO, WORLD" |
   +------+--------+------+----------------+      +------+----------------+
   
也可以看看：
LengthFieldPrepender

Netty ReplayingDecoder 源码分析与特性解读

在介绍ReplayingDecoder之前想看一下它的用法，构建一个服务端和客户端的模型：

ReplayingDecoder java doc

例如，以下ByteToMessageDecoder实现：

使用ReplayingDecoder了如下简化：

这是如何运作的？

局限性

改善表现

用Enum调用checkpoint(T)

没有参数调用checkpoint()

用流水线中的另一个解码器替换一个解码器

几种常见的编解码器:

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

Netty ReplayingDecoder 源码分析与特性解读

在介绍ReplayingDecoder之前 想看一下它的用法，构建一个服务端和客户端的模型：

ReplayingDecoder java doc

例如，以下ByteToMessageDecoder实现：

使用ReplayingDecoder了如下简化：

这是如何运作的？

局限性

改善表现

用Enum调用checkpoint(T)

没有参数调用checkpoint()

用流水线中的另一个解码器替换一个解码器

几种常见的编解码器:

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

在介绍ReplayingDecoder之前想看一下它的用法，构建一个服务端和客户端的模型：