10.netty客户端与服务器使用protobuf传输报文

Posted 2022-09-05 PacosonSWJTU

【README】

本文总结自B站《尚硅谷-netty》，很不错； 内容如下：

•  netty的编码器与解码器；
• netty客户端与服务器通过 protobuf  传输报文的开发方式；
• 文末po出了所有代码；

---

【1】netty的编码器与解码器 codec

1）编解码器应用场景

• 编写网络应用程序时，因为数据在网络中传输的都是二进制字节码数据，在发送数据时就需要编码，接收数据时就需要解码 ；如下图

                                                  【图1】 编码与解码流程图

2） codec(编解码器) 的组成部分有两个：decoder(解码器)和encoder(编码器)。

• encoder 负责把业务数据（如pojo对象）转换成字节码数据；
• decoder 负责把字节码数据转换成业务数据（如pojo对象）；

补充： 编解码器对应的英文“codec”（compress和decompress简化而成的合成词语 )，from wikipedia；

3）java序列化的问题（所以才引入了 protobuf）：

• Netty 本身自带的 ObjectDecoder 和 ObjectEncoder 可以用来实现POJO对象或各种业务对象的编码和解码，底层使用的仍是 Java 序列化技术 , 而Java 序列化技术本身效率就不高，存在如下问题
  •  无法跨语言；
  •  序列化后的体积太大，是二进制编码的 5 倍多；
  •  序列化性能太低；

引入新的解决方案： google 的 Protobuf ；

---

【2】Protobuf

【2.1】概述

0） intro2 protobuf, refer2 https://developers.google.com/protocol-buffers ;

1）ProtoBuf定义：

• 是 Protocol Buffer 的缩写，即协议缓冲区；它是google实现的一种开源跨平台的序列化资料结构的协议。

2）ProtoBuf 是google提出的结构数据序列化方法，可简单类比于 XML，其具有以下特点：

• ① 语言无关、平台无关。即 ProtoBuf 支持 Java、C++、Python 等多种语言，支持多个平台；
• ② 高效。即比 XML 更小（3 ~ 10倍）、更快（20 ~ 100倍）、更为简单；
• ③ 扩展性、兼容性好。你可以更新数据结构，而不影响和破坏原有的旧程序；

【小结】

• 说的直接点， protobuf序列化性能 优于  java序列化；
• 因为protobuf 很适合做数据存储或 RPC[远程过程调用]数据交换格式，所以目前有些公司把远程调用的报文传输方式从 http+json 修改为 tcp+protobuf ；

3）基于protobuf的编码与解码流程

【图解】 使用protobuf序列化 java对象的步骤；

• 步骤1：编写 proto 文件；
• 步骤2：使用 protoc.exe 编译把 proto文件编译为 java文件（pojo文件）；
• 步骤3：创建pojo对象并赋值，发送到服务器；

【补充】关于 protobuf的更多介绍，refer2  https://www.jianshu.com/p/a24c88c0526a   

---

 【2.2】 netty客户端服务器通过protobuf传输报文

说明：

• 为了清晰展示 protobuf 的开发方式，下文使用了截图；但 netty通过 protobuf 传输报文的所有代码，会在文末po出；

【步骤1】 引入 protobuf 依赖

<dependency>
    <groupId>com.google.protobuf</groupId>    
    <artifactId>protobuf-java</artifactId>
    <version>3.6.1</version>
</dependency>

【步骤2】编写 proto 文件 

Student.proto

syntax="proto3"; // 版本
option java_outer_classname = "StudentPOJO"; // 生成的外部类名，同时也是文件名
// protobuf 使用 message 管理数据
message Student  // 会在 StudentPOJO 外部类生成一个内部类 Student，他是真正方发送的POJO对象
    int32 id = 1; // Student类中有一个属性，属性名为 id，类型为 int32; 1 表示属性序号并不是属性值；
    string name = 2;

【步骤3】编译 proto文件到 POJO javabean 文件 

编译命令：

protoc.exe --java_out=. Student.proto

【补充】

• protoc.exe 编译器到这个地方 Central Repository: com/google/protobuf/protoc/3.6.1 下载，并录入到path（系统环境变量）
• 生成的 javabean文件为：  StudentPOJO.java

---

【步骤4】 客户端发送 Student 对象到服务器

1）客户端添加 protobuf 编码器 ProtobufEncoder;

2） 客户端处理器

• 根据 protobuf编译成的 POJO类文件，创建对应javabean对象，并发送到服务器；

---

【步骤5】 服务器解码protobuf字节码

添加 protobuf解码器 ProtobufDecoder ；

netty服务器 处理器，接收POJO类对象，pojo类是由protobuf编译而成；

 【演示结果】netty客户端与服务器使用 probobuf传输报文

---

【3】netty 使用 protobuf传输报文的客户端与服务器代码（源代码）

【3.1】netty服务器

1）netty服务器 （注意它添加的protobuf解码器 ProtobufDecoder ）

public class ProtobufNettyServer76 

    public static void main(String[] args) throws InterruptedException 
        // 创建 BossGroup 和 WorkerGroup
        // 1. 创建2个线程组 bossGroup， workerGroup
        // 2 bossGroup 仅处理连接请求； 真正的业务逻辑，交给workerGroup完成；
        // 3 两个线程组都是无限循环
        // 4 bossGroup 和 workerGroup 含有的子线程（NIOEventLoop）个数
        // 默认是 cpu核数 * 2
        EventLoopGroup boosGroup = new NioEventLoopGroup();
        EventLoopGroup workerGruop = new NioEventLoopGroup();

        try 
            // 创建服务器端的启动对象， 配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            bootstrap.group(boosGroup, workerGruop) // 设置2个线程组
                    .channel(NioserverSocketChannel.class) // 使用NIOSocketChannel 作为服务器的通道实现
                    .option(ChannelOption.SO_BACKLOG, 128) // 设置线程队列等待连接的个数
                    .childOption(ChannelOption.SO_KEEPALIVE, true) // 设置保持活动连接状态
                    .childHandler(new ChannelInitializer<SocketChannel>()  // 创建一个通道初始化对象
                        // 给 pipeline 设置处理器
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception 
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            // 添加 protobuf 解码器, 指定对哪种对象进行解码
                            pipeline.addLast("decoder", new ProtobufDecoder(StudentPOJO.Student.getDefaultInstance()));
                            // 添加业务处理器
//                            pipeline.addLast(new ProtobufNettyServerHandler());
                            pipeline.addLast(new ProtobufNettyServerHandler2());
                        
                    );  // 给我们的workerGroup 的 EventLoop 对应的管道设置处理器
            System.out.println("... server is ready.");

            // 启动服务器， 绑定端口并同步处理 ，生成一个 ChannelFuture对象
            ChannelFuture channelFuture = bootstrap.bind(6668).sync();
            channelFuture.addListener((future1) -> System.out.println("Finish binding"));

            // 给 channelFuture 注册监听器，监听我们关心的事件
            channelFuture.addListener(future -> 
                if (future.isSuccess()) 
                    System.out.println("监听端口6668 成功");
                else 
                    System.out.println("监听端口6668 失败");
                
            );

            // 对关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
         finally 
            // 优雅关闭
            boosGroup.shutdownGracefully();
            workerGruop.shutdownGracefully();
        
    

2）netty服务器处理器（channelRead0方法中的 StudentPOJO.Student javabean就是由 protobuf编译生成的java类）

public class ProtobufNettyServerHandler2 extends SimpleChannelInboundHandler<StudentPOJO.Student> 

    // 读写数据事件（读取客户端发送的消息）
    // 1. ChannelHandlerContext ctx： 上下文信息，包括管道pipeline，通道channel，地址
    // 2. Object msg： 客户端发送的数据，默认是 Object
    @Override
    public void channelRead0(ChannelHandlerContext ctx, StudentPOJO.Student student) throws Exception 
        System.out.println("SimpleChannelInboundHandler子类handler-客户端发送的数据 id=" + student.getId() + " name=" + student.getName());
    

    // 数据读取完毕，回复客户端
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception 
        // writeAndFlush 是 write + flush ;把数据写入到缓冲，并刷新
        ChannelFuture channelFuture = ctx.writeAndFlush(Unpooled.copiedBuffer("hello, 客户端", StandardCharsets.UTF_8));
        channelFuture.addListener(future -> System.out.println("回复成功"));
    

    // 处理异常，关闭通道
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception 
        ctx.channel().close();
    

---

【3.2】netty客户端

1）netty客户端 （注意其添加的 Protobuf编码器 ProtobufEncoder）

public class ProtobufNettyClient76 
    public static void main(String[] args) throws InterruptedException 
        // 客户端需要一个事件循环组
        EventLoopGroup eventLoopGroup = new NioEventLoopGroup();
        try 
            // 创建客户端启动对象， 注意是 BootStrap
            Bootstrap bootstrap = new Bootstrap();
            // 设置相关参数
            bootstrap.group(eventLoopGroup) // 设置线程组
                    .channel(NioSocketChannel.class) // 设置客户端通道实现类(反射)
                    .handler(new ChannelInitializer<SocketChannel>() 
                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception 
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            // 添加 ProtobufEncoder 编码器
                            pipeline.addLast(new ProtobufEncoder());
                            // 添加业务处理器
                            pipeline.addLast(new ProtobufNettyClientHandler()); // 加入自己的处理器
                        
                    );
            System.out.println("client is ok");
            // 启动客户端去连接服务器
            // ChannelFuture， 设计到netty的异步模型
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();
            // 给关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
         finally 
            eventLoopGroup.shutdownGracefully();
        
    

2）netty客户端处理器 （注意 channelActive方法中的 StudentPOJO.Student，是由Protobuf编译生成的POJO类） 

public class ProtobufNettyClientHandler extends ChannelInboundHandlerAdapter 
    // 当通道就绪就会触发该方法
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception 
        // 发送一个 StudentPOJO 对象到服务器
        StudentPOJO.Student student = StudentPOJO.Student.newBuilder().setId(4).setName("zhangsan").build();
        ctx.writeAndFlush(student);
    

    // 当通道有读取事件时，会触发
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception 
        ByteBuf byteBuf = (ByteBuf) msg;
        System.out.println("服务器回复消息：" + byteBuf.toString(StandardCharsets.UTF_8) + ", 当前时间=" + DateUtils.getNowTimestamp());
        System.out.println("服务器地址：" + ctx.channel().remoteAddress());
    

    // 捕获异常
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception 
        cause.printStackTrace();
        ctx.close();