2019-05-24 pipeline 事件及异常的传播

netty中事件的传播主要包含inBound事件和outBound事件

 ChannelInboundHandler extends ChannelHandler 
 ChannelOutboundHandler extends ChannelHandler

首先我们看下ChannelInboundHandler接口，主要包含以下方法，基本都是用和连接事件相关的

    /**
     * channel 注册到NioEventLoop上的回调
     */
    void channelRegistered(ChannelHandlerContext ctx) throws Exception;
    /**
     * channel 解除注册到NioEventLoop上的回调
     */
    void channelUnregistered(ChannelHandlerContext ctx) throws Exception;
    /**
     * channel在激活之后的回调
     */
    void channelActive(ChannelHandlerContext ctx) throws Exception;
    /**
     * channel失效之后的回调
     */
    void channelInactive(ChannelHandlerContext ctx) throws Exception;
    /**
     * channel在读取数据,或者接收 到链接之后的回调,
     * 对于服务端channel,这里的msg是一个链接
     * 对于客户端channel,这里的msg是一个ByteBuf
     */
    void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception;
    /**
     * 数据读取完成后的一个回调
     */
    void channelReadComplete(ChannelHandlerContext ctx) throws Exception;
    /**
     * 用户可以自定义的一些事件
     * Gets called if an user event was triggered.
     */
    void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception;
    /**
     * 异常事件的传播
     */
    @Override
    @SuppressWarnings("deprecation")
    void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception;

而InBoundHandler有其对应的对应的实现类

class ChannelInboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelInboundHandler

接下来基于ChannelInboundHandlerAdapter讲一下read事件在pipeline中的传播流程，这是我们案例的服务端代码

public static void main(String[] args) throws Exception {
       // Configure the server.
       EventLoopGroup bossGroup = new NioEventLoopGroup(1);
       EventLoopGroup workerGroup = new NioEventLoopGroup();
       try {
           ServerBootstrap b = new ServerBootstrap();
           b.group(bossGroup, workerGroup)
                   .channel(NioServerSocketChannel.class)
                   .option(ChannelOption.SO_BACKLOG, 100)
                   .handler(new LoggingHandler(LogLevel.INFO))
                   .childHandler(new ChannelInitializer<SocketChannel>() {
                       @Override
                     /**
                     添加3个InBoundHandler
                     */
                       public void initChannel(SocketChannel ch) throws Exception {
                           ch.pipeline().addLast(new InBoundHandlerA());
                           ch.pipeline().addLast(new InBoundHandlerB());
                           ch.pipeline().addLast(new InBoundHandlerC());
                       }
                   });

           // Start the server.
           ChannelFuture f = b.bind(8007).sync();

           // Wait until the server socket is closed.
           f.channel().closeFuture().sync();
       } finally {
           // Shut down all event loops to terminate all threads.
           bossGroup.shutdownGracefully();
           workerGroup.shutdownGracefully();
       }
   }

InBoundHandlerA的实现如下:

public class InBoundHandlerA extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println(this.getClass().getName() + " read msg:" + msg);
        super.channelRead(ctx, msg);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {    
       ctx.channel().pipeline().fireChannelRead("hello");
    }
}

InBoundHandlerB,InBoundHandlerC的实现如下:

public class InBoundHandlerB extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println(this.getClass().getName() + " read msg:" + msg);
        super.channelRead(ctx, msg);
    }
}

public class InBoundHandlerC extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        System.out.println(this.getClass().getName() + " read msg:" + msg);
        super.channelRead(ctx, msg);
    }
}

启动server，本地通过telnet命令telnet localhost 8007 触发后，我们可以看到控制台会输出以下结果:

com.tyust.netty.inbound.InBoundHandlerA read msg:hello
com.tyust.netty.inbound.InBoundHandlerB read msg:hello
com.tyust.netty.inbound.InBoundHandlerC read msg:hello
17:21:48.214 [nioEventLoopGroup-3-1] DEBUG io.netty.channel.DefaultChannelPipeline - Discarded inbound message hello that reached at the tail of the pipeline. Please check your pipeline configuration.

从控制台输出的日志可以得知,read事件在pipeline中的传播是基于InBoundHandler在pipeline中的添加顺序来的.接下来从InBoundHandlerA的这行代码ctx.channel().pipeline().fireChannelRead("hello")入手,基于源码分析一下这个中间的执行流程.

代码位置:io.netty.channel.DefaultChannelPipeline#fireChannelRead

image.png

可以看出这个事件是在Head节点开始传播的。

基于前面的文章,我们都知道,实例中的代码的pipeline结构是这样的

image.png

而现在我们在IA这个节点触发了一个read事件，流程也就是这样的

image.png

根据代码可知，最开始是在head节点开始传播的，从head节点触发之后，我们继续跟代码，看到代码后会进入HeadContext进行处理

image.png

image.png

代码进入这个位置:io.netty.channel.DefaultChannelPipeline.HeadContext#channelRead

image.png

image.png

这时候会开始去找寻下一个InboundHandler，我们看下找寻的逻辑，轮训pipeline中的Handler，碰到inbound的就返回.
image.png

inbound和outbound的标示是在构建context的时候就定好了的

image.png

继续跟代码，这时候就得到了InBoundHandlerA，继续调用其invokeChannelRead，就进入了我们的InBoundHandlerA#channelRead方法

image.png

image.png

同样的，InBoundHandlerB和InBoundHandlerB的执行也是一样的思路.
最后，事件会传播到我们的TailContext节点

image.png

看下我们TailContext中的read逻辑,会打印出前面我们控制台中显示的那一段
Discarded inbound message hello that reached at the tail of the pipeline. Please check your pipeline configuration.的日志,最后msg进行回收,避免内存泄漏.

image.png

image.png

好的，我们的InBound事件的传播就分析到这里，接下来我们看outBound事件。

看到ChannelOutboundHandler接口的定义，可以看出基本都是跟IO读写相关的事件

public interface ChannelOutboundHandler extends ChannelHandler {
    void bind(ChannelHandlerContext ctx, SocketAddress localAddress, ChannelPromise promise) throws Exception;

    void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception;

    void disconnect(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;

    void close(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;

    void deregister(ChannelHandlerContext ctx, ChannelPromise promise) throws Exception;

    void read(ChannelHandlerContext ctx) throws Exception;

    void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception;

    void flush(ChannelHandlerContext ctx) throws Exception;
}

而OutBoundHandler也有其对应的对应的实现类

class ChannelOutboundHandlerAdapter extends ChannelHandlerAdapter implements ChannelOutboundHandler

接下来基于ChannelOutboundHandlerAdapter讲一下write事件在pipeline中的传播流程，基于之前的代码，我们改下添加handler的部分

...
.childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new OutBoundHandlerA());
                            ch.pipeline().addLast(new OutBoundHandlerB());
                            ch.pipeline().addLast(new OutBoundHandlerC());
                        }
                    });

InBoundHandlerA的实现如下：

public class OutBoundHandlerA extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        System.out.println(this.getClass().getName() + " write msg: " + msg);
        super.write(ctx, msg, promise);
    }

    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        ctx.executor().schedule(() -> ctx.channel().write("hello,world"), 3, TimeUnit.SECONDS);
    }
}

InBoundHandlerB，InBoundHandlerC的代码实现如下：

public class OutBoundHandlerB extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        System.out.println(this.getClass().getName() + " write msg: " + msg);
        super.write(ctx, msg, promise);
    }
}

public class OutBoundHandlerC extends ChannelOutboundHandlerAdapter {

    @Override
    public void write(ChannelHandlerContext ctx, Object msg, ChannelPromise promise) throws Exception {
        System.out.println(this.getClass().getName() + " write msg: " + msg);
        super.write(ctx, msg, promise);
    }
}

启动server，本地通过telnet命令telnet localhost 8007 触发后，我们可以看到控制台会输出以下结果：

com.tyust.netty.outbound.OutBoundHandlerC write msg: hello,world
com.tyust.netty.outbound.OutBoundHandlerB write msg: hello,world
com.tyust.netty.outbound.OutBoundHandlerA write msg: hello,world

从控制台输出的日志可以得知, write事件在pipeline中的传播是基于outBoundHandler在pipeline中的添加顺序逆向顺序来的。接下来从OutBoundHandlerA的这行代码ctx.channel().pipeline().fireChannelRead("hello")入手，基于源码分析一下这个中间的执行流程。

从代码流程来看，可以看出事件传播是从TailContext开始传播

image.png

image.png

接着，会去pipeline中开始寻找下一个节点OutBoundHandlerC

image.png

image.png
image.png

接着代码就会进入OutBoundHandlerC#write方法中，OutBoundHandlerC中事件会继续沿着pipeline往下进行传播，最终会传播到HeadContext

image.png

流程就是如图所示:


image.png

最后我们看下在HeadContext中对write事件的处理，他会调用unsafe的write方法，unsafe#write主要是将数据写会到客户端，这里对unsafe不做过多的解析，后面我们会详细讲unsafe。

image.png

ok，outBound事件就分析到这里，接下来我们分析异常的传播

修改我们server端的代码变成如下:

    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new InBoundHandlerA());
                            ch.pipeline().addLast(new InBoundHandlerB());
                            ch.pipeline().addLast(new InBoundHandlerC());
                            ch.pipeline().addLast(new OutBoundHandlerA());
                            ch.pipeline().addLast(new OutBoundHandlerB());
                            ch.pipeline().addLast(new OutBoundHandlerC());
                        }
                    });

其中InBoundHandlerB的代码如下,调用channelRead方法的时候会抛出一个异常:

public class InBoundHandlerB extends ChannelInboundHandlerAdapter {

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
        throw new RuntimeException(this.getClass().getName() + " happen error");
    }


    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println(this.getClass().getName() + " exceptionCaught exec ");
        ctx.fireExceptionCaught(cause);
    }
}

InBoundHandlerA，InBoundHandlerC，OutBoundHandlerA，OutBoundHandlerB，OutBoundHandlerC 重些exceptionCaught方法，代码如下:

@Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println(this.getClass().getName() + " exceptionCaught ");
        super.exceptionCaught(ctx, cause);
    }

启动server端代码，本地通过telnet命令telnet localhost 8007 触发后，随便输入字符，我们可以看到控制台会输出以下结果：

com.tyust.netty.exception.InBoundHandlerB exceptionCaught exec 
com.tyust.netty.exception.InBoundHandlerC exceptionCaught 
com.tyust.netty.exception.OutBoundHandlerA exceptionCaught 
com.tyust.netty.exception.OutBoundHandlerB exceptionCaught 
com.tyust.netty.exception.OutBoundHandlerC exceptionCaught 
21:09:05.814 [nioEventLoopGroup-3-1] WARN io.netty.channel.DefaultChannelPipeline - An exceptionCaught() event was fired, and it reached at the tail of the pipeline. It usually means the last handler in the pipeline did not handle the exception.
java.lang.RuntimeException: com.tyust.netty.exception.InBoundHandlerB happen error

从日志显示得出,异常是随着handler的添加顺序进行传播，接下来我们进行断点分析;在调用完InBoundHandlerB#channelRead方法后,事件会往下一个节点进行传播，但由于出现了异常，代码会进入这个位置io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead(java.lang.Object)，紧接着，他会去找pipeline中下一个重写了exceptionCaught的方法
找到了InboundHandlerB

image.png
image.png
image.png

也就出现了我们控制台中显示的com.tyust.netty.exception.InBoundHandlerB exceptionCaught exec 日志输出；

接下来他会继续找下一个重写了exceptionCaught的方法也就是InBoundHandlerC，以此类推，最后会执行到TailContext的exceptionCaught方法

image.png

image.png

最后我们看下TailContext的exceptionCaught方法，它什么事情都没做，只是把日志进行输出,然后进行一场回收

image.png

image.png

这样其实很不友好,异常是反映我们系统是否出问题最重要的一个因素，我们需要将其捕获进行处理，因此常用的处理流程是调整我们的代码添加一个异常的handler

public class ExceptionHandler extends ChannelInboundHandlerAdapter {

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        System.out.println(this.getClass().getName() + " 异常处理,e:" + cause);
    }
}

...
 .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) throws Exception {
                            ch.pipeline().addLast(new InBoundHandlerA());
                            ch.pipeline().addLast(new InBoundHandlerB());
                            ch.pipeline().addLast(new InBoundHandlerC());
                            ch.pipeline().addLast(new OutBoundHandlerA());
                            ch.pipeline().addLast(new OutBoundHandlerB());
                            ch.pipeline().addLast(new OutBoundHandlerC());
                            ch.pipeline().addLast(new ExceptionHandler());
                        }
                    });

好了，我们的事件及异常传播到这里就结束了，留给大家两个问题，大家可以沿着我们上面的分析去解决这两个问题:

  • 在outbound事件传播中，如果OutBoundHandlerA#handlerAdded使用的case2中的代码，事件会是怎么样在pipeline中传播的?
   @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        //case 1
        ctx.executor().schedule(() -> ctx.channel().write("hello,world"), 3, TimeUnit.SECONDS);

        //case 2
        ctx.executor().schedule(() -> ctx.write("hello,world"), 3, TimeUnit.SECONDS);
    }
  • 同样的，在 inBound事件传播中，如果InBoundHandlerA#channelActive方法中调用的是case2中的代码，那事件是如何传播的?
@Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        //case  1
        ctx.channel().pipeline().fireChannelRead("hello");
        //case  2
        ctx.fireChannelRead("hello");
    }
最后编辑于 ：
©著作权归作者所有,转载或内容合作请联系作者
【社区内容提示】社区部分内容疑似由AI辅助生成，浏览时请结合常识与多方信息审慎甄别。
平台声明：文章内容（如有图片或视频亦包括在内）由作者上传并发布，文章内容仅代表作者本人观点，简书系信息发布平台，仅提供信息存储服务。