Netty分布式pipeline管道異常傳播事件的示例分析

這篇文章主要介紹了Netty分布式pipeline管道異常傳播事件的示例分析，具有一定借鑒價值，感興趣的朋友可以參考下，希望大家閱讀完這篇文章之后大有收獲，下面讓小編帶著大家一起了解一下。

傳播異常事件

簡單的異常處理的場景

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    throw new Exception("throw Exception");
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    System.out.println(cause.getMessage());
}

我們在handler的channelRead方法中主動拋出異常, 模擬程序中出現異常的場景, 經測試會發現, 程序最終會走到exceptionCaught方法中, 獲取異常對象并打印其信息

那么拋出異常之后, 是如何走到exceptionCaught方法的呢?

我們回顧之前小節channelRead事件的傳播流程, channelRead方法是在AbstractChannelHandlerContext類的invokeChannelRead方法中被調用

我們跟到invokeChannelRead這個方法

private void invokeChannelRead(Object msg) {
    if (invokeHandler()) {
        try {
            //調用了當前handler的channelRead方法, 其實就是head對象調用自身的channelRead方法
            ((ChannelInboundHandler) handler()).channelRead(this, msg);
        } catch (Throwable t) {
            //發生異常的時候在這里捕獲異常
            notifyHandlerException(t);
        }
    } else {
        fireChannelRead(msg);
    }
}

這里不難看出, 當調用戶自定義的handler的channelRead方法發生異常之后, 會被捕獲, 并調用notifyHandlerException方法, 并傳入異常對象, 也就是我們示例中拋出的異常

我們跟到fireChannelRead方法中:

private void notifyHandlerException(Throwable cause) {
    //代碼省略
    invokeExceptionCaught(cause);
}

再繼續跟到invokeExceptionCaught方法中:

private void invokeExceptionCaught(final Throwable cause) {
    if (invokeHandler()) {
        try {
            //當前handler調用exceptionCaught()方法
            handler().exceptionCaught(this, cause);
        } catch (Throwable error) {
            //代碼省略
        }
    } else {
        fireExceptionCaught(cause);
    }
}

走到這里一切都明白了, 這里調用了當前handler的exceptionCaught方法, 也就是我們重寫的exceptionCaught方法

知道了為什么會走到exceptionCaught方法之后, 我們再進行剖析異常事件的傳播流程

我還是通過兩種寫法來進行剖析

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    System.out.println(cause.getMessage());
}
@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    //寫法1
    ctx.fireChannelRead(cause);
    //寫法2
    ctx.pipeline().fireExceptionCaught(cause);
}

這兩種寫法我們并不陌生, 可能我們能直接猜到, 第一種寫法是從當前節點進行傳播, 第二種寫法則從頭結點或者尾節點進行轉播, 那么和傳播inbound事件或outbound事件有什么區別呢?我們先以第二種寫法為例, 剖析異常事件傳輸的整個流程

跟到DefualtChannelPipeline的fireExceptionCaught方法中:

public final ChannelPipeline fireExceptionCaught(Throwable cause) {
    AbstractChannelHandlerContext.invokeExceptionCaught(head, cause);
    return this;
}

我們看到invokeExceptionCaught傳入了head節點, 我們可以猜測, 異常事件的傳播是從head節點開始的

跟進invokeExceptionCaught方法

static void invokeExceptionCaught(final AbstractChannelHandlerContext next, final Throwable cause) {
    ObjectUtil.checkNotNull(cause, "cause");
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        //執行下一個節點的異常方法
        next.invokeExceptionCaught(cause);
    } else {
        try {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeExceptionCaught(cause);
                }
            });
        } catch (Throwable t) {
            //忽略代碼
        }
    }
}

因為這里是傳入的是head節點, 所以這里的next指向head節點

我們跟到invokeExceptionCaught方法中, 這里其實是headContext的父類AbstractChannelHandlerContext中的方法:

private void invokeExceptionCaught(final Throwable cause) {
    if (invokeHandler()) {
        try {
            //當前handler調用exceptionCaught()方法
            handler().exceptionCaught(this, cause);
        } catch (Throwable error) {
            //代碼省略
        }
    } else {
        fireExceptionCaught(cause);
    }
}

這里又是我們熟悉的邏輯, 調用當前handler的exceptionCaught方法, 因為當前handler是head, 所以首先會調用headContext的exceptionCaught方法

跟進exceptionCaught方法:

public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    ctx.fireExceptionCaught(cause);
}

這里僅僅是繼續傳播異常事件, 這時候我們發現, 這個寫法和我們剛才提到傳播異常事件的兩種寫法的第一種寫法一樣:

@Override
public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    //寫法1
    ctx.fireChannelRead(cause);
    //寫法2
    ctx.pipeline().fireExceptionCaught(cause);
}

根據我們之前的學習, 我們知道第一種寫法是從當前節點傳播, 而第二種寫法是從頭傳播, 并且要求傳播事件一定要使用第一種寫法, 否則事件到這里會重新從頭傳播進而引發不可預知錯誤, 這個結論在異常傳播同樣適用, 同學們一定要注意這點

我們繼續跟fireExceptionCaught方法, 這里會走到AbstractChannelHandlerContex類的fireExceptionCaught方法:

public ChannelHandlerContext fireExceptionCaught(final Throwable cause) {
    //傳播異常事件的時候, 直接拿了當前節點的下一個節點
    invokeExceptionCaught(next, cause);
    return this;
}

這個時候我們發現, 這里并沒有去獲取下一個的inbound節點還是outbound節點, 而是直接通過next拿到下一個節點, 這就說明在異常事件傳播的過程中是不區分inbound事件還是outbound事件的, 都是直接從head節點按照鏈表結構往下傳播,

跟到invokeExceptionCaught方法中

static void invokeExceptionCaught(final AbstractChannelHandlerContext next, final Throwable cause) {
    ObjectUtil.checkNotNull(cause, "cause");
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) { 
        next.invokeExceptionCaught(cause);
    } else {
        try {
            executor.execute(new Runnable() {
                @Override
                public void run() {
                    next.invokeExceptionCaught(cause);
                }
            });
        } catch (Throwable t) {
            //代碼省略
        }
    }
}

這里又是我們熟悉的邏輯, 我們知道invokeExceptionCaught中執行了next的exceptionCaught, 這里的next, 因為我們是從head節點開始剖析的, 所以這里很有可能就是用戶自定義的handler, 如果用戶沒有重寫exceptionCaught方法, 則會交給用戶handler的父類處理

我們以ChannelInboundHandlerAdapter為例看它的該方法實現:

public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause)
        throws Exception {
    ctx.fireExceptionCaught(cause);
}

我們看到這里繼續向下傳播了異常事件

走到這里我們會知道, 如果我們沒有重寫exceptionCaught方法, 異常事件會一直傳播到鏈表的底部, 就是tail節點

我們跟到TailConext的exceptionCaught方法:

public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
    onUnhandledInboundException(cause);
}

感謝你能夠認真閱讀完這篇文章，希望小編分享的“Netty分布式pipeline管道異常傳播事件的示例分析”這篇文章對大家有幫助，同時也希望大家多多支持億速云，關注億速云行業資訊頻道，更多相關知識等著你來學習!