17-高并发- 编码和解码-Decoder原理与实践

淡淡的烟草味﹌ 2023-06-12 09:25 56阅读 0赞

一、解码和编码的作用

*  netty从底层java通道读到ByteBuf二进制数据，传入Netty通道的流水线。随后开始入站处理。
 *  在入站处理的过程中，需要将ByteBuf二进制数据解码成POJO对象。这个解码过程，通过Netty的Decoder解码器实现。
 *  在出站处理过程中，业务处理后的结果，需要从POJO编码为最终的ByteBuf二进制数据，然后通过底层java通道发送到对端。这个编码过程，通过Netty的Encoder解码器实现。

二、Decoder原理与实践

**一个标准的解码器将输入类型为ByteBuf缓冲区的数据进行解码，输出一个一个的Java POJO对象。Netty内置了这个解码器，叫作ByteToMessageDecoder，位在Netty的io.netty.handler.codec包中。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZG9uZ2xpYW5nemhpY2Fp_size_16_color_FFFFFF_t_70][]

*  作为解码器的父类，ByteToMessageDecoder仅仅提供了一个流程性质的框架：它仅仅将子类的decode方法解码之后的Object结果，放入自己内部的结果列表List<Object>中，最终，父类会负责将List<Object>中的元素，一个一个地传递给下一个站。
 *  如果要实现一个自己的解码器，首先继承ByteToMessageDecoder抽象类。然后，实现其基类的decode抽象方法。将解码的逻辑，写入此方法。
 *  ReplayingDecoder类是ByteToMessageDecoder的子类。其作用是：  
    · 在读取ByteBuf缓冲区的数据之前，需要检查缓冲区是否有足够的字节。  
    · 若ByteBuf中有足够的字节，则会正常读取；反之，如果没有足够的字节，则会停止解码。

三、整数分包解码案例

**在Java的NIO中，由于NIO的非阻塞性，就会出现图中这样的问题。怎样保证一次性读取到完整的数据，就成了一个大问题。**

答：如果发现当前ByteBuf中所有可读数据不够，需要结束解析，直到可读数据是足够的。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZG9uZ2xpYW5nemhpY2Fp_size_16_color_FFFFFF_t_70 1][]

解码整数序列-解码两个整数，并求和作为解码结果

相当于设置一个游戏存档

父类中有两个重要的变量：
            private S state;
            private int checkpoint;//保存当前读取位置//相当于一个游戏存档
    
            //...
            public class IntegerAddDecoder extends ReplayingDecoder<IntegerAddDecoder.Status>             
            {
                enum Status {
                  PARSE_1,
                  PARSE_2
                }
                private int first;
                private int second;
    
                public IntegerAddDecoder() {
                //在构造函数中，需要初始化父类的state属性，表示当前阶段
                  super(Status.PARSE_1);
                }
    
                @Override
                protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception 
                {
                  switch (state()) {
                      case PARSE_1:
                          //从装饰器ByteBuf中读取数据
                          first = in.readInt();
                          //第一步解析成功，
                          //进入第二步，并且设置“读断点指针”为当前的读取位置
                          checkpoint(Status.PARSE_2);
                          break;
    
                      case PARSE_2:
                          second = in.readInt();
                          Integer sum = first + second;
                          out.add(sum);
                          checkpoint(Status.PARSE_1);
                          break;
                      default:
                          break;
                  }
                }
            }

四、字符串分包解码器

原理上，整数分包和字符串分包传输解码是一样的。不同的是：整数的长度是固定的，java中是4个字节。而字符串的长度是不固定的，是可变长度。

一般来说，netty中进行字符串传输，可用采用普通的Header-Content协议。

*  在协议Head中放置字符串的长度，一个int即可。
 *  在协议Content中放置字符串的字节数组。

**看代码可能会有误解，每一次解析就是一次IO读事件触发的。当ByteBuf长度不够时，本次解析就会终止，同时还原存档。下一个IO事件读取时候触发，从上一次存档开始读取 。**

那么，一个Header-Content包，在发送端会被编码成一个ByteBuf发送包。当到达接收端后，可能被分成很多ByteBuf接收包。对于这些参差不齐的接收包，如何解码成最初的ByteBuf包？

使用ReplayingDecoder解码器。ReplayingDecoder更多的是应用于数据解析逻辑简单的场景。在数据解析复杂的应用场景，建议使用在前文介绍的解码器ByteToMessageDecoder或者其子类，它们会更加合适。

下面为代码：

public class StringReplayDecoderextends ReplayingDecoder<StringReplayDecoder.Status> {
              //每一个上层Header-Content内容的读取阶段
              enum Status {
                  PARSE_1, PARSE_2
              }
              private int length;
              private byte[] inBytes;
              //在构造函数中，需要初始化父类的state属性，表示当前阶段
              public StringReplayDecoder() {
                  super(Status.PARSE_1);
              }
              @Override
              protected void decode(ChannelHandlerContextctx, ByteBuf in,
                                  List<Object> out) throws Exception {
    
                  switch (state()) {
                      case PARSE_1:
                        //第一步，从装饰器ByteBuf中读取长度
                        length = in.readInt();
                        inBytes = new byte[length];
                        // 进入第二步，读取内容
                        // 并且设置“读断点指针”为当前的读取位置，同时设置下一个阶段
                        checkpoint(Status.PARSE_2);
                        break;
                      case PARSE_2:
                        //第二步，从装饰器ByteBuf中读取内容数组
                        in.readBytes( inBytes, 0, length);
                        out.add(new String(inBytes, "UTF-8"));
                        // 第二步解析成功，
                        // 进入第一步，读取下一个字符串的长度
                        // 并且设置“读断点指针”为当前的读取位置，同时设置下一个阶段
                         checkpoint(Status.PARSE_1);
                        break;
                      default:
                        break;
                  }
              }
    }

**在数据解析逻辑复杂的应用场景，ReplayingDecoder在解析速度上相对较差。原因是什么呢？**

答：在ByteBuf中长度不够时，ReplayingDecoder会捕获一个ReplayError异常，这时会把ByteBuf中的读指针还原到之前的读断点指针（checkpoint），然后结束这次解析操作，等待下一次IO读事件。在网络条件比较糟糕时，一个数据包的解析逻辑会被反复执行多次，如果解析过程是一个消耗CPU的操作，那么这对CPU是个大负担。

五、ByteToMessageDecoder解码器实践

//…
            public class StringIntegerHeaderDecoder extends ByteToMessageDecoder {
                @Override
                protected void decode(ChannelHandlerContext channelHandlerContext,
                          ByteBuf buf, List<Object> out) throws Exception {
                  //可读字节小于4，消息头还没读满，返回
                  if (buf.readableBytes() < 4) {
                      return;
                  }
                  //消息头已经完整
                  //在真正开始从缓冲区读取数据之前，调用markReaderIndex()设置回滚点
                  // 回滚点为消息头的readIndex读指针位置
                  buf.markReaderIndex();
                  int length = buf.readInt();
                  //从缓冲区中读出消息头的大小，这会使得readIndex读指针前移
                  //剩余长度不够消息体，重置读指针
                  if (buf.readableBytes() < length) {
                      //读指针回滚到消息头的readIndex位置处，未进行状态的保存
                      buf.resetReaderIndex();
                      return;
                  }
                  // 读取数据，编码成字符串
                  byte[] inBytes = new byte[length];
                  buf.readBytes(inBytes, 0, length);
                  out.add(new String(inBytes, "UTF-8"));
                }
            }

*  在读取数据之前，需要调用buf.markReaderIndex()记录当前的位置指针，当可读内容不够，**也就是buf.readableBytes() < length时，需要调用buf.resetReaderIndex()方法将读指针回滚到旧的读起始位置。（没全部读完就回档）**
 *  在ByteToMessageDecoder的内部，有一个二进制字节的累积器cumulation，它用来保存没有解析完的二进制内容。所以，ByteToMessageDecoder及其子类是有状态的业务处理器，也就是说，它不能共享，在每次初始化通道的流水线时，都要重新创建一个ByteToMessageDecoder或者它的子类的实例。

六、Netty提供的Decoder解码器  
**（1）固定长度数据包解码器——FixedLengthFrameDecoder**

*  适用场景：每个接收到的数据包的长度，都是固定的，例如100个字节。
 *  注：在这种场景下，只需要把这个解码器加到流水线中，它会把入站ByteBuf数据包拆分成一个个长度为100的数据包，然后发往下一个channelHandler入站处理器。补充说明一下：**这里所指的一个数据包，在Netty中就是一个ByteBuf实例。**
 *  注：数据帧（Frame），本书也通称为数据包。

**（2）行分割数据包解码器——LineBasedFrameDecoder**

*  适用场景：每个ByteBuf数据包，使用换行符（或者回车换行符）作为数据包的边界分割符。
 *  如果每个接收到的数据包，都以换行符/回车换行符作为分隔。在这种场景下，只需要把这个解码器加到流水线中，Netty会使用换行分隔符，把ByteBuf数据包分割成一个一个完整的应用层ByteBuf数据包，再发送到下一站。

**（3）自定义分隔符数据包解码器——DelimiterBasedFrameDecoder**

*  DelimiterBasedFrameDecoder是LineBasedFrameDecoder按照行分割的通用版本。不同之处在于，这个解码器更加灵活，可以自定义分隔符，而不是局限于换行符。如果使用这个解码器，那么接收到的数据包，末尾必须带上对应的分隔符。

**（4）自定义长度数据包解码器——LengthFieldBasedFrameDecoder**

*  这是一种基于灵活长度的解码器。在ByteBuf数据包中，加了一个长度字段，保存了原始数据包的长度。解码的时候，会按照这个长度进行原始数据包的提取。

//....
            public class NettyOpenBoxDecoder {
                public static final int VERSION = 100;
                static String content = "以上以上以上以上以上以上以上以上以上以上！";
                //...
                /＊＊
                ＊ LengthFieldBasedFrameDecoder使用实例 1
                ＊/
                @Test
                public void testLengthFieldBasedFrameDecoder1() {
                  try {
                      final LengthFieldBasedFrameDecoder spliter =new LengthFieldBasedFrameDecoder(1024, 0, 4,0,4);
                      ChannelInitializeri = new ChannelInitializer<EmbeddedChannel>() {
                          protected void initChannel(EmbeddedChannelch) {
                              ch.pipeline().addLast(spliter);
                              ch.pipeline().addLast(new StringDecoder(Charset.forName("UTF-8")));
                              ch.pipeline().addLast(new StringProcessHandler());
                          }
                      };
                      EmbeddedChannel channel = new EmbeddedChannel(i);
    
                      for (int j = 1; j <= 100; j++) {
                          ByteBufbuf = Unpooled.buffer();
                          String s = j + "次发送->"+content;
                          byte[] bytes = s.getBytes("UTF-8");
                          buf.writeInt(bytes.length );
                          buf.writeBytes(bytes);
                          channel.writeInbound(buf);
                      }
                      Thread.sleep(Integer.MAX_VALUE);
                  } catch (InterruptedException e) {
                      e.printStackTrace();
                  } catch (UnsupportedEncodingException e) {
                      e.printStackTrace();
                  }
                }
            }

ByteBuf[58字节] = 长度字段[4字节]+版本字段[2字节]+content字段[52字节]
    
    长度字段：就是指Head
    
    
    （1）maxFrameLength：发送的数据包的最大长度。示例程序中该值为1024，表示一个数据包最多可发送1024个字节。
    （2）lengthFieldOffset：长度字段偏移量。指的是长度字段位于整个数据包内部的字节数组中的下标值。
    （3）lengthFieldLength：长度字段所占的字节数。如果长度字段是一个int整数，则为4，如果长度字段是一个short整数，则为2。
    （4）lengthAdjustment：长度的矫正值。这个参数最为难懂。在传输协议比较复杂的情况下，例如包含了长度字段、协议版本号、魔数等等。那么，解码时，就需要进行长度矫正。长度矫正值的计算公式为：内容字段偏移量 - 长度字段偏移量 - 长度字段的字节数。就是上面的版本字段[2字节]，即版本号的长度。
    （5）initialBytesToStrip：丢弃的起始字节数。在有效数据字段Content前面，还有一些其他字段的字节，作为最终的解析结果，可以丢弃。例如，上面的示例程序中，前面有4个节点的长度字段，它起辅助的作用，最终的结果中不需要这个长度，所以丢弃的字节数为4。         
    
    
    
    
    
             public LengthFieldBasedFrameDecoder(
                      int maxFrameLength,         //发送的数据包最大长度
                      int lengthFieldOffset,     //长度字段偏移量
                      int lengthFieldLength,     //长度字段自己占用的字节数
                      int lengthAdjustment,      //长度字段的偏移量矫正
                      int initialBytesToStrip)  //丢弃的起始字节数
             {
                //...
             }
    
    
    多字段Head-Content协议数据帧解析

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZG9uZ2xpYW5nemhpY2Fp_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20191112091612394.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZG9uZ2xpYW5nemhpY2Fp,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZG9uZ2xpYW5nemhpY2Fp_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20191112110724379.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZG9uZ2xpYW5nemhpY2Fp,size_16,color_FFFFFF,t_70