使用 Netty 整合 protobuf

Posted 2021-08-10 马士兵

protobuf 是谷歌的 Protocol Buffers 的简称，用于结构化数据和字节码之间互相转换（序列化、反序列化），一般应用于网络传输，可支持多种编程语言。

protobuf 如何使用这里不再介绍，本文主要介绍在 Netty 中如何使用 protobuf。

在前一篇文章（）中，有介绍到一种用一个固定为4字节的前缀Header来指定Body的字节数的一种消息分割方式，在这里同样要使用到。只是其中 Body 的内容不再是字符串，而是 protobuf 字节码。

在处理业务逻辑时，肯定不希望还要对数据进行序列化和反序列化，而是希望直接操作一个对象，那么就需要有相应的编码器和解码器，将序列化和反序列化的逻辑写在编码器和解码器中。有关编码器和解码器的实现，上一篇文章中有介绍。

Netty 包中已经自带针对 protobuf 的编码器和解码器，那么就不用再自己去实现了。

这里定义一个protobuf数据结构，用于描述一个学生的信息，保存为StudentMsg.proto文件：

message Student {
   // ID
   required int32 id = 1;  

   // 姓名
   required string name = 2;

   // email
   optional string email = 3;

   // 朋友
   repeated string friends = 4;
}

用 StudentMsg.proto 分别生成 Java 和 Python 代码，将代码加入到相应的项目中。生成的代码就不再贴上来了。下面介绍在 Netty 中如何使用 protobuf 来传输 Student 信息。

Netty 自带 protobuf 的编码器和解码器，分别是 ProtobufEncoder 和 ProtobufDecoder。需要注意的是，ProtobufEncoder 和 ProtobufDecoder 只负责 protobuf 的序列化和反序列化，而处理消息 Header 前缀和消息分割的还需要 LengthFieldBasedFrameDecoder 和 LengthFieldPrepender。LengthFieldBasedFrameDecoder 即用于解析消息 Header 前缀，根据 Header 中指定的 Body 字节数截取 Body，LengthFieldPrepender 用于在wirte消息时在消息前面添加一个 Header 前缀来指定 Body 字节数。

public class TcpServer {

   public static void main(String[] args) throws InterruptedException {
       EventLoopGroup bossGroup = new NioEventLoopGroup();
       EventLoopGroup workerGroup = new NioEventLoopGroup();
       try {
           ServerBootstrap b = new ServerBootstrap();
           b.group(bossGroup, workerGroup)
                   .channel(NioserverSocketChannel.class)
                   .childHandler(new ChannelInitializer<SocketChannel>() {
                       @Override
                       public void initChannel(SocketChannel ch)
                               throws Exception {
                           ChannelPipeline pipeline = ch.pipeline();

             // 负责通过4字节Header指定的Body长度将消息切割
             pipeline.addLast("frameDecoder",
                 new LengthFieldBasedFrameDecoder(1048576, 0, 4, 0, 4));

             // 负责将frameDecoder处理后的完整的一条消息的protobuf字节码转成Student对象
             pipeline.addLast("protobufDecoder",
                 new ProtobufDecoder(StudentMsg.Student.getDefaultInstance()));

             // 负责将写入的字节码加上4字节Header前缀来指定Body长度
             pipeline.addLast("frameEncoder", new LengthFieldPrepender(4));

             // 负责将Student对象转成protobuf字节码
             pipeline.addLast("protobufEncoder", new ProtobufEncoder());

                           pipeline.addLast(new TcpServerHandler());
                       }
                   });
           ChannelFuture f = b.bind(8080).sync();
           f.channel().closeFuture().sync();
       } finally {
           workerGroup.shutdownGracefully();
           bossGroup.shutdownGracefully();
       }
   }
}

处理事件时，接收和发送的参数直接就是Student对象。

public class TcpServerHandler extends ChannelInboundHandlerAdapter {

   @Override
   public void channelRead(ChannelHandlerContext ctx, Object msg) {

       // 读取客户端传过来的Student对象
     StudentMsg.Student student = (StudentMsg.Student) msg;
       System.out.println("ID:" + student.getId());
       System.out.println("Name:" + student.getName());
       System.out.println("Email:" + student.getEmail());
       System.out.println("Friends:");
       List<String> friends = student.getFriendsList();
       for(String friend : friends) {
         System.out.println(friend);
       }

       // 新建一个Student对象传到客户端
       StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder();
       builder.setId(9);
       builder.setName("服务器");
       builder.setEmail("[e-mail]");
       builder.addFriends("X");
       builder.addFriends("Y");
       StudentMsg.Student student2 = builder.build();
       ctx.writeAndFlush(student2);
   }

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

下面是Java编写的一个客户端测试程序：

public class TcpClient {

   public static void main(String[] args) throws IOException {

       Socket socket = null;
       DataOutputStream out = null;
       DataInputStream in = null;

       try {

           socket = new Socket("localhost", 8080);
           out = new DataOutputStream(socket.getOutputStream());
           in = new DataInputStream(socket.getInputStream());

           // 创建一个Student传给服务器
           StudentMsg.Student.Builder builder = StudentMsg.Student.newBuilder();
           builder.setId(1);
           builder.setName("客户端");
           builder.setEmail("[e-mail]");
           builder.addFriends("A");
           builder.addFriends("B");
           StudentMsg.Student student = builder.build();
           byte[] outputBytes = student.toByteArray(); // Student转成字节码
           out.writeInt(outputBytes.length); // write header
           out.write(outputBytes); // write body
           out.flush();

           // 获取服务器传过来的Student
           int bodyLength = in.readInt();  // read header
           byte[] bodyBytes = new byte[bodyLength];
           in.readFully(bodyBytes);  // read body
           StudentMsg.Student student2 = StudentMsg.Student.parseFrom(bodyBytes); // body字节码解析成Student
           System.out.println("Header:" + bodyLength);
           System.out.println("Body:");
           System.out.println("ID:" + student2.getId());
           System.out.println("Name:" + student2.getName());
           System.out.println("Email:" + student2.getEmail());
           System.out.println("Friends:");
           List<String> friends = student2.getFriendsList();
           for(String friend : friends) {
             System.out.println(friend);
           }

       } finally {
           // 关闭连接
           in.close();
           out.close();
           socket.close();
       }
   }
}

 
   
   
 

  
    
    
  
 
   
 
    
 
     
 
      
 
       
 
        
 
         
 
          
 
           
 
            
 
             
 
               
             
 
            
 
           
 
          
 
         
 
         
如有收获请划至底部
 
         
点击“在看”支持，谢谢！
 
         

 
         
 
          

 
         
 
         
关注马士兵
 
         
每天分享技术干货
 
         

 
         
 
        
 
       
 
      
 
     
 
    
 
   
 
  
 
   
   
 



点赞是最大的支持