Netty分布式pipeline傳播inbound事件源碼分析

更新時間：2022年03月28日 10:59:19 作者：向南是個萬人迷

這篇文章主要為大家介紹了Netty分布式pipeline傳播inbound事件的源碼分析，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進步

傳播inbound事件

有關于inbound事件, 在概述中做過簡單的介紹, 就是以自己為基準, 流向自己的事件, 比如最常見的channelRead事件, 就是對方發(fā)來數(shù)據(jù)流的所觸發(fā)的事件, 己方要對這些數(shù)據(jù)進行處理, 這一小節(jié), 以激活channelRead為例講解有關inbound事件的處理流程

在業(yè)務代碼中, 我們自己的handler往往會通過重寫channelRead方法來處理對方發(fā)來的數(shù)據(jù), 那么對方發(fā)來的數(shù)據(jù)是如何走到channelRead方法中了呢, 也是我們這一小節(jié)要剖析的內(nèi)容

在業(yè)務代碼中, 傳遞channelRead事件方式是通過fireChannelRead方法進行傳播的

這里給大家看兩種寫法

@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //寫法1:
    ctx.fireChannelRead(msg);
    //寫法2
    ctx.pipeline().fireChannelRead(msg);
}

這里重寫了channelRead方法, 并且方法體內(nèi)繼續(xù)通過fireChannelRead方法進行傳播channelRead事件, 那么這兩種寫法有什么異同?

我們先以寫法2為例, 將這種寫法進行剖析

這里首先獲取當前context的pipeline對象, 然后通過pipeline對象調(diào)用自身的fireChannelRead方法進行傳播, 因為默認創(chuàng)建的DefaultChannelpipeline

我們跟到DefaultChannelpipeline的fireChannelRead方法中:

public final ChannelPipeline fireChannelRead(Object msg) {
    AbstractChannelHandlerContext.invokeChannelRead(head, msg);
    return this;
}

這里首先調(diào)用的是AbstractChannelHandlerContext類的靜態(tài)方法invokeChannelRead, 參數(shù)傳入head節(jié)點和事件的消息

我們跟進invokeChannelRead方法:

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRead(m);
    } else {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                next.invokeChannelRead(m);
            }
        });
    }
}

這里的Object m m通常就是我們傳入的msg, 而next, 目前是head節(jié)點, 然后再判斷是否為當前eventLoop線程, 如果不是則將方法包裝成task交給eventLoop線程處理

我們跟到invokeChannelRead方法中

private void invokeChannelRead(Object msg) {
    if (invokeHandler()) {
        try {
            ((ChannelInboundHandler) handler()).channelRead(this, msg);
        } catch (Throwable t) {
            notifyHandlerException(t);
        }
    } else {
        fireChannelRead(msg);
    }
}

首先通過invokeHandler()判斷當前handler是否已添加, 如果添加, 則執(zhí)行當前handler的chanelRead方法, 其實這里我們基本上就明白了, 通過fireChannelRead方法傳遞事件的過程中, 其實就是找到相關handler執(zhí)行其channelRead方法, 由于我們在這里的handler就是head節(jié)點, 所以我們跟到HeadContext的channelRead方法中:

HeadContext的channelRead方法:

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //向下傳遞channelRead事件
    ctx.fireChannelRead(msg);
}

在這里我們看到, 這里通過fireChannelRead方法繼續(xù)往下傳遞channelRead事件, 而這種調(diào)用方式, 就是我們剛才分析用戶代碼的第一種調(diào)用方式:

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //寫法1:
    ctx.fireChannelRead(msg);
    //寫法2
    ctx.pipeline().fireChannelRead(msg);
}

這里直接通過context對象調(diào)用fireChannelRead方法, 那么和使用pipeline調(diào)用有什么區(qū)別的, 我會回到HeadConetx的channelRead方法, 我們來剖析ctx.fireChannelRead(msg)這句, 大家就會對這個問題有答案了, 跟到ctx的fireChannelRead方法中, 這里會走到AbstractChannelHandlerContext類中的fireChannelRead方法中

跟到AbstractChannelHandlerContext類中的fireChannelRead方法:

public ChannelHandlerContext fireChannelRead(final Object msg) {
    invokeChannelRead(findContextInbound(), msg);
    return this;
}

這里我們看到, invokeChannelRead方法中傳入了一個findContextInbound()參數(shù), 而這findContextInbound方法其實就是找到當前Context的下一個節(jié)點

跟到findContextInbound方法

private AbstractChannelHandlerContext findContextInbound() {
    AbstractChannelHandlerContext ctx = this;
    do {
        ctx = ctx.next;
    } while (!ctx.inbound);
    return ctx;
}

這里的邏輯也比較簡單, 是通過一個doWhile循環(huán), 找到當前handlerContext的下一個節(jié)點, 這里要注意循環(huán)的終止條件, while (!ctx.inbound)表示下一個context標志的事件不是inbound的事件, 則循環(huán)繼續(xù)往下找, 言外之意就是要找到下一個標注inbound事件的節(jié)點

有關事件的標注, 之前的小節(jié)已經(jīng)剖析過了, 如果是用戶定義的handler, 是通過handler繼承的接口而定的, 如果tail或者head, 那么是在初始化的時候就已經(jīng)定義好, 這里不再贅述

回到AbstractChannelHandlerContext類的fireChannelRead方法中:

public ChannelHandlerContext fireChannelRead(final Object msg) {
    invokeChannelRead(findContextInbound(), msg);
    return this;
}

找到下一個節(jié)點后, 繼續(xù)調(diào)用invokeChannelRead方法, 傳入下一個和消息對象:

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
    final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
    //第一次執(zhí)行next其實就是head
    EventExecutor executor = next.executor();
    if (executor.inEventLoop()) {
        next.invokeChannelRead(m);
    } else {
        executor.execute(new Runnable() {
            @Override
            public void run() {
                next.invokeChannelRead(m);
            }
        });
    }
}

這里的邏輯我們又不陌生了, 因為我們傳入的是當前context的下一個節(jié)點, 所以這里會調(diào)用下一個節(jié)點invokeChannelRead方法, 因我們剛才剖析的是head節(jié)點, 所以下一個節(jié)點有可能是用戶添加的handler的包裝類HandlerConext的對象

這里我們跟進invokeChannelRead方法中去:

private void invokeChannelRead(Object msg) {
    if (invokeHandler()) {
        try { 
            ((ChannelInboundHandler) handler()).channelRead(this, msg);
        } catch (Throwable t) {
            //發(fā)生異常的時候在這里捕獲異常
            notifyHandlerException(t);
        }
    } else {
        fireChannelRead(msg);
    }
}

又是我們熟悉的邏輯, 調(diào)用了自身handler的channelRead方法, 如果是用戶自定義的handler, 則會走到用戶定義的channelRead()方法中去, 所以這里就解釋了為什么通過傳遞channelRead事件, 最終會走到用戶重寫的channelRead方法中去

同樣, 也解釋了該小節(jié)最初提到過的兩種寫法的區(qū)別:

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    //寫法1:
    ctx.fireChannelRead(msg);
    //寫法2
    ctx.pipeline().fireChannelRead(msg);
}

寫法1是通過當前節(jié)點往下傳播事件

寫法2是通過頭節(jié)點往下傳遞事件

所以, 在handler中如果如果要在channelRead方法中傳遞channelRead事件, 一定要采用寫法2的方式向下傳遞, 或者交給其父類處理, 如果采用1的寫法則每次事件傳輸?shù)竭@里都會繼續(xù)從head節(jié)點傳輸, 從而陷入死循環(huán)或者發(fā)生異常

這里有一點需要注意, 如果用戶代碼中channelRead方法, 如果沒有顯示的調(diào)用ctx.fireChannelRead(msg)那么事件則不會再往下傳播, 則事件會在這里終止, 所以如果我們寫業(yè)務代碼的時候要考慮有關資源釋放的相關操作

如果ctx.fireChannelRead(msg)則事件會繼續(xù)往下傳播, 如果每一個handler都向下傳播事件, 當然, 根據(jù)我們之前的分析channelRead事件只會在標識為inbound事件的HandlerConetext中傳播, 傳播到最后, 則最終會調(diào)用到tail節(jié)點的channelRead方法

我們跟到tailConext的channelRead方法中

public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
    onUnhandledInboundMessage(msg);
}

我們跟進到onUnhandledInboundMessage方法中:

protected void onUnhandledInboundMessage(Object msg) {
    try {
        logger.debug(
                "Discarded inbound message {} that reached at the tail of the pipeline. " +
                        "Please check your pipeline configuration.", msg);
    } finally {
        //釋放資源
        ReferenceCountUtil.release(msg);
    }
}

這里做了釋放資源的相關的操作

至此, channelRead事件傳輸相關羅輯剖析完整, 其實對于inbound事件的傳輸流程都會遵循這一邏輯, 小伙伴們可以自行剖析其他inbound事件的傳輸流程，更多關于Netty分布式pipeline傳播inbound事件的資料請關注腳本之家其它相關文章！

您可能感興趣的文章: