当前位置：网站首页 > 热门文章 > 正文

您懂Netty高性能消息传输GoogleProtobuf吗

bigegpt 2024-11-24 12:00 4 浏览

编码和解码的基本介绍

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

2、codec(编解码器)的组成部分有两个：decoder（解码器）和encoder（编码器），encoder负责把业务数据转换成字节码数据，decoder负责把字节码数据转换成业务数据。

Netty本身的编码解码机制问题分析

1、Netty自身提供了一些codec(编解码器)

2、Netty提供的编码器如下：

① StringEncoder，对字符串数据进行编码；

② ObjectEncoder，对Java对象进行编码；

......

3、Netty提供的解码器

① StringDecoder,对字符串数据进行解码；

② ObjectDecoder，对Java对象进行解码；

......

4、Netty本身自带的ObjectDecoder和ObjectEncoder可以用来实现POJO对象或各种业务对象的编码和解码，底层使用的仍是Java系列化技术，而Java系列化技术本身效率就不高，存在如下问题：

① 无法跨语言；

② 系列化后的体积太大，是二进制编码的5倍多；

③ 系列化性能太低

=>引出新的解决方案【Google的Protobuf】

Protobuf基本介绍和使用意图

1、Protobuf是Google发布的开源项目，全称Google Protocol Buffers，是一种轻便高效的结构化数据存储格式，可以用于结构化数据串行化，或者说系列化。它很适合做数据存储或RPC【远程过程调用remote procedure call】数据交换格式,目前很多公司将 http+json -> tcp+protobuf

2、参考文档：https://developers.google.com/protocol-buffers/docs/proto

3、Protobuf是以message的方式来管理数据的；

4、支持跨平台、跨语言，即【客户端和服务端可以是不同的语言写的】（支持目前绝大多数语言，例如：C++，C#，java，python等）；

5、高性能、高可靠性；

6、使用protobuf编译器能自动生成代码，Protobuf是将类的定义使用.proto文件进行描述。说明在idea中编写.proto文件时，会自动提示是否下载.protobuf编写插件可以让语法高亮；

7、然后通过protoc.exe编译器根据.proto自动生成.java文件；

Protobuf快速入门实例1

1、客户端可以发送一个Student POJO对象到服务器（通过Protobuf编码）；

2、服务端能接收Studen POJO对象，并显示信息（通过Protobuf解码）

3、代码演示

① 定义传输的消息格式，我们这里采用了谷歌的Protobuf进行消息传输和解析，这里我们定义一个传输对象为Student.proto，里面的对象属性类型参考protobuf官方文档

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;
}

② 利用官网提供的编译工具编译消息传输对象，编译工具在官网可以下载https://github.com/protocolbuffers/protobuf/releases

③ 将编写好的Student.proto 拷贝到protoc.exe 所在的目录下D:\xbcai\jar\protoc-3.13.0-win64\bin

④ 打开cmd,在该目录下输入该命令对proto文件进行编译 protoc.exe --java_out=. Student.proto

编译后生成的文件为StudentPOJO.java，然后将该Java文件拷贝回项目中

⑤ 在项目里面引入protobuf-java jar包

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

⑥ 编写服务端，设置服务端解码器为ProtobufDecoder，并设置指定对哪种对象进行解码

public class NettyServer {
    @SuppressWarnings("all")
    public static void main(String[] args) throws Exception {
        /**
         * 说明
         * 1、创建两个线程组bossGroup 和 workerGroup
         * 2、bossGroup只是处理连接请求，真正和客户端业务处理的会交给workerGroup完成
         * 3、两个都是无限循环
         * 4、bossGroup和workerGroup含有的子线程（NioEventLoop）的个数默认实际 cpu核数*2
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //创建服务器端的启动对象，配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            //使用链式编程来进行设置
            //设置两个线程组
           bootstrap.group(bossGroup, workerGroup)
                    //使用NioServerSocketChannel作为服务器的通道实现
                    .channel(NioServerSocketChannel.class)
                    //设置线程队列得到的连接个数
                    .option(ChannelOption.SO_BACKLOG, 128)
                    //设置保持获得的连接状态
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    //创建一个通道测试对象（匿名对象）
                    //给我们的workerGroup的EventLoop对应的管道设置处理器
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        //给pipeline设置处理器
                        @Override
                        protected void initChannel(SocketChannel sc) throws Exception {
                            ChannelPipeline pipeline = sc.pipeline();
                            //在pipeline加入ProtoBufDecoder解码器，指定对哪种对象进行解码
                            pipeline.addLast("decoder",new ProtobufDecoder(StudentPOJO.Student.getDefaultInstance()));
                            //可以使用一个集合管理SocketChannel,在推送消息时，可以将业务加入到各个channel对应的NIOEventLoop的taskQueue或者scheduleTaskQueue
                            pipeline.addLast(new NettyServerHandler());
                        }
                    });
            System.out.println("......服务器 is ready...");
            //绑定一个端口并且同步，生成了一个ChannelFuture对象
            //启动服务器（并绑定端口）
            ChannelFuture cf = bootstrap.bind(6668).sync();
            //给cf注册监听器，监控我们关心的事件
            cf.addListener((ChannelFutureListener) channelFuture -> {
                if(cf.isSuccess()){
                    System.out.println("监听端口6668成功");
                }else{
                    System.out.println("监听端口6668失败");
                }
            });
            //对关闭通道进行监听
            cf.channel().closeFuture().sync();
        } finally {
            //优雅关闭
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }

}

⑦ 编写服务端的业务处理器，读取传输的proto格式的消息对象

/**
 * 1、我们自定义一个Handler，需要继承netty规定好的某个handlerAdapter(规范)
 * 2、这时我们自定义一个Handler，才能称为一个Handler
 */
public class NettyServerHandler extends ChannelInboundHandlerAdapter {
    /**
     * 读取数据（这里我们可以读取客户端发送的消息）
     * @param ctx 上下文对象，含有管道pipeline，通道channel,地址
     * @param msg 就是客户端发送的数据，默认是Object
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
      //读取从客户端发送的StudentPOJO.Student
        StudentPOJO.Student student = (StudentPOJO.Student)msg;
        System.out.println("客户端发送的数据，ID="+student.getId()+",名字="+student.getName());
    }

    /**
     * 数据读取完毕
     * @param ctx
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        /**
         * writeAndFlush 是write+flush
         * 将数据写入到缓冲区，并刷新，一般讲，我们对这个发送的数据进行编码
         */
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello,客户端",CharsetUtil.UTF_8));
    }

    /**
     * 处理异常，一般需要关闭通道
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

⑧ 编写客户端，在客户端管道pipeline中加入ProtoBufEncoder编码器

public class NettyClient {
    @SuppressWarnings("all")
    public static void main(String[] args) {
        //客户端需要一个事件循环组
        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            //创建客户端启动对象，注意客户端使用不是ServerBootstarp而是Bootstrap
            Bootstrap bootstrap = new Bootstrap();
            //设置相关参数
            //设置线程组
            bootstrap.group(group)
                    //设置客户端通道的实现类（反射）
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel sc) throws Exception {
                            //加入自己的处理器
                            ChannelPipeline pipeline = sc.pipeline();
                            //在pipeline中加入ProtoBufEncoder编码器
                            pipeline.addLast("encoder",new ProtobufEncoder());
                            pipeline.addLast(new NettyClientHandler());
                        }
                    });
            System.out.println("客户端 ok...");
            //启动客户端去连接服务器端
            //关于ChannelFuture要分析，涉及到netty的异步模型
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();
            //给关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }
}

⑨ 编写客户端业务逻辑处理器，这里我们让客户端一连上服务端就发送采用proto生成的对象的消息

public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    /**
     * 当通道就绪就会触发该方法
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client "+ctx);
        //发送一个Student对象到服务器
        StudentPOJO.Student student = StudentPOJO.Student.newBuilder().setId(4).setName("开封府 包拯").build();
        ctx.writeAndFlush(student);
    }

    /**
     *当通道有读取事件时会触发
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("服务器回复的消息："+buf.toString(CharsetUtil.UTF_8));
        System.out.println("服务器的地址："+ctx.channel().remoteAddress());
    }

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

⑩ 启动服务端，启动客户端看效果

Protobuf快速入门实例2

1、客户端可以随机发送Student POJO/Worker POJO对象到服务器（通过Protobuf编码）；

2、服务端能接收Studen POJO/Worker POJO对象(需要判断是哪种类型)，并显示信息（通过Protobuf解码）

3、代码演示

① 定义传输的消息体对象Student.proto

syntax = "proto3";
option optimize_for = SPEED;//加快解析
option java_package="com.xbcai.netty_project.netty.codec2";//指定生成到哪个包下面
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;
}

② 服务端代码

public class NettyServer {
    @SuppressWarnings("all")
    public static void main(String[] args) throws Exception {
        /**
         * 说明
         * 1、创建两个线程组bossGroup 和 workerGroup
         * 2、bossGroup只是处理连接请求，真正和客户端业务处理的会交给workerGroup完成
         * 3、两个都是无限循环
         * 4、bossGroup和workerGroup含有的子线程（NioEventLoop）的个数默认实际 cpu核数*2
         */
        EventLoopGroup bossGroup = new NioEventLoopGroup();
        EventLoopGroup workerGroup = new NioEventLoopGroup();
        try {
            //创建服务器端的启动对象，配置参数
            ServerBootstrap bootstrap = new ServerBootstrap();
            //使用链式编程来进行设置
            //设置两个线程组
           bootstrap.group(bossGroup, workerGroup)
                    //使用NioServerSocketChannel作为服务器的通道实现
                    .channel(NioServerSocketChannel.class)
                    //设置线程队列得到的连接个数
                    .option(ChannelOption.SO_BACKLOG, 128)
                    //设置保持获得的连接状态
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    //创建一个通道测试对象（匿名对象）
                    //给我们的workerGroup的EventLoop对应的管道设置处理器
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        //给pipeline设置处理器
                        @Override
                        protected void initChannel(SocketChannel sc) throws Exception {
                            ChannelPipeline pipeline = sc.pipeline();
                            //在pipeline加入ProtoBufDecoder解码器，指定对哪种对象进行解码
                            pipeline.addLast("decoder",new ProtobufDecoder(MyDataInfo.MyMessage.getDefaultInstance()));
                            //可以使用一个集合管理SocketChannel,在推送消息时，可以将业务加入到各个channel对应的NIOEventLoop的taskQueue或者scheduleTaskQueue
                            pipeline.addLast(new NettyServerHandler());
                        }
                    });
            System.out.println("......服务器 is ready...");
            //绑定一个端口并且同步，生成了一个ChannelFuture对象
            //启动服务器（并绑定端口）
            ChannelFuture cf = bootstrap.bind(6668).sync();
            //给cf注册监听器，监控我们关心的事件
            cf.addListener((ChannelFutureListener) channelFuture -> {
                if(cf.isSuccess()){
                    System.out.println("监听端口6668成功");
                }else{
                    System.out.println("监听端口6668失败");
                }
            });
            //对关闭通道进行监听
            cf.channel().closeFuture().sync();
        } finally {
            //优雅关闭
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }

    }

}

③ 服务端业务逻辑代码

/**
 * 1、我们自定义一个Handler，需要继承netty规定好的某个handlerAdapter(规范)
 * 2、这时我们自定义一个Handler，才能称为一个Handler
 */
public class NettyServerHandler extends SimpleChannelInboundHandler<MyDataInfo.MyMessage> {
    /**
     * 读取数据（这里我们可以读取客户端发送的消息）
     * @param ctx 上下文对象，含有管道pipeline，通道channel,地址
     * @param msg 就是客户端发送的数据，默认是Object
     */
    @Override
    public void channelRead0(ChannelHandlerContext ctx, MyDataInfo.MyMessage msg) throws Exception {
        //根据dataType来显示不同的信息
        MyDataInfo.MyMessage.DataType dataType = msg.getDataType();
        if(dataType==MyDataInfo.MyMessage.DataType.StudentType){
            MyDataInfo.Student student = msg.getStudent();
            System.out.println("学生ID="+student.getId()+",学生姓名="+student.getName());
        }else if(dataType==MyDataInfo.MyMessage.DataType.WorkerType){
            MyDataInfo.Worker worker = msg.getWorker();
            System.out.println("工人姓名="+worker.getName()+",年龄="+worker.getAge());
        }else{
            System.out.println("传输的类型不正确");
        }
    }

    /**
     * 数据读取完毕
     * @param ctx
     */
    @Override
    public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
        /**
         * writeAndFlush 是write+flush
         * 将数据写入到缓冲区，并刷新，一般讲，我们对这个发送的数据进行编码
         */
        ctx.writeAndFlush(Unpooled.copiedBuffer("hello,客户端",CharsetUtil.UTF_8));
    }

    /**
     * 处理异常，一般需要关闭通道
     */
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.close();
    }
}

④ 客户端代码

public class NettyClient {
    @SuppressWarnings("all")
    public static void main(String[] args) {
        //客户端需要一个事件循环组
        NioEventLoopGroup group = new NioEventLoopGroup();
        try {
            //创建客户端启动对象，注意客户端使用不是ServerBootstarp而是Bootstrap
            Bootstrap bootstrap = new Bootstrap();
            //设置相关参数
            //设置线程组
            bootstrap.group(group)
                    //设置客户端通道的实现类（反射）
                    .channel(NioSocketChannel.class)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel sc) throws Exception {
                            //加入自己的处理器
                            ChannelPipeline pipeline = sc.pipeline();
                            //在pipeline中加入ProtoBufEncoder编码器
                            pipeline.addLast("encoder",new ProtobufEncoder());
                            pipeline.addLast(new NettyClientHandler());
                        }
                    });
            System.out.println("客户端 ok...");
            //启动客户端去连接服务器端
            //关于ChannelFuture要分析，涉及到netty的异步模型
            ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6668).sync();
            //给关闭通道进行监听
            channelFuture.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            group.shutdownGracefully();
        }
    }
}

⑤ 客户端逻辑代码，发送消息

public class NettyClientHandler extends ChannelInboundHandlerAdapter {
    /**
     * 当通道就绪就会触发该方法
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println("client "+ctx);
        //发送一个Student对象到服务器
        //随机的发送Student 或者 Worker对象
        int random = new Random().nextInt(3);
        MyDataInfo.MyMessage myMessage = null;
        //发送Student对象
        if(0==random){
          myMessage = MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.StudentType).
                    setStudent(MyDataInfo.Student.newBuilder().setId(5).setName("小红，小花").build()).build();
        }else{
            //发送一个Worker对象
            myMessage = MyDataInfo.MyMessage.newBuilder().setDataType(MyDataInfo.MyMessage.DataType.WorkerType).
                    setWorker(MyDataInfo.Worker.newBuilder().setAge(20).setName("老李，老王").build()).build();
        }
        ctx.writeAndFlush(myMessage);
    }

    /**
     *当通道有读取事件时会触发
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        ByteBuf buf = (ByteBuf) msg;
        System.out.println("服务器回复的消息："+buf.toString(CharsetUtil.UTF_8));
        System.out.println("服务器的地址："+ctx.channel().remoteAddress());
    }

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

⑥ 启动服务端，多次启动客户端看看服务端接收到的消息

protobufpython

上一篇：Google Protobuf vs. Apache Avro
下一篇：初识Protobuf及其编译工具的安装

您懂Netty高性能消息传输GoogleProtobuf吗

相关推荐

idea本地配置连接远程hadoop集群的一些网络问题解决汇总

无缓存不行?例行升级的入门级阿斯加特AN2 SSD装机点评

Ceph运维手册(基于P版本)

大数据开发前要做什么准备?8台Hadoop服务器进行集群规划前配置

Tensorflow分类loss函数总结 tensorflow绘制loss曲线

R语言学习笔记(七) -离散型数据的模型预测2

iOS Runtime详解

7 个对 Java 意义重大的性能指标，你知道几个?

PHP 远程调试最佳实践

Laravel框架使用图片处理简单教程