Netty 整合 MessagePack 序列化框架 + LengthFieldBasedFrameDecoder 自定义解码器

超、凢脫俗 2022-04-14 06:15 378阅读 0赞

**目录**

环境准备及说明

MessagePack 编解码器

MessagePack 编码器

MessagePack 解码器

POJO  User

Netty 网络通信

服务端

客户端

运行测试

--------------------

# 环境准备及说明 #

如果是导入二进制开发包，则如下所示：

![20181125150000635.png][]

需要开发包的可以参考《[ MessagePack 开发入门详解][MessagePack]》。

如果是 Maven 项目，则添加如下依赖：

<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>
    
    		
    		<dependency>
    			<groupId>org.javassist</groupId>
    			<artifactId>javassist</artifactId>
    			<version>3.24.0-GA</version>
    		</dependency>

> 1）netty-all：是 Netty 开发包
> 
> 2）msgpack：是 Messagepck 序列化开发包
> 
> 3）javassist：是 msgpack 自己的依赖包

本文示例项目结构如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70][]

> 1）User：网络传输的 POJO 对象，注意：序列化 POJO 必须加 org.msgpack.annotation.Message 注解：@Message
> 
> 2）echo：包中为 netty 通信的客户端与服务端
> 
> 3）messagepack：包中为 MessagePack 编解码器

特别提醒：

> 1）虽然 MessagePack 用于序列化对象，但是普通 String、Integer 等等同样也是对象，所以照样可以传输普通的字符串等消息  
> 2）需要序列化的 POJO 对象上必须加上 org.msgpack.annotation.Message 注解：@Message，否则传输会失败，而且也不报错，很难排查  
> 3）MessagePack 序列化对象后的消息，经过发送后，接收端 channelRead(ChannelHandlerContext ctx, Object msg)  
>     3.1）即使发送的是 User 对象，接收端的 msg 也不能进行 User user = (User)msg 强转，否则客户端会被强制断开连接  
>     3.2）如果发送的是 User 对象，接收端可以转为 List<Object> objects = (List<Object>) msg，list 中的元素对应 User 的属性值  
>     3.3）如果发送的不是 POJO 对象，而是简单的 String 对象，则不能转为 List<Object>，否则客户端也会被强制断开

# MessagePack 编解码器 #

利用 Netty 的编解码框架可以非常方便的集成第三方序列化框架，Netty 预集成了几种常用的编解码框架，用户也可以根据自己项目的实际情况集成其它编解码框架，或者进行自定义。

## MessagePack 编码器 ##

package com.example.messagepack;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToByteEncoder;
    import org.msgpack.MessagePack;
    
    /**
     * Created by Administrator on 2018/11/25 0025.
     * MessagePack 编码器 —— 继承 Netty 的 MessageToByteEncoder，比重写方法
     */
    public class MsgpackEncoder extends MessageToByteEncoder<Object> {
    
        /**
         * 重写方法，负责将 Object 类型的 POJO 对象编码为 byte 数组，然后写入 ByteBuf 中
         *
         * @param channelHandlerContext
         * @param o
         * @param byteBuf
         * @throws Exception
         */
    
        @Override
        protected void encode(ChannelHandlerContext channelHandlerContext, Object o, ByteBuf byteBuf) throws Exception {
            MessagePack messagePack = new MessagePack();
    
            /** 序列化对象*/
            byte[] raw = messagePack.write(o);
            byteBuf.writeBytes(raw);
        }
    }

## MessagePack 解码器 ##

package com.example.messagepack;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToMessageDecoder;
    import org.msgpack.MessagePack;
    
    import java.util.List;
    
    /**
     * Created by Administrator on 2018/11/25 0025.
     * MessagePack 解码器 - 继承 Netty 的 MessageToMessageDecoder,并重写方法
     */
    public class MsgpackDecoder extends MessageToMessageDecoder<ByteBuf> {
        /**
         * 重写方法，首先从数据报 byteBuf 中获取需要解码的 byte 数组，
         * 然后调用 MessagePack 的 read 方法将其反序列化为 Object 对象，将解码后的对象加入到解码列表 list 中，
         * 这样就完成了 MessagePack 的解码操作
         *
         * @param channelHandlerContext
         * @param byteBuf
         * @param list
         * @throws Exception
         */
        @Override
        protected void decode(ChannelHandlerContext channelHandlerContext, ByteBuf byteBuf, List<Object> list) throws Exception {
            int length = byteBuf.readableBytes();
            byte[] array = new byte[length];
            byteBuf.getBytes(byteBuf.readerIndex(), array, 0, length);
    
            MessagePack messagePack = new MessagePack();
            list.add(messagePack.read(array));
        }
    }

## POJO  User ##

package com.example.domain;
    
    import org.msgpack.annotation.Message;
    
    import java.util.Date;
    
    /**
     * Created by Administrator on 2018/11/25 0025.
     * 用户 实体
     * 需要序列化的 POJO 对象上必须加上 org.msgpack.annotation.Message 注解：@Message
     */
    @Message
    public class User {
        private Integer pId;
        private String pName;
        private Date birthday;
        private Boolean isMarry;
    
        public Date getBirthday() {
            return birthday;
        }
    
        public void setBirthday(Date birthday) {
            this.birthday = birthday;
        }
    
        public Integer getpId() {
            return pId;
        }
    
        public void setpId(Integer pId) {
            this.pId = pId;
        }
    
        public String getpName() {
            return pName;
        }
    
        public void setpName(String pName) {
            this.pName = pName;
        }
    
        public Boolean getIsMarry() {
            return isMarry;
        }
    
        public void setIsMarry(Boolean isMarry) {
            this.isMarry = isMarry;
        }
    
        @Override
        public String toString() {
            return "User{" +
                    "birthday=" + birthday +
                    ", pId=" + pId +
                    ", pName='" + pName + '\'' +
                    ", isMarry=" + isMarry +
                    '}';
        }
    }

# Netty 网络通信 #

首先模拟的情况是：客户端连接上服务器后，给服务器连发消息，服务器接收后会将原信息进回复，同时会解决 TCP 粘包与拆包。会使用 Netty 的 LengthFieldPrepender、LengthFieldBasedFrameDecoder 编解码器处理半包消息，不会出现 TCP 粘包/拆包。

## 服务端 ##

**EchoServer 内容如下：**

package com.example.echo;
    
    import com.example.messagepack.MsgpackDecoder;
    import com.example.messagepack.MsgpackEncoder;
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    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.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldPrepender;
    import io.netty.handler.codec.string.StringDecoder;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    
    /**
     * Created by Administrator on 2018/11/11 0011.
     * Echo 服务器
     */
    public class EchoServer {
        public static void main(String[] args) {
            int port = 9898;
            new EchoServer().bind(port);
        }
    
        public void bind(int port) {
            /**
             * interface EventLoopGroup extends EventExecutorGroup extends ScheduledExecutorService extends ExecutorService
             * 配置服务端的 NIO 线程池,用于网络事件处理，实质上他们就是 Reactor 线程组
             * bossGroup 用于服务端接受客户端连接，workerGroup 用于进行 SocketChannel 网络读写*/
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                /** ServerBootstrap 是 Netty 用于启动 NIO 服务端的辅助启动类，用于降低开发难度
                 * */
                final ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .option(ChannelOption.SO_BACKLOG, 1024)
                                // 设置TCP连接超时时间
                        .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                        .handler(new LoggingHandler(LogLevel.INFO))
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                System.out.println(Thread.currentThread().getName() + ",服务器初始化通道...");
                                /**
                                 * 为了处理半包消息，添加如下两个 Netty 内置的编解码器
                                 * LengthFieldPrepender：前置长度域编码器——放在MsgpackEncoder编码器前面
                                 * LengthFieldBasedFrameDecoder：长度域解码器——放在MsgpackDecoder解码器前面
                                 * 关于 长度域编解码器处理半包消息，本文不做详细讲解，会有专门篇章进行说明
                                 */
                                ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(2));
                                ch.pipeline().addLast("MessagePack encoder", new MsgpackEncoder());
                                ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 2));
                                ch.pipeline().addLast("MessagePack Decoder", new MsgpackDecoder());
                                ch.pipeline().addLast(new EchoServerHandler());
                            }
                        });
    
                /**服务器启动辅助类配置完成后，调用 bind 方法绑定监听端口，调用 sync 方法同步等待绑定操作完成*/
                ChannelFuture f = b.bind(port).sync();
    
                System.out.println(Thread.currentThread().getName() + ",服务器开始监听端口，等待客户端连接.........");
    
                /**下面会进行阻塞，等待服务器连接关闭之后 main 方法退出，程序结束* */
                f.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                /**优雅退出，释放线程池资源*/
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

**EchoServerHandler 内容如下：**

package com.example.echo;
    
    import com.example.domain.User;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    import java.util.List;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /**
     * Created by Administrator on 2017/5/16.
     * ChannelInboundHandlerAdapter extends ChannelHandlerAdapter 用于对网络事件进行读写操作
     */
    public class EchoServerHandler extends ChannelInboundHandlerAdapter {
    
        /**
         * 因为多线程，所以使用原子操作类来进行计数
         */
        private static AtomicInteger atomicInteger = new AtomicInteger();
    
        /**
         * 收到客户端消息，自动触发
         *
         * @param ctx
         * @param msg
         * @throws Exception
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println((atomicInteger.addAndGet(1)) + "--->" + Thread.currentThread().getName() + ",The server receive  order : " + msg);
    
            /**
             * 如果传输的是 POJO 对象，则可以转成 List<Object>
             * list 中的每一个元素都是发送来的 POJO 对象的属性值
             * 注意：如果对方传输只是简单的 String 对象，则不能强转为 List<Object>
             */
           
            /* List<Object> objects = (List<Object>) msg;
            for (Object obj : objects) {
                System.out.println("属性：" + obj);
            }*/
    
            /**
             * 服务端接收到客户端发送来的数据后，再回发给客户端
             */
            ctx.writeAndFlush(msg);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            System.out.println("-----客户端关闭:" + ctx.channel().remoteAddress());
            /**当发生异常时，关闭 ChannelHandlerContext，释放和它相关联的句柄等资源 */
            ctx.close();
        }
    }

## 客户端 ##

**EchoClient 内容如下：**

package com.example.echo;
    
    import com.example.messagepack.MsgpackDecoder;
    import com.example.messagepack.MsgpackEncoder;
    import io.netty.bootstrap.Bootstrap;
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    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.DelimiterBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldBasedFrameDecoder;
    import io.netty.handler.codec.LengthFieldPrepender;
    import io.netty.handler.codec.string.StringDecoder;
    
    /**
     * Created by Administrator on 2017/5/16.
     * Echo 客户端
     */
    public class EchoClient {
    
        /**
         * 使用 2 个线程模拟 2 个客户端
         *
         * @param args
         */
        public static void main(String[] args) {
            for (int i = 0; i < 2; i++) {
                new Thread(new MyThread()).start();
            }
        }
    
        static class MyThread implements Runnable {
    
            @Override
            public void run() {
                connect("192.168.1.20", 9898);
            }
    
            public void connect(String host, int port) {
                /**配置客户端 NIO 线程组/池*/
                EventLoopGroup group = new NioEventLoopGroup();
                try {
                    /**Bootstrap 与 ServerBootstrap 都继承(extends)于 AbstractBootstrap
                     * 创建客户端辅助启动类,并对其配置,与服务器稍微不同，这里的 Channel 设置为 NioSocketChannel
                     * 然后为其添加 Handler，这里直接使用匿名内部类，实现 initChannel 方法
                     * 作用是当创建 NioSocketChannel 成功后，在进行初始化时,将它的ChannelHandler设置到ChannelPipeline中，用于处理网络I/O事件*/
                    Bootstrap b = new Bootstrap();
                    b.group(group).channel(NioSocketChannel.class)
                            .option(ChannelOption.TCP_NODELAY, true)
                                    // 设置TCP连接超时时间
                            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000)
                            .handler(new ChannelInitializer<SocketChannel>() {
                                @Override
                                public void initChannel(SocketChannel ch) throws Exception {
                                    System.out.println(Thread.currentThread().getName() + ",客户端初始化管道...");
                                    /**
                                     * 为了处理半包消息，添加如下两个 Netty 内置的编解码器
                                     * LengthFieldPrepender：前置长度域编码器——放在MsgpackEncoder编码器前面
                                     * LengthFieldBasedFrameDecoder：长度域解码器——放在MsgpackDecoder解码器前面
                                     * 关于 长度域编解码器处理半包消息，本文不做详细讲解，会有专门篇章进行说明
                                     */
                                    ch.pipeline().addLast("frameEncoder", new LengthFieldPrepender(2));
                                    ch.pipeline().addLast("MessagePack encoder", new MsgpackEncoder());
                                    ch.pipeline().addLast("frameDecoder", new LengthFieldBasedFrameDecoder(65535, 0, 2, 0, 2));
                                    ch.pipeline().addLast("MessagePack Decoder", new MsgpackDecoder());
                                    ch.pipeline().addLast(new EchoClientHandler());
                                }
                            });
    
                    /**connect：发起异步连接操作，调用同步方法 sync 等待连接成功*/
                    ChannelFuture channelFuture = b.connect(host, port).sync();
                    System.out.println(Thread.currentThread().getName() + ",客户端发起异步连接..........");
    
                    /**等待客户端链路关闭*/
                    channelFuture.channel().closeFuture().sync();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    /**优雅退出，释放NIO线程组*/
                    group.shutdownGracefully();
                }
            }
        }
    }

**EchoClientHandler 内容如下：**

package com.example.echo;
    
    import com.example.domain.User;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    import java.util.Arrays;
    import java.util.Date;
    import java.util.List;
    import java.util.concurrent.atomic.AtomicInteger;
    
    /**
     * Created by Administrator on 2017/5/17.
     * 用于对网络事件进行读写操作
     */
    public class EchoClientHandler extends ChannelInboundHandlerAdapter {
    
        /**
         * 因为 Netty 采用线程池，所以这里使用原子操作类来进行计数
         */
        private static AtomicInteger atomicInteger = new AtomicInteger();
    
        /**
         * 当客户端和服务端 TCP 链路建立成功之后，Netty 的 NIO 线程会调用 channelActive 方法
         */
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
    
            /**
             * 多余 数组、List、Set、Map 等，对立面的元素逐个进行发送，则对方也是逐个接收
             * 否则如果直接发送 数组、List、Set、Map 等，则对方会统一接收
             * 注意：因为使用LengthFieldPrepender、LengthFieldBasedFrameDecoder编解码器处理半包消息
             * 所以这里连续发送也不会出现 TCP 粘包/拆包
             */
            List<User> users = getUserArrayData();
            for (User user : users) {
                ctx.writeAndFlush(user);
            }
            ctx.writeAndFlush("我是普通的字符串消息" + Thread.currentThread().getName());
        }
    
        /**
         * 当服务端返回应答消息时，channelRead 方法被调用，从 Netty 的 ByteBuf 中读取并打印应答消息
         */
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println((atomicInteger.addAndGet(1)) + "---" + Thread.currentThread().getName() + ",Server return Message：" + msg);
        }
    
        /**
         * 当发生异常时，打印异常 日志，释放客户端资源
         */
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            /**释放资源*/
            ctx.close();
        }
    
        /**
         * 设置网络传输的 POJO 对象数组/列表
         *
         * @return
         */
        public List<User> getUserArrayData() {
            User[] users = new User[5];
            User loopUser = null;
            for (int i = 0; i < 5; i++) {
                loopUser = new User();
                loopUser.setpId(i + 1);
                loopUser.setpName("华安" + Thread.currentThread().getName());
                loopUser.setIsMarry(true);
                loopUser.setBirthday(new Date());
                users[i] = loopUser;
            }
            return Arrays.asList(users);
        }
    }

# 运行测试 #

先运行服务器，再运行客户端。

![20181126103711497.gif][]

[20181125150000635.png]: /images/20220414/ec4ba4d3495b4ddda4b4360f3253fb11.png
[MessagePack]: https://blog.csdn.net/wangmx1993328/article/details/84477073
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dhbmdteDE5OTMzMjg_size_16_color_FFFFFF_t_70]: /images/20220414/09879664446441c1a21c048f1e5fd57c.png
[20181126103711497.gif]: /images/20220414/fe3aa28be24e4a7788c745aef7d67bed.png