22 - 高并发- 序列化Protobuf协议通讯

r囧r小猫 2023-06-13 03:28 31阅读 0赞

一、Protobuf简介

与JSON、XML相比，Protobuf算是后起之秀，是Google开源的一种数据格式。只是Protobuf更加适合于高性能、快速响应的数据传输应用场景。另外，JSON、XML是文本格式，数据具有可读性；而Protobuf是二进制数据格式，数据本身不具有可读性，只有反序列化之后才能得到真正可读的数据。正因为Protobuf是二进制数据格式，数据序列化之后，体积相比JSON和XML要小，更加适合网络传输。

总体来说，在一个需要大量数据传输的应用场景中，因为数据量很大，那么选择Protobuf可以明显地减少传输的数据量和提升网络IO的速度。对于打造一款高性能的通信服务器来说，Protobuf传输协议是最高性能的传输协议之一。微信的消息传输就采用了Protobuf协议。

二、proto文件来预先定义的消息格式

Protobuf使用proto文件来预先定义的消息格式。数据包是按照proto文件所定义的消息格式完成二进制码流的编码和解码。proto文件，简单地说，就是一个消息的协议文件，这个协议文件的后缀文件名为“.proto”。

（1）定义“.proto”文件

// [开始头部声明]
            //语法版本号
            syntax = "proto3";
            //指定包名，防止冲突
            package com.crazymakercircle.netty.protocol;
            // [结束头部声明]
    
            // [开始java选项配置]
            // 打包方式-两种方式：方式1.一个消息对应一个独立的java类。方式2.所有消息作为内部类，打包到一个外部类。默认2，false。
            option java_package = "com.crazymakercircle.netty.protocol";
            //方式2外部类名
            option java_outer_classname = "MsgProtos";
            // [结束java选项配置]
    
            // [开始消息定义]
            // 限定修饰符|数据类型|字段名称|分配标识号
            message Msg {
              uint32 id = 1;  //消息ID
              string content = 2; //消息内容
            }
    
            // [结束消息定义]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpZG9uZ2xpYW5nemhpY2Fp_size_16_color_FFFFFF_t_70][]

（2）生成POJO和Builder

<plugin>
                <groupId>org.xolstice.maven.plugins</groupId>
                <artifactId>protobuf-maven-plugin</artifactId>
                <version>0.5.0</version>
                <extensions>true</extensions>
                <configuration>
                  <! --proto文件路径-->
                    <protoSourceRoot>${project.basedir}/protobuf</protoSourceRoot>
    
                    <! --目标路径-->
                    <outputDirectory>${project.build.sourceDirectory}</outputDirectory>
                    <! --设置是否在生成Java文件之前清空outputDirectory的文件-->
                      <clearOutputDirectory>false</clearOutputDirectory>
                        <! --临时目录-->
                        <temporaryProtoFileDirectory>
                            ${project.build.directory}/protoc-temp
                        </temporaryProtoFileDirectory>
                        <! --protoc可执行文件路径-->
                        <protocExecutable>
                            ${project.basedir}/protobuf/protoc3.6.1.exe
                        </protocExecutable>
                </configuration>
                <executions>
                  <execution>
                      <goals>
                          <goal>compile</goal>
                          <goal>test-compile</goal>
                      </goals>
                </execution>
                </executions>
         </plugin>

配置好之后，执行插件的compile命令，Java代码就利索生成了。或者在Maven的项目编译时，POJO类和Builder类也会自动生成。

三、实例

（1）

<dependency>
                <groupId>com.google.protobuf</groupId>
                <artifactId>protobuf-java</artifactId>
                <version>${protobuf.version}</version>
            </dependency>

（2）使用Builder构造者，构造POJO消息对象

//...
            public class ProtobufDemo {
                public static MsgProtos.MsgbuildMsg() {
                  MsgProtos.Msg.Builder personBuilder = MsgProtos.Msg.newBuilder();
                  personBuilder.setId(1000);
                  personBuilder.setContent("内容@@@@@@@@@@@@@@@@@@@");
                  MsgProtos.Msg message = personBuilder.build();
                  return message;
                }
             //…..
            }

（3）将POJO序列化成二进制字节，将二进制字节反序列化为POJO

方式一、以对象为主体，常规处理。

*  序列化：POJO.toByteArray()
 *  反序列化：POJO.parseFrom(byte\[\] data)

//...
            public class ProtobufDemo {
    
                //第1种方式：序列化serialization &反序列化Deserialization
                @Test
                public void serAndDesr1() throws IOException {
                  MsgProtos.Msg message = buildMsg();
                  //将Protobuf对象序列化成二进制字节数组
                  byte[] data = message.toByteArray();
                  //可以用于网络传输，保存到内存或外存
                  ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
                  outputStream.write(data);
    
                  data = outputStream.toByteArray();
                  //二进制字节数组反序列化成Protobuf对象
                  MsgProtos.MsginMsg = MsgProtos.Msg.parseFrom(data);
                  Logger.info("id:=" + inMsg.getId());
                  Logger.info("content:=" + inMsg.getContent());
                }
            //….
            }

方式二、 以流为主体，存在半包问题。

*  序列化：POJO.writeTo(OutputStream)
 *  反序列化：POJO.parseFrom(InputStream)

//...
            public class ProtobufDemo {
             //…
                //第2种方式：序列化serialization &反序列化Deserialization
                @Test
                public void serAndDesr2() throws IOException {
                  MsgProtos.Msg message = buildMsg();
                  //序列化到二进制码流
                  ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
                  message.writeTo(outputStream);
    
                  ByteArrayInputStream inputStream = new ByteArrayInputStream(outputStream.toByteArray());
    
                  //从二进码流反序列化成Protobuf对象
                  MsgProtos.MsginMsg = MsgProtos.Msg.parseFrom(inputStream);
                  Logger.info("id:=" + inMsg.getId());
                  Logger.info("content:=" + inMsg.getContent());
                }
            //….
            }

方式三、以流为主体，解决半包问题

序列化的字节码之前加了字节数组的长度，长度为varint32类型。

*  序列化：POJO.writeDelimitedTo(outputStream)
 *  反序列化：POJO.parseDelimitedFrom(inputStream)

//...
            public class ProtobufDemo {
             //…
                //第3种方式：序列化serialization &反序列化Deserialization
                //带字节长度：[字节长度][字节数据]，解决粘包/半包问题
                @Test
                public void serAndDesr3() throws IOException {
                  MsgProtos.Msg message = buildMsg();
                  //序列化到二进制码流
                  ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
                  message.writeDelimitedTo(outputStream);
                  ByteArrayInputStream inputStream= new ByteArrayInputStream(outputStream.toByteArray());
                  //从二进码流反序列化成Protobuf对象
                  MsgProtos.Msg inMsg = MsgProtos.Msg.parseDelimitedFrom(inputStream);
                  Logger.info("id:=" + inMsg.getId());
                  Logger.info("content:=" + inMsg.getContent());
                }
            }

四、Protobuf编码解码实践

Netty默认支持Protobuf编码解码，内置了一套基础的Protobuf编码解码器。

*  ProtobufEncoder编码器：直接使用POJO.ByteArray()实现序列化。
 *  ProtobufDecoder解码器：指定POJO消息的原型实例，根据实例找到对应的Parser解析器，实现反序列化。
 *  ProtobufVarint32LengthFieldPrepender长度编码器：在ProtobufEncoder生成的字节数组之前，前置一个varint32数字，表示序列化的二进制字节数。
 *  ProtobufVarint32FrameDecoder长度解码器：根据数据包中varint32中的长度值，解码一个足额的字节数组。然后将字节数组交给下一站的解码器ProtobufDecoder。

服务端：

//...
            public class ProtoBufServer {
    
                //..．省略成员属性，构造器
    
                public void runServer() {
                  //创建反应器线程组
                  EventLoopGroup bossLoopGroup = new NioEventLoopGroup(1);
                  EventLoopGroup workerLoopGroup = new NioEventLoopGroup();
    
                  try {
                      //..．省略：启动器的反应器线程，设置配置项
                      //5 装配子通道流水线
                      b.childHandler(new ChannelInitializer<SocketChannel>() {
                          //有连接到达时会创建一个通道
                          protected void initChannel(SocketChannelch) throws Exception {
                              // 流水线管理子通道中的Handler业务处理器
                              // 向子通道流水线添加3个Handler业务处理器
                              ch.pipeline().addLast(new ProtobufVarint32FrameDecoder());
                                //传入参数为解码原型
                              ch.pipeline().addLast(new ProtobufDecoder(MsgProtos.Msg.getDefaultInstance()));
                              ch.pipeline().addLast(new ProtobufBussinessDecoder());
                          }
                      });
                      //...．省略端口绑定，服务监听，从容关闭
                }
                //服务器端的业务处理器
                static class ProtobufBussinessDecoder extends ChannelInboundHandlerAdapter
                {
                  @Override
                  public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                      MsgProtos.MsgprotoMsg = (MsgProtos.Msg) msg;
                      //经过流水线的各个解码器，到此Person类型已经可以断定
                      Logger.info("收到一个MsgProtos.Msg数据包 =》");
                      Logger.info("protoMsg.getId():=" + protoMsg.getId());
                      Logger.info("protoMsg.getContent():=" + protoMsg.getContent());
                  }
                }
    
                public static void main(String[] args) throws InterruptedException {
                  int port = NettyDemoConfig.SOCKET_SERVER_PORT;
                  new ProtoBufServer(port).runServer();
                }
            }

客户端：

//...
            public class ProtoBufSendClient {
                static String content = "内容内容内容内容内容内容内容内容内容！";
    
                //..．省略成员属性，构造器
    
                public void runClient() {
                  //创建反应器线程组
                  EventLoopGroup workerLoopGroup = new NioEventLoopGroup();
                  try {
                      //1 设置反应器线程组
                      b.group(workerLoopGroup);
                      //2 设置nio类型的通道
                      b.channel(NioSocketChannel.class);
                      //3 设置监听端口
                      b.remoteAddress(serverIp, serverPort);
                      //4 设置通道的参数
                      b.option(ChannelOption.ALLOCATOR,PooledByteBufAllocator.DEFAULT);
    
                      //5 装配通道流水线
                      b.handler(new ChannelInitializer<SocketChannel>() {
                        //初始化客户端通道
                        protected void initChannel(SocketChannel ch) throws Exception {
                            // 客户端太多流水线添加2个Handler业务处理器
                            ch.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
                            ch.pipeline().addLast(new ProtobufEncoder());
                        }
                      });
                      ChannelFuture f = b.connect();
                      //...
                      // 阻塞，直到连接完成
                      f.sync();
                      Channel channel = f.channel();
    
                      //发送Protobuf对象
                      for (int i = 0; i< 1000; i++) {
                        MsgProtos.Msg user = build(i, i + "->" + content);
                        channel.writeAndFlush(user);
                        Logger.info("发送报文数：" + i);
                      }
                      channel.flush();
    
                    //省略关闭等待，从容关闭
    
              }
              //构建ProtoBuf对象
              public MsgProtos.Msg build(int id, String content) {
                  MsgProtos.Msg.Builder builder = MsgProtos.Msg.newBuilder();
                  builder.setId(id);
                  builder.setContent(content);
                  return builder.build();
              }
    
              public static void main(String[] args) throws InterruptedException {
                  int port = NettyDemoConfig.SOCKET_SERVER_PORT;
                  String ip = NettyDemoConfig.SOCKET_SERVER_IP;
                  new ProtoBufSendClient(ip, port).runClient();
              }
            }

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