利用Netty构建自定义协议的通信

Dear 丶 2021-11-13 23:22 256阅读 0赞

在复杂的网络世界中，各种应用之间通信需要依赖各种各样的协议，比如：`HTTP`,`Telnet`,`FTP`,`SMTP`等等。

在开发过程中，有时候我们需要构建一些适应自己业务的应用层协议，`Netty`作为一个非常优秀的网络通信框架，可以帮助我们完成自定义协议的通信。

一般而言，我们制定的协议需要两个部分：

*  Header : 协议头部，放置一些Meta信息。
 *  Content : 应用之间交互的信息主体。

添加maven坐标(使用经典的netty4):

<dependency>
                <groupId>io.netty</groupId>
                <artifactId>netty-all</artifactId>
                <version>4.1.36.Final</version>
    </dependency>

例如：

其中`Version`,`Content-Length`,`SessionName`就是Header信息，`Content`就是交互的主体。给这个协议起一个名字叫做

MyProtocol,依照 MyProtocol 协议，我们构建一个类。

package com.onyx.rpc.orange.cat.netty;
    
    // 消息的头部
    public class MyProtocolHeader {
    
        // 协议版本
        private int version;
        // 消息内容长度
        private int contentLength;
        // 服务名称
        private String serverName;
    
    
        public MyProtocolHeader(int version, int contentLength, String serverName) {
            this.version = version;
            this.contentLength = contentLength;
            this.serverName = serverName;
        }
    
        public int getVersion() {
            return version;
        }
    
        public void setVersion(int version) {
            this.version = version;
        }
    
        public int getContentLength() {
            return contentLength;
        }
    
        public void setContentLength(int contentLength) {
            this.contentLength = contentLength;
        }
    
        public String getServerName() {
            return serverName;
        }
    
        public void setServerName(String serverName) {
            this.serverName = serverName;
        }
    
    
        @Override
        public String toString() {
            return "MyProtocolHeader{" +
                    "version=" + version +
                    ", contentLength=" + contentLength +
                    ", serverName='" + serverName + '\'' +
                    '}';
        }
    }

package com.onyx.rpc.orange.cat.netty;
    
    /**
     * 消息的主体
     */
    public class MyProtocolMessage {
    
        private MyProtocolHeader myProtocolHeader;
        private String content;
    
        public MyProtocolMessage(MyProtocolHeader myProtocolHeader, String content) {
            this.myProtocolHeader = myProtocolHeader;
            this.content = content;
        }
    
        public MyProtocolHeader getMyProtocolHeader() {
            return myProtocolHeader;
        }
    
        public void setMyProtocolHeader(MyProtocolHeader myProtocolHeader) {
            this.myProtocolHeader = myProtocolHeader;
        }
    
        public String getContent() {
            return content;
        }
    
        public void setContent(String content) {
            this.content = content;
        }
    
        @Override
        public String toString() {
            return "MyProtocolMessage{" +
                    "myProtocolHeader=" + myProtocolHeader +
                    ", content='" + content + '\'' +
                    '}';
        }
    }

那么我们在`Netty`中如何去对这种自定义的协议**编码(Encode)**呢?

在`Netty`中对数据进行编码解码需要利用`Codec`组件，`Codec`组件中分为：

*  Encoder : 编码器，将**出站**的数据从一种格式转换成另外一种格式。
 *  Decoder : 解码器，将**入站**的数据从一种格式转换成另外一种格式。

package com.onyx.rpc.orange.cat.netty;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToByteEncoder;
    
    /**
     *  编码器，将出站的数据从一种格式转换成另外一种格式。
     */
    public class MyEncoder extends MessageToByteEncoder<MyProtocolMessage> {
    
        @Override
        protected void encode(ChannelHandlerContext ctx, MyProtocolMessage message, ByteBuf out) throws Exception {
    
            // 将Message转换成二进制数据
            MyProtocolHeader header = message.getMyProtocolHeader();
    
            // 这里写入的顺序就是协议的顺序.
            // 写入Header信息
            out.writeInt(header.getVersion());
            out.writeInt(message.getContent().length());
            out.writeBytes(header.getServerName().getBytes());
    
            // 写入消息主体信息
            out.writeBytes(message.getContent().getBytes());
        }
    }

package com.onyx.rpc.orange.cat.netty;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ByteToMessageDecoder;
    
    import java.util.List;
    
    /**
     * 解码器，将入站的数据从一种格式转换成另外一种格式
     */
    public class MyDecoder extends ByteToMessageDecoder {
    
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    
            // 获取协议的版本
            int version = in.readInt();
            // 获取消息长度
            int contentLength = in.readInt();
            // 获取SessionId
            byte[] sessionByte = new byte[36];
            in.readBytes(sessionByte);
            String name = new String(sessionByte);
    
            // 组装协议头
            MyProtocolHeader header = new MyProtocolHeader(version, contentLength, name);
    
            // 读取消息内容
            ByteBuf buf = in.readBytes(in.readableBytes());
            byte[] content = new byte[buf.readableBytes()];
            buf.readBytes(content);
    
            MyProtocolMessage message = new MyProtocolMessage(header, new String(content));
            out.add(message);
        }
    }

编写一个服务端启动类:

package com.onyx.rpc.orange.cat.netty;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.ChannelOption;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    import io.netty.handler.logging.LogLevel;
    import io.netty.handler.logging.LoggingHandler;
    
    public class Server {
    
        // 指定端口号
        private static final int PORT = 8888;
    
        public static void main(String args[]) throws InterruptedException {
    
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup workerGroup = new NioEventLoopGroup();
            try {
    
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                // 指定socket的一些属性
                serverBootstrap.option(ChannelOption.SO_BACKLOG, 1024);
                serverBootstrap.group(bossGroup, workerGroup)
                        .channel(NioServerSocketChannel.class)  // 指定是一个NIO连接通道
                        .handler(new LoggingHandler(LogLevel.INFO))
                        .childHandler(new ServerInitializer());
    
                // 绑定对应的端口号,并启动开始监听端口上的连接
                Channel ch = serverBootstrap.bind(PORT).sync().channel();
    
                System.out.println("自定义协议启动地址：127.0.0.1:" + PORT);
    
                // 等待关闭,同步端口
                ch.closeFuture().sync();
            } finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

package com.onyx.rpc.orange.cat.netty;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.channel.socket.SocketChannel;
    
    public class ServerInitializer extends ChannelInitializer<SocketChannel> {
    
        @Override
        protected void initChannel(SocketChannel channel) throws Exception {
    
            ChannelPipeline pipeline = channel.pipeline();
    
            // 添加编解码器, 由于ByteToMessageDecoder的子类无法使用@Sharable注解,
            // 这里必须给每个Handler都添加一个独立的Decoder.
            pipeline.addLast(new MyEncoder());
            pipeline.addLast(new MyDecoder());
    
            // 添加逻辑控制层
            pipeline.addLast(new ServerHandler());
    
        }
    
        static class ServerHandler extends SimpleChannelInboundHandler<MyProtocolMessage> {
    
            @Override
            protected void channelRead0(ChannelHandlerContext channelHandlerContext, MyProtocolMessage myProtocolMessage) throws Exception {
                // 简单地打印出server接收到的消息
                System.out.println(myProtocolMessage.toString());
            }
        }
    
    }

最后一步了,.写一个客户端类:

package com.onyx.rpc.orange.cat.netty;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.Channel;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    import java.util.UUID;
    
    public class Client {
    
        public static void main(String args[]) throws InterruptedException {
    
            EventLoopGroup group = new NioEventLoopGroup();
            try {
                Bootstrap b = new Bootstrap();
                b.group(group)
                  .channel(NioSocketChannel.class)
                  .handler(new ClientInitializer());
    
                // Start the connection attempt.
                Channel ch = b.connect("127.0.0.1", 8888).sync().channel();
    
                int version = 1;
                String name = UUID.randomUUID().toString();
                String content = "自定义的协议....哈哈";
    
                MyProtocolHeader header = new MyProtocolHeader(version, content.length(), name);
                MyProtocolMessage message = new MyProtocolMessage(header, content);
                ch.writeAndFlush(message);
    
                ch.close();
            } finally {
                group.shutdownGracefully();
            }
        }
    }

package com.onyx.rpc.orange.cat.netty;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.channel.socket.SocketChannel;
    
    public class ClientInitializer extends ChannelInitializer<SocketChannel> {
    
        @Override
        protected void initChannel(SocketChannel channel) throws Exception {
    
            ChannelPipeline pipeline = channel.pipeline();
    
            // 这里必须给每个Handler都添加一个独立的Decoder.
            pipeline.addLast(new MyEncoder());
            pipeline.addLast(new MyDecoder());
    
            // and then business logic.
            pipeline.addLast(new ClientHandler());
    
        }
    
    
        static class ClientHandler extends SimpleChannelInboundHandler<MyProtocolMessage>{
    
            @Override
            protected void channelRead0(ChannelHandlerContext ctx, MyProtocolMessage msg) throws Exception {
                System.out.println(msg);
            }
        }
    
    
    }

依次启动server 和client就可以了.