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

Posted PacosonSWJTU

tags:

篇首语:本文由小常识网(cha138.com)小编为大家整理,主要介绍了10.netty客户端与服务器使用protobuf传输报文相关的知识,希望对你有一定的参考价值。

【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

【补充】


【步骤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();
    

以上是关于10.netty客户端与服务器使用protobuf传输报文的主要内容,如果未能解决你的问题,请参考以下文章

Netty4.XUnity与Netty使用protoBuf

protoBuf 实现客户端与服务端

如何将 protobuf-net 与通过 ASP.NET Web Reference 生成的类一起使用,同时避免代码重复?

SuperSocket与Netty之实现protobuf协议,包括服务端和客户端

python 3.7使用protobuf-3.17.3版本

Erik Bergstedt的工具库简化了在.NET中使用Protobuf的方式