Netty序列化问题解决方案（十二）

r囧r小猫 2022-10-29 13:26 365阅读 0赞

今天分享Netty序列化问题解决方案

**一、Java** **序列化方案：**

1、Java  序列化的目的主要有两个：

1). 网络传输

2). 对象持久化

当选行远程跨迸程服务调用时，需要把被传输的  Java  对象编码为字节数组或者 ByteBuffer 对象。而当远程服务读取到  ByteBuffer  对象或者字节数组时，需要将其解码为发 送时的 Java  对象。这被称为  Java  对象编解码技术。 Java 序列化仅仅是  Java  编解码技术的一种，由于它的种种缺陷，衍生出了多种编解码技术和框架

**2、Java**  **序列化的缺点**

Java  序列化从  JDK1.1  版本就已经提供，它不需要添加额外的类库，只需实现 java.io.Serializable 并生成序列  ID  即可，因此，它从诞生之初就得到了广泛的应用。

但是在远程服务调用（ RPC ）时，很少直接使用  Java  序列化进行消息的编解码和传输， 这又是什么原因呢？下面通过分析.Tava  序列化的缺点来找出答案。

1 ） 无法跨语言

对于跨进程的服务调用，服务提供者可能会使用  C  十＋或者其他语言开发，当我们需要 和异构语言进程交互时 Java  序列化就难以胜任。由于  Java  序列化技术是  Java  语言内部的私 有协议，其他语言并不支持，对于用户来说它完全是黑盒。对于 Java  序列化后的字节数组， 别的语言无法进行反序列化，这就严重阻碍了它的应用。

2）  序列化后的码流太大

通过一个实例看下  Java  序列化后的字节数组大小。

3 ） 序列化性能太低

无论是序列化后的码流大小，还是序列化的性能，JDK 默认的序列化机制表现得都很差。 因此，我们边常不会选择 Java  序列化作为远程跨节点调用的编解码框架。

4）JDK序列化与ByteBuffer 二进制编码的性能对比：

首先、pom文件引入jar

<dependency>
                <groupId>com.google.protobuf</groupId>
                <artifactId>protobuf-java</artifactId>
                <version>2.6.1</version>
            </dependency>

对象核心类：

public class UserInfo implements Serializable {
    
        /**
         * 默认的序列号
         */
        private static final long serialVersionUID = 1L;
    
        private String userName;
    
        private int userID;
    
    	public UserInfo buildUserName(String userName) {
    		this.userName = userName;
    		return this;
        }
    
        public UserInfo buildUserID(int userID) {
    		this.userID = userID;
    		return this;
        }
    
        public final String getUserName() {
    		return userName;
        }
    
    
        public final void setUserName(String userName) {
    		this.userName = userName;
        }
    
    
        public final int getUserID() {
    		return userID;
        }
    
    
        public final void setUserID(int userID) {
    		this.userID = userID;
        }
    
        //自行序列化
        public byte[] codeC() {
    		ByteBuffer buffer = ByteBuffer.allocate(1024);
    		byte[] value = this.userName.getBytes();//userName转换为字节数组value
    		buffer.putInt(value.length);//写入字节数组value的长度
    		buffer.put(value);//写入字节数组value的值
    		buffer.putInt(this.userID);//写入userID的值
    		buffer.flip();//准备读取buffer中的数据
    		value = null;
    		byte[] result = new byte[buffer.remaining()];
    		buffer.get(result);//buffer中的数据写入字节数组并作为结果返回
    		return result;
        }
    
    	//自行序列化方法2
        public byte[] codeC(ByteBuffer buffer) {
    		buffer.clear();
    		byte[] value = this.userName.getBytes();
    		buffer.putInt(value.length);
    		buffer.put(value);
    		buffer.putInt(this.userID);
    		buffer.flip();
    		value = null;
    		byte[] result = new byte[buffer.remaining()];
    		buffer.get(result);
    		return result;
        }
    }

测试序列化后的流的大小对比：

public class TestUserInfo {
    
        /**
         * @param args
         * @throws IOException
         */
        public static void main(String[] args) throws IOException {
    		UserInfo info = new UserInfo();
    		info.buildUserID(100).buildUserName("Welcome to Netty");
    		ByteArrayOutputStream bos = new ByteArrayOutputStream();
    		ObjectOutputStream os = new ObjectOutputStream(bos);
    		os.writeObject(info);
    		os.flush();
    		os.close();
    		byte[] b = bos.toByteArray();
    		System.out.println("The jdk serializable length is : " + b.length);
    		bos.close();
    		System.out.println("-------------------------------------");
    		System.out.println("The byte array serializable length is : "
    			+ info.codeC().length);
    		ByteBuffer buffer = ByteBuffer.allocate(1024);
    		System.out.println("The ByteBuffer byte array serializable length is : "
    				+ info.codeC(buffer).length);
    
        }
    
    }

执行结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70][]

测试序列化过程所用时间对比：

public class PerformTestUserInfo {
    
        public static void main(String[] args) throws IOException {
    	UserInfo info = new UserInfo();
    	info.buildUserID(100).buildUserName("Welcome to Netty");
    	int loop = 1000000;
    	ByteArrayOutputStream bos = null;
    	ObjectOutputStream os = null;
    	long startTime = System.currentTimeMillis();
    	for (int i = 0; i < loop; i++) {
    	    bos = new ByteArrayOutputStream();
    	    os = new ObjectOutputStream(bos);
    	    os.writeObject(info);
    	    os.flush();
    	    os.close();
    	    byte[] b = bos.toByteArray();
    	    bos.close();
    	}
    	long endTime = System.currentTimeMillis();
    	System.out.println("The jdk serializable cost time is  : "
    		+ (endTime - startTime) + " ms");
    	System.out.println("-------------------------------------");
    
    	ByteBuffer buffer = ByteBuffer.allocate(1024);
    	startTime = System.currentTimeMillis();
    	for (int i = 0; i < loop; i++) {
    	    byte[] b = info.codeC(buffer);
    	}
    	endTime = System.currentTimeMillis();
    	System.out.println("The byte array serializable cost time is : "
    		+ (endTime - startTime) + " ms");
        }
    }

执行对比结果：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1][]

**二、序列化 – 内置和第三方的**  **MessagePack**  **实战**

***1、*** Netty ***内置 *** 编解码器

Netty  内置了对  JBoss Marshalling  和  Protocol Buffers  的支持Protocol Buffers 序列化机制

Google protobuf 编解码，主要 依据 ProtobufDecoder 解码器 和 ProtobufEncoder 编码器原理：

测试使用示例如下：

原生的对象代码：

public class Person {
        String name;
        int id;
        String email;
    }

Protobuf格式的对象转换成的java 代码,如图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2][]

服务端代码：

public class ProtoBufServer {
        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, 100)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            public void initChannel(SocketChannel ch) {
                                /*去除消息长度部分，同时根据这个消息长度读取实际的数据*/
                                ch.pipeline().addLast(
                                        new ProtobufVarint32FrameDecoder());
                                ch.pipeline().addLast(new ProtobufDecoder(
                                        PersonProto.Person.getDefaultInstance()
                                ));
                                ch.pipeline().addLast(new ProtoBufServerHandler());
                    }
                });
    
                // 绑定端口，同步等待成功
                ChannelFuture f = b.bind(port).sync();
    
                System.out.println("init start");
                // 等待服务端监听端口关闭
                f.channel().closeFuture().sync();
            } finally {
                // 优雅退出，释放线程池资源
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
            int port = 8080;
            new ProtoBufServer().bind(port);
        }
    }

服务端业务代码：

public class ProtoBufServerHandler  extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception {
            PersonProto.Person req = (PersonProto.Person)msg;
            System.out.println("get data name = "+req.getName());
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            if(cause instanceof IOException){
                System.out.println("远程客户端强迫关闭了一个现有的连接。");
            }
            ctx.close();
        }
    }

客户端代码：

public class ProtoBufClient {
        public void connect(int port, String host) throws Exception {
            // 配置客户端NIO线程组
            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
                            public void initChannel(SocketChannel ch)
                                    throws Exception {
                                /*加一个消息长度，由netty自动计算*/
                                ch.pipeline().addLast(
                                        new ProtobufVarint32LengthFieldPrepender()
                                );
                                /*负责编码,序列化*/
                                ch.pipeline().addLast(new ProtobufEncoder());
                                ch.pipeline().addLast(new ProtoBufClientHandler());
                            }
                        });
    
                ChannelFuture f = b.connect(host, port).sync();
                f.channel().closeFuture().sync();
            } finally {
                // 优雅退出，释放NIO线程组
                group.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
            int port = 8080;
            new ProtoBufClient().connect(port, "127.0.0.1");
        }
    }

客户端业务代码：

public class ProtoBufClientHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelActive(ChannelHandlerContext ctx) {
           System.out.println("Prepare to make data........");
           PersonProto.Person.Builder builder = PersonProto.Person.newBuilder();
            builder.setName("Mark");
            builder.setId(1);
            builder.setEmail("Mark@enjoyedu.com");
            System.out.println("send data........");
            ctx.writeAndFlush(builder.build());
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            cause.printStackTrace();
            ctx.close();
        }
    }

执行测试结果：

客户端：

![20210219135315962.png][]

服务端：

![20210219135329828.png][]

[protobuf使用参考][protobuf]文章。

***二、集成第三方***  ***MessagePack***  ***实战（*** ***LengthFieldBasedFrame***  ***详解）***

LengthFieldBasedFrame  详解

maxFrameLength ：表示的是包的最大长度，

lengthFieldOffset ：指的是长度域的偏移量，表示跳过指定个数字节之后的才是长度域；

lengthFieldLength ：记录该帧数据长度的字段，也就是长度域本身的长度；

lengthAdjustment ：长度的一个修正值，可正可负， Netty  在读取到数据包的长度值  N  后，认为接下来的 N  个字节都是需要读取的，但是根据实际情况，有可能需要增加  N  的值，也有可能需要减少 N  的值，具体增加多少，减少多少，写在这个参数里；

initialBytesToStrip：从数据帧中跳过的字节数，表示得到一个完整的数据包之后，扔掉 这个数据包中多少字节数，才是后续业务实际需要的业务数据。

failFast ：如果为  true ，则表示读取到长度域， TA  的值的超过  maxFrameLength ，就抛出 一个 TooLongFrameException ，而为  false  表示只有当真正读取完长度域的值表示的字节之后，才会抛出 TooLongFrameException ，默认情况下设置为  true ，建议不要修改，否则可能会造成内存溢出。

**数据包大小** **: 14B =**  **长度域**  **2B + "HELLO, WORLD"** **（单词**  **HELLO+** **一个逗号** **+** **一个空格** **+** **单** **词**  **WORLD** **）**

![20210215124322679.png][]

长度域的值为  12B(0x000c) 。希望解码后保持一样，根据上面的公式 , 参数应该为：

1. lengthFieldOffset = 0

2. lengthFieldLength = 2

3. lengthAdjustment  无需调整

4. initialBytesToStrip = 0 -  解码过程中，没有丢弃任何数据

**数据包大小** **: 14B =**  **长度域**  **2B + "HELLO, WORLD"**

![20210215124339268.png][]

长度域的值为  12B(0x000c) 。解码后，希望丢弃长度域  2B  字段，所以，只要initialBytesToStrip = 2 即可。

1. lengthFieldOffset = 0

2. lengthFieldLength = 2

3. lengthAdjustment  无需调整

4. initialBytesToStrip = 2  解码过程中，丢弃  2  个字节的数据

**数据包大小** **: 14B =**  **长度域**  **2B + "HELLO, WORLD"** **。长度域的值为**  **14(0x000E)**

![20210215124353245.png][]

长度域的值为  14(0x000E) ，包含了长度域本身的长度。希望解码后保持一样，根据上面的公式，参数应该为：

1. lengthFieldOffset = 0

2. lengthFieldLength = 2

3. lengthAdjustment = -2  因为长度域为  14 ，而报文内容为  12 ，为了防止读取报文超出报文本体，和将长度字段一起读取进来，需要告诉 netty ，实际读取的报文长度比长度域中的要少 2 （ 12-14=-2 ）

4. initialBytesToStrip = 0 -  解码过程中，没有丢弃任何数据

**在长度域前添加**  **2**  **个字节的**  **Header** **。长度域的值** **(0x00000C) = 12** **。总数据包长度** **:**  **17=Header(2B) +**  **长度域** **(3B) + "HELLO, WORLD"**

![20210215124409164.png][]

长度域的值为  12B(0x000c) 。编码解码后，长度保持一致，所以  initialBytesToStrip = 0 。

参数应该为:

1. lengthFieldOffset = 2

2. lengthFieldLength = 3

3. lengthAdjustment = 0  无需调整

4. initialBytesToStrip = 0 -  解码过程中，没有丢弃任何数据

**Header**  **与长度域的位置换了。总数据包长度** **: 17=** **长度域** **(3B) + Header(2B) + "HELLO,**  **WORLD"**

![20210215124425305.png][]

长度域的值为  12B(0x000c) 。编码解码后，长度保持一致，所以  initialBytesToStrip = 0 。

参数应该为 :

1. lengthFieldOffset = 0

2. lengthFieldLength = 3

3. lengthAdjustment = 2  因为长度域为  12 ，而报文内容为  12 ，但是我们需要把  Header 的值一起读取进来，需要告诉 netty ，实际读取的报文内容长度比长度域中的要多  2 （ 12+2=14 ）

4. initialBytesToStrip = 0 -  解码过程中，没有丢弃任何数据

**带有两个**  **header** **。** **HDR1**  **丢弃，长度域丢弃，只剩下第二个**  **header**  **和有效包体，这种** **协议中，一般**  **HDR1**  **可以表示**  **magicNumber** **，表示应用只接受以该**  **magicNumber**  **开头的二** **进制数据，**

**rpc**  **里面用的比较多。总数据包长度** **: 16=HDR1(1B)+** **长度域** **(2B) +HDR2(1B) + "HELLO,**  **WORLD"**

![20210215124454818.png][]

长度域的值为  12B(0x000c)

1. lengthFieldOffset = 1 (HDR1  的长度 )

2. lengthFieldLength = 2

3. lengthAdjustment =1  因为长度域为  12 ，而报文内容为  12 ，但是我们需要把  HDR2  的值一起读取进来，需要告诉 netty ，实际读取的报文内容长度比长度域中的要多  1 （ 12+1=13 ）

4. initialBytesToStrip = 3  丢弃了  HDR1  和长度字段

**带有两个**  **header** **，** **HDR1**  **丢弃，长度域丢弃，只剩下第二个**  **header**  **和有效包体。总数** **据包长度** **: 16=HDR1(1B)+** **长度域** **(2B) +HDR2(1B) + "HELLO, WORLD"**

![20210215124509416.png][]

长度域的值为  16B(0x0010) ，长度为  2 ， HDR1  的长度为  1 ， HDR2  的长度为  1 ，包体的长度为 12 ， 1+1+2+12=16 。

1. lengthFieldOffset = 1

2. lengthFieldLength = 2

3. lengthAdjustment = -3  因为长度域为  16 ，需要告诉  netty ，实际读取的报文内容长度比长度域中的要 少 3 （ 13-16= -3 ）

4. initialBytesToStrip = 3  丢弃了  HDR1  和长度字段

三、MessagePack  集成测试代码；

首先、pom文件引用jar:

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

用户实体类：

@Message
    public class User {
        private String id;
        private String userName;
        private int age;
        private UserContact userContact;
    
        public User(String userName, int age, String id) {
            this.userName = userName;
            this.age = age;
            this.id = id;
        }
    
        public User() {
        }
    
        public String getUserName() {
            return userName;
        }
    
        public void setUserName(String userName) {
            this.userName = userName;
        }
    
        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 UserContact getUserContact() {
            return userContact;
        }
    
        public void setUserContact(UserContact userContact) {
            this.userContact = userContact;
        }
    
        @Override
        public String toString() {
            return "User{" +
                    "userName='" + userName + '\'' +
                    ", age=" + age +
                    ", id='" + id + '\'' +
                    ", userContact=" + userContact +
                    '}';
        }
    }

用户相关实体类：

@Message//MessagePack提供的注解，表明这是一个需要序列化的实体类
    public class UserContact {
        private String mail;
        private String phone;
    
        public UserContact() {
        }
    
        public UserContact(String mail, String phone) {
            this.mail = mail;
            this.phone = phone;
        }
    
        public String getMail() {
            return mail;
        }
    
        public void setMail(String mail) {
            this.mail = mail;
        }
    
        public String getPhone() {
            return phone;
        }
    
        public void setPhone(String phone) {
            this.phone = phone;
        }
    
        @Override
        public String toString() {
            return "UserContact{" +
                    "mail='" + mail + '\'' +
                    ", phone='" + phone + '\'' +
                    '}';
        }
    }

编码器代码：

/*基于MessagePack的编码器，序列化*/
    public class MsgPackEncode extends MessageToByteEncoder<User> {
        @Override
        protected void encode(ChannelHandlerContext ctx, User msg, ByteBuf out)
                throws Exception {
            MessagePack messagePack = new MessagePack();
            byte[] raw = messagePack.write(msg);
            out.writeBytes(raw);
        }
    }

解码器代码：

/*基于MessagePack的解码器，反序列化*/
    public class MsgPackDecoder extends MessageToMessageDecoder<ByteBuf> {
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf msg, List<Object> out)
                throws Exception {
            final int length = msg.readableBytes();
            final byte[] array = new byte[length];
            msg.getBytes(msg.readerIndex(),array,0,length);
            MessagePack messagePack = new MessagePack();
            out.add(messagePack.read(array,User.class));
        }
    }

1、服务端代码：

public class ServerMsgPackEcho {
    
        public static final int PORT = 9995;
    
        public static void main(String[] args) throws InterruptedException {
            ServerMsgPackEcho serverMsgPackEcho = new ServerMsgPackEcho();
            System.out.println("服务器即将启动");
            serverMsgPackEcho.start();
        }
        public void start() throws InterruptedException {
            final MsgPackServerHandler serverHandler = new MsgPackServerHandler();
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                ServerBootstrap b = new ServerBootstrap();/*服务端启动必须*/
                b.group(group)/*将线程组传入*/
                    .channel(NioServerSocketChannel.class)/*指定使用NIO进行网络传输*/
                    .localAddress(new InetSocketAddress(PORT))/*指定服务器监听端口*/
                    /*服务端每接收到一个连接请求，就会新启一个socket通信，也就是channel，
                    所以下面这段代码的作用就是为这个子channel增加handle*/
                    .childHandler(new ChannelInitializerImp());
                ChannelFuture f = b.bind().sync();/*异步绑定到服务器，sync()会阻塞直到完成*/
                System.out.println("服务器启动完成，等待客户端的连接和数据.....");
                f.channel().closeFuture().sync();/*阻塞直到服务器的channel关闭*/
            } finally {
                group.shutdownGracefully().sync();/*优雅关闭线程组*/
            }
        }
    
        private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(65535,
                        0,2,0,
                        2));
                ch.pipeline().addLast(new MsgPackDecoder());
                ch.pipeline().addLast(new MsgPackServerHandler());
            }
        }
    }

业务代码：

@ChannelHandler.Sharable
    public class MsgPackServerHandler extends ChannelInboundHandlerAdapter {
    
        private AtomicInteger counter = new AtomicInteger(0);
    
        /*** 服务端读取到网络数据后的处理*/
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            //将上一个handler生成的数据强制转型
            User user = (User)msg;
            System.out.println("Server Accept["+user
                    +"] and the counter is:"+counter.incrementAndGet());
            //服务器的应答
            String resp = "I process user :"+user.getUserName()
                    + System.getProperty("line.separator");
            ctx.writeAndFlush(Unpooled.copiedBuffer(resp.getBytes()));
            ctx.fireChannelRead(user);
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

2、客户端代码：

public class ClientMsgPackEcho {
    
        private final String host;
    
        public ClientMsgPackEcho(String host) {
            this.host = host;
        }
    
        public void start() throws InterruptedException {
            EventLoopGroup group = new NioEventLoopGroup();/*线程组*/
            try {
                final Bootstrap b = new Bootstrap();;/*客户端启动必须*/
                b.group(group)/*将线程组传入*/
                        .channel(NioSocketChannel.class)/*指定使用NIO进行网络传输*/
                        /*配置要连接服务器的ip地址和端口*/
                        .remoteAddress(
                                new InetSocketAddress(host, ServerMsgPackEcho.PORT))
                        .handler(new ChannelInitializerImp());
                ChannelFuture f = b.connect().sync();
                System.out.println("已连接到服务器.....");
                f.channel().closeFuture().sync();
            } finally {
                group.shutdownGracefully().sync();
            }
        }
    
        private static class ChannelInitializerImp extends ChannelInitializer<Channel> {
    
            @Override
            protected void initChannel(Channel ch) throws Exception {
                /*告诉netty，计算一下报文的长度，然后作为报文头加在前面*/
                ch.pipeline().addLast(new LengthFieldPrepender(2));
                /*对服务器的应答也要解码，解决粘包半包*/
                ch.pipeline().addLast(new LineBasedFrameDecoder(1024));
                /*对我们要发送的数据做编码-序列化*/
               ch.pipeline().addLast(new MsgPackEncode());
               ch.pipeline().addLast(new MsgPackClientHandler(5));
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            new ClientMsgPackEcho("127.0.0.1").start();
        }
    }

业务代码：

public class MsgPackClientHandler extends SimpleChannelInboundHandler<ByteBuf> {
        private final int sendNumber;
    
        public MsgPackClientHandler(int sendNumber) {
            this.sendNumber = sendNumber;
        }
    
        private AtomicInteger counter = new AtomicInteger(0);
    
        /*** 客户端读取到网络数据后的处理*/
        protected void channelRead0(ChannelHandlerContext ctx, ByteBuf msg) throws Exception {
            System.out.println("client Accept["+msg.toString(CharsetUtil.UTF_8)
                    +"] and the counter is:"+counter.incrementAndGet());
        }
    
        /*** 客户端被通知channel活跃后，做事*/
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            User[] users = makeUsers();
            //发送数据
            for(User user:users){
                System.out.println("Send user:"+user);
                ctx.write(user);
            }
            ctx.flush();
        }
    
        /*** 发生异常后的处理*/
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
        /*生成用户实体类的数组，以供发送*/
        private User[] makeUsers(){
            User[] users=new User[sendNumber];
            User user =null;
            for(int i=0;i<sendNumber;i++){
                user=new User();
                user.setAge(i);
                String userName = "ABCDEFG --->"+i;
                user.setUserName(userName);
                user.setId("No:"+(sendNumber-i));
                user.setUserContact(
                        new UserContact(userName+"@xiangxue.com","133"));
                users[i]=user;
            }
            return users;
        }
    }

3、执行结果：

客户端：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3][]

服务端：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4][]

到此，Netty序列化问题解决方案分析完成，实际项目中可以根据具体业务选择不同的方案，下篇我们分析Netty 单元测试问题，敬请期待。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70]: /images/20221024/c73bebd444834b4888508d2e0b5ebc3f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 1]: /images/20221024/cce26bb8d7a34db0be6fae37913bc61a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 2]: /images/20221024/ee314e0dee394f15bf7271498f24d281.png
[20210219135315962.png]: /images/20221024/ec80f8f45a6b4138b460ea86670449ec.png
[20210219135329828.png]: /images/20221024/27511dc1e5c84f6bbe304e439466ec8f.png
[protobuf]: https://blog.csdn.net/nandao158/article/details/108349526
[20210215124322679.png]: /images/20221024/34da59050507477fa0bd8dbbccd30983.png
[20210215124339268.png]: /images/20221024/ad120b56e4db47768183aed1ed1db503.png
[20210215124353245.png]: /images/20221024/d06bd87492a74224bf957818547b1ff8.png
[20210215124409164.png]: /images/20221024/d7aedf1b52964f1da00bf528b95f619c.png
[20210215124425305.png]: /images/20221024/f4109b90569544f383e60387b6055c04.png
[20210215124454818.png]: /images/20221024/a7bba08343e84bbd856c08a5ed84b8a2.png
[20210215124509416.png]: /images/20221024/fb1aaa9a08064e019932583a89377bce.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 3]: /images/20221024/7a35f51dbbd941d1b7b714f35362a632.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L25hbmRhbzE1OA_size_16_color_FFFFFF_t_70 4]: /images/20221024/49c241e4a0ad4e469599f98bf878091a.png