Netty学习

文章10 |   阅读 5319 |   点赞0

来源：https://blog.csdn.net/a953713428/category_6806288.html

Netty学习(七)-Netty编解码技术以及ProtoBuf和Thrift的介绍

转x33g5p2x  于2021-12-21 转载在 其他  
字(8.2k)|赞(0)|评价(0)|浏览(657)

在前几节我们学习过处理粘包和拆包的问题，用到了Netty提供的几个解码器对不同情况的问题进行处理。功能很是强大。我们有没有去想这么强大的功能是如何实现的呢？背后又用到了什么技术？这一节我们就来处理这个问题。了解一下编码解码到底是如何处理的。

通常说的编码(Encoder)也就是发生在发送消息的时候需要将消息编译成字节对象，在Netty中即编译成ByteBuf对象。在java中我们将这种编译称之为序列化(Serializable),即将对象序列化为字节数组，然后用于传输或是持久化啊之类的。那么自然解码（Decoder）就是一个反序列化的过程，使用相应的编码格式对接收到的对做一个解码，以正确解析该对象。

1. java序列化的弱点

谈到序列化我们自然想到java提供的Serializable接口，在java中我们如果需要序列化只需要继承该接口就可以通过输入输出流进行序列化和反序列化。但是在提供很用户简单的调用的同时他也存在很多问题：

  • 无法跨语言。当我们进行跨应用之间的服务调用的时候如果另外一个应用使用c语言来开发，这个时候我们发送过去的序列化对象，别人是无法进行反序列化的因为其内部实现对于别人来说完全就是黑盒。
  • 序列化之后的码流太大。这个我们可以做一个实验还是上一节中的Message类，我们分别用java的序列化和使用二进制编码来做一个对比，下面我写了一个测试类：
  1. @Test
  2. public void testSerializable(){
  3.     String str = "哈哈,我是一条消息";
  4.     Message msg = new Message((byte)0xAD,35,str);
  5.     ByteArrayOutputStream out = new ByteArrayOutputStream();
  6.     try {
  7.         ObjectOutputStream os = new ObjectOutputStream(out);
  8.         os.writeObject(msg);
  9.         os.flush();
  10.         byte[] b = out.toByteArray();
  11.         System.out.println("jdk序列化后的长度： "+b.length);
  12.         os.close();
  13.         out.close();
  15.         ByteBuffer buffer = ByteBuffer.allocate(1024);
  16.         byte[] bt = msg.getMsgBody().getBytes();
  17.         buffer.put(msg.getType());
  18.         buffer.putInt(msg.getLength());
  19.         buffer.put(bt);
  20.         buffer.flip();
  22.         byte[] result = new byte[buffer.remaining()];
  23.         buffer.get(result);
  24.         System.out.println("使用二进制序列化的长度："+result.length);
  26.     } catch (IOException e) {
  27.         e.printStackTrace();
  28.     }
  29. }

输出结果为：

我们可以看到差距是挺大的，目前的主流编解码框架序列化之后的码流也都比java序列化要小太多。

  • 序列化效率差，这个我们也可以做一个对比，还是上面写的测试代码我们循环跑100000次对比一下时间：
  1. @Test
  2. public void testSerializable(){
  3.     String str = "哈哈,我是一条消息";
  4.     Message msg = new Message((byte)0xAD,35,str);
  5.     ByteArrayOutputStream out = new ByteArrayOutputStream();
  6.     try {
  7.         long startTime = System.currentTimeMillis();
  8.         for(int i = 0;i < 100000;i++){
  9.             ObjectOutputStream os = new ObjectOutputStream(out);
  10.             os.writeObject(msg);
  11.             os.flush();
  12.             byte[] b = out.toByteArray();
  13.             /*System.out.println("jdk序列化后的长度： "+b.length);*/
  14.             os.close();
  15.             out.close();
  16.         }
  17.         long endTime = System.currentTimeMillis();
  18.         System.out.println("jdk序列化100000次耗时：" +(endTime - startTime));
  20.         long startTime1 = System.currentTimeMillis();
  21.         for(int i = 0;i < 100000;i++){
  22.             ByteBuffer buffer = ByteBuffer.allocate(1024);
  23.             byte[] bt = msg.getMsgBody().getBytes();
  24.             buffer.put(msg.getType());
  25.             buffer.putInt(msg.getLength());
  26.             buffer.put(bt);
  27.             buffer.flip();
  29.             byte[] result = new byte[buffer.remaining()];
  30.             buffer.get(result);
  31.             /*System.out.println("使用二进制序列化的长度："+result.length);*/
  32.         }
  33.         long endTime1 = System.currentTimeMillis();
  34.         System.out.println("使用二进制序列化100000次耗时：" +(endTime1 - startTime1));
  36.     } catch (IOException e) {
  37.         e.printStackTrace();
  38.     }
  39. }

结果为：

结果为毫秒数，这个差距也是不小的。

结合以上我们看到：目前的序列化过程中使用java本身的肯定是不行，使用二进制编码的话又的我们自己去手写，所以为了让我们少搬砖前辈们早已经写好了工具让我们调用，目前社区比较活跃的有google的Protobuf和Apache的Thrift。

2. Protobuf序列化的使用

我们先来使用Protobuf进行序列化，他和XML，json一样都有自己的语法，xml的后缀是.xml，json文件的后缀是.json，自然Protobuf文件的后缀就是.proto（哈哈，当然不是全称）。

下面我们使用Protobuf来封装一段消息，通过一个案例简单介绍一下它的使用。

首先我们用Protobuf的语法格式来写一段需要序列化的对象，命名格式为：Msg.proto

  1. option java_package = "cn.edu.hust.netty.demo10";
  2. option java_outer_classname = "MessageProto";
  4. message RequestMsg{
  5.     required bytes msgType = 1;
  6.     required string receiveOne = 2;
  7.     required string msg = 3;
  8. }
  10. message ResponseMsg{
  11.     required bytes msgType = 1;
  12.     required string receiveOne = 2;
  13.     required string msg = 3;
  14. }

关于Message.proto中的语法格式，详情大家google一下相关的说明，网上很多介绍，再次简单就上面的语法说明一下：

  • option java_package：表示生成的.java文件的包名
  • option java_outer_classname：生成的java文件的文件名
  • message ： 为他的基本类型，如同java中的class一样

字段修饰符：

  • required：一个格式良好的消息一定要含有1个这种字段。表示该值是必须要设置的；
  • optional：消息格式中该字段可以有0个或1个值（不超过1个）。
  • repeated：在一个格式良好的消息中，这种字段可以重复任意多次（包括0次）。重复的值的顺序会被保留。表示该值可以重复，相当于java中的List。

字符类型稍微有些不同：double,float,int32,int64,bool(boolean)
,string,bytes。稍微有些不同，String，boolean，int有差别。

另外我们看到上面3个字段分别赋值了，这个值是什么意思呢？消息定义中，每个字段都有唯一的一个数字标识符。这些标识符是用来在消息的二进制格式中识别各个字段的，一旦开始使用就不能够再改变。注：[1,15]之内的标识号在编码的时候会占用一个字节。[16,2047]之内的标识号则占用2个字节。所以应该为那些频繁出现的消息元素保留 [1,15]之内的标识号。

关于Protobuf 的语法我们就简单的介绍这么多，更多细节大家自己去查阅文档吧。下面我们开始使用Protobuf 来进行序列化。

首先我们的在工程中引入protobuf的jar包，目前官方版本最高3.2，我们用3.0的吧：

  1. <dependency>
  2.     <groupId>com.google.protobuf</groupId>
  3.     <artifactId>protobuf-java</artifactId>
  4.     <version>3.0.2</version>
  5. </dependency>

Protobuf的文件已经定义好了，下就需要把它编译成java代码，这里我们的借助到google为我们提供的脚本工具protoc，链接在这里，点击下载这里提供的是protoc-3.0.2。要注意protoc的版本需要和Protobuf的版本对应上，不然不同的版本之间会有一些差异解析可能会有问题。现在知道我们为啥非得选用protobuf3.0.2版本吧，因为我没有找到别的版本的protoc。。。

下载好了我们解压缩然后把刚才写好的Msg.proto文件复制进去。

接着我们进cmd输入如下命令：

主要是第三句命令。如果你输入没有报错的话你的proto文件夹应该会生成一个子文件夹：

进去该文件夹你会看到已经生成了MessageProto.java文件，恭喜你，这时候你已经完成了protobuf序列化文件的生成。然后你把该文件拷贝至工程目录下。接下来我们用生成的文件去发消息吧。还是老套路服务端和客户端。

服务端：

  1. public class ProtoBufServer {
  2.     private int port;
  4.     public ProtoBufServer(int port) {
  5.         this.port = port;
  6.     }
  8.     public void start(){
  9.         EventLoopGroup bossGroup = new NioEventLoopGroup();
  10.         EventLoopGroup workGroup = new NioEventLoopGroup();
  12.         ServerBootstrap server = new ServerBootstrap().group(bossGroup,workGroup)
  13.                                     .channel(NioServerSocketChannel.class)
  14.                                     .childHandler(new ServerChannelInitializer());
  16.         try {
  17.             ChannelFuture future = server.bind(port).sync();
  18.             future.channel().closeFuture().sync();
  19.         } catch (InterruptedException e) {
  20.             e.printStackTrace();
  21.         }finally {
  22.             bossGroup.shutdownGracefully();
  23.             workGroup.shutdownGracefully();
  24.         }
  25.     }
  27.     public static void main(String[] args) {
  28.         ProtoBufServer server = new ProtoBufServer(7788);
  29.         server.start();
  30.     }
  31. }

服务端Initializer：

  1. public class ServerChannelInitializer extends ChannelInitializer<SocketChannel> {
  2.     @Override
  3.     protected void initChannel(SocketChannel socketChannel) throws Exception {
  4.         ChannelPipeline pipeline = socketChannel.pipeline();
  6.         pipeline.addLast(new ProtobufVarint32FrameDecoder());
  7.         pipeline.addLast(new ProtobufDecoder(MessageProto.RequestMsg.getDefaultInstance()));
  8.         pipeline.addLast(new ProtoBufServerHandler());
  9.     }
  10. }

服务端handler：

  1. public class ProtoBufServerHandler extends ChannelInboundHandlerAdapter {
  2.     @Override
  3.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  4.         MessageProto.ResponseMsg.Builder builder = MessageProto.ResponseMsg.newBuilder();
  5.         builder.setMsgType(ByteString.copyFromUtf8("CBSP"));
  6.         builder.setReceiveOne("小红");
  7.         builder.setMsg("你好，你有啥事");
  9.         ctx.writeAndFlush(builder.build());
  10.     }
  12.     @Override
  13.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  14.         MessageProto.RequestMsg m = (MessageProto.RequestMsg)msg;
  15.         System.out.println("Client say: "+m.getReceiveOne()+","+m.getMsg());
  16.     }
  18.     @Override
  19.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
  20.         super.exceptionCaught(ctx, cause);
  21.         ctx.close();
  22.     }
  23. }

客户端：

  1. public class ProtoBufClient {
  2.     private  int port;
  3.     private  String address;
  5.     public ProtoBufClient(int port, String address) {
  6.         this.port = port;
  7.         this.address = address;
  8.     }
  10.     public void start(){
  11.         EventLoopGroup group = new NioEventLoopGroup();
  13.         Bootstrap bootstrap = new Bootstrap();
  14.         bootstrap.group(group)
  15.                 .channel(NioSocketChannel.class)
  16.                 .handler(new ClientChannelInitializer());
  18.         try {
  20.             ChannelFuture future = bootstrap.connect(address,port).sync();
  21.             future.channel().closeFuture().sync();
  22.         } catch (Exception e) {
  23.             e.printStackTrace();
  24.         }finally {
  25.             group.shutdownGracefully();
  26.         }
  28.     }
  30.     public static void main(String[] args) {
  31.         ProtoBufClient client = new ProtoBufClient(7788,"127.0.0.1");
  32.         client.start();
  33.     }
  34. }

客户端Initializer：

  1. public class ClientChannelInitializer extends ChannelInitializer<SocketChannel> {
  3.     protected void initChannel(SocketChannel socketChannel) throws Exception {
  4.         ChannelPipeline pipeline = socketChannel.pipeline();
  6.         pipeline.addLast(new ProtobufVarint32LengthFieldPrepender());
  7.         pipeline.addLast(new ProtobufEncoder());
  8.         pipeline.addLast(new ProtoBufClientHandler());
  9.     }
  10. }
  12. 客户端handler：
  14. public class ProtoBufClientHandler extends ChannelInboundHandlerAdapter {
  15.     @Override
  16.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
  17.         MessageProto.ResponseMsg m = (MessageProto.ResponseMsg)msg;
  18.         System.out.println("Server say: "+m.getReceiveOne()+","+m.getMsg());
  19.     }
  21.     @Override
  22.     public void channelActive(ChannelHandlerContext ctx) throws Exception {
  23.         MessageProto.RequestMsg.Builder builder = MessageProto.RequestMsg.newBuilder();
  24.         builder.setMsgType(ByteString.copyFromUtf8("CBSP"));
  25.         builder.setReceiveOne("小明");
  26.         builder.setMsg("你好，我找你有事");
  28.         ctx.writeAndFlush(builder.build());
  29.     }
  31.     @Override
  32.     public void channelInactive(ChannelHandlerContext ctx) throws Exception {
  33.         System.out.println("Client is close");
  34.     }
  35. }

启动服务端和客户端，输出如下：

最简单的protoBuf应用案例我们就写完了，真实的使用场景大同小异，随机应变即可。

3. thrift序列化的使用

哈哈，我本来是打算讲thrift的安装和使用的，但是现在却讲不了，因为这玩意儿的安装是个问题。由于我没有linux环境，thrift如果在linux环境下安装使用是挺简单的，但是在windows环境下挺麻烦。thrift在windows下，还使用C++，搭环境是最难的。 libthrift依赖boost libthriftnb依赖boost，libevent 等于你得安装boost，libevent 除此之外，还需要openssl 装openssl，又需要perl，nasm 期间，还会涉及版本兼容问题，总而言之，比较折磨，而这还仅是安装编译。

所以暂时我就跳过这一部分，等我安装linux环境之后再来讲解吧。

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/a953713428/article/details/69236148

内容来源于网络，如有侵权，请联系作者删除！

netty编码netty学习netty教程

相关文章

    查看更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com