Netty出入站机制与编解码器

刺骨的言语ヽ痛彻心扉 2022-09-09 00:23 231阅读 0赞

从Socket读取数据到客户端/服务器为入站操作，入站操作管道中事件传播顺序是从前往后；将客户端/服务器数据写入到Socket为出站操作，出站操作管道中事件传播顺序是从后往前。编码器为出站处理器，将业务数据进行编码，使用编码器收到的数据类型必须与处理的类型一致，否则encode方法不会被调用；解码器为入站处理器，对编码后的数据进行解码为我们的业务数据，解码器需要判断ByteBuf中可读取的字节数是否足够。

出入站操作：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_20_color_FFFFFF_t_70_g_se_x_16][]

接下来实现通过自定义编码器解码器实现字节码和Long类型的转换，客户端发送long类型数据给服务端，服务端返回long数据给客户端

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_20_color_FFFFFF_t_70_g_se_x_16 1][]

package com.tech.netty.netty.inboundleandoutboudle;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /**
     * @author lw
     * @since 2021/8/27
     */
    public class MyServer {
        public static void main(String[] args) throws InterruptedException {
            NioEventLoopGroup bossGroup = new NioEventLoopGroup(1);
            NioEventLoopGroup workerGroup = new NioEventLoopGroup();
            try{
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup,workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new MyServerInitializer());
                ChannelFuture channelFuture = serverBootstrap.bind(7000).sync();
                channelFuture.channel().closeFuture().sync();
            }finally {
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    }

package com.tech.netty.netty.inboundleandoutboudle;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    /**
     * SimpleChannelInboundHandler<Long> 该处理器只读取long 其他类型不读取
     * @author lw
     * @since 2021/8/27
     */
    public class MyServerHandler extends SimpleChannelInboundHandler<Long> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
            System.out.println("从客户端"+ctx.channel().remoteAddress()+"读取到long"+msg);
            //给客户端发送一个long
            ctx.writeAndFlush(98765L);
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }

package com.tech.netty.netty.inboundleandoutboudle;
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.string.StringDecoder;
    
    /**
     * @author lw
     * @since 2021/8/27
     */
    public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            //收发数据时，如果不设置编码和解码器就需要用ByteBuf发送和接收数据，
            //收发数据时，如果设置编解码器，则编码解码器要使用相同类型和字节码进行转换
            //入站的handler进行解码
            pipeline.addLast("MyByteToLongDecoder",new MyByteToLongDecoder())
                    //出站的handler进行编码
                    .addLast("MyLongToByteEncoder",new MyLongToByteEncoder())
                    //自定义的handler处理业务逻辑
                    .addLast("MyServerHandler",new MyServerHandler());
        }
    }

package com.tech.netty.netty.inboundleandoutboudle;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.MessageToByteEncoder;
    
    /**
     * encoder 传入的类型要和处理的类型Long保持一致 否则不会调用encode方法
     * @author lw
     * @since 2021/8/27
     */
    public class MyLongToByteEncoder extends MessageToByteEncoder<Long> {
        //编码方法
        @Override
        protected void encode(ChannelHandlerContext ctx, Long msg, ByteBuf out) throws Exception {
            System.out.println("MyLongToByteEncoder 被调用");
            System.out.println("msg="+msg);
            out.writeLong(msg);
        }
    }

package com.tech.netty.netty.inboundleandoutboudle;
    
    import io.netty.bootstrap.Bootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioSocketChannel;
    
    /**
     * @author lw
     * @since 2021/8/27
     */
    public class MyClient {
        public static void main(String[] args) throws InterruptedException {
            EventLoopGroup group = new NioEventLoopGroup();
    
            try{
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group)
                        .channel(NioSocketChannel.class)
                        .handler(new MyClientInitializer()); //自定义一个初始化类
                ChannelFuture channelFuture = bootstrap.connect("127.0.0.1", 7000).sync();
                channelFuture.channel().closeFuture().sync();
            }finally {
                group.shutdownGracefully();
            }
        }
    }

package com.tech.netty.netty.inboundleandoutboudle;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    /**
     * @author lw
     * @since 2021/8/27
     */
    public class MyClientHandler extends SimpleChannelInboundHandler<Long> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
            System.out.println("服务器地址 "+ctx.channel().remoteAddress());
            System.out.println("收到服务器消息 msg="+msg);
        }
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("MyClientHandler 发送数据");
            ctx.writeAndFlush(123456L); //发送的是一个long
            //分析
            //1. abcdabcdabcdabcd是16个字节
            //2. 该处理器的前一个handler是MessageToByteEncoder
            //3. MessageToByteEncoder父类是MessageToByteEncoder
            /*
             public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            ByteBuf buf = null;
            try {
                if (acceptOutboundMessage(msg)) { //判断当前msg 是不是应该处理的类型，如果是就处理，不是就跳过encode
                    @SuppressWarnings("unchecked")
                    I cast = (I) msg;
                    buf = allocateBuffer(ctx, cast, preferDirect);
                    try {
                        encode(ctx, cast, buf);
                    } finally {
                        ReferenceCountUtil.release(cast);
                    }
    
                    if (buf.isReadable()) {
                        ctx.write(buf, promise);
                    } else {
                        buf.release();
                        ctx.write(Unpooled.EMPTY_BUFFER, promise);
                    }
                    buf = null;
                } else {
                    ctx.write(msg, promise);
                }
            }
             */
            //因此我们编写encoder传入的类型要和处理的类型一致 否则不会调用MessageToByteEncoder的encode方法，
            // 如下将string添加到缓冲区中进行发送，不会调用MessageToByteEncoder的encode方法
    //        ctx.writeAndFlush(Unpooled.copiedBuffer("abcdabcdabcdabcd",CharsetUtil.UTF_8));
        }
    }

package com.tech.netty.netty.inboundleandoutboudle;
    
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.handler.codec.string.StringEncoder;
    
    /**
     * @author lw
     * @since 2021/8/27
     */
    public class MyClientInitializer extends ChannelInitializer<SocketChannel> {
    
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();
            //加入一个出站的handler 对数据进行编码
            pipeline.addLast(new MyLongToByteEncoder())
                    //加入一个入站的解码器
                    .addLast(new MyByteToLongDecoder())
                    //加入一个自定义的handler处理业务
                    .addLast(new MyClientHandler());
        }
    }

package com.tech.netty.netty.inboundleandoutboudle;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ByteToMessageDecoder;
    
    import java.util.List;
    
    /**
     * @author lw
     * @since 2021/8/27
     */
    public class MyByteToLongDecoder extends ByteToMessageDecoder {
    
        /**
         * 解码器
         * decode方法会根据接收的数据被多次调用，直到list中没有新的元素加入或者入站缓冲区没有可读的字节为止
         * （例子中当可读取字节不足8时候decode会被重复调用几次才停止，只要list没有元素就不会调用后面的入站handler）
         * 只要list不为空就会将list中元素逐一传递给下一个入站处理器，有一个元素下一个处理器就会被调用一次
         * @param ctx 上下文对象
         * @param in 入站的ByteBuf
         * @param out List集合 将解码后的数据传递给下一个入站Handler
         * @throws Exception
         */
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
            //因为long有8个字节 需要判断够8个字节可读才进行数据读取
            System.out.println("MyByteToLongDecoder 被调用 可读取字节数="+in.readableBytes());
            if(in.readableBytes()>=8){
                out.add(in.readLong());
            }
        }
    }

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220829/0fc1ae060c2847afa8759aa0428a19db.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_20_color_FFFFFF_t_70_g_se_x_16 1]: /images/20220829/1939c3b56a484bf6a40b0a345a050b6e.png