Netty 权威指南之Google protobuf 编解码

一时失言乱红尘 2023-10-16 21:50 53阅读 0赞

本章相关知识点：

Google 的protobuf 在业界非常流行，很多的商业项目选址使用protobuf作为编解码框架，这里一起回顾一下protobuf 框架的优点：

1、结构化数据存储（类似XML和JSON）

2、高效编码性能

3、语言无法、平台无法、拓展性好

4、官方支持Java、C++和Python

本章学习目标：

1、Protobuf 开发入门

2、开发支持Protobuf 的Netty 服务端

3、开发支持Protobuf 的Netty 客户端

4、运行基于Netty 开发的Protobuf 应用

第一节:Protobuf 开发入门

Protobuf 入门

protobuf 是一个灵活、高效、结构化的数据序列化框架。相比XML等传统序列化工具，它更小、更快、更简单.Protobuf支持数据结构化一次编译可以到处使用，甚至是跨语言使用，通过代码生成工具可以自动生成不同语言版本的源代码，甚至可以在不同的版本数据结构进程中进行数据传递，实现数据结构的前向兼容。

Protobuf 开发环境搭建

1、首先先下载protobuf 的windows 版本

Center

2、protoc.exe 主要根据.proto文件生成代码，我们以图书订购流程为例，定义SubscribeReq.proto和SubscribeResp.proto，数据定义格式如下：

SubscribeResp.proto 

  1. option java_package="com.nio.serlizable";
  2. option java_outer_classname="SubscribeRespProto";
  4. message SubscribeResp{
  5.         required int32 subReqId = 1;
  6.         required int32 respCode = 2;
  7.         required string desc = 3;        
  8. }

SubscribeReq.proto

  1. option java_package="com.nio.serlizable";
  2. option java_outer_classname="SubscribeReqProto";
  4. message SubscribeReq{
  5.         required int32 subReqId = 1;
  6.         required string userName = 2;
  7.         required string productName = 3;
  8.         repeated string address = 4;
  9. }

3、通过protoc.exe命令生成java代码，命令行如图所示：

Center 1

4、将生成的的POJO代码SubscribeReqProto.java和SubscribeRespProto.java复制到对应的IDEA14 项目当中。

Center 2

5、到此，我们已经完成对google protobuf 开发环境搭建工作，我们接下来通过一个简单的Demo 来了解Protobuf 类库使用。

  1. package com.nio.protobuf;
  3. import java.util.ArrayList;
  4. import java.util.List;
  6. /**
  7.  * Created by vixuan-008 on 2015/6/24.
  8.  */
  9. public class TestSubscribeReq {
  10.     public static void main(String[] args)throws Exception{
  11.         SubscribeReqProto.SubscribeReq req=createSubscribeReq();
  12.         System.out.println("Before encode:"+req.toString());
  14.         SubscribeReqProto.SubscribeReq result=decode(encode(req));
  15.         System.out.println("decode cotent is:"+result.toString());
  16.     }
  17.     private static byte[] encode(SubscribeReqProto.SubscribeReq req){
  18.             return req.toByteArray();
  19.     }
  20.     private static  SubscribeReqProto.SubscribeReq decode(byte[] body) throws Exception{
  21.         return SubscribeReqProto.SubscribeReq.parseFrom(body);
  22.     }
  24.     private static SubscribeReqProto.SubscribeReq createSubscribeReq(){
  25.         SubscribeReqProto.SubscribeReq.Builder builder= SubscribeReqProto.SubscribeReq.newBuilder();
  26.         builder.setSubReqId(1);
  27.         builder.setUserName("zhouzhigang");
  28.         builder.setProductName("Netty Book");
  29.         List<String> address=new ArrayList<String>();
  30.         address.add("湖南长沙");
  31.         address.add("湖南株洲");
  32.         address.add("湖南湘潭");
  33.         builder.addAllAddress(address);
  34.         return builder.build();
  36.     }
  37. }

首先我们看如何创建 SubscribeReqProto.SubscribeReq的实例，通过SubscribeReqProto.SubscribeReq的静态方法newBuilder创建SubscribeReqProto.SubscribeReq的Builder实例,通过Builder构造器对SubscribeReq的属性进行相关设置，对于集合类型，通过addAllXXX()方法可以将集合对象添加到对象属性当中。

编码通过调用SubscribeReqProto.SubscribeReq实例的toByteArray方法，即可将SubscribeReq对象编码为byte数组，使用非常方便。

解码通过调用SubscribeReqProto.SubscribeReq的静态方法parseFrom将二进制数组解码为原始数据对象。

6、Protobuf 测试程序效果截图:

Center 3

第二节:开发支持Protobuf 的Netty 服务端

SubRespProServer.java源代码（Handler 存在问题可能需要修改） 

  1. package com.nio.server;
  3. import com.nio.handler.SubRespProHandler;
  4. import com.nio.protobuf.SubscribeReqProto;
  5. import io.netty.bootstrap.ServerBootstrap;
  6. import io.netty.channel.ChannelFuture;
  7. import io.netty.channel.ChannelInitializer;
  8. import io.netty.channel.ChannelOption;
  9. import io.netty.channel.EventLoopGroup;
  10. import io.netty.channel.nio.NioEventLoopGroup;
  11. import io.netty.channel.socket.SocketChannel;
  12. import io.netty.channel.socket.nio.NioServerSocketChannel;
  13. import io.netty.handler.codec.protobuf.ProtobufDecoder;
  14. import io.netty.handler.codec.protobuf.ProtobufEncoder;
  15. import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
  16. import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
  17. import io.netty.handler.logging.LogLevel;
  18. import io.netty.handler.logging.LoggingHandler;
  20. /**
  21.  * Created by vixuan-008 on 2015/6/24.
  22.  */
  23. public class SubRespProServer {
  24.     public static void main(String[] args)throws  Exception{
  25.         int port=15444;
  26.         new SubRespProServer().bind(port);
  28.     }
  30.     public void bind(int port)throws Exception{
  31.         //配置服务端的NIO线程池
  32.         EventLoopGroup bossGroup=new NioEventLoopGroup();
  33.         EventLoopGroup workGroup=new NioEventLoopGroup();
  34.         try{
  35.             ServerBootstrap b=new ServerBootstrap();
  36.             b.group(bossGroup, workGroup);
  37.             b.channel(NioServerSocketChannel.class);
  38.             b.option(ChannelOption.SO_BACKLOG, 100);
  39.             b.handler(new LoggingHandler(LogLevel.INFO));
  40.             b.childHandler(new ChannelInitializer<SocketChannel>() {
  41.                 @Override
  42.                 protected void initChannel(SocketChannel socketChannel) throws Exception {
  43.                    socketChannel.pipeline().addLast(new ProtobufVarint32FrameDecoder());
  44.                     socketChannel.pipeline().addLast(new ProtobufDecoder(SubscribeReqProto.SubscribeReq.getDefaultInstance()));
  45.                     socketChannel.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
  46.                     socketChannel.pipeline().addLast(new ProtobufEncoder());
  47.                     socketChannel.pipeline().addLast(new SubRespProHandler());
  49.                 }
  50.             });
  51.             //绑定端口，等待同步成功
  52.             ChannelFuture f=b.bind(port).sync();
  53.             //等待服务端关闭监听端口
  54.             f.channel().closeFuture().sync();
  56.         }finally {
  57.             //释放线程池资源
  58.             bossGroup.shutdownGracefully();
  59.             workGroup.shutdownGracefully();
  61.         }
  64.     }
  65. }
  67. package com.nio.handler;
  69. import com.nio.protobuf.SubscribeReqProto;
  70. import com.nio.protobuf.SubscribeRespProto;
  71. import io.netty.channel.ChannelHandlerAdapter;
  72. import io.netty.channel.ChannelHandlerContext;
  74. /**
  75.  * Created by vixuan-008 on 2015/6/24.
  76.  */
  77. public class SubRespProHandler extends ChannelHandlerAdapter {
  79.     @Override
  80.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  81.        ctx.close();
  82.     }
  84.     @Override
  85.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  86.         super.channelActive(ctx);
  87.     }
  89.     @Override
  90.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  91.         SubscribeReqProto.SubscribeReq req=(SubscribeReqProto.SubscribeReq)msg;
  92.         System.out.println("server receiver client message is:"+req.toString());
  93.         ctx.writeAndFlush(resp(req.getSubReqId()));
  94.     }
  96.     private SubscribeRespProto.SubscribeResp resp(int subReqId)throws Exception{
  97.         SubscribeRespProto.SubscribeResp.Builder resp= SubscribeRespProto.SubscribeResp.newBuilder();
  98.         resp.setSubReqId(subReqId);
  99.         resp.setRespCode(0);
  100.         resp.setDesc("Netty Book order succeed 3 day later,sent to the designated adderss");
  101.         return resp.build();
  102.     }
  104.     @Override
  105.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  106.         super.channelReadComplete(ctx);
  107.     }
  108. }

第三节: 开发支持Protobuf 的Netty 客户端

  1. package com.nio.client;
  3. import com.nio.handler.SubReqProHandler;
  4. import com.nio.handler.SubReqServerHandler;
  5. import com.nio.protobuf.SubscribeRespProto;
  6. import io.netty.channel.ChannelFuture;
  7. import io.netty.channel.ChannelInitializer;
  8. import io.netty.channel.ChannelOption;
  9. import io.netty.channel.EventLoopGroup;
  10. import io.netty.channel.nio.NioEventLoopGroup;
  11. import io.netty.channel.socket.SocketChannel;
  12. import io.netty.channel.socket.nio.NioSocketChannel;
  13. import io.netty.handler.codec.protobuf.ProtobufDecoder;
  14. import io.netty.handler.codec.protobuf.ProtobufEncoder;
  15. import io.netty.handler.codec.protobuf.ProtobufVarint32FrameDecoder;
  16. import io.netty.handler.codec.protobuf.ProtobufVarint32LengthFieldPrepender;
  17. import io.netty.handler.codec.serialization.ClassResolvers;
  18. import io.netty.handler.codec.serialization.ObjectDecoder;
  19. import io.netty.handler.codec.serialization.ObjectEncoder;
  21. /**
  22.  * Created by vixuan-008 on 2015/6/24.
  23.  */
  24. public class SubReqProClient {
  25.     public static void main(String[] args)throws Exception{
  26.         int port=15444;
  27.         new SubReqProClient().bind(port, "127.0.0.1");
  28.     }
  30.     public void bind(int port,String host)throws Exception{
  31.         //配置客户端NIO线程池
  32.         EventLoopGroup workGroup=new NioEventLoopGroup();
  33.         try{
  34.             io.netty.bootstrap.Bootstrap b=new io.netty.bootstrap.Bootstrap();
  35.             b.group(workGroup);
  36.             b.channel(NioSocketChannel.class);
  37.             b.option(ChannelOption.TCP_NODELAY,true);
  38.             b.handler(new ChannelInitializer<SocketChannel>() {
  39.                 @Override
  40.                 protected void initChannel(SocketChannel socketChannel) throws Exception {
  41.                     socketChannel.pipeline().addLast(new ProtobufVarint32FrameDecoder());
  42.                     socketChannel.pipeline().addLast(new ProtobufDecoder(SubscribeRespProto.SubscribeResp.getDefaultInstance()));
  43.                     socketChannel.pipeline().addLast(new ProtobufVarint32LengthFieldPrepender());
  44.                     socketChannel.pipeline().addLast(new ProtobufEncoder());
  45.                     socketChannel.pipeline().addLast(new SubReqProHandler());
  46.                 }
  47.             });
  48.             //发起异步连接操作
  49.             ChannelFuture f=b.connect(host,port).sync();
  50.             //等待客户端链路关闭
  51.             f.channel().closeFuture().sync();
  53.         }finally {
  54.             //释放NIO 线程组
  55.             workGroup.shutdownGracefully();
  57.         }
  60.     }
  61. }
  63. package com.nio.handler;
  65. import com.nio.protobuf.SubscribeReqProto;
  66. import com.nio.protobuf.SubscribeRespProto;
  67. import io.netty.channel.ChannelHandlerAdapter;
  68. import io.netty.channel.ChannelHandlerContext;
  70. import java.util.ArrayList;
  71. import java.util.List;
  73. /**
  74.  * Created by vixuan-008 on 2015/6/24.
  75.  */
  76. public class SubReqProHandler extends ChannelHandlerAdapter {
  77.     @Override
  78.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  79.       ctx.close();
  80.     }
  82.     @Override
  83.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  84.         for(int i=0;i<10;i++){
  85.             ctx.write(subReq(i));
  86.         }
  87.         ctx.flush();
  88.     }
  90.     private SubscribeReqProto.SubscribeReq subReq(int i){
  91.         SubscribeReqProto.SubscribeReq.Builder req=SubscribeReqProto.SubscribeReq.newBuilder();
  92.        req.setProductName("Netty Book");
  93.         req.setUserName("zhouzhigang");
  94.         req.setSubReqId(i);
  95.       List<String> address=new ArrayList<String>();
  96.         address.add("china");
  97.         address.add("usa");
  98.         address.add("france");
  99.         req.addAllAddress(address);
  100.         return req.build();
  105.     }
  107.     @Override
  108.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  109.        System.out.println("Receiver server message is:"+msg);
  110.     }
  112.     @Override
  113.     public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
  114.         ctx.flush();
  115.     }
  116. }

第四节: 运行基于Netty 开发的Protobuf 应用

服务端截图:

Center 4

Protobuf 使用注意事项：

1、ProtobufDecoder仅仅负责解码，它不支持读半包。因此，在ProtobufDecode前面，一定要有能够处理读半包的解码器，有三种方式可以选择.

使用Netty提供的ProtobufVarint32FrameDecoder，它可以处理半包消息：

集成Netty提供的通用半包解码器LengthFieldBasedFrameDecoder;

继承ByteMessageDecoder类，自己处理半包消息。

如果你只使用ProtobufDecoder解码器而忽视对半包消息处理，程序是不能正常运行的。今天的学习就到这里

发表评论

表情：
评论列表 （有 0 条评论，53人围观）

还没有评论，来说两句吧...

相关阅读

    相关 ProtoBuf 编解码

    在网络程序开发过程中，数据在网络中是以二进制字节码的格式传输的，在发送业务数据前，先对业务数据进行编码，经过网络传输，收到数据后对数据进行解码，Netty提供了一些编解码器，它

    ﹏ヽ暗。殇╰゛Y﹏ヽ暗。殇╰゛Y/ 0 赞/ 400 阅读