Netty编解码（MessagePack实现）

我会带着你远行 2022-05-14 14:12 293阅读 0赞

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

MessagePack特点：

1.编解码高效，性能高；

2.序列化后的码流小；

3.支持跨语言。

实现：

首先需要的jar:

<dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
                <version>4.1.30.Final</version>
            </dependency>        
            <dependency>
                <groupId>org.msgpack</groupId>
                <artifactId>msgpack</artifactId>
                <version>0.6.12</version>
            </dependency>

解码器：

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 {
    //      获取要解码的byte数组
            final byte[] bytes;
            final int length = byteBuf.readableBytes();
            bytes = new byte[length];
            byteBuf.getBytes(byteBuf.readerIndex(),bytes,0,length);
    //      调用MessagePack 的read方法将其反序列化为Object对象
            MessagePack msgPack = new MessagePack();
            list.add(msgPack.read(bytes));
        }
    }

编码器：

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 {
            MessagePack msgPack = new MessagePack();
    //      编码，然后转为ButyBuf传递
            byte[] bytes = msgPack.write(o);
            byteBuf.writeBytes(bytes);
        }
    }

这里传输User类，User类代码：

import org.msgpack.annotation.Message;
    
    @Message
    public class User {
        private String name;
        private int age;
        private String id;
        private String sex;
    
        public int getAge() {
            return age;
        }
    
        public void setAge(int age) {
            this.age = age;
        }
    
        public String getId() {
            return id;
        }
    
        public void setId(String id) {
            this.id = id;
        }
    
        public String getSex() {
            return sex;
        }
    
        public void setSex(String sex) {
            this.sex = sex;
        }
    
        public String getName() {
            return name;
        }
    
        public void setName(String name) {
            this.name = name;
        }
    
        @Override
        public String toString() {
            return "User{" +
                    "name='" + name + '\'' +
                    ", age=" + age +
                    ", id='" + id + '\'' +
                    ", sex='" + sex + '\'' +
                    '}';
        }
    }

服务端代码：

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.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    
    public class EchoServer {
    
        public void bind(int port) throws Exception {
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try{
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 100)
                        .handler(new LoggingHandler(LogLevel.INFO))
                        .childHandler(new ChannelInitializer<SocketChannel>() {
    
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                // TODO Auto-generated method stub
                                ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535,0,
                                        2,0,2));
                                ch.pipeline().addLast(new MsgPackDecoder());
                                ch.pipeline().addLast(new LengthFieldPrepender(2));
                                ch.pipeline().addLast(new MsgPackEncoder());
                                ch.pipeline().addLast(new EchoServerHandler());
                            }
                        });
    
                //bind port
                ChannelFuture f = b.bind(port).sync();
                //wait
                f.channel().closeFuture().sync();
            }finally{
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
        /**
         * @param args
         */
        public static void main(String[] args) {
            int port = 8080;
            try {
                new EchoServer().bind(port);
            } catch (Exception e) {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
    
        }
    
    }

import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("服务器接收msgpack消息 : "+msg+"");
            // 原路返回给客户端
            ctx.writeAndFlush(msg);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

客户端代码：

import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    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.NioSocketChannel;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldPrepender;
    
    public class EchoClient {
        public void connection(int port,String host) throws InterruptedException {
            NioEventLoopGroup workGroup = new NioEventLoopGroup();
            try {
                Bootstrap b = new Bootstrap();
                b.group(workGroup)
                        .channel(NioSocketChannel.class)
                        .option(ChannelOption.TCP_NODELAY,true)
                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS,3000)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel socketChannel) throws Exception {
                                socketChannel.pipeline().addLast("frameDecoder",new LengthFieldBasedFrameDecoder(65535,0,
                                        2,0,2));
                                socketChannel.pipeline().addLast("msgpack译码器",new MsgPackDecoder());
                                socketChannel.pipeline().addLast("frameEncoder",new LengthFieldPrepender(2));
                                socketChannel.pipeline().addLast("msgpack编码器",new MsgPackEncoder());
                                socketChannel.pipeline().addLast(new EchoClientHandler());
    //
                            }
                        });
    //            发起异步连接操作
                ChannelFuture f = b.connect(host,port).sync();
    //                          等待客户端链路关闭
                f.channel().closeFuture().sync();
            } finally {
                workGroup.shutdownGracefully();
            }
        }
        public static void main(String[] args) throws InterruptedException {
            int port = 8080;
            new EchoClient().connection(port,"127.0.0.1");
        }
    }

import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    public class EchoClientHandler extends ChannelInboundHandlerAdapter {
        private int count;
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
           /* User user = getUser();
            ctx.writeAndFlush(user);*/
            User[] users = getUsers();
            for (User u : users) {
                System.out.println(u);
                ctx.writeAndFlush(u);
            }
    
        }
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("这是客户端接收的消息【  " + ++count + "  】时间:【" + msg + "】");
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    
        private User[] getUsers() {
            User[] users = new User[5];
            for (int i = 0; i < 5; i++) {
                User user = new User();
                user.setId(String.valueOf(i));
                user.setAge(18 + i);
                user.setName("迪迦" + i);
                user.setSex("男" + String.valueOf(i * 2));
                users[i] = user;
            }
            return users;
        }
    
        private User getUser() {
            User user = new User();
            user.setId("11");
            user.setAge(18);
            user.setName("张元");
            user.setSex("男");
            return user;
        }
    }

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_27_color_FFFFFF_t_70][]

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_27_color_FFFFFF_t_70 1][]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_27_color_FFFFFF_t_70]: /images/20220504/8de07b91b4af46c3b08c3212a3aeadb1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_27_color_FFFFFF_t_70 1]: /images/20220504/7e72cca6b7974c9bbc9074be4a4a338b.png