ProtoBuf 编解码

﹏ヽ暗。殇╰゛Y 2022-09-07 12:27 326阅读 0赞

在网络程序开发过程中，数据在网络中是以二进制字节码的格式传输的，在发送业务数据前，先对业务数据进行编码，经过网络传输，收到数据后对数据进行解码，Netty提供了一些编解码器，它们底层采用了java序列号技术，有一定的弊端：首先不支持跨语言；序列化体积较大；序列化性能不高。针对这些问题，可以采用ProtoBuf来解决：ProtoBuf是一种支持跨语言传输、高性能、高可靠的数据存储格式，支持序列化，适合用作RPC远程调用的数据交换格式。ProtoBuf的使用是先编写.proto文件来描述消息，然后使用protoBuf编译器生成代码来使用，在Netty的管道中需要指定ProtoBuf编码器和解码器。

# **示例：测试传输单个对象** #

下载ProtoBuf编译器

protoc-3.6.1-win32.zip

解压后目录

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpYW5nd2VubWFpbA_size_16_color_FFFFFF_t_70][]

编写.proto文件(Student.proto)

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

生成java代码

拷贝Student.proto到ProtoBuf解压包的bin目录下 执行：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpYW5nd2VubWFpbA_size_16_color_FFFFFF_t_70 1][]

拷贝StudentPOJO到代码中：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpYW5nd2VubWFpbA_size_16_color_FFFFFF_t_70 2][]

package com.tech.netty.netty.codec;
    
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.protobuf.ProtobufDecoder;
    
    import java.util.HashSet;
    import java.util.Set;
    
    /**
     * @author lw
     * @since 2021/8/11
     **/
    public class NettyServer {
    
        public static void main(String[] args) throws InterruptedException {
    
            //1 创建两个线程组bossGroup workerGroup
            //2 bossGroup处理客户端连接请求 workerGroup处理客户端数据读写请求
            //3 他们都是事件循环组 默认含有CPU核数*2个事件循环NioEventLoop
            NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
            NioEventLoopGroup workerGroup = new NioEventLoopGroup(2);
            try {
                //创建服务器端启动对象，进行参数配置
                ServerBootstrap serverBootstrap = new ServerBootstrap();
    
                serverBootstrap.group(bossGroup, workerGroup)  //设置两个线程组
                        .channel(NioServerSocketChannel.class)  //设置服务器端通道为NioServerSocketChannel
                        .option(ChannelOption.SO_BACKLOG, 128) //设置连接队列允许的连接个数
                        .childOption(ChannelOption.SO_KEEPALIVE, true) //设置连接连接状态为活动连接
                        .childHandler(new ChannelInitializer<SocketChannel>() { //给workerGroup NioEventLoop通过管道添加处理器
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                //加入protoBuf的解码器
                                socketChannel.pipeline().addLast("decoder",new ProtobufDecoder(StudentPOJO.Student.getDefaultInstance()));
                                socketChannel.pipeline().addLast(new NettyServerHandler());
                            }
                        });
                System.out.println("...服务器 is ready...");
                //启动服务器并绑定一个端口
                System.out.println("ready");
                ChannelFuture channelFuture = serverBootstrap.bind(7001).sync();
                System.out.println("ok");
                channelFuture.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                        if(channelFuture.isSuccess()){
                            System.out.println("服务启动成功");
                        }else{
                            System.out.println("服务启动失败");
                        }
                    }
                });
                //对通道的关闭事件进行监听
                channelFuture.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

package com.tech.netty.netty.codec;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.channel.ChannelPipeline;
    import io.netty.util.CharsetUtil;
    import lombok.extern.slf4j.Slf4j;
    
    /**
     * @author lw
     * @since 2021/8/11
     **/
    //自定义一个Handler
    @Slf4j
    public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            log.info("channel={} | handlerAdded",ctx.channel().id());
        }
    
        //读取客户端发送的消息
        // ChannelHandlerContext 是一个上下文对象，可以获取通道和管道信息
        // msg 是客户端发送的消息
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            StudentPOJO.Student student = (StudentPOJO.Student) msg;
            System.out.println("接收到客户端消息：id="+student.getId()+",name="+student.getName());
        }
    
        //数据读取完毕
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            //发送数据
            ctx.writeAndFlush(Unpooled.copiedBuffer("hello 客户端", CharsetUtil.UTF_8));
        }
    
        //处理异常 一般是需要关闭通道
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

package com.tech.netty.netty.codec;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.protobuf.ProtobufEncoder;
    
    /**
     * @author lw
     * @since 2021/8/11
     **/
    public class NettyClient {
        public static void main(String[] args) throws InterruptedException {
            //客户端需要一个事件循环组
            NioEventLoopGroup eventExecutors = new NioEventLoopGroup();
            try {
                // 创建客户端启动对象 并配置参数
    
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(eventExecutors)  //设置线程组
                        .channel(NioSocketChannel.class)  //设置客户端通道
                        .handler(new ChannelInitializer<SocketChannel>() {  //设置处理器
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                //加入protoBuf的编码器
                                socketChannel.pipeline().addLast("encoder",new ProtobufEncoder());
                                socketChannel.pipeline().addLast(new NettyClientHandler());
                            }
                        });
                System.out.println("...客户端 is ready...");
                //连接服务器
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 7001).sync();
                //对通道关闭事件进行监听
                channelFuture.channel().closeFuture().sync();
            } finally {
                eventExecutors.shutdownGracefully();
            }
    
        }
    }

package com.tech.netty.netty.codec;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    /**
     * @author lw
     * @since 2021/8/11
     **/
    public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    
        //当通道就绪会触发该方法 给服务器发送数据
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            StudentPOJO.Student student = StudentPOJO.Student.newBuilder().setId(1).setName("张三").build();
            ctx.writeAndFlush(student);
            System.out.println("发送成功");
        }
    
        //当通道有读取事件时触发
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf byteBuf = (ByteBuf) msg;
            System.out.println("服务器回复的消息："+byteBuf.toString(CharsetUtil.UTF_8));
            System.out.println("服务器的地址："+ctx.channel().remoteAddress());
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

需要在管道中指定ProtoBuf编解码器

这样实现客户端与服务端使用protoBuf交换数据。

# **示例：测试传输多个对象** #

Student.proto  注意包路径的修改

syntax="proto3";
    option optimize_for=SPEED; //加快解析
    option java_package="com.tech.netty.netty.codecs"; //指定生成到哪个包下
    option java_outer_classname="MyDataInfo"; //外部类名，文件名
    
    //protoBuf 可以使用message管理其他的message
    message MyMessage{
        enum DataType{
            StudentType=0; //在proto3要求enum的编号从0开始
            WorkerType=1;
        }
    
        //用data_type来标识传的是哪一个枚举类型
        DataType data_type=1;
    
        //表示枚举类型最多只能出现其中的一个，节省空间
        oneof dataBody{
            Student student=2;
            Worker worker=3;
        }
    }
    
    message Student{
        int32 id=1; //Student类的属性
        string name=2;
    }
    
    message Worker{
        string name=1;
        int32 age=2;
    }

按照上面介绍的使用protoBuf编译器来生成代码MyDataInfo.java

package com.tech.netty.netty.codecs;
    
    
    import com.tech.netty.netty.codec.StudentPOJO;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelFutureListener;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.protobuf.ProtobufDecoder;
    
    /**
     * @author lw
     * @since 2021/8/11
     **/
    public class NettyServer {
    
        public static void main(String[] args) throws InterruptedException {
    
            //1 创建两个线程组bossGroup workerGroup
            //2 bossGroup处理客户端连接请求 workerGroup处理客户端数据读写请求
            //3 他们都是事件循环组 默认含有CPU核数*2个事件循环NioEventLoop
            NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
            NioEventLoopGroup workerGroup = new NioEventLoopGroup(2);
            try {
                //创建服务器端启动对象，进行参数配置
                ServerBootstrap serverBootstrap = new ServerBootstrap();
    
                serverBootstrap.group(bossGroup, workerGroup)  //设置两个线程组
                        .channel(NioServerSocketChannel.class)  //设置服务器端通道为NioServerSocketChannel
                        .option(ChannelOption.SO_BACKLOG, 128) //设置连接队列允许的连接个数
                        .childOption(ChannelOption.SO_KEEPALIVE, true) //设置连接连接状态为活动连接
                        .childHandler(new ChannelInitializer<SocketChannel>() { //给workerGroup NioEventLoop通过管道添加处理器
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                //加入protoBuf的解码器
                                socketChannel.pipeline().addLast("decoder",new ProtobufDecoder(MyDataInfo.MyMessage.getDefaultInstance()));
                                socketChannel.pipeline().addLast(new NettyServerHandler());
                            }
                        });
                System.out.println("...服务器 is ready...");
                //启动服务器并绑定一个端口
                System.out.println("ready");
                ChannelFuture channelFuture = serverBootstrap.bind(7001).sync();
                System.out.println("ok");
                channelFuture.addListener(new ChannelFutureListener() {
                    @Override
                    public void operationComplete(ChannelFuture channelFuture) throws Exception {
                        if(channelFuture.isSuccess()){
                            System.out.println("服务启动成功");
                        }else{
                            System.out.println("服务启动失败");
                        }
                    }
                });
                //对通道的关闭事件进行监听
                channelFuture.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

package com.tech.netty.netty.codecs;
    
    import com.tech.netty.netty.codec.StudentPOJO;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.util.CharsetUtil;
    import lombok.extern.slf4j.Slf4j;
    
    /**
     * @author lw
     * @since 2021/8/11
     **/
    //自定义一个Handler
    @Slf4j
    public class NettyServerHandler extends SimpleChannelInboundHandler<MyDataInfo.MyMessage> {
    
        @Override
        public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
            log.info("channel={} | handlerAdded",ctx.channel().id());
        }
    
        //读取客户端发送的消息
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, MyDataInfo.MyMessage msg) throws Exception {
            if(MyDataInfo.MyMessage.DataType.StudentType==msg.getDataType()){
                System.out.println("学生：id="+msg.getStudent().getId()+",name="+msg.getStudent().getName());
            }else if(MyDataInfo.MyMessage.DataType.WorkerType==msg.getDataType()){
                System.out.println("工人：age="+msg.getWorker().getAge()+",name="+msg.getWorker().getName());
            }else{
                System.out.println("传输的类型不正确");
            }
        }
    
        //数据读取完毕
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            //发送数据
            ctx.writeAndFlush(Unpooled.copiedBuffer("hello 客户端", CharsetUtil.UTF_8));
        }
    
        //处理异常 一般是需要关闭通道
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

package com.tech.netty.netty.codecs;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.protobuf.ProtobufEncoder;
    
    /**
     * @author lw
     * @since 2021/8/11
     **/
    public class NettyClient {
        public static void main(String[] args) throws InterruptedException {
            //客户端需要一个事件循环组
            NioEventLoopGroup eventExecutors = new NioEventLoopGroup();
            try {
                // 创建客户端启动对象 并配置参数
    
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(eventExecutors)  //设置线程组
                        .channel(NioSocketChannel.class)  //设置客户端通道
                        .handler(new ChannelInitializer<SocketChannel>() {  //设置处理器
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                //加入protoBuf的编码器
                                socketChannel.pipeline().addLast("encoder",new ProtobufEncoder());
                                socketChannel.pipeline().addLast(new NettyClientHandler());
                            }
                        });
                System.out.println("...客户端 is ready...");
                //连接服务器
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 7001).sync();
                //对通道关闭事件进行监听
                channelFuture.channel().closeFuture().sync();
            } finally {
                eventExecutors.shutdownGracefully();
            }
    
        }
    }

package com.tech.netty.netty.codecs;
    
    import com.tech.netty.netty.codec.StudentPOJO;
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.util.CharsetUtil;
    
    import java.util.Random;
    
    /**
     * @author lw
     * @since 2021/8/11
     **/
    public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    
        //当通道就绪会触发该方法 给服务器发送数据
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            int random = new Random().nextInt(3);
            MyDataInfo.MyMessage msg=null;
            if(0 ==random){ //发送student对象
                msg=MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.StudentType)
                        .setStudent(MyDataInfo.Student.newBuilder().setId(1).setName("张三").build()).build();
            }else{ //发送一个worker对象
                msg=MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.WorkerType)
                        .setWorker(MyDataInfo.Worker.newBuilder().setAge(10).setName("李四").build()).build();
            }
            ctx.writeAndFlush(msg);
            System.out.println("发送成功");
        }
    
        //当通道有读取事件时触发
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf byteBuf = (ByteBuf) msg;
            System.out.println("服务器回复的消息："+byteBuf.toString(CharsetUtil.UTF_8));
            System.out.println("服务器的地址："+ctx.channel().remoteAddress());
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpYW5nd2VubWFpbA_size_16_color_FFFFFF_t_70]: /images/20220829/b04125bbb28e4e51bc3af0d4478d31b6.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpYW5nd2VubWFpbA_size_16_color_FFFFFF_t_70 1]: /images/20220829/87da1ce13e544baa8e798b8a02ba3de4.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xpYW5nd2VubWFpbA_size_16_color_FFFFFF_t_70 2]: /images/20220829/581c42a92efd4eb4aa20a3c6b358ef8e.png