Netty之协议设计

谁践踏了优雅 2023-10-11 23:58 21阅读 0赞

**目录**

为什么需要协议

redis协议示例

http协议举例

自定义协议

要素

编解码器

测试

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

### 为什么需要协议 ###

TCP/IP 中消息传输基于流的方式，没有边界。

协议的目的就是划定消息的边界，制定通信双方要共同遵守的通信规则

例如：在网络上传输

下雨天留客天留我不留

是中文一句著名的无标点符号句子，在没有标点符号情况下，这句话有数种拆解方式，而意思却是完全不同，所以常被用作讲述标点符号的重要性

一种解读

下雨天留客，天留，我不留

另一种解读

下雨天，留客天，留我不？留

如何设计协议呢？其实就是给网络传输的信息加上“标点符号”。但通过分隔符来断句不是很好，因为分隔符本身如果用于传输，那么必须加以区分。因此，下面一种协议较为常用

定长字节表示内容长度 + 实际内容

例如，假设一个中文字符长度为 3，按照上述协议的规则，发送信息方式如下，就不会被接收方弄错意思了

0f下雨天留客06天留09我不留

#### redis协议示例 ####

>     *3
>     $3
>     SET
>     $5
>     mykey
>     $7
>     myvalue
> 
> 对于上面的内容发出的命令为SET mykey myvalue
> 
> \*3表示要发三个数组
> 
> $3表示第一个数组的长度为3
> 
> 后接具体的指令为SET
> 
> 后面的$5 $7也是同样的道理

public class redisHttp {
        public static void main(String[] args) {
            NioEventLoopGroup worker = new NioEventLoopGroup();
            byte[] LINE={13,10};
            try {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.channel(NioSocketChannel.class);
                bootstrap.group(worker);
                bootstrap.handler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new LoggingHandler());
                        ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                            public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                set(ctx);
                                get(ctx);
                            };
                            private void get(ChannelHandlerContext ctx){
                                ByteBuf buf = ctx.alloc().buffer();
                                buf.writeBytes("*2".getBytes());
                                buf.writeBytes(LINE);
                                buf.writeBytes("$3".getBytes());
                                buf.writeBytes(LINE);
                                buf.writeBytes("get".getBytes());
                                buf.writeBytes(LINE);
                                buf.writeBytes("aaa".getBytes());
                                buf.writeBytes(LINE);
                                ctx.writeAndFlush(buf);
                            }
    
                            private  void  set(ChannelHandlerContext ctx){
                                ByteBuf buf = ctx.alloc().buffer();
                                buf.writeBytes("*3".getBytes());
                                buf.writeBytes(LINE);
                                buf.writeBytes("$3".getBytes());
                                buf.writeBytes(LINE);
                                buf.writeBytes("set".getBytes());
                                buf.writeBytes(LINE);
                                buf.writeBytes("$3".getBytes());
                                buf.writeBytes(LINE);
                                buf.writeBytes("aaa".getBytes());
                                buf.writeBytes(LINE);
                                buf.writeBytes("xxx".getBytes());
                                buf.writeBytes(LINE);
                                ctx.writeAndFlush(buf);
                            }
                        });
                    }
                });
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 6379);
                channelFuture.channel().closeFuture().sync();
    
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                worker.shutdownGracefully();
            }
        }
    }

#### http协议举例 ####

public class httpSimple {
        static final Logger log = LoggerFactory.getLogger(HelloWordServer.class);
        public static void main(String[] args) {
           
            NioEventLoopGroup boss=new NioEventLoopGroup();
            NioEventLoopGroup worker=new NioEventLoopGroup();
            try {
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.channel(NioServerSocketChannel.class);
                serverBootstrap.group(boss,worker);
                serverBootstrap.childHandler(new ChannelInitializer<SocketChannel>() {
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(new LoggingHandler(LogLevel.DEBUG));
                        ch.pipeline().addLast(new HttpServerCodec());
                        ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {
                            @Override
                            protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
                                //获取请求
                                log.debug(msg.uri());
                                // 返回响应
                                DefaultFullHttpResponse response = new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
                                byte[] bytes = "<h1>Hello,world!</h1>".getBytes();
                                response.headers().setInt(CONTENT_LENGTH,bytes.length);
                                response.content().writeBytes(bytes);
                                //写回响应
                                ctx.writeAndFlush(response);
                            }
                        });
    
                    }
                });
                ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
                channelFuture.channel().closeFuture().sync();
    
            }catch (Exception e){
                e.printStackTrace();
            }finally {
                boss.shutdownGracefully();
                worker.shutdownGracefully();
            }
    
        }
    }

启动服务端在浏览器中输入[localhost:8080][localhost_8080]

![d994e85a018f4502bad059d0cfbc5436.png][] ![3eabc9b1d527445d802210a9ff039a73.png][]

### 自定义协议 ###

#### 要素 ####

*  魔数，用来在第一时间判定是否是无效数据包
 *  版本号，可以支持协议的升级
 *  序列化算法，消息正文到底采用哪种序列化反序列化方式，可以由此扩展，例如：json、protobuf、hessian、jdk
 *  指令类型，是登录、注册、单聊、群聊... 跟业务相关
 *  请求序号，为了双工通信，提供异步能力
 *  正文长度
 *  消息正文

#### 编解码器 ####

@Slf4j
    public class MessageCodec extends ByteToMessageCodec<Message> {
    
        @Override
        protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
         
        }
        
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    
        }
    }

重写编码方法

@Override
        protected void encode(ChannelHandlerContext ctx, Message msg, ByteBuf out) throws Exception {
      // 1. 4 字节的魔数
            out.writeBytes(new byte[]{1, 2, 3, 4});
            // 2. 1 字节的版本,
            out.writeByte(1);
            // 3. 1 字节的序列化方式 jdk 0 , json 1
            out.writeByte(0);
            // 4. 1 字节的指令类型
            out.writeByte(msg.getMessageType());
            // 5. 4 个字节
            out.writeInt(msg.getSequenceId());
            // 无意义，对齐填充
            out.writeByte(0xff);
            // 6. 获取内容的字节数组
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(bos);
            oos.writeObject(msg);
            byte[] bytes = bos.toByteArray();
            // 7. 长度
            out.writeInt(bytes.length);
            // 8. 写入内容
            out.writeBytes(bytes);
         
        }

用于将自定义Message对象编码成二进制数据流发送给远程服务器。具体解释如下：

1.  4字节的魔数：这个魔数是用来标志协议的，客户端和服务端都要保持一致，表示这是同一种协议。
2.  1字节的版本：表示当前数据流的版本号。
3.  1字节的序列化方式：表示使用哪种序列化方式将Message对象转为二进制数据流，其中0代表JDK序列化方式，1代表JSON序列化方式。
4.  1字节的指令类型：表示Message对象中的指令类型，也就是表示这个消息是干什么用的。
5.  4字节的序列号：表示该消息的序列号，用于检测是否有消息丢失或重复等问题。
6.  无意义，8位填充：由于前面魔数、版本、序列化方式、指令类型、序列号已经使用了12个字节的长度，而长度字段需要占用4个字节的长度，为了对其，需要在这里填充一个字节，使得总长度为13个字节。
7.  4字节的消息体长度：表示消息体的长度。
8.  消息内容：将Message对象序列化为字节数组，再写到输出流中。

最终，这个编码器将Message对象转化为了一个二进制数据流，方便通过网络传输到远程服务器。

重写解码方法

@Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
          int magicNum = in.readInt();
            byte version = in.readByte();
            byte serializerType = in.readByte();
            byte messageType = in.readByte();
            int sequenceId = in.readInt();
            in.readByte();
            int length = in.readInt();
            byte[] bytes = new byte[length];
            in.readBytes(bytes, 0, length);
            ObjectInputStream ois = new ObjectInputStream(new ByteArrayInputStream(bytes));
            Message message = (Message) ois.readObject();
            log.debug("{}, {}, {}, {}, {}, {}", magicNum, version, serializerType, messageType, sequenceId, length);
            log.debug("{}", message);
            out.add(message);
        }

用于将接收到的二进制数据流解码成自定义的Message对象。具体解释如下：

1.  读取4字节的魔数。
2.  读取1字节的版本。
3.  读取1字节的序列化方式。
4.  读取1字节的指令类型。
5.  读取4字节的序列号。
6.  读取1字节，这个字节被视为无意义填充。
7.  读取4字节的消息体长度，也就是消息内容的字节长度。
8.  根据消息体长度创建一个字节数组，并从输入流中读取相应的字节数据。
9.  将字节数组反序列化成一个Message对象。
10. 输出相应的日志信息，包括魔数、版本、序列化方式、指令类型、序列号、消息体长度以及反序列化后的Message对象。
11. 把反序列化后的Message对象添加到out列表中。

最终，这个解码器将二进制数据流转化为了自定义的Message对象，方便在业务逻辑中使用。

#### 测试 ####

EmbeddedChannel channel = new EmbeddedChannel(
        new LoggingHandler(),
        new LengthFieldBasedFrameDecoder(
            1024, 12, 4, 0, 0),
        new MessageCodec()
    );
    // encode
    LoginRequestMessage message = new LoginRequestMessage("zhangsan", "123", "张三");
    //        channel.writeOutbound(message);
    // decode
    ByteBuf buf = ByteBufAllocator.DEFAULT.buffer();
    new MessageCodec().encode(null, message, buf);
    
    ByteBuf s1 = buf.slice(0, 100);
    ByteBuf s2 = buf.slice(100, buf.readableBytes() - 100);
    s1.retain(); // 引用计数 2
    channel.writeInbound(s1); // release 1
    channel.writeInbound(s2);

![392c26ebfe4c423392532b8a9d3341cf.png][]

[localhost_8080]: http://localhost:8080/
[d994e85a018f4502bad059d0cfbc5436.png]: https://img-blog.csdnimg.cn/d994e85a018f4502bad059d0cfbc5436.png
[3eabc9b1d527445d802210a9ff039a73.png]: https://img-blog.csdnimg.cn/3eabc9b1d527445d802210a9ff039a73.png
[392c26ebfe4c423392532b8a9d3341cf.png]: https://img-blog.csdnimg.cn/392c26ebfe4c423392532b8a9d3341cf.png