Netty4.0学习笔记系列之六：多种通讯协议支持

绝地灬酷狼 2022-04-11 09:11 272阅读 0赞

上文介绍了如何应用Netty开发自定义通讯协议，本文在此基础上进一步深化，研究如何同时支持不同的通讯协议。

此处所谓的通讯协议，指的是把Netty通讯管道中的二进制流转换为对象、把对象转换成二进制流的过程。转换过程追根究底还是ChannelInboundHandler、ChannelOutboundHandler的实现类在进行处理。ChannelInboundHandler负责把二进制流转换为对象，ChannelOutboundHandler负责把对象转换为二进制流。

接下来要构建一个Server，同时支持Person通讯协议和String通讯协议。

Person通讯协议：二进制流与Person对象间的互相转换。

String通讯协议：二进制流与有固定格式要求的String的相互转换。String格式表示的也是一个Person对象，格式规定为：name:xx;age:xx;sex:xx;

这时候，来自客户端的请求，会依次传递给两个通讯解析接口进行解析，每个通讯接口判断是否是匹配的协议，如果是则进行解析，如果不是则传递给其它通讯接口进行解析。

实体类Person.java

package com.bijian.netty.dto;
    
    import java.io.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、StringDecoder.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;
    
    // 测试coder 和 handler 的混合使用  
    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 StringDecoder());
                                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把二进制流转换成Person对象

3.StringDecoder.java把满足条件的字符串转换成Person对象

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.dto.Person;
    import com.bijian.netty.util.ByteBufToBytes;
    
    public class StringDecoder extends ByteToMessageDecoder {
        
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
            
            // 判断是否是String协议
            byte n = "n".getBytes()[0];
            byte p = in.readByte();
            // 把读取的起始位置重置
            in.resetReaderIndex();
            if (n == p) {
                ByteBufToBytes reader = new ByteBufToBytes();
                String msg = new String(reader.read(in));
                Person person = buildPerson(msg);
                out.add(person);
                //in.release();
            } else {
                ctx.fireChannelRead(in);
            }
        }
    
        private Person buildPerson(String msg) {
            
            Person person = new Person();
            String[] msgArray = msg.split(";|:");
            person.setName(msgArray[1]);
            person.setAge(Integer.parseInt(msgArray[3]));
            person.setSex(msgArray[5]);
            return person;
        }
    }

4.BusinessHandler.java展现客户端请求的内容

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);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            ctx.flush();
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            ctx.close();
        }
    }

客户端1发送Person格式的协议：Client.java、ClientInitHandler.java、PersonEncoder.java

1.Client.java

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;
    
    import com.bijian.netty.dto.Person;
    
    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());
                        Person person = new Person();
                        person.setName("test");
                        person.setSex("man");
                        person.setAge(30);
                        ch.pipeline().addLast(new ClientInitHandler(person));
                    }
                });
    
                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向服务端发送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);
        private Person person;
    
        public ClientInitHandler(Person person) {
            this.person = person;
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            logger.info("ClientInitHandler.channelActive");
            ctx.write(person);
            ctx.flush();
        }
    }

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

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 {
            out.writeBytes(ByteObjConverter.ObjectToByte(msg));
        }
    }

客户端2发送String格式的协议：Client2.java、StringEncoder.java同样使用了客户端1中定义的ClientInitHandler 进行数据发送操作。

1.Client2.java

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;
    
    import com.bijian.netty.dto.Person;
    
    public class Client2 {
        
        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 StringEncoder());
                        Person person = new Person();
                        person.setName("test2");
                        person.setSex("girl");
                        person.setAge(4);
                        ch.pipeline().addLast(new ClientInitHandler(person));
                    }
                });
    
                ChannelFuture f = b.connect(host, port).sync();
                f.channel().closeFuture().sync();
            } finally {
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws Exception {
            Client2 client = new Client2();
            client.connect("127.0.0.1", 8000);
        }
    }

2.StringEncoder.java把Person对象转换成固定格式的String的二进制流进行传送

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;
    
    public class StringEncoder extends MessageToByteEncoder<Person> {
    
        @Override
        protected void encode(ChannelHandlerContext ctx, Person msg, ByteBuf out) throws Exception {
            // 转成字符串：name:xx;age:xx;sex:xx;
            StringBuffer sb = new StringBuffer();
            sb.append("name:").append(msg.getName()).append(";");
            sb.append("age:").append(msg.getAge()).append(";");
            sb.append("sex:").append(msg.getSex()).append(";");
            out.writeBytes(sb.toString().getBytes());
        }
    }

其它：工具类ByteBufToBytes（读取ByteBuf数据的工具类）、ByteObjConverter（Object与byte互转的工具类）在以前的文章中已经存在，在此省略。

**注意事项：**

1.该段代码能运行出结果，但是运行的时候会报 io.netty.util.IllegalReferenceCountException: refCnt: 0, decrement: 1 异常，已经解决。日志中的提示信息为：

An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception

说明缺少exceptionCaught方法，在server端最后一个Handler中增加这个方法即可。

2.PersonDecoder和StringDecoder中有一个if判断，是为了判断消息究竟是什么协议。如果是String协议的话，格式是【name:xx;age:xx;sex:xx;】，第一个字母是英文字母n，所以判断协议类型时候是读取二进制流的第一个字符进行判断，当然这种判断方式非常幼稚，以后有机会可以进行改善。

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

[http_blog.csdn.net_u013252773_article_details_22108385]: http://blog.csdn.net/u013252773/article/details/22108385