Netty中序列化框架MessagePack的简单实现

ゝ一纸荒年。 2022-02-20 06:15 291阅读 0赞

  MessagePack是一个高效的二进制序列化框架，它像JSON一样支持不同语言间的数据交换，但是它的性能更快，序列化之后的码流也更小。MessagePack的特点如下:

  1. 编解码高效，性能高；
  2. 序列化之后码流小
  3. 支持跨语言

MessagePack使用

1.依赖

  使用maven构建项目

  1. <dependency>
  2.     <groupId>org.msgpack</groupId>
  3.     <artifactId>msgpack</artifactId>
  4.     <version>0.6.12</version>
  5. </dependency>

2.创建编码和解码器

编码器

  1. /** * @param ctx 上下文 * @param msg 需要编码的对象 * @param out 编码后的数据 */
  2. @Override
  3. protected void encode(ChannelHandlerContext ctx, Object msg, ByteBuf out) throws Exception { 
  5.     MessagePack msgpack = new MessagePack();
  6.     // 对对象进行序列化
  7.     byte[] raw = msgpack.write(msg);
  8.     // 返回序列化的数据
  9.     out.writeBytes(raw);
  10. }

解码器

  1. /** * @param ctx 上下文 * @param msg 需要解码的数据 * @param out 解码列表 */
  2. @Override
  3. protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out) throws Exception { 
  4.     final byte[] array;
  5.     final int length = msg.readableBytes();
  6.     array = new byte[length];
  7.     // 获取需要解码的字节数组
  8.     msg.getBytes(msg.readerIndex(), array,0,length);
  9.     MessagePack msgpack = new MessagePack();
  10.     // 反序列化并将结果保存到了解码列表中
  11.     out.add(msgpack.read(array));
  12. }

3.客户端

EchoClient

  1. /** * MsgPack 编解码器 * * @author 波波烤鸭 * @email dengpbs@163.com * */
  2. public class EchoClient { 
  4.     public static void main(String[] args) throws Exception { 
  5.         int port = 8080;
  6.         if (args != null && args.length > 0) { 
  7.             try { 
  8.                 port = Integer.valueOf(args[0]);
  9.             } catch (NumberFormatException e) { 
  10.                 // 采用默认值
  11.             }
  12.         }
  13.         new EchoClient().connector(port, "127.0.0.1",10);
  14.     }
  16.     public void connector(int port, String host,final int sendNumber) throws Exception { 
  17.         // 配置客户端NIO线程组
  18.         EventLoopGroup group = new NioEventLoopGroup();
  19.         try { 
  20.             Bootstrap b = new Bootstrap();
  21.             b.group(group).channel(NioSocketChannel.class)
  22.                             .option(ChannelOption.TCP_NODELAY, true)
  23.                             .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
  24.                             .handler(new ChannelInitializer<SocketChannel>() { 
  26.                 @Override
  27.                 protected void initChannel(SocketChannel ch) throws Exception { 
  28.                     //这里设置通过增加包头表示报文长度来避免粘包
  29.                     ch.pipeline().addLast("frameDecoder",new LengthFieldBasedFrameDecoder(1024, 0, 2,0,2));
  30.                     //增加解码器
  31.                     ch.pipeline().addLast("msgpack decoder",new MsgpackDecoder());
  32.                     //这里设置读取报文的包头长度来避免粘包
  33.                     ch.pipeline().addLast("frameEncoder",new LengthFieldPrepender(2));
  34.                     //增加编码器
  35.                     ch.pipeline().addLast("msgpack encoder",new MsgpackEncoder());
  36.                     // 4.添加自定义的处理器
  37.                     ch.pipeline().addLast(new EchoClientHandler(sendNumber));
  38.                 }
  39.             });
  41.             // 发起异步连接操作
  42.             ChannelFuture f = b.connect(host, port).sync();
  43.             // 等待客户端链路关闭
  44.             f.channel().closeFuture().sync();
  45.         }catch(Exception e){ 
  46.             e.printStackTrace();
  47.         }  finally { 
  48.             // 优雅退出，释放NIO线程组
  49.             group.shutdownGracefully();
  50.         }
  51.     }
  53. }

EchoClientHandler

  1. /** * DelimiterBasedFrameDecoder 案例 * 自定义处理器 * @author 波波烤鸭 * @email dengpbs@163.com * */
  2. public class EchoServerHandler extends ChannelHandlerAdapter{ 
  5.     @Override
  6.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
  7.         //UserInfo user = (UserInfo) msg;
  8.         System.out.println("server receive the msgpack message :"+msg);
  9.         //ctx.writeAndFlush(user);
  10.         ctx.writeAndFlush(msg);
  12.     }
  14.     @Override
  15.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
  16.         cause.printStackTrace();
  17.         ctx.close(); // 发生异常关闭链路
  18.     }
  19. }

4.服务端

EchoServer

  1. /** * MsgPack 编解码器 * 服务端 * @author 波波烤鸭 * @email dengpbs@163.com * */
  2. public class EchoServer { 
  4.     public void bind(int port) throws Exception { 
  5.         // 配置服务端的NIO线程组
  6.         // 服务端接受客户端的连接
  7.         NioEventLoopGroup bossGroup = new NioEventLoopGroup();
  8.         // 进行SocketChannel的网络读写
  9.         NioEventLoopGroup workerGroup = new NioEventLoopGroup();
  10.         try { 
  11.             ServerBootstrap b = new ServerBootstrap();
  12.             b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
  13.                     .option(ChannelOption.SO_BACKLOG, 100)
  14.                     .handler(new LoggingHandler(LogLevel.INFO))
  15.                     .childHandler(new ChannelInitializer<SocketChannel>() { 
  17.                         @Override
  18.                         protected void initChannel(SocketChannel ch) throws Exception { 
  19.                             ch.pipeline().addLast("frameDecoder",new LengthFieldBasedFrameDecoder(65535, 0, 2,0,2));
  20.                             // 添加msgpack的编码和解码器
  21.                             ch.pipeline().addLast("msgpack decoder",new MsgpackDecoder());
  22.                             ch.pipeline().addLast("frameEncoder",new LengthFieldPrepender(2));
  23.                             ch.pipeline().addLast("msgpack encoder",new MsgpackEncoder());
  24.                             // 添加自定义的处理器
  25.                             ch.pipeline().addLast(new EchoServerHandler());
  27.                         }
  28.                     });
  30.             // 绑定端口，同步等待成功
  31.             ChannelFuture f = b.bind(port).sync();
  32.             // 等待服务端监听端口关闭
  33.             f.channel().closeFuture().sync();
  34.         }catch(Exception e){ 
  35.             e.printStackTrace();
  36.         } finally { 
  37.             // 优雅退出，释放线程池资源
  38.             bossGroup.shutdownGracefully();
  39.             workerGroup.shutdownGracefully();
  40.         }
  41.     }
  43.     public static void main(String[] args) throws Exception { 
  44.         int port = 8080;
  45.         if(args!=null && args.length > 0){ 
  46.             try{ 
  47.                 port = Integer.valueOf(args[0]);
  48.             }catch(NumberFormatException e){ 
  49.                 // 采用默认值
  50.             }
  51.         }
  52.         new EchoServer().bind(port);
  53.     }
  55. }

EchoServerHandler

  1. /** * DelimiterBasedFrameDecoder 案例 * 自定义处理器 * @author 波波烤鸭 * @email dengpbs@163.com * */
  2. public class EchoServerHandler extends ChannelHandlerAdapter{ 
  5.     @Override
  6.     public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception { 
  7.         //UserInfo user = (UserInfo) msg;
  8.         System.out.println("server receive the msgpack message :"+msg);
  9.         //ctx.writeAndFlush(user);
  10.         ctx.writeAndFlush(msg);
  12.     }
  14.     @Override
  15.     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception { 
  16.         cause.printStackTrace();
  17.         ctx.close(); // 发生异常关闭链路
  18.     }
  19. }

5.注意点（POJO）

  消息类上加上注解Message，还有就是必须要有默认的无参构造器

  1. /** * Msgpack 中必须添加@Message注解 及 无参构造方法 * @author 波波烤鸭 * @email dengpbs@163.com * */
  2. @Message
  3. public class UserInfo { 
  6.     private String name;
  8.     private int age;
  10.     public String getName() { 
  11.         return name;
  12.     }
  14.     public void setName(String name) { 
  15.         this.name = name;
  16.     }
  18.     public int getAge() { 
  19.         return age;
  20.     }
  22.     public void setAge(int age) { 
  23.         this.age = age;
  24.     }
  26.     @Override
  27.     public String toString() { 
  28.         return "UserInfo [name=" + name + ", age=" + age + "]";
  29.     }
  30. }

6.测试

服务端输出

  1. server receive the msgpack message :["bobo烤鸭:0",0]
  2. server receive the msgpack message :["bobo烤鸭:1",1]
  3. server receive the msgpack message :["bobo烤鸭:2",2]
  4. server receive the msgpack message :["bobo烤鸭:3",3]
  5. server receive the msgpack message :["bobo烤鸭:4",4]
  6. server receive the msgpack message :["bobo烤鸭:5",5]
  7. server receive the msgpack message :["bobo烤鸭:6",6]
  8. server receive the msgpack message :["bobo烤鸭:7",7]
  9. server receive the msgpack message :["bobo烤鸭:8",8]
  10. server receive the msgpack message :["bobo烤鸭:9",9]

客户端输出

  1. Client receive the msgpack message :["bobo烤鸭:0",0]
  2. Client receive the msgpack message :["bobo烤鸭:1",1]
  3. Client receive the msgpack message :["bobo烤鸭:2",2]
  4. Client receive the msgpack message :["bobo烤鸭:3",3]
  5. Client receive the msgpack message :["bobo烤鸭:4",4]
  6. Client receive the msgpack message :["bobo烤鸭:5",5]
  7. Client receive the msgpack message :["bobo烤鸭:6",6]
  8. Client receive the msgpack message :["bobo烤鸭:7",7]
  9. Client receive the msgpack message :["bobo烤鸭:8",8]
  10. Client receive the msgpack message :["bobo烤鸭:9",9]

至此Netty中就可以通过MessagePack来处理序列化的情况了~

发表评论

表情：
评论列表 （有 0 条评论，291人围观）

还没有评论，来说两句吧...

相关阅读

    相关 基于Netty实现简单的RPC框架

            Dubbo采用Netty作为基础通信组件，模仿Dubbo实现简单版的RPC框架。服务消费者和服务提供者约定接口和协议，用于远程调用和TCP请求验证。服务提供者作

    女爷i女爷i/ 0 赞/ 267 阅读