Netty源码分析之ChannelPipeline—异常事件的传播

ChannelHandler中异常的获取与处理是通过继承重写exceptionCaught方法来实现的，本篇文章我们对ChannelPipeline中exceptionCaught异常事件的传播进行梳理分析

1、出站事件的传播示例

首先我们继续在之前的代码上进行改造，模拟异常事件的传播

public class ServerApp {

    public static void main(String[] args) {

        EventLoopGroup boss = new NioEventLoopGroup();

        EventLoopGroup work = new NioEventLoopGroup(2);

        try {

            ServerBootstrap bootstrap = new ServerBootstrap();

            bootstrap.childOption(ChannelOption.SO_SNDBUF,2);

            bootstrap.group(boss, work).channel(NioServerSocketChannel.class)

                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override

                        public void initChannel(SocketChannel ch) throws Exception {

                            ChannelPipeline p = ch.pipeline();

                            // p.addLast(new LoggingHandler(LogLevel.INFO));

                            // 向ChannelPipeline中添加自定义channelHandler

                            p.addLast(new OutHandlerA());

                            p.addLast(new ServerHandlerA());

                            p.addLast(new ServerHandlerB());

                            p.addLast(new ServerHandlerC());

                            p.addLast(new OutHandlerB());

                            p.addLast(new OutHandlerC());

                        }

                    });

            bootstrap.bind(8050).sync();

        } catch (Exception e) {

            // TODO: handle exception

        }

    }

}

public class OutHandlerA extends ChannelOutboundHandlerAdapter {

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)

            throws Exception {

        System.err.println(this.getClass().getName()+"---"+cause.getMessage());

        ctx.fireExceptionCaught(cause);

    }

}

public class OutHandlerB extends ChannelOutboundHandlerAdapter {

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)

            throws Exception {

        System.err.println(this.getClass().getName()+"---"+cause.getMessage());

        ctx.fireExceptionCaught(cause);

    }

}

public class OutHandlerC extends ChannelOutboundHandlerAdapter {

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)

            throws Exception {

        System.err.println(this.getClass().getName()+"---"+cause.getMessage());

        ctx.fireExceptionCaught(cause);

    }

}

public class ServerHandlerB extends ChannelInboundHandlerAdapter {

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)

            throws Exception {

        System.err.println(this.getClass().getName()+"---"+cause.getMessage());

        ctx.fireExceptionCaught(cause);

    }

}

public class ServerHandlerC extends ChannelInboundHandlerAdapter {

    @Override

    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)

            throws Exception {

        System.err.println(this.getClass().getName()+"---"+cause.getMessage());

        ctx.fireExceptionCaught(cause);

    }

}

然后我们在ServerHandlerA的channelRead方法中执行ctx的write方法，模拟异常事件的发生。

    @Override

    public void channelRead(ChannelHandlerContext ctx, Object object) {

        ctx.fireExceptionCaught(new Throwable("出现异常"));

        //ctx.pipeline().fireExceptionCaught(new Throwable("出现异常"));

    }

我们首先看下运行结果

ctx.fireExceptionCaught

io.netty.example.echo.my.ServerHandlerB---出现异常

io.netty.example.echo.my.ServerHandlerC---出现异常

io.netty.example.echo.my.OutHandlerB---出现异常

io.netty.example.echo.my.OutHandlerC---出现异常

18:34:17.147 [nioEventLoopGroup-3-1] WARN  i.n.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.Throwable: 出现异常

    at io.netty.example.echo.my.ServerHandlerA.channelRead(ServerHandlerA.java:39)

    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)

    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:338)

    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1424)

    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)

    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:944)

    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163)

    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:709)

    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:639)

    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:553)

    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:510)

    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:912)

    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)

    at java.lang.Thread.run(Thread.java:748)

ctx.pipeline().fireExceptionCaught

io.netty.example.echo.my.OutHandlerA---出现异常

io.netty.example.echo.my.ServerHandlerA---出现异常

io.netty.example.echo.my.ServerHandlerB---出现异常

io.netty.example.echo.my.ServerHandlerC---出现异常

io.netty.example.echo.my.OutHandlerB---出现异常

io.netty.example.echo.my.OutHandlerC---出现异常

20:08:53.723 [nioEventLoopGroup-3-1] WARN  i.n.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.Throwable: 出现异常

    at io.netty.example.echo.my.ServerHandlerA.channelRead(ServerHandlerA.java:40)

    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)

    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

    at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:338)

    at io.netty.channel.DefaultChannelPipeline$HeadContext.channelRead(DefaultChannelPipeline.java:1424)

    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:363)

    at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:348)

    at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:944)

    at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:163)

    at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:709)

    at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:639)

    at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:553)

    at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:510)

    at io.netty.util.concurrent.SingleThreadEventExecutor$5.run(SingleThreadEventExecutor.java:912)

    at io.netty.util.concurrent.FastThreadLocalRunnable.run(FastThreadLocalRunnable.java:30)

    at java.lang.Thread.run(Thread.java:748)

根据输出结果可以看出ctx.fireExceptionCaught 会从异常产生的ChannelHandler一直往后传播到tail尾节点，ctx.pipeline().fireExceptionCaught会从管道中第一个节点一直往后传播到tail尾节点，而上面结果中打印的异常信息则是在TailContext尾节点中统一处理的。

2、异常事件传播的分析

ctx.pipeline().fireExceptionCaught与ctx.fireExceptionCaught两种传播异常方法

前者调用的是DefaultChannelPipeline 的 fireExceptionCaught方法

    @Override

    public final ChannelPipeline fireExceptionCaught(Throwable cause) {

        AbstractChannelHandlerContext.invokeExceptionCaught(head, cause);

        return this;

    }

后者调用的是AbstractChannelHandlerContext 的 fireExceptionCaught方法

    @Override

    public ChannelHandlerContext fireExceptionCaught(final Throwable cause) {

        invokeExceptionCaught(next, cause);

        return this;

    }

可以看到DefaultChannelPipeline的fireExceptionCaught方法中默认传入了head头部节点，所以ctx.pipeline().fireExceptionCaught会从管道中第一个节点开始向后传播。

我们进入invokeExceptionCaught方法内部看下具体实现

    static void invokeExceptionCaught(final AbstractChannelHandlerContext next, final Throwable cause) {

        ObjectUtil.checkNotNull(cause, "cause");//检查异常是否为空

        EventExecutor executor = next.executor();

        if (executor.inEventLoop()) {//判断是否与当前线程一直

            next.invokeExceptionCaught(cause);//触发回调，触发下一个AbstractChannelHandlerContext节点中handler的异常处理事件

        } else {

            try {

                executor.execute(new Runnable() {//如果线程不一致，由其绑定的executor执行

                    @Override

                    public void run() {

                        next.invokeExceptionCaught(cause);

                    }

                });

            } catch (Throwable t) {

                if (logger.isWarnEnabled()) {

                    logger.warn("Failed to submit an exceptionCaught() event.", t);

                    logger.warn("The exceptionCaught() event that was failed to submit was:", cause);

                }

            }

        }

invokeExceptionCaught方法内部实现

    private void invokeExceptionCaught(final Throwable cause) {

        if (invokeHandler()) {//判断当前handler的状态

            try {

                handler().exceptionCaught(this, cause);//调用exceptionCaught方法实现

            } catch (Throwable error) {

                if (logger.isDebugEnabled()) {

                    logger.debug(

                        "An exception {}" +

                        "was thrown by a user handler's exceptionCaught() " +

                        "method while handling the following exception:",

                        ThrowableUtil.stackTraceToString(error), cause);

                } else if (logger.isWarnEnabled()) {

                    logger.warn(

                        "An exception '{}' [enable DEBUG level for full stacktrace] " +

                        "was thrown by a user handler's exceptionCaught() " +

                        "method while handling the following exception:", error, cause);

                }

            }

        } else {

            fireExceptionCaught(cause);

        }

    }

3、异常处理机制的设计

通过上面的分析我们可以看到如果通过ctx.fireExceptionCaught一直向后传递异常事件，最终会触发尾节点的exceptionCaught事件打印异常日志；

        @Override

        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {

            onUnhandledInboundException(cause);

        }

    protected void onUnhandledInboundException(Throwable cause) {

        try {

            logger.warn(

                    "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.",

                    cause);

        } finally {

            ReferenceCountUtil.release(cause);

        }

    }

在实际项目中我们可以在ChannelPipeline尾部增加一个异常处理handle用来统一处理异常信息;

        public void initChannel(SocketChannel ch) throws Exception {

                            ChannelPipeline p = ch.pipeline();

                            // p.addLast(new LoggingHandler(LogLevel.INFO));

                            // 向ChannelPipeline中添加自定义channelHandler

                            p.addLast(new OutHandlerA());

                            p.addLast(new ServerHandlerA());

                            p.addLast(new ServerHandlerB());

                            p.addLast(new ServerHandlerC());

                            p.addLast(new OutHandlerB());

                            p.addLast(new OutHandlerC());

                            p.addLast(new ExceptionHandler());

                        }

通过以上三点内容我们对异常信息在ChannelPipeline中的传播进行了模拟，梳理事件的传播流程以及应该怎样统一处理异常信息，其中如有不足与不正确的地方还望指出与海涵。

关注微信公众号，查看更多技术文章。

巴特西

Netty源码分析之ChannelPipeline—异常事件的传播

1、出站事件的传播示例

2、异常事件传播的分析

3、异常处理机制的设计

最新文章

热门文章