Netty4.0学习笔记系列之五：自定义通讯协议

Bertha 。 2022-04-11 09:11 182阅读 0赞

Netty中，通讯的双方建立连接后，会把数据按照ByteBuf的方式进行传输，例如http协议中，就是通过HttpRequestDecoder对ByteBuf数据流进行处理，转换成http的对象。基于这个思路，我自定义一种通讯协议：Server和客户端直接传输Java对象。

实现的原理是通过Encoder把java对象转换成ByteBuf流进行传输，通过Decoder把ByteBuf转换成java对象进行处理，处理逻辑如下图所示：

![37049191-d8b6-3d53-9331-997e670e16fc.jpg][]  
        传输的Java bean为Person.java

package com.bijian.netty.dto;
    
    import java.io.Serializable;
    
    //必须实现Serializable接口  
    public class Person implements Serializable {
        
        private static final long serialVersionUID = 1L;
        private String name;
        private String sex;
        private int age;
    
        public String toString() {
            return "name:" + name + " sex:" + sex + " age:" + age;
        }
    
        public String getName() {
            return name;
        }
    
        public void setName(String name) {
            this.name = name;
        }
    
        public String getSex() {
            return sex;
        }
    
        public void setSex(String sex) {
            this.sex = sex;
        }
    
        public int getAge() {
            return age;
        }
    
        public void setAge(int age) {
            this.age = age;
        }
    }

Server端类：Server.java、PersonDecoder.java、BusinessHandler.java

1.Server.java启动netty服务

package com.bijian.netty.server;
    
    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;
    
    public class Server {
        
        public void start(int port) throws Exception {
            
            EventLoopGroup bossGroup = new NioEventLoopGroup();
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                ServerBootstrap b = new ServerBootstrap();
                b.group(bossGroup, workerGroup).channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            public void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new PersonDecoder());
                                ch.pipeline().addLast(new BusinessHandler());
                            }
                        }).option(ChannelOption.SO_BACKLOG, 128).childOption(ChannelOption.SO_KEEPALIVE, true);
    
                ChannelFuture f = b.bind(port).sync();
                f.channel().closeFuture().sync();
            } finally {
                workerGroup.shutdownGracefully();
                bossGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
            Server server = new Server();
            server.start(8000);
        }
    }

2.PersonDecoder.java把ByteBuf流转换成Person对象，其中ByteBufToBytes是读取ButeBuf的工具类，上一篇文章中提到过，在此不在详述。ByteObjConverter是byte和obj的互相转换的工具。

package com.bijian.netty.server;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ByteToMessageDecoder;
    
    import java.util.List;
    
    import com.bijian.netty.util.ByteBufToBytes;
    import com.bijian.netty.util.ByteObjConverter;
      
    public class PersonDecoder extends ByteToMessageDecoder {
        
        @Override  
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
            
            ByteBufToBytes read = new ByteBufToBytes();  
            Object obj = ByteObjConverter.ByteToObject(read.read(in));  
            out.add(obj);
        }
    }

3.BusinessHandler.java读取Person信息，并打印

package com.bijian.netty.server;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import com.bijian.netty.dto.Person;
    
    public class BusinessHandler extends ChannelInboundHandlerAdapter {
        
        private Logger logger = LoggerFactory.getLogger(BusinessHandler.class);
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            
            Person person = (Person) msg;
            logger.info("BusinessHandler read msg from client :" + person);
            ctx.write("ok");
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    
        }
    }

Client端的类：Client ClientInitHandler PersonEncoder

1.Client.java建立与Server的连接

package com.bijian.netty.client;
    
    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;
    
    public class Client {
        
        public void connect(String host, int port) throws Exception {
            
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
                Bootstrap b = new Bootstrap();
                b.group(workerGroup);
                b.channel(NioSocketChannel.class);
                b.option(ChannelOption.SO_KEEPALIVE, true);
                b.handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    public void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new PersonEncoder());
                        ch.pipeline().addLast(new ClientInitHandler());
                    }
                });
    
                ChannelFuture f = b.connect(host, port).sync();
                f.channel().closeFuture().sync();
            } finally {
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
            Client client = new Client();
            client.connect("127.0.0.1", 8000);
        }
    }

2.ClientInitHandler.java向Server发送Person对象

package com.bijian.netty.client;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    import org.slf4j.Logger;
    import org.slf4j.LoggerFactory;
    
    import com.bijian.netty.dto.Person;
    
    public class ClientInitHandler extends ChannelInboundHandlerAdapter {
        
        private static Logger logger = LoggerFactory.getLogger(ClientInitHandler.class);
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            
            logger.info("ClientInitHandler.channelActive");
            Person person = new Person();
            person.setName("test");
            person.setSex("man");
            person.setAge(30);
            ctx.write(person);
            ctx.flush();
        }
    }

3.PersonEncoder.java把Person对象转换成ByteBuf进行传送

package com.bijian.netty.client;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToByteEncoder;
    
    import com.bijian.netty.dto.Person;
    import com.bijian.netty.util.ByteObjConverter;
      
    public class PersonEncoder extends MessageToByteEncoder<Person> {  
      
        @Override  
        protected void encode(ChannelHandlerContext ctx, Person msg, ByteBuf out) throws Exception {
            
            byte[] datas = ByteObjConverter.ObjectToByte(msg);  
            out.writeBytes(datas);  
            ctx.flush();
        }
    }

工具类：ByteObjConverter.java

package com.bijian.netty.util;
    
    import java.io.ByteArrayInputStream;
    import java.io.ByteArrayOutputStream;
    import java.io.IOException;
    import java.io.ObjectInputStream;
    import java.io.ObjectOutputStream;
    
    public class ByteObjConverter {
        
        public static Object ByteToObject(byte[] bytes) {
            
            Object obj = null;
            ByteArrayInputStream bi = new ByteArrayInputStream(bytes);
            ObjectInputStream oi = null;
            try {
                oi = new ObjectInputStream(bi);
                obj = oi.readObject();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    bi.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                try {
                    oi.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            return obj;
        }
    
        public static byte[] ObjectToByte(Object obj) {
            
            byte[] bytes = null;
            ByteArrayOutputStream bo = new ByteArrayOutputStream();
            ObjectOutputStream oo = null;
            try {
                oo = new ObjectOutputStream(bo);
                oo.writeObject(obj);
                bytes = bo.toByteArray();
            } catch (Exception e) {
                e.printStackTrace();
            } finally {
                try {
                    bo.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
                try {
                    oo.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            return (bytes);
        }
    }

工具类：ByteBufToBytes.java

package com.bijian.netty.util;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    
    public class ByteBufToBytes {
        
        private ByteBuf temp;
    
        private boolean end = true;
    
        public ByteBufToBytes() {
            
        }
        
        public ByteBufToBytes(int length) {
            temp = Unpooled.buffer(length);
        }
    
        public void reading(ByteBuf datas) {
            datas.readBytes(temp, datas.readableBytes());
            if (this.temp.writableBytes() != 0) {
                end = false;
            } else {
                end = true;
            }
        }
    
        public boolean isEnd() {
            return end;
        }
    
        public byte[] readFull() {
            if (end) {
                byte[] contentByte = new byte[this.temp.readableBytes()];
                this.temp.readBytes(contentByte);
                this.temp.release();
                return contentByte;
            } else {
                return null;
            }
        }
    
        public byte[] read(ByteBuf datas) {
            byte[] bytes = new byte[datas.readableBytes()];
            datas.readBytes(bytes);
            return bytes;
        }
    }

通过上述代码，实现了Server端与Client端直接使用person对象进行通信的目的。基于此，可以构建更为复杂的场景：Server端同时支撑多种协议，不同的协议采用不同的Decoder进行解析，解析结果保持统一，这样业务处理类可以保持接口一致。下一节将编写这样一个案例。

本例中需要注意的事项是：

1.Person对象必须实现Serializable接口，否则不能进行序列化。

2.PersonDecoder读取ByteBuf数据的时候，并没有对多次流式数据进行处理，而是简单的一次性接收，如果数据量大的情况下，可能会出现数据不完整，这个问题会在后续的学习中解决。

文章来源：[http://blog.csdn.net/u013252773/article/details/21608951][http_blog.csdn.net_u013252773_article_details_21608951]

[37049191-d8b6-3d53-9331-997e670e16fc.jpg]: /images/20220411/3d5d036dcdb443d38528d39e5c0143d2.png
[http_blog.csdn.net_u013252773_article_details_21608951]: http://blog.csdn.net/u013252773/article/details/21608951