記一次有趣的 Netty 源碼問題

harriszh 發(fā)布于2019-08-14 15:04 / 944人閱讀

摘要：背景起因是一個朋友問我的一個關于啟動的問題相關他的問題我復述一下的綁定流程如下在中可能會調(diào)用即并且在中也會有的調(diào)用即那么有沒有可能造成了兩次的調(diào)用我的回答是不會為什么呢對于直接想知道答案的朋友可以直接閱讀到最后面的回答與總

背景

起因是一個朋友問我的一個關于 ServerBootstrap 啟動的問題.
相關 issue
他的問題我復述一下:
ServerBootstrap 的綁定流程如下:

ServerBootstrap.bind ->
    AbstractBootstrap.bind ->
        AbstractBootstrap.doBind ->
            AbstractBootstrap.initAndRegister ->
                AbstractChannel#AbstractUnsafe.register ->
                    eventLoop.execute( () -> AbstractUnsafe.register0)
            doBind0() ->
                channel.eventLoop().execute( () -> channel.bind) ->
                    AbstractUnsafe.bind

在 AbstractUnsafe.register0 中可能會調(diào)用 pipeline.fireChannelActive(), 即:

private void register0(ChannelPromise promise) {
    try {
        ...
        boolean firstRegistration = neverRegistered;
        doRegister();
        ...
        if (firstRegistration && isActive()) {
            pipeline.fireChannelActive();
        }
    } catch (Throwable t) {
        ...
    }
}

并且在 AbstractUnsafe.bind 中也會有 pipeline.fireChannelActive() 的調(diào)用, 即:

public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
    ...
    boolean wasActive = isActive();
    try {
        doBind(localAddress);
    } catch (Throwable t) {
        ...
    }

    if (!wasActive && isActive()) {
        invokeLater(new OneTimeTask() {
            @Override
            public void run() {
                pipeline.fireChannelActive();
            }
        });
    }
    ...
}

那么有沒有可能造成了兩次的 pipeline.fireChannelActive() 調(diào)用?

我的回答是不會. 為什么呢? 對于直接想知道答案的朋友可以直接閱讀到最后面的回答與總結兩節(jié)..

下面我們就來根據(jù)代碼詳細分析一下.

分析

首先, 根據(jù)我們上面所列出的調(diào)用流程, 會有 AbstractBootstrap.doBind 的調(diào)用, 它的代碼如下:

private ChannelFuture doBind(final SocketAddress localAddress) {
        // 步驟1
        final ChannelFuture regFuture = initAndRegister();
        ...
        // 步驟2
        if (regFuture.isDone()) {
            ...
            doBind0(regFuture, channel, localAddress, promise);
            ...
        } else {
            regFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) throws Exception {
                    ...
                    doBind0(regFuture, channel, localAddress, promise);
                }
            });
        }
}

首先在 doBind 中, 執(zhí)行步驟1, 即調(diào)用 initAndRegister 方法, 這個方法會最終調(diào)用到AbstractChannel#AbstractUnsafe.register. 而在 AbstractChannel#AbstractUnsafe.register 中, 會通過 eventLoop.execute 的形式將 AbstractUnsafe.register0 的調(diào)用提交到任務隊列中(即提交到 eventLoop 線程中, 而當前代碼所在的線程是 main 線程), 即:

@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
    // 當前線程是主線程, 因此這個判斷是 false
    if (eventLoop.inEventLoop()) {
        register0(promise);
    } else {
        try {
            eventLoop.execute(new OneTimeTask() {
                @Override
                public void run() {
                    // register0 在 eventLoop 線程中執(zhí)行.
                    register0(promise);
                }
            });
        } catch (Throwable t) {
           ...
        }
    }
}

接著 AbstractBootstrap.initAndRegister 返回, 回到 AbstractBootstrap.doBind 中, 于是執(zhí)行到步驟2. 注意, 因為 AbstractUnsafe.register0 是在 eventLoop 中執(zhí)行的, 因此有可能主線程執(zhí)行到步驟2 時, AbstractUnsafe.register0 已經(jīng)執(zhí)行完畢了, 此時必然有 regFuture.isDone() == true; 但也有可能 AbstractUnsafe.register0 沒有來得及執(zhí)行, 因此此時 regFuture.isDone() == false. 所以上面的步驟2 考慮到了這兩種情況, 因此分別針對這兩種情況做了區(qū)分, 即:

// 步驟2
if (regFuture.isDone()) {
    ...
    doBind0(regFuture, channel, localAddress, promise);
    ...
} else {
    regFuture.addListener(new ChannelFutureListener() {
        @Override
        public void operationComplete(ChannelFuture future) throws Exception {
            ...
            doBind0(regFuture, channel, localAddress, promise);
        }
    });
}

一般情況下, regFuture.isDone() 為 false, 因為綁定操作是比較費時的, 因此很大幾率會執(zhí)行到 else 分支, 并且 if 分支和 else 分支從結果上說沒有不同, 而且 if 分支邏輯還更簡單一些, 因此我們以 else 分支來分析吧. 在 else 分支中, 會為 regFuture 設置一個回調(diào)監(jiān)聽器. regFuture 是一個 ChannelFuture, 而 ChannelFuture 代表了一個 Channel 的異步 IO 的操作結果, 因此這里 regFuture 代表了 Channel 注冊(register) 的這個異步 IO 的操作結果.
Netty 這里之所以要為 regFuture 設置一個回調(diào)監(jiān)聽器, 是為了保證 register 和 bind 的時序上的正確性: Channel 的注冊必須要發(fā)生在 Channel 的綁定之前.
(關于時序的正確性的問題, 我們在后面有證明)

接下來我們來看一下 AbstractUnsafe.register0 方法:

private void register0(ChannelPromise promise) {
    try {
        ....
        // neverRegistered 一開始是 true, 因此 firstRegistration == true
        boolean firstRegistration = neverRegistered;
        doRegister();
        neverRegistered = false;
        registered = true;
        safeSetSuccess(promise);
        pipeline.fireChannelRegistered();
        // Only fire a channelActive if the channel has never been registered. This prevents firing
        // multiple channel actives if the channel is deregistered and re-registered.
        // firstRegistration == true, 而 isActive() == false, 
        // 因此不會執(zhí)行到 pipeline.fireChannelActive()
        if (firstRegistration && isActive()) {
            pipeline.fireChannelActive();
        }
    } catch (Throwable t) {
        // Close the channel directly to avoid FD leak.
        closeForcibly();
        closeFuture.setClosed();
        safeSetFailure(promise, t);
    }
}

注意, 我需要再強調(diào)一下, 這里 AbstractUnsafe.register0 是在 eventLoop 中執(zhí)行的.
AbstractUnsafe.register0 中會調(diào)用 doRegister() 注冊 NioServerSocketChannel, 然后調(diào)用 safeSetSuccess() 設置 promise 的狀態(tài)為成功. 而這個 promise 變量是什么呢? 我將 AbstractBootstrap.doBind 的調(diào)用鏈寫詳細一些:

AbstractBootstrap.doBind ->
    AbstractBootstrap.initAndRegister ->
        MultithreadEventLoopGroup.register ->
            SingleThreadEventLoop.register -> 
                AbstractChannel#AbstractUnsafe.register ->
                    eventLoop.execute( () -> AbstractUnsafe.register0)

在 SingleThreadEventLoop.register 中會實例化一個 DefaultChannelPromise, 即:

@Override
public ChannelFuture register(Channel channel) {
    return register(channel, new DefaultChannelPromise(channel, this));
}

接著調(diào)用重載的 SingleThreadEventLoop.register 方法:

@Override
public ChannelFuture register(final Channel channel, final ChannelPromise promise) {
    if (channel == null) {
        throw new NullPointerException("channel");
    }
    if (promise == null) {
        throw new NullPointerException("promise");
    }

    channel.unsafe().register(this, promise);
    return promise;
}

我們看到, 實例化的 DefaultChannelPromise 最終會以方法返回值的方式返回到調(diào)用方, 即返回到 AbstractBootstrap.doBind 中:

final ChannelFuture regFuture = initAndRegister();

因此我們這里有一個共識: regFuture 是一個在 SingleThreadEventLoop.register 中實例化的 DefaultChannelPromise 對象.

再回到 SingleThreadEventLoop.register 中, 在這里會調(diào)用 channel.unsafe().register(this, promise), 將 promise 對象傳遞到 AbstractChannel#AbstractUnsafe.register 中, 因此在 AbstractUnsafe.register0 中的 promise 就是 AbstractBootstrap.doBind 中的 regFuture.
promise == regFuture 很關鍵.

既然我們已經(jīng)確定了 promise 的身份, 那么調(diào)用的 safeSetSuccess(promise); 我們也知道是干嘛的了. safeSetSuccess 方法設置一個 Promise 的狀態(tài)為成功態(tài), 而 Promise 的 成功態(tài) 是最終狀態(tài), 即此時 promise.isDone() == true. 那么設置 promise 為成功態(tài)后, 會發(fā)生什么呢?
還記得不 promise == regFuture, 而我們在 AbstractBootstrap.doBind 的 else 分支中設置了一個回調(diào)監(jiān)聽器:

final PendingRegistrationPromise promise = new PendingRegistrationPromise(channel);
regFuture.addListener(new ChannelFutureListener() {
    @Override
    public void operationComplete(ChannelFuture future) throws Exception {
        Throwable cause = future.cause();
        if (cause != null) {
            // Registration on the EventLoop failed so fail the ChannelPromise directly to not cause an
            // IllegalStateException once we try to access the EventLoop of the Channel.
            promise.setFailure(cause);
        } else {
            // Registration was successful, so set the correct executor to use.
            // See https://github.com/netty/netty/issues/2586
            promise.executor = channel.eventLoop();
        }
        doBind0(regFuture, channel, localAddress, promise);
    }
});

因此當 safeSetSuccess(promise); 調(diào)用時, 根據(jù) Netty 的 Promise/Future 機制, 會觸發(fā)上面的 operationComplete 回調(diào), 在回調(diào)中調(diào)用 doBind0 方法:

private static void doBind0(
        final ChannelFuture regFuture, final Channel channel,
        final SocketAddress localAddress, final ChannelPromise promise) {
    // This method is invoked before channelRegistered() is triggered.  Give user handlers a chance to set up
    // the pipeline in its channelRegistered() implementation.
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (regFuture.isSuccess()) {
                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                promise.setFailure(regFuture.cause());
            }
        }
    });
}

注意到, 有一個關鍵的地方, 代碼中將 **channel.bind** 的調(diào)用放到了 eventLoop 中執(zhí)行. doBind0 返回后, 代碼繼續(xù)執(zhí)行 AbstractUnsafe.register0 方法的剩余部分代碼, 即:

private void register0(ChannelPromise promise) {
    try {
        ....
        safeSetSuccess(promise);
        // safeSetSuccess 返回后, 繼續(xù)執(zhí)行如下代碼
        pipeline.fireChannelRegistered();
        // Only fire a channelActive if the channel has never been registered. This prevents firing
        // multiple channel actives if the channel is deregistered and re-registered.
        // firstRegistration == true, 而 isActive() == false, 
        // 因此不會執(zhí)行到 pipeline.fireChannelActive()
        if (firstRegistration && isActive()) {
            pipeline.fireChannelActive();
        }
    } catch (Throwable t) {
        // Close the channel directly to avoid FD leak.
        closeForcibly();
        closeFuture.setClosed();
        safeSetFailure(promise, t);
    }
}

當 AbstractUnsafe.register0 方法執(zhí)行完畢后, 才執(zhí)行到 channel.bind 方法.

而 channel.bind 方法最終會調(diào)用到 AbstractChannel#AbstractUnsafe.bind 方法, 源碼如下:

@Override
public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
    boolean wasActive = isActive();
    logger.info("---wasActive: {}---", wasActive);

    try {
        // 調(diào)用 NioServerSocketChannel.bind 方法, 
        // 將底層的 Java NIO SocketChannel 綁定到指定的端口.
        // 當 SocketChannel 綁定到端口后, isActive() 才為真.
        doBind(localAddress);
    } catch (Throwable t) {
        ...
    }

    boolean activeNow = isActive();
    logger.info("---activeNow: {}---", activeNow);

    // 這里 wasActive == false
    // isActive() == true
    if (!wasActive && isActive()) {
        invokeLater(new OneTimeTask() {
            @Override
            public void run() {
                pipeline.fireChannelActive();
            }
        });
    }

    safeSetSuccess(promise);
}

上面的代碼中, 調(diào)用了 doBind(localAddress) 將底層的 Java NIO SocketChannel 綁定到指定的端口. 并且當 SocketChannel 綁定到端口后, isActive() 才為真.
因此我們知道, 如果 SocketChannel 第一次綁定時, 在調(diào)用 doBind 前, wasActive == false == isActive(), 而當調(diào)用了 doBind 后, isActive() == true, 因此第一次綁定端口時, if 判斷成立, 會調(diào)用 pipeline.fireChannelActive().

關于 Channel 注冊與綁定的時序問題

我們在前的分析中, 直接認定了 Channel 注冊 在 Channel 的綁定 之前完成, 那么依據(jù)是什么呢?
其實所有的關鍵在于 EventLoop 的任務隊列機制.
不要閑我啰嗦哦. 我們需要繼續(xù)回到 AbstractUnsafe.register0 的調(diào)用中(再次強調(diào)一下, 在 eventLoop 線程中執(zhí)行AbstractUnsafe.register0), 這個方法我們已經(jīng)分析了, 它會調(diào)用 safeSetSuccess(promise), 并由 Netty 的 Promise/Future 機制, 導致了AbstractBootstrap.doBind 中的 regFuture 所設置的回調(diào)監(jiān)聽器的 operationComplete 方法調(diào)用, 而 operationComplete 中調(diào)用了 AbstractBootstrap.doBind0:

private static void doBind0(
        final ChannelFuture regFuture, final Channel channel,
        final SocketAddress localAddress, final ChannelPromise promise) {
    channel.eventLoop().execute(new Runnable() {
        @Override
        public void run() {
            if (regFuture.isSuccess()) {
                channel.bind(localAddress, promise).addListener(ChannelFutureListener.CLOSE_ON_FAILURE);
            } else {
                promise.setFailure(regFuture.cause());
            }
        }
    });
}

在 doBind0 中, 根據(jù) EventLoop 的任務隊列機制, 會使用 eventLoop().execute 將 channel.bind 封裝為一個 Task, 放到 eventLoop 的 taskQueue 中.
如下用一幅圖表示上面的過程:

原圖在此

而當 channel.bind 被調(diào)度時, AbstractUnsafe.register0 早就已經(jīng)調(diào)用結束了.

因此由于 EventLoop 的任務隊列機制, 我們知道, 在執(zhí)行 AbstractUnsafe.register0 時, 是在 EventLoop 線程中的, 而 channel.bind 的調(diào)用是以 task 的形式添加到 taskQueue 隊列的末尾, 因此必然是有 EventLoop 線程先執(zhí)行完 AbstractUnsafe.register0 方法后, 才有機會從 taskQueue 中取出一個 task 來執(zhí)行, 因此這個機制從根本上保證了 Channel 注冊發(fā)生在綁定 之前.

回答

你的疑惑是, AbstractChannel#AbstractUnsafe.register0 中, 可能會調(diào)用 pipeline.fireChannelActive(), 即:

private void register0(ChannelPromise promise) {
    try {
        ...
        boolean firstRegistration = neverRegistered;
        doRegister();
        ...
        if (firstRegistration && isActive()) {
            pipeline.fireChannelActive();
        }
    } catch (Throwable t) {
        ...
    }
}

并且在 AbstractChannel#AbstractUnsafe.bind 中也可能會調(diào)用到pipeline.fireChannelActive(), 即:

public final void bind(final SocketAddress localAddress, final ChannelPromise promise) {
    ...
    boolean wasActive = isActive();
    try {
        doBind(localAddress);
    } catch (Throwable t) {
        ...
    }

    if (!wasActive && isActive()) {
        invokeLater(new OneTimeTask() {
            @Override
            public void run() {
                pipeline.fireChannelActive();
            }
        });
    }
    ...
}

我覺得是不會. 因為根據(jù)上面我們分析的結果可知, Netty 的 Promise/Future 與 EventLoop 的任務隊列機制保證了 NioServerSocketChannel 的注冊和 NioServerSocketChannel 的綁定的時序: Channel 的注冊必須要發(fā)生在 Channel 的綁定之前, 而當一個 NioServerSocketChannel 沒有綁定到具體的端口前, 它是不活躍的(Inactive), 因此在 register0 中, if (firstRegistration && isActive()) 就不成立, 進而就不會執(zhí)行到 pipeline.fireChannelActive() 了.
而執(zhí)行完注冊操作后, 在 AbstractChannel#AbstractUnsafe.bind 才會調(diào)用pipeline.fireChannelActive(), 因此最終只有一次 fireChannelActive 調(diào)用.

總結

有兩點需要注意的:

isActive() == true 成立的關鍵是此 NioServerSocketChannel 已經(jīng)綁定到端口上了.

由 Promise/Future 與 EventLoop 機制, 導致了 Channel 的注冊 發(fā)生在 Channel 的綁定 之前, 因此在 AbstractChannel#AbstractUnsafe.register0 中的 isActive() == false, if 判斷不成立, 最終就是 register0 中的 pipeline.fireChannelActive() 不會被調(diào)用.

云服務器 GPU云服務器記一次在阿里云搭建服務器記一次阿里云服務器集體中毒 java源碼javajava類的問題這篇文章手把手帶你閱讀Netty源碼【附

文章版權歸作者所有，未經(jīng)允許請勿轉載,若此文章存在違規(guī)行為，您可以聯(lián)系管理員刪除。

轉載請注明本文地址：http://www.ezyhdfw.cn/yun/65271.html

記一次XX前端面試

摘要：面試官說那我問你一個哲學的問題，為什么有數(shù)據(jù)結構這種東西哇，這是啥，巴拉巴拉扯了一通，大致就是物以類聚，人以群分，先人積累下來的經(jīng)驗，這些讓我們更方便處理數(shù)據(jù)啥的。前因，沒有比摸魚有趣的事了距離自己被外派(俗稱外包)出去，已經(jīng)過了快五個月，工作的話，很閑。人啊，一定保持好的習慣，懶惰是會上癮，日常摸魚，懷疑人生，我是誰，我在哪，我要干什么。中午吃飯的時候，收到了boss直聘的一條...

Shisui 2019-08-22 17:21 評論0 收藏0
如果連鐵將軍都不再可靠--記一次排查使用分布式輪候鎖+SESSION防訂單重復仍然加鎖失效問題經(jīng)歷

摘要：盡可能地將數(shù)據(jù)寫入，例如創(chuàng)建設置的都會將數(shù)據(jù)立即的寫入再來看看文檔怎么描述的看看這可愛的默認值我們終于知道了當我們不做任何設置時，默認采用的是方式顯而易見，使用方式能最大限度的減少與的交互，而在大多數(shù)場景下都是沒有問題的。 0.問題背景此次問題源于一次挺嚴重的生產(chǎn)事故：客戶的訂單被重復生成了，而出問題的代碼其實很簡單： // .... redisLockUtil.lock(membe...

econi 2019-08-15 17:50 評論0 收藏0
記一次開源學習--D2Admin 人人企業(yè)版

摘要：前言上個月月底開源組開源了使用適配人人企業(yè)版專業(yè)版的前端工程具體詳情見人人企業(yè)版適配發(fā)布。當然，也督促自己產(chǎn)出一篇相關的文章，來記錄這次有趣的學習之旅。 Created by huqi at 2019-5-5 13:01:14 Updated by huqi at 2019-5-20 15:57:37 前言上個月月底@D2開源組開源了使用 D2Admin 適配人人企業(yè)版(專業(yè)版) 的...

notebin 2019-05-21 19:46 評論0 收藏0

發(fā)表評論

登陸后可評論

0條評論

harriszh

男|高級講師

我要關注我要私信

TA的文章

BUI Webapp用于項目中的一點小心得

閱讀 1755·2019-08-30 15:54
前端面試題總結——綜合問題(持續(xù)更新中)

閱讀 3408·2019-08-26 17:15
在瀏覽器調(diào)起本地應用的方法

閱讀 3604·2019-08-26 13:49
leetcode 鏈表相關題目解析

閱讀 2630·2019-08-26 13:38
【刷算法】丑數(shù)

閱讀 2363·2019-08-26 12:08
webstorm預覽html配置localhost為本機ip地址

閱讀 3294·2019-08-26 10:41
籃球即時比分api接口調(diào)用示例代碼

閱讀 1424·2019-08-26 10:24
Webpack包教不包會

閱讀 3433·2019-08-23 18:35

亚洲中字慕日产2020,大陆极品少妇内射AAAAAA,无码av大香线蕉伊人久久,久久精品国产亚洲av麻豆网站

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

記一次有趣的 Netty 源碼問題

相關文章

記一次XX前端面試

**如果連鐵將軍都不再可靠--記一次排查使用分布式輪候鎖+SESSION防訂單重復仍然加鎖失效問題經(jīng)歷**

記一次開源學習--D2Admin 人人企業(yè)版

發(fā)表評論

0條評論

harriszh

男|高級講師

TA的文章

BUI Webapp用于項目中的一點小心得

前端面試題總結——綜合問題(持續(xù)更新中)

在瀏覽器調(diào)起本地應用的方法

leetcode 鏈表相關題目解析

【刷算法】丑數(shù)

webstorm預覽html配置localhost為本機ip地址

籃球即時比分api接口調(diào)用示例代碼

Webpack包教不包會

最新活動

資訊專欄INFORMATION COLUMN

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！

記一次有趣的 Netty 源碼問題

相關文章

發(fā)表評論

0條評論

男|高級講師

TA的文章

最新活動

上云采購季！| 2核2G4M爆款云服務器低至59元/年，更有多臺、長期優(yōu)惠，快來選購！