美文网首页
Netty中inboundHandler与outboundHan

Netty中inboundHandler与outboundHan

作者: lietobrain | 来源:发表于2020-04-06 22:36 被阅读0次

    一个疑问

    首先一切的一切,是从一次意外开始。

    在写一个netty的server的时候,这里有四个handler,inboundHandler实现的类EchoInHandler1与EchoInHandler2,outboundHandler实现的类EchoOutHandler1与EchoOutHandler2;

    在添加到pipeline的时候,如果这些handler的存放到pipeline的位置为EchoOutHandler1-EchoOutHandler2-EchoInHandler1-EchoInHandler2,那么一切就正常了。

    开始监听,端口为:/127.0.0.1:20000
    in1
    in2
    接收客户端数据:QUERY TIME ORDER
    server向client发送数据
    out2
    out1
    Complete1
    

    但是如果存放的顺序是EchoInHandler1-EchoInHandler2-EchoOutHandler1-EchoOutHandler2,那么会出现在出站的时候,EchoOutHandler1与EchoOutHandler2却没有执行。

    开始监听,端口为:/127.0.0.1:20000
    in1
    in2
    接收客户端数据:QUERY TIME ORDER
    server向client发送数据
    Complete1
    

    这是为什么呢?
    PS:如果想知道答案可以直接看最后一节

    public void start() throws Exception {
        EventLoopGroup eventLoopGroup = null;
        try {
            //server端引导类
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            //连接池处理数据
            eventLoopGroup = new NioEventLoopGroup();
            serverBootstrap.group(eventLoopGroup)
                .channel(NioServerSocketChannel.class)
                //指定通道类型为NioServerSocketChannel,一种异步模式,OIO阻塞模式为OioServerSocketChannel
                .localAddress("localhost",port)
                //设置InetSocketAddress让服务器监听某个端口已等待客户端连接。
                .childHandler(new ChannelInitializer<Channel>() {
                    //设置childHandler执行所有的连接请求
                    @Override
                    protected void initChannel(Channel ch) throws Exception {
               // 注册两个InboundHandler,执行顺序为注册顺序,所以应该是InboundHandler1 InboundHandler2
               // 注册两个OutboundHandler,执行顺序为注册顺序的逆序,所以应该是OutboundHandler2 OutboundHandler1
                        ch.pipeline().addLast(new EchoInHandler1());
                        ch.pipeline().addLast(new EchoInHandler2());
                        ch.pipeline().addLast(new EchoOutHandler1());
                        ch.pipeline().addLast(new EchoOutHandler2());
                    }
                });
            // 最后绑定服务器等待直到绑定完成,调用sync()方法会阻塞直到服务器完成绑定,
            // 然后服务器等待通道关闭,因为使用sync(),所以关闭操作也会被阻塞。
            ChannelFuture channelFuture = serverBootstrap.bind().sync();
            System.out.println("开始监听,端口为:" + channelFuture.channel().localAddress());
            channelFuture.channel().closeFuture().sync();
        } finally {
            eventLoopGroup.shutdownGracefully().sync();
        }
    }
    

    EchoInHandler1

    package com.aguicode.practice.netty.mutilhandler;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    /**
     * @author aguicode
     * @since 2020-3-8
     */
    public class EchoInHandler1 extends ChannelInboundHandlerAdapter {
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception {
            System.out.println("in1");
            // 通知执行下一个InboundHandler
            ctx.fireChannelRead(msg);
            //ctx.writeAndFlush(msg);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            System.out.println("Complete1");
            //ctx.flush();//刷新后才将数据发出到SocketChannel
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }
    

    EchoInHandler2

    package com.aguicode.practice.netty.mutilhandler;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    import java.util.Date;
    
    /**
     * @author aguicode
     * @since 2020-3-8
     */
    public class EchoInHandler2 extends ChannelInboundHandlerAdapter {
    
    
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg)
                throws Exception {
            System.out.println("in2");
            ByteBuf buf = (ByteBuf) msg;
            byte[] req = new byte[buf.readableBytes()];
            buf.readBytes(req);
            String body = new String(req, "UTF-8");
            System.out.println("接收客户端数据:" + body);
            //向客户端写数据
            System.out.println("server向client发送数据");
            String currentTime = new Date(System.currentTimeMillis()).toString();
            ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
            //ctx.write(resp);
            ctx.writeAndFlush(resp);
        }
    
        @Override
        public void channelReadComplete(ChannelHandlerContext ctx) throws Exception {
            System.out.println("Complete2");
            //ctx.flush();//刷新后才将数据发出到SocketChannel
        }
    
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
                throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    
    }
    

    EchoOutHandler1

    package com.aguicode.practice.netty.mutilhandler;
    
    import io.netty.buffer.ByteBuf;
    import io.netty.buffer.Unpooled;
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelOutboundHandlerAdapter;
    import io.netty.channel.ChannelPromise;
    
    import java.util.Date;
    
    /**
     * @author aguicode
     * @since 2020-3-8
     */
    public class EchoOutHandler1 extends ChannelOutboundHandlerAdapter {
    
        @Override
        // 向client发送消息
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            System.out.println("out1");
            /*System.out.println(msg);*/
    
            String response = "\nI am ok!\n";
            ByteBuf encoded = ctx.alloc().buffer(4 * response.length());
            encoded.writeBytes(response.getBytes());
    
            String currentTime = new Date(System.currentTimeMillis()).toString();
            ByteBuf resp = Unpooled.copiedBuffer(currentTime.getBytes());
            ctx.write(resp);
            ctx.writeAndFlush(encoded);
            ctx.flush();
        }
    }
    
    

    EchoOutHandler2

    package com.aguicode.practice.netty.mutilhandler;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelOutboundHandlerAdapter;
    import io.netty.channel.ChannelPromise;
    
    /**
     * @author aguicode
     * @since 2020-3-8
     */
    public class EchoOutHandler2 extends ChannelOutboundHandlerAdapter {
    
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            System.out.println("out2");
            // 执行下一个OutboundHandler
                /*System.out.println("at first..msg = "+msg);
                msg = "hi newed in out2";*/
            // 通知执行下一个OutboundHandler
            super.write(ctx, msg, promise);
            super.flush(ctx);
        }
    }
    

    几个重要的概念

    工作原理 channelHandler双向链表

    netty中有以下几个重要的概念,首先是server与client,它们中有channel、channelPipeline、channelHandler、channelHandlerContext、ServerBootStrap、bootStrap、channelFuture、selector、Eventloop;

    关于Netty的组件中的介绍会安排到另外一篇详细解答,这里只是分析in与out boundHandler执行顺序

    Netty负责人演讲用的PPT

    原理解析

    channelHandler 中定义outboundhandler和inboundhandler,表示一个请求进来时通过入站inboundhandler,而内部进行一些业务的逻辑处理之后出站使用outboundhandler,

    这里handler是定义在channelPipeline里边的,handler之间是一种双向链表的关系,inBound事件从head节点传播到tail节点,outBound事件从tail节点传播到head节点。

    /**
     *                                                 I/O Request
     *                                            via {@link Channel} or
     *                                        {@link ChannelHandlerContext}
     *                                                      |
     *  +---------------------------------------------------+---------------+
     *  |                           ChannelPipeline         |               |
     *  |                                                  \|/              |
     *  |    +---------------------+            +-----------+----------+    |
     *  |    | Inbound Handler  N  |            | Outbound Handler  1  |    |
     *  |    +----------+----------+            +-----------+----------+    |
     *  |              /|\                                  |               |
     *  |               |                                  \|/              |
     *  |    +----------+----------+            +-----------+----------+    |
     *  |    | Inbound Handler N-1 |            | Outbound Handler  2  |    |
     *  |    +----------+----------+            +-----------+----------+    |
     *  |              /|\                                  .               |
     *  |               .                                   .               |
     *  | ChannelHandlerContext.fireIN_EVT() ChannelHandlerContext.OUT_EVT()|
     *  |        [ method call]                       [method call]         |
     *  |               .                                   .               |
     *  |               .                                  \|/              |
     *  |    +----------+----------+            +-----------+----------+    |
     *  |    | Inbound Handler  2  |            | Outbound Handler M-1 |    |
     *  |    +----------+----------+            +-----------+----------+    |
     *  |              /|\                                  |               |
     *  |               |                                  \|/              |
     *  |    +----------+----------+            +-----------+----------+    |
     *  |    | Inbound Handler  1  |            | Outbound Handler  M  |    |
     *  |    +----------+----------+            +-----------+----------+    |
     *  |              /|\                                  |               |
     *  +---------------+-----------------------------------+---------------+
     *                  |                                  \|/
     *  +---------------+-----------------------------------+---------------+
     *  |               |                                   |               |
     *  |       [ Socket.read() ]                    [ Socket.write() ]     |
     *  |                                                                   |
     *  |  Netty Internal I/O Threads (Transport Implementation)            |
     *  +-------------------------------------------------------------------+
    */
    

    例如在建立三次握手之后,开始读数据,从head节点发起,准确来说是head的unsafe方法发起,inbound寻找下一个inbound时,调用invokeChannelActive(next),一个个递归调用,直到最后一个inBound节点—即tail节点,并且tail节点作为尾节点,会终止inbound事件的传播,读事件就结束了,

    这个时候,经过一段业务逻辑的处理,就需要处理outbound事件,转而反向传播,outbound则调用的是writeAndFlush(),直到head节点,数据最终会落在head节点的unsafe.write方法。


    我是分割线


    执行顺序的分析

    那么原理都懂了,这里就重点分析一下inboundHandler与outboundHandler添加顺序不同,带来执行顺序的问题

    1. inbound事件在pipeline中传输方向是head->tail,即从头到尾,而且会忽略outbound事件
    invokeChannelRead(findContextInbound(MASK_CHANNEL_READ), msg);
    

    重要的是find方法

     private AbstractChannelHandlerContext findContextInbound(int mask) {
            AbstractChannelHandlerContext ctx = this;
            do {
                ctx = ctx.next;
            } while ((ctx.executionMask & mask) == 0);
            return ctx;
        }
    

    或者类似这样子:


    忽略非inbound
    1. outbound事件在pipeline传输方向正好相反,会从tail->head,即从尾到头,同时也会忽略inbound事件
     final AbstractChannelHandlerContext next = findContextOutbound(flush ?
                    (MASK_WRITE | MASK_FLUSH) : MASK_WRITE);
    

    重要的是find方法

     private AbstractChannelHandlerContext findContextOutbound(int mask) {
            AbstractChannelHandlerContext ctx = this;
            do {
                ctx = ctx.prev;
            } while ((ctx.executionMask & mask) == 0);
            return ctx;
        }
    

    或者类似这样子:


    忽略非outbound

    但是 需要关注的是:AbstractChannelHandlerContext ctx = this;

    其实AbstractChannelHandlerContext是上下文都共享的,所以,

    如果是EchoInHandler1-EchoInHandler2-EchoOutHandler1-EchoOutHandler2,那么一开始入站执行了EchoInHandler1-EchoInHandler2,因为do-while循环跳出,ctx留在了EchoInHandler2的位置,在出站的时候,在EchoInHandler2的位置反向遍历,只会遍历EchoInHandler2-EchoInHandler1,那么自然就不会去读取-EchoOutHandler1-EchoOutHandler2了。

    相反,如果是EchoOutHandler1-EchoOutHandler2-EchoInHandler1-EchoInHandler2的顺序,一开始入站ctx到了EchoInHandler2的位置,反向遍历就会经过EchoInHandler2-EchoInHandler1-EchoOutHandler2-EchoOutHandler1

    相关文章

      网友评论

          本文标题:Netty中inboundHandler与outboundHan

          本文链接:https://www.haomeiwen.com/subject/ekuiphtx.html