Netty学习之八—— MessagePack 编解码

今天药忘吃喽~ 2022-02-21 18:49 331阅读 0赞

# 1、MessagePack 介绍 #

1、编解码高效、性能高；

2、序列化之后的码流小；

3、支持跨语言。

<dependency>
        <groupId>org.msgpack</groupId>
        <artifactId>msgpack</artifactId>
        <version>0.6.12</version>
    </dependency>

## MessagePack 编解码代码 ##

package com.netty.codec.msgpack;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToByteEncoder;
    import org.msgpack.MessagePack;
    
    public class MsgPackEncoder extends MessageToByteEncoder<Object> {
        @Override
        protected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf byteBuf) throws Exception {
            // Object编译成 byte 数组，然后写入到 ByteBuf中
            MessagePack pack = new MessagePack();
            byte[] raw = pack.write(o);
            byteBuf.writeBytes(raw);
        }
    }

package com.netty.codec.msgpack;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToMessageDecoder;
    import org.msgpack.MessagePack;
    
    import java.util.List;
    
    public class MsgPackDecoder extends MessageToMessageDecoder<ByteBuf> {
    
        @Override
        protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
            final byte[] array;
            final int length = byteBuf.readableBytes();
            array = new byte[length];
            byteBuf.getBytes(byteBuf.readerIndex(), array, 0, length);
            MessagePack pack = new MessagePack();
            list.add(pack.read(array));
        }
    }

pojo

package com.netty.codec.pojo;
    
    import org.msgpack.annotation.Message;
    
    @Message
    public class UserInfo{
        private String name;
        private int age;
    
        public String getName() {
            return name;
        }
    
        public void setName(String name) {
            this.name = name;
        }
    
        public int getAge() {
            return age;
        }
    
        public void setAge(int age) {
            this.age = age;
        }
    
        @Override
        public String toString() {
            return "User [name=" + name + ", age=" + age + "]";
        }
    }

## 服务端代码： ##

package com.netty.codec.msgpack;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldPrepender;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    
    public class MsgPackEchoServer {
        public void bind(int port) throws Exception {
            // 配置服务端的 NIO 线程组
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
    
                            @Override
                            protected void initChannel(SocketChannel channel) throws Exception {
                                // 解决粘包问题
                                channel.pipeline().addLast("frameDecoder",new LengthFieldBasedFrameDecoder(65535, 0, 2,0,2));
                                channel.pipeline().addLast("msgpack decoder", new MsgPackDecoder());
                                // 解决粘包问题
                                channel.pipeline().addLast("frameEncoder",new LengthFieldPrepender(2));
                                channel.pipeline().addLast("msgpack encoder", new MsgPackEncoder());
                                channel.pipeline().addLast(new MsgPackEchoServerHandler());
                            }
                        });
                // 绑定端口，同步等待成功
                ChannelFuture f = b.bind(port).sync();
    
                // 等待服务端监听端口关闭
                f.channel().closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
            int port = 8080;
            new MsgPackEchoServer().bind(port);
        }
    }

package com.netty.codec.msgpack;
    
    
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class MsgPackEchoServerHandler extends ChannelHandlerAdapter {
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("Server receive the msgpack message : " + msg);
            ctx.write(msg);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    }

## 客户端代码： ##

package com.netty.codec.msgpack;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioSocketChannel;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldPrepender;
    import io.netty.handler.codec.LineBasedFrameDecoder;
    import io.netty.handler.codec.string.StringDecoder;
    
    public class MsgPackEchoClient {
        public void connect(int port, String host, int sendNumber) throws Exception {
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                Bootstrap b = new Bootstrap();
                b.group(group).channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY, true)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel channel) throws Exception {
                                // 解决粘包问题
                                channel.pipeline().addLast("frameDecoder",new LengthFieldBasedFrameDecoder(65535, 0, 2,0,2));
                                channel.pipeline().addLast("msgpack decoder", new MsgPackDecoder());
                                // 解决粘包问题
                                channel.pipeline().addLast("frameEncoder",new LengthFieldPrepender(2));
                                channel.pipeline().addLast("msgpack encoder", new MsgPackEncoder());
                                channel.pipeline().addLast(new MsgPackEchoClientHandler(sendNumber));
                            }
                        });
                // 发起异步连接操作
                ChannelFuture f = b.connect(host, port).sync();
    
                // 等待客户端链路关闭
                f.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
            int port = 8080;
            new MsgPackEchoClient().connect(port, "127.0.0.1", 1000);
    
        }
    }

package com.netty.codec.msgpack;
    
    import com.netty.codec.pojo.UserInfo;
    import io.netty.channel.ChannelHandlerAdapter;
    import io.netty.channel.ChannelHandlerContext;
    
    public class MsgPackEchoClientHandler extends ChannelHandlerAdapter {
    
        private int sendNumber;
    
        public MsgPackEchoClientHandler(int sendNumber) {
            this.sendNumber = sendNumber;
        }
    
        /**
         * 构建长度为userNum的User对象数组
         * @param userNum
         * @return
         */
        private UserInfo[] getUserArray(int userNum) {
            UserInfo[] users = new UserInfo[userNum];
            UserInfo user = null;
            for(int i = 0; i < userNum; i++) {
                user = new UserInfo();
                user.setName("ABCDEFG --->" + i);
                user.setAge(i);
                users[i] = user;
            }
            return users;
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
            UserInfo[] users = getUserArray(sendNumber);
            for (UserInfo user : users) {
                ctx.writeAndFlush(user);
            }
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("Client receive the msgpack message : " + msg);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

## LengthFieldBasedFrameDecoder  和  LengthFieldPrepender 解决了粘包问题。 ##

LengthFieldPrepender 在MessagePack编码器编码前增加了两个字节的消息长度字段，

![20190411161900530.png][]

LengthFieldBasedFrameDecoder 在MessagePack 解码器解码前可以读取到消息长度字段，这样 解码器永远接到的都是整包。

![20190411162041348.png][]

## 总结：在消息头中增加报文长度字段，利用该字段进行半包的编解码，解决粘包/半包问题。 ##

[20190411161900530.png]: /images/20220221/ce9a9e3d082349ee8fda10ca2c7fe73a.png
[20190411162041348.png]: /images/20220221/103b4761a95e491fa118a9c2ea25ae3f.png