Netty源碼分析NioEventLoop怎么執行select

本篇內容主要講解“Netty源碼分析NioEventLoop怎么執行select”，感興趣的朋友不妨來看看。本文介紹的方法操作簡單快捷，實用性強。下面就讓小編來帶大家學習“Netty源碼分析NioEventLoop怎么執行select”吧!

select操作的入口,NioEventLoop的run方法:

protected void run() {
    for (;;) {
        try {
            switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                case SelectStrategy.CONTINUE:
                    continue;
                case SelectStrategy.SELECT:
                    //輪詢io事件(1)
                    select(wakenUp.getAndSet(false));
                    if (wakenUp.get()) {
                        selector.wakeup();
                    }
                default:
            }
            cancelledKeys = 0;
            needsToSelectAgain = false;
            //默認是50
            final int ioRatio = this.ioRatio; 
            if (ioRatio == 100) {
                try {
                    processSelectedKeys();
                } finally {
                    runAllTasks();
                }
            } else {
                //記錄下開始時間
                final long ioStartTime = System.nanoTime();
                try {
                    //處理輪詢到的key(2)
                    processSelectedKeys();
                } finally {
                    //計算耗時
                    final long ioTime = System.nanoTime() - ioStartTime;
                    //執行task(3)
                    runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
                }
            }
        } catch (Throwable t) {
            handleLoopException(t);
        }
        //代碼省略
    }
}

代碼比較長, 其實主要分為三部分:

1. 輪詢io事件

2. 處理輪詢到的key

3. 執行task

這一小節, 主要剖析第一部分

輪詢io事件

首先switch塊中默認會走到SelectStrategy.SELECT中, 執行select(wakenUp.getAndSet(false))方法

參數wakenUp.getAndSet(false)代表當前select操作是未喚醒狀態

進入到select(wakenUp.getAndSet(false))方法中

private void select(boolean oldWakenUp) throws IOException {
    Selector selector = this.selector;
    try {
        int selectCnt = 0;
        //當前系統的納秒數
        long currentTimeNanos = System.nanoTime();
        //截止時間=當前時間+隊列第一個任務剩余執行時間
        long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
        for (;;) {
            //阻塞時間(毫秒)=(截止時間-當前時間+0.5毫秒)
            long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
            if (timeoutMillis <= 0) {
                if (selectCnt == 0) {
                    selector.selectNow();
                    selectCnt = 1;
                }
                break;
            }
            if (hasTasks() && wakenUp.compareAndSet(false, true)) {
                selector.selectNow();
                selectCnt = 1;
                break;
            }
            //進行阻塞式的select操作
            int selectedKeys = selector.select(timeoutMillis);
            //輪詢次數
            selectCnt ++;
            //如果輪詢到一個事件(selectedKeys != 0), 或者當前select操作需要喚醒(oldWakenUp), 
            //或者在執行select操作時已經被外部線程喚醒(wakenUp.get()), 
            //或者任務隊列已經有任務(hasTask), 或者定時任務隊列中有任務(hasScheduledTasks())
            if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
                break;
            }
            //省略
            //記錄下當前時間
            long time = System.nanoTime();
            //當前時間-開始時間>=超時時間(條件成立, 執行過一次select操作, 條件不成立, 有可能發生空輪詢)
            if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
                //代表已經進行了一次阻塞式select操作, 操作次數重置為1
                selectCnt = 1;
            } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 && selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
                //省略日志代碼
                //如果空輪詢的次數大于一個閾值(512), 解決空輪詢的bug
                rebuildSelector();
                selector = this.selector;
                selector.selectNow();
                selectCnt = 1;
                break;
            }
            currentTimeNanos = time;
        }
        //代碼省略
    } catch (CancelledKeyException e) {
        //省略
    }
}

首先通過 long currentTimeNanos = System.nanoTime() 獲取系統的納秒數

繼續往下看:

long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos)；

delayNanos(currentTimeNanos)代表距定時任務中第一個任務剩余多長時間, 這個時間+當前時間代表這次操作不能超過的時間, 因為超過之后定時任務不能嚴格按照預定時間執行, 其中定時任務隊列是已經按照執行時間有小到大排列好的隊列, 所以第一個任務則是最近需要執行的任務, selectDeadLineNanos就代表了當前操作不能超過的時間

然后就進入到了無限for循環

for循環中我們關注:

long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L

selectDeadLineNanos - currentTimeNanos+500000L 代表截止時間-當前時間+0.5毫秒的調整時間, 除以1000000表示將計算的時間轉化為毫秒數

最后算出的時間就是selector操作的阻塞時間, 并賦值到局部變量的timeoutMillis中

后面有個判斷 if(imeoutMillis<0) , 代表當前時間已經超過了最后截止時間+0.5毫秒,  selectCnt == 0 代表沒有進行select操作, 滿足這兩個條件, 則執行selectNow()之后, 將selectCnt賦值為1之后跳出循環

如果沒超過截止時間, 就進行了 if(hasTasks() && wakenUp.compareAndSet(false, true)) 判斷

這里我們關注hasTasks()方法, 這里是判斷當前NioEventLoop所綁定的taskQueue是否有任務, 如果有任務, 則執行selectNow()之后, 將selectCnt賦值為1之后跳出循環(跳出循環之后去執行任務隊列中的任務)

hasTasks()方法可以自己跟一下, 非常簡單

如果沒有滿足上述條件, 就會執行 int selectedKeys = selector.select(timeoutMillis) 進行阻塞式輪詢, 并且自增輪詢次數, 而后會進行如下判斷:

if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
    break;
}

selectedKeys != 0代表已經有輪詢到的事件, oldWakenUp代表當前select操作是否需要喚醒, wakenUp.get()說明已經被外部線程喚醒, hasTasks()代表任務隊列是否有任務, hasScheduledTasks()代表定時任務隊列是否任務, 滿足條件之一, 就跳出循環

long time = System.nanoTime() 記錄了當前的時間, 之后有個判斷:

 if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) 這里的意思是當前時間-阻塞時間>方法開始執行的時間, 這里說明已經完整的執行完成了一個阻塞的select()操作, 將selectCnt設置成1

如果此條件不成立, 說明沒有完整執行select()操作, 可能觸發了一次空輪詢, 根據前一個selectCnt++這步我們知道, 每觸發一次空輪詢selectCnt都會自增

之后會進入第二個判斷 SELECTOR_AUTO_REBUILD_THRESHOLD > 0 && selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD 

其中SELECTOR_AUTO_REBUILD_THRESHOLD默認是512, 這個判斷意思就是空輪詢的次數如果超過512次, 則會認為是發生了epoll bug, 這樣會通過rebuildSelector()方法重新構建selector, 然后將重新構建的selector賦值到局部變量selector, 執行一次selectNow(), 將selectCnt初始化1, 跳出循環

rebuildSelector()方法中, 看netty是如何解決epoll bug的

public void rebuildSelector() {
    //是否是由其他線程發起的
    if (!inEventLoop()) {
        //如果是其他線程發起的, 將rebuildSelector()封裝成任務隊列, 由NioEventLoop進行調用
        execute(new Runnable() {
            @Override
            public void run() {
                rebuildSelector();
            }
        });
        return;
    }
    final Selector oldSelector = selector;
    final Selector newSelector;
    if (oldSelector == null) {
        return;
    }
    try {
        //重新創建一個select
        newSelector = openSelector();
    } catch (Exception e) {
        logger.warn("Failed to create a new Selector.", e);
        return;
    }
    int nChannels = 0;
    for (;;) {
        try {
            //拿到舊select中所有的key
            for (SelectionKey key: oldSelector.keys()) {
                Object a = key.attachment();
                try {
                    Object a = key.attachment();
                    //代碼省略

                    //獲取key注冊的事件
                    int interestOps = key.interestOps();
                    //將key注冊的事件取消
                    key.cancel();
                    //注冊到重新創建的新的selector中
                    SelectionKey newKey = key.channel().register(newSelector, interestOps, a);
                    //如果channel是NioChannel
                    if (a instanceof AbstractNioChannel) {
                        //重新賦值
                        ((AbstractNioChannel) a).selectionKey = newKey;
                    }
                    nChannels ++;
                } catch (Exception e) {
                    //代碼省略
                }
            }
        } catch (ConcurrentModificationException e) {
            continue;
        }
        break;
    }
    selector = newSelector;
    //代碼省略
}

首先會判斷是不是當前NioEventLoop線程執行的, 如果不是, 則將構建方法封裝成task由當前NioEventLoop執行

 final Selector oldSelector = selector 表示拿到舊的selector

然后通過 newSelector = openSelector() 創建新的selector

通過for循環遍歷所有注冊在selector中的key

 Object a = key.attachment() 是獲取channel, 第一章講過, 在注冊時, 將自身作為屬性綁定在key上

for循環體中, 通過 int interestOps = key.interestOps() 獲取其注冊的事件

key.cancel()將注冊的事件進行取消

 SelectionKey newKey = key.channel().register(newSelector, interestOps, a) 將channel以及注冊的事件注冊在新的selector中

 if (a instanceof AbstractNioChannel) 判斷是不是NioChannel

如果是NioChannel, 則通過 ((AbstractNioChannel) a).selectionKey = newKey 將自身的屬性selectionKey賦值為新返回的key

 selector = newSelector 將自身NioEventLoop屬性selector賦值為新創建的newSelector

至此, 就是netty解決epoll bug的步驟, 其實就是創建一個新的selector, 將舊selector中注冊的channel和事件重新注冊到新的selector中, 然后將自身selector屬性替換成新創建的selector

到此，相信大家對“Netty源碼分析NioEventLoop怎么執行select”有了更深的了解，不妨來實際操作一番吧！這里是億速云網站，更多相關內容可以進入相關頻道進行查詢，關注我們，繼續學習！