Netty进阶篇——协议设计与解析

淩亂°似流年 2023-10-13 07:24 8阅读 0赞

#### 1、测试使用redis协议进行与redis的通信 ####

> Reids通信协议格式是：
> 
> *`set name zhangsan`*
> 
> *`*3 三个部分组成：命令、key、value`*
> 
> *`$3`*
> 
> *`set`*
> 
> *`$8`*
> 
> *`zhangsan`*

**代码示例**

public static void main(String[] args) {
            final byte[] LINE = {13,10};
            NioEventLoopGroup group = new NioEventLoopGroup();
            try {
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new LoggingHandler());
                                ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                                    @Override
                                    public void channelActive(ChannelHandlerContext ctx) throws Exception {
                                        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("$4".getBytes());
                                        buf.writeBytes(LINE);
                                        buf.writeBytes("name".getBytes());
                                        buf.writeBytes(LINE);
                                        buf.writeBytes("$8".getBytes());
                                        buf.writeBytes(LINE);
                                        buf.writeBytes("zhangsan".getBytes());
                                        buf.writeBytes(LINE);
                                        ctx.writeAndFlush(buf);
                                    }
                                    @Override
                                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                        ByteBuf buf = (ByteBuf) msg;
                                        System.out.println(buf.toString(Charset.defaultCharset()));
                                    }
                                });
                            }
                        });
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1",6379).sync();
                channelFuture.channel().closeFuture().sync();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                group.shutdownGracefully();
            }
        }

#### 2、HTTP协议进行通信 ####

*  `HttpServerCodec()` 对HTTP协议请求进行解析，包括了接收和发送
    
     *  结尾为Codec代表接收和发送
 *  `SimpleChannelInboundHandler`() 表示接收特定格式的HTTP请求，可以避免进一步去判断是什么请求

public static void main(String[] args) {
            NioEventLoopGroup boss = new NioEventLoopGroup();
            NioEventLoopGroup worker = new NioEventLoopGroup();
            ServerBootstrap bootstrap = new ServerBootstrap();
            try {
                bootstrap.group(boss,worker)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
                                ch.pipeline().addLast(new LoggingHandler());   //日志打印
                                ch.pipeline().addLast(new HttpServerCodec());  //接收HTTP协议的解析器
                                ch.pipeline().addLast(new SimpleChannelInboundHandler<HttpRequest>() {  //接收特定格式的HTTP协议的解析器
                                    @Override
                                    protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
    //                                   获取请求头
                                        log.info("{}",msg.uri());
    
    //                                    返回浏览器的响应对象
                                        DefaultFullHttpResponse response =
                                                new DefaultFullHttpResponse(msg.protocolVersion(), HttpResponseStatus.OK);
                                        byte[] bytes = ("<h1>Hello, world!</h1></br><h2>"+new Date() +"</h2>").getBytes();
    //                                      设置返回的消息长度，避免浏览器一直继续等待响应
                                        response.headers().setInt(CONTENT_LENGTH,bytes.length);
                                        response.content().writeBytes(bytes);
                                        ctx.writeAndFlush(response);
                                    }
                                });
    //                            不设置接收http格式时，则需要对不同的请求头进行判断，分别处理
    /*                            ch.pipeline().addLast(new ChannelInboundHandlerAdapter(){
                                    @Override
                                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                                        log.debug("******请求类型：{} **********",msg.getClass());
                                        if (msg instanceof HttpRequest){  //请求头
                                        }else if (msg instanceof HttpContent) //请求体
                                        super.channelRead(ctx, msg);
                                    }
                                });*/
                            }
                        });
                ChannelFuture channelFuture = bootstrap.bind(8888).sync();
                channelFuture.channel().closeFuture().sync();
                
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                boss.shutdownGracefully();
                worker.shutdownGracefully();
            }
        }

#### 3、自定义协议 ####

**要素**

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

> 固定字节一般需要设计成2的n次方

**常用编解码器：**

*  `MessageToMessageCodec` \*\*：\*\*可以用于在多线程环境下多个channel共享使用，可以使用`Sharable标识`
 *  `ByteToMessageCodec`：不可以用与在多线程环境下多个channel共享使用

> 两者在使用时都需要配合`new LengthFieldBasedFrameDecoder()`一起使用，去避免出现半包粘包问题

**代码实例**

@Slf4j
    public class MessageCodeC extends ByteToMessageCodec<Message> {
    //      编码
        @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:1,json:2
            out.writeByte(0);
    //        4、1字节的指令类型
            out.writeByte(msg.getMessageType());
    //        5、4个字节请求序列
            out.writeInt(msg.getSequenceId());
    //        无意义填充
            out.writeByte(0xff);
    
    //              把对象转为byte数组
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            ObjectOutputStream oos = new ObjectOutputStream(bos);
            oos.writeObject(msg);
            byte[] bytes = bos.toByteArray();
    
    //        7、4个字节的正文长度
            out.writeInt(bytes.length);
    //        8、获取内容的字节数组
            out.writeBytes(bytes);
        }
    
    //    解码
        @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);
        }
    }

测试使用

public static void main(String[] args) {
            NioEventLoopGroup boss = new NioEventLoopGroup();
            NioEventLoopGroup whrker = new NioEventLoopGroup();
            LoggingHandler logging = new LoggingHandler();
            MessageCodeC message = new MessageCodeC();
            try {
                ServerBootstrap serverBootstrap = new ServerBootstrap()
                        .group(boss, whrker)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
    
                                ch.pipeline().addLast(logging);
                                ch.pipeline().addLast(new LengthFieldBasedFrameDecoder(
                                        1024, 12, 4, 0, 0));
                                ch.pipeline().addLast(message);
                            }
                        });
                ChannelFuture channelFuture = serverBootstrap.bind(8080).sync();
                channelFuture.channel().closeFuture().sync();
    
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                boss.shutdownGracefully();
                whrker.shutdownGracefully();
            }
        }