Protobuf3 + Netty4: 在socket上传输多种类型的protobuf数据

快来打我* 2021-11-09 09:30 278阅读 0赞

Protobuf序列化的字节流数据是不能自描述的，当我们通过socket把数据发送到Client时，Client必须知道发送的是什么类型的数据，才能正确的反序列化它。这严重影响限制了C/S功能的实现，不解决的话信道事实上只能传输一种类型的数据。本文讲解一下我用的解决办法，虽然我觉得应该有官方的实现更合理，即原生支持Protobuf的自描述。

（在金融领域，有一个叫FAST的协议，基本原理和Protobuf相同，并且有更高的压缩率，并且序列化后的字节流是自描述的，可以自动反序列化为对应的模板的数据（模板相当于.proto文件），但是时间效率比protobuf差，大家也可以关注一下。）

## 解决方案一 ##

首先，介绍另外一种实现，在protobuf官方wiki中描述的一种workaround，通过定义一种用于自描述的类型：

message SelfDescribingMessage {
      // Set of .proto files which define the type.
      required FileDescriptorSet proto_files = 1;
    
      // Name of the message type.  Must be defined by one of the files in
      // proto_files.
      required string type_name = 2;
    
      // The message data.
      required bytes message_data = 3;
    }

（参考：[https://developers.google.com/protocol-buffers/docs/techniques\#self-description][https_developers.google.com_protocol-buffers_docs_techniques_self-description]）

把实际要传输的类型的字节数组放在message\_data字段中，用proto\_files和type\_name字段来描述它的proto文件和类型。这样，信道上传输的都是SelfDescribingMessage类型，但是其上的负载可以是任何类型的数据。

我没有试过这种方式。我不太愿意使用这种方式的原因是，很显然，这样做需要进行2次序列化和2次反序列化，byte数组也要被创建2次。如果对应时延和性能敏感的系统，这样做不够好。

## 解决方案二 ##

今天主要要介绍的方案。在protobuf序列化的前面，加上一个自定义的头，这个头包含序列化的长度和它的类型。在解压的时候根据包头来反序列化。

假设socket上要传输2个类型的数据，股票行情信息和期权行情信息：

股票的.proto定义：

syntax = "proto3";
    
    package test.model.protobuf;
    
    option java_package = "test.model.protobuf";
    
    message StockTick {
        string stockId = 1;
        int price = 2;
    }

期权的.proto定义：

syntax = "proto3";
    
    package test.model.protobuf;
    
    option java_package = "test.model.protobuf";
    
    message OptionTick {
        string optionId = 1;
        string securityId = 2;
        int price = 3;
    }

netty4官方事实上已经实现了protobuf的编解码的插件，但是只能用于传输单一类型的protobuf序列化。我这里截取一段netty代码，熟悉netty的同学马上就能理解它的作用：

@Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast(new ProtobufVarint32FrameDecoder());
                pipeline.addLast(new ProtobufDecoder(StockTickOuterClass.StockTick.getDefaultInstance()));
                pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
                pipeline.addLast(new ProtobufEncoder());
                pipeline.addLast(new CustomProtoServerHandler());
            }

看以上代码高亮部分，netty4官方的编解码器必须指定单一的protobuf类型才行。具体每个类的作用：

**ProtobufEncoder**：用于对Probuf类型序列化。  
**ProtobufVarint32LengthFieldPrepender**：用于在序列化的字节数组前加上一个简单的包头，只包含序列化的字节长度。  
**ProtobufVarint32FrameDecoder**：用于decode前解决半包和粘包问题（利用包头中的包含数组长度来识别半包粘包）  
**ProtobufDecoder**：反序列化**指定的**Probuf字节数组为protobuf类型。

我们可以参考以上官方的编解码代码，将实现我们客户化的protobuf编解码插件，但是要支持多种不同类型protobuf数据在一个socket上传输：

### 编码器CustomProtobufEncoder： ###

import com.google.protobuf.MessageLite;
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandler.Sharable;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToByteEncoder;
    
    /**
     * 参考ProtobufVarint32LengthFieldPrepender 和 ProtobufEncoder
     */
    @Sharable
    public class CustomProtobufEncoder extends MessageToByteEncoder<MessageLite> {
        
    HangqingEncoder hangqingEncoder;
        
        public CustomProtobufEncoder(HangqingEncoder hangqingEncoder)
        {
            this.hangqingEncoder = hangqingEncoder;
        }
        
        @Override
        protected void encode(
                ChannelHandlerContext ctx, MessageLite msg, ByteBuf out) throws Exception {
            
    
            byte[] body = msg.toByteArray();
            byte[] header = encodeHeader(msg, (short)body.length);
            
            out.writeBytes(header);
            out.writeBytes(body);
            
            return;
        }
        
        private byte[] encodeHeader(MessageLite msg, short bodyLength) {
            byte messageType = 0x0f;
            
            if (msg instanceof StockTickOuterClass.StockTick) {
                messageType = 0x00;
            } else if (msg instanceof OptionTickOuterClass.OptionTick) {
                messageType = 0x01;
            }
            
            byte[] header = new byte[4];
            header[0] = (byte) (bodyLength & 0xff);
            header[1] = (byte) ((bodyLength >> 8) & 0xff);
            header[2] = 0; // 保留字段
            header[3] = messageType;
    
            return header;
    
        }
    }

CustomProtobufEncoder序列化传入的protobuf类型，并且为它创建了一个4个字节的包头，格式如下

<table> 
 <tbody> 
  <tr> 
   <td>body长度（low）</td> 
   <td>body长度<br>（high）</td> 
   <td>保留字节</td> 
   <td>类型</td> 
  </tr> 
 </tbody> 
</table>

其中的encodeHeader方法具体的实现要根据你要传输哪些protobuf类型来修改代码，也可以稍加设计避免使用太多的if…else。

### 解码器CustomProtobufDecoder： ###

import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ByteToMessageDecoder;
    import java.util.List;
    import com.google.protobuf.MessageLite;
    
    /**
     * 参考ProtobufVarint32FrameDecoder 和 ProtobufDecoder
     */
    
    public class CustomProtobufDecoder extends ByteToMessageDecoder {
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
            while (in.readableBytes() > 4) { // 如果可读长度小于包头长度，退出。
                in.markReaderIndex();
    
                // 获取包头中的body长度
                byte low = in.readByte();
                byte high = in.readByte();
                short s0 = (short) (low & 0xff);
                short s1 = (short) (high & 0xff);
                s1 <<= 8;
                short length = (short) (s0 | s1);
    
                // 获取包头中的protobuf类型
                in.readByte();
                byte dataType = in.readByte();
    
                // 如果可读长度小于body长度，恢复读指针，退出。
                if (in.readableBytes() < length) {
                    in.resetReaderIndex();
                    return;
                }
    
                // 读取body
                ByteBuf bodyByteBuf = in.readBytes(length);
    
                byte[] array;
                int offset;
    
                int readableLen= bodyByteBuf.readableBytes();
                if (bodyByteBuf.hasArray()) {
                    array = bodyByteBuf.array();
                    offset = bodyByteBuf.arrayOffset() + bodyByteBuf.readerIndex();
                } else {
                    array = new byte[readableLen];
                    bodyByteBuf.getBytes(bodyByteBuf.readerIndex(), array, 0, readableLen);
                    offset = 0;
                }
                
                //反序列化
                MessageLite result = decodeBody(dataType, array, offset, readableLen);
                out.add(result);
            }
        }
    
        public MessageLite decodeBody(byte dataType, byte[] array, int offset, int length) throws Exception {
            if (dataType == 0x00) {
                return StockTickOuterClass.StockTick.getDefaultInstance().
                        getParserForType().parseFrom(array, offset, length);
    
            } else if (dataType == 0x01) {
                return OptionTickOuterClass.OptionTick.getDefaultInstance().
                        getParserForType().parseFrom(array, offset, length);
            }
    
            return null; // or throw exception
        }
    }

CustomProtobufDecoder实现了2个功能，1）通过包头中的长度信息来解决半包和粘包。 2）把消息body反序列化为对应的protobuf类型（根据包头中的类型信息）。

其中的decodeBody方法具体的实现要根据你要传输哪些protobuf类型来修改代码，也可以稍加设计避免使用太多的if…else。

### 在Netty服务器上应用编解码器 ###

如何把我们自定义的编解码用于netty Server:

@Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ChannelPipeline pipeline = ch.pipeline();
                pipeline.addLast("decoder",new CustomProtobufDecoder());
                pipeline.addLast("encoder",new CustomProtobufEncoder());
                pipeline.addLast(new CustomProtoServerHandler());
            }

[Binhua Liu][]原创文章，转载请注明原地址[http://www.cnblogs.com/Binhua-Liu/p/5577622.html][http_www.cnblogs.com_Binhua-Liu_p_5577622.html]

转载于:https://www.cnblogs.com/Binhua-Liu/p/5577622.html

[https_developers.google.com_protocol-buffers_docs_techniques_self-description]: https://developers.google.com/protocol-buffers/docs/techniques#self-description
[Binhua Liu]: http://www.cnblogs.com/binhua-liu
[http_www.cnblogs.com_Binhua-Liu_p_5577622.html]: http://www.cnblogs.com/Binhua-Liu/p/5577622.html