zoukankan      html  css  js  c++  java
  • Netty 编解码器和 handler 的调用机制

    1.基本说明
    1) netty 的组件设计: Netty 的主要组件有 ChannelEventLoopChannelFutureChannelHandlerChannelPipe
    2) ChannelHandler 充当了处理入站和出站数据的应用程序逻辑的容器。 例如, 实现 ChannelInboundHandler 接口(或
       ChannelInboundHandlerAdapter) , 你就可以接收入站事件和数据, 这些数据会被业务逻辑处理。 当要给客户端
       发 送 响 应 时 , 也 可 以 从 ChannelInboundHandler 冲 刷 数 据 。 业 务 逻 辑 通 常 写 在 一 个 或 者 多 个
       ChannelInboundHandler 中。 ChannelOutboundHandler 原理一样, 只不过它是用来处理出站数据的
    3) ChannelPipeline 提供了 ChannelHandler 链的容器。 以客户端应用程序为例, 如果事件的运动方向是从客户端到
      服务端的, 那么我们称这些事件为出站的, 即客户端发送给服务端的数据会通过 pipeline 中的一系列ChannelOutboundHandler, 并被这些 Handler 处理, 反之则称为入站的 


      2.编码解码器 

    1) Netty 发送或者接受一个消息的时候, 就将会发生一次数据转换。 入站消息会被解码: 从字节转换为另一种格式(比如 java 对象) ; 如果是出站消息, 它会被编码成字节。

    2) Netty 提供一系列实用的编解码器, 他们都实现了 ChannelInboundHadnler 或者 ChannelOutboundHandler 接口。在这些类中, channelRead 方法已经被重写了。 以入站为例, 对于每个从入站 Channel 读取的消息, 这个方法会
    被调用。 随后, 它将调用由解码器所提供的 decode()方法进行解码, 并将已经解码的字节转发给 ChannelPipeline中的下一个 ChannelInboundHandler。 

    注意:进站和出站都是想对而言


     3 解码器-ByteToMessageDecoder

    1) 关系继承图

     2) 由于不可能知道远程节点是否会一次性发送一个完整的信息, tcp 有可能出现粘包拆包的问题, 这个类会对入站数据进行缓冲, 直到它准备好被处理.
     3) 一个关于 ByteToMessageDecoder 实例分析

     


    4 Netty handler 链的调用机制

    实例要求:
    1) 使用自定义的编码器和解码器来说明 Netty handler 调用机制
      客户端发送 long -> 服务器
      服务端发送 long -> 客户端
    2) 案例演示

     3) 结论

      不论解码器 handler 还是 编码器 handler 即接收的消息类型必须与待处理的消息类型一致, 否则该 handler 不会被执行
      在解码器 进行数据解码时, 需要判断 缓存区(ByteBuf)的数据是否足够 , 否则接收到的结果会期望结果可能不一致 

    代码:

    MyServer
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.EventLoopGroup;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    public class MyServer {
        public static void main(String[] args) throws Exception{
    
            EventLoopGroup bossGroup = new NioEventLoopGroup(1);
            EventLoopGroup 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();
            }
    
        }
    }
    View Code
    MyServerInitializer
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    
    
    public class MyServerInitializer extends ChannelInitializer<SocketChannel> {
    
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
            ChannelPipeline pipeline = ch.pipeline();//一会下断点
    
            //入站的handler进行解码 MyByteToLongDecoder
            //pipeline.addLast(new MyByteToLongDecoder());
            pipeline.addLast(new MyByteToLongDecoder2());
            //出站的handler进行编码
            pipeline.addLast(new MyLongToByteEncoder());
            //自定义的handler 处理业务逻辑
            pipeline.addLast(new MyServerHandler());
            System.out.println("xx");
        }
    }
    View Code
    MyServerHandler
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    
    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();
        }
    }
    View Code
    MyClient
    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;
    
    public class MyClient {
        public static void main(String[] args)  throws  Exception{
    
            EventLoopGroup group = new NioEventLoopGroup();
    
            try {
    
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(group).channel(NioSocketChannel.class)
                        .handler(new MyClientInitializer()); //自定义一个初始化类
    
                ChannelFuture channelFuture = bootstrap.connect("localhost", 7000).sync();
    
                channelFuture.channel().closeFuture().sync();
    
            }finally {
                group.shutdownGracefully();
            }
        }
    }
    View Code
    MyClientInitializer
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.ChannelPipeline;
    import io.netty.channel.socket.SocketChannel;
    
    
    public class MyClientInitializer extends ChannelInitializer<SocketChannel> {
        @Override
        protected void initChannel(SocketChannel ch) throws Exception {
    
            ChannelPipeline pipeline = ch.pipeline();
    
            //加入一个出站的handler 对数据进行一个编码
            pipeline.addLast(new MyLongToByteEncoder());
    
            //这时一个入站的解码器(入站handler )
            //pipeline.addLast(new MyByteToLongDecoder());
            pipeline.addLast(new MyByteToLongDecoder2());
            //加入一个自定义的handler , 处理业务
            pipeline.addLast(new MyClientHandler());
    
    
        }
    }
    View Code
    MyClientHandler
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.SimpleChannelInboundHandler;
    import io.netty.util.CharsetUtil;
    
    import java.nio.charset.Charset;
    
    public class MyClientHandler  extends SimpleChannelInboundHandler<Long> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, Long msg) throws Exception {
    
            System.out.println("服务器的ip=" + ctx.channel().remoteAddress());
            System.out.println("收到服务器消息=" + msg);
    
        }
    
        //重写channelActive 发送数据
    
        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            System.out.println("MyClientHandler 发送数据");
            //ctx.writeAndFlush(Unpooled.copiedBuffer(""))
            ctx.writeAndFlush(123456L); //发送的是一个long
    
            //分析
            //1. "abcdabcdabcdabcd" 是 16个字节
            //2. 该处理器的前一个handler 是  MyLongToByteEncoder
            //3. MyLongToByteEncoder 父类  MessageToByteEncoder
            //4. 父类  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);
                }
            }
            4. 因此我们编写 Encoder 是要注意传入的数据类型和处理的数据类型一致
            */
           // ctx.writeAndFlush(Unpooled.copiedBuffer("abcdabcdabcdabcd",CharsetUtil.UTF_8));
    
        }
    }
    View Code
    MyByteToLongDecoder
    解码器
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ByteToMessageDecoder;
    
    import java.util.List;
    
    public class MyByteToLongDecoder extends ByteToMessageDecoder {
        /**
         *
         * decode 会根据接收的数据,被调用多次, 直到确定没有新的元素被添加到list
         * , 或者是ByteBuf 没有更多的可读字节为止
         * 如果list out 不为空,就会将list的内容传递给下一个 channelinboundhandler处理, 该处理器的方法也会被调用多次
         *
         * @param ctx 上下文对象
         * @param in 入站的 ByteBuf
         * @param out List 集合,将解码后的数据传给下一个handler
         * @throws Exception
         */
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    
            System.out.println("MyByteToLongDecoder 被调用");
            //因为 long 8个字节, 需要判断有8个字节,才能读取一个long
            if(in.readableBytes() >= 8) {
                out.add(in.readLong());
            }
        }
    }
    View Code

    解码器-ReplayingDecoder

     1)  public abstract class ReplayingDecoder<S> extends ByteToMessageDecoder
     2)  ReplayingDecoder 扩展了 ByteToMessageDecoder 类, 使用这个类, 我们不必调用 readableBytes()方法。 参数 S指定了用户状态管理的类型, 其中 Void 代表不需要状态管理
     3)  应用实例: 使用 ReplayingDecoder 编写解码器, 对前面的案例进行简化 [案例演示 

    MyByteToLongDecoder2
    import io.netty.buffer.ByteBuf;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.handler.codec.ReplayingDecoder;
    
    import java.util.List;
    
    public class MyByteToLongDecoder2 extends ReplayingDecoder<Void> {
        @Override
        protected void decode(ChannelHandlerContext ctx, ByteBuf in, List<Object> out) throws Exception {
    
            System.out.println("MyByteToLongDecoder2 被调用");
            //在 ReplayingDecoder 不需要判断数据是否足够读取,内部会进行处理判断
            out.add(in.readLong());
    
    
        }
    }
    View Code

    4)  ReplayingDecoder 使用方便, 但它也有一些局限性:
      1. 并 不 是 所 有 的 ByteBuf 操 作 都 被 支 持 , 如 果 调 用 了 一 个 不 被 支 持 的 方 法 , 将 会 抛 出 一 个UnsupportedOperationException
      2. ReplayingDecoder 在某些情况下可能稍慢于 ByteToMessageDecoder, 例如网络缓慢并且消息格式复杂时,
          消息会被拆成了多个碎片, 速度变慢 。


    6 其它编解码器

    1) LineBasedFrameDecoder: 这个类在 Netty 内部也有使用, 它使用行尾控制字符( 或者 ) 作为分隔符来解析数据。
    2) DelimiterBasedFrameDecoder: 使用自定义的特殊字符作为消息的分隔符。
    3) HttpObjectDecoder: 一个 HTTP 数据的解码器
    4) LengthFieldBasedFrameDecoder: 通过指定长度来标识整包消息, 这样就可以自动的处理黏包和半包消息。


    7 其它编码器


  • 相关阅读:
    AGC 044 A
    example
    python3遇到的问题
    构建开发环境
    pandas处理数据
    pandas.DataFrame对象解析
    pandas再次学习
    监督式学习
    机器学习的基础概念
    赖世雄老师的音标课,旋元佑老师的语法书
  • 原文地址:https://www.cnblogs.com/cb1186512739/p/12774529.html
Copyright © 2011-2022 走看看