事件在ChannelHandler的传播顺序

女爷i 2022-09-12 09:48 92阅读 0赞

pipeline是由ChannelHandlerContext组成的双向链表，事件在入站处理器上是从前往后传播，在出站处理器上是从后往前传播。

通过ChannelHandlerContext调用fireXXX方法，事件传递给当前Handler后面的下一个入站处理器，如果通过pipeline调用fireXXX方法，则事件传递给管道中从队首开始往后的第一个入站处理器。

通过ChannelHandlerContext调用write等方法，事件传递给当前Handler前面的一个出站处理器，如果通过pipeline或者channel调用write等方法，则事件传递给管道从队尾开始往前的第一个出站处理器。

**当入站处理器在出站处理器前面时**：

处理器顺序：
    
    InboundleHandlerA（入A）->InboundleHandlerB（入B）->OutboundleHandlerA（出A）->OutboundleHandlerB（出B）

package com.tech.netty.echo;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoServer {
    
        private int port;
    
        public EchoServer(int port) {
            this.port = port;
        }
    
        /**
         * 启动流程
         * @throws InterruptedException
         */
        public void run() throws InterruptedException {
            //配置服务端线程组
            NioEventLoopGroup bossGroup = new NioEventLoopGroup();
            NioEventLoopGroup workerGroup = new NioEventLoopGroup();
            try{
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup,workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
    //                            ch.pipeline().addLast(new EchoServerHandler());
    
                                ch.pipeline().addLast(new InboundleHandlerA());
                                ch.pipeline().addLast(new InboundleHandlerB());
                                ch.pipeline().addLast(new OutboundleHandlerA());
                                ch.pipeline().addLast(new OutboundleHandlerB());
    
                            }
                        });
                System.out.println("Echo 服务器启动ing");
                //绑定端口 同步等待成功
                ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
                //等待服务端监听端口关闭
                channelFuture.channel().closeFuture().sync();
            }finally {
                //优雅退出释放线程池
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            int port=8080;
            if(args.length>0){
                port=Integer.parseInt(args[0]);
            }
            new EchoServer(port).run();
        }
    }

执行结果：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_20_color_FFFFFF_t_70_g_se_x_16][]

package com.tech.netty.echo;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    /**
     * @author lw
     * @since 2021/9/23
     */
    public class InboundleHandlerB extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("InboundleHandlerB");
            //会将消息传递给InboundleHandlerB前面的一个出站处理器，由于前面没有出站处理器，
            // 所以OutboundleHandlerA OutboundleHandlerB都不会处理
    //        ctx.writeAndFlush(msg);
            //使用channel或者pipeline调用writeAndFlush(msg)，事件会传递到OutboundleHandlerB
            ctx.pipeline().writeAndFlush(msg);
        }
    }

package com.tech.netty.echo;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.channel.ChannelOutboundHandlerAdapter;
    import io.netty.channel.ChannelPromise;
    
    /**
     * @author lw
     * @since 2021/9/23
     */
    public class OutboundleHandlerB extends ChannelOutboundHandlerAdapter {
    
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            System.out.println("OutboundleHandlerB");
            //由于OutboundleHandlerB没有调用ctx.writeAndFlush 所以没有传递给OutboundleHandlerA
        }
    }

执行结果：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_11_color_FFFFFF_t_70_g_se_x_16][]

package com.tech.netty.echo;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.channel.ChannelOutboundHandlerAdapter;
    import io.netty.channel.ChannelPromise;
    
    /**
     * @author lw
     * @since 2021/9/23
     */
    public class OutboundleHandlerA extends ChannelOutboundHandlerAdapter {
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            System.out.println("OutboundleHandlerA");
        }
    }

执行结果：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_14_color_FFFFFF_t_70_g_se_x_16][]

**当出站处理器在入站处理器前面时**：

处理器顺序：
    
    OutboundleHandlerA（出A）->OutboundleHandlerB（出B）->InboundleHandlerA（入A）->InboundleHandlerB（入B）

package com.tech.netty.echo;
    
    import io.netty.bootstrap.ServerBootstrap;
    import io.netty.channel.ChannelFuture;
    import io.netty.channel.ChannelInitializer;
    import io.netty.channel.nio.NioEventLoopGroup;
    import io.netty.channel.socket.SocketChannel;
    import io.netty.channel.socket.nio.NioServerSocketChannel;
    
    /**
     * @author lw
     * @since 2021/9/17
     */
    public class EchoServer {
    
        private int port;
    
        public EchoServer(int port) {
            this.port = port;
        }
    
        /**
         * 启动流程
         * @throws InterruptedException
         */
        public void run() throws InterruptedException {
            //配置服务端线程组
            NioEventLoopGroup bossGroup = new NioEventLoopGroup();
            NioEventLoopGroup workerGroup = new NioEventLoopGroup();
            try{
                ServerBootstrap serverBootstrap = new ServerBootstrap();
                serverBootstrap.group(bossGroup,workerGroup)
                        .channel(NioServerSocketChannel.class)
                        .childHandler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel ch) throws Exception {
    //                            ch.pipeline().addLast(new EchoServerHandler());
                                ch.pipeline().addLast(new OutboundleHandlerA());
                                ch.pipeline().addLast(new OutboundleHandlerB());
                                ch.pipeline().addLast(new InboundleHandlerA());
                                ch.pipeline().addLast(new InboundleHandlerB());
                            }
                        });
                System.out.println("Echo 服务器启动ing");
                //绑定端口 同步等待成功
                ChannelFuture channelFuture = serverBootstrap.bind(port).sync();
                //等待服务端监听端口关闭
                channelFuture.channel().closeFuture().sync();
            }finally {
                //优雅退出释放线程池
                bossGroup.shutdownGracefully();
                workerGroup.shutdownGracefully();
            }
        }
    
        public static void main(String[] args) throws InterruptedException {
            int port=8080;
            if(args.length>0){
                port=Integer.parseInt(args[0]);
            }
            new EchoServer(port).run();
        }
    }

package com.tech.netty.echo;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    
    /**
     * @author lw
     * @since 2021/9/23
     */
    public class InboundleHandlerB extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            System.out.println("InboundleHandlerB");
            // 事件传递给当前处理器往前的第一个出站处理器 outboundleHandlerB
            ctx.writeAndFlush(msg);
        }
    }

package com.tech.netty.echo;
    
    import io.netty.channel.ChannelHandlerContext;
    import io.netty.channel.ChannelInboundHandlerAdapter;
    import io.netty.channel.ChannelOutboundHandlerAdapter;
    import io.netty.channel.ChannelPromise;
    
    /**
     * @author lw
     * @since 2021/9/23
     */
    public class OutboundleHandlerA extends ChannelOutboundHandlerAdapter {
        @Override
        public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
            System.out.println("OutboundleHandlerA");
        }
    }

执行结果：

![watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_15_color_FFFFFF_t_70_g_se_x_16][]

[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_20_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/cc8cadf9e8e745f88e0bba0cc632f090.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_11_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/1aa550d3053f4235ad41bdc5b70d0ab9.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_14_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/e70aee90cbd04e60ae657027d55f10ec.png
[watermark_type_ZHJvaWRzYW5zZmFsbGJhY2s_shadow_50_text_Q1NETiBA6KaB5LqJ5rCU_size_15_color_FFFFFF_t_70_g_se_x_16]: /images/20220828/7d9c7f0442cb4a49a14c7eacdec37e44.png