Netty5.0的NioEventLoop源碼詳細分析

標簽： JAVA Netty 源碼線程 IO

了解Netty線程模型的小伙伴應該都知道，Netty的線程有兩個NioEventLoopGroup線程池，一個是boss線程池，一個是worker線程池，其中worker線程池的任務如下：

a.異步讀取通訊對端的消息，向ChannelPipeline發出讀事件

b.異步向通訊對端發送消息，調用ChannelPipeline發送消息接口

c.執行系統Task任務

d.執行定時任務

系統Task

通過調用NioEventLoop的execute(Runnable task)方法實現，創建它們的原因是當IO線程和用戶線程都在操作同一個資源時，

會發生鎖競爭的問題，所以將用戶線程封裝為一個Task，交給IO線程串行處理，實現局部無鎖化

定時Task

通過調用NioEventLoop的schedule(Runnable command,long delay,TimeUnit unit)實現，主要用于監控和檢查等定時動作

所以Netty的NioEventLoop并不是一個純粹的I/O線程，它還負責調度執行Task任務

下面看看NioEventLoop的類圖

作為NIO框架的Reactor線程，NioEventLoop需要處理網絡I/O讀寫事件，因此它必須聚合一個多路復用器對象--Selector

selector的初始化方法就是直接調用openSelector()方法

從上圖中可以看到，Netty對Selector的selectedKeys進行了優化，用戶可以通過io.netty.noKeySetOptimization開關決定

是否啟用該優化項，默認不打開優化。如果沒有開啟該優化，則由provider.openSelector()創建并打開selector之后就直接返回，

如果設置了開啟優化，則通過反射機制獲取到selectedKeys和publicSelectedKeys，并將這兩個屬性設為可寫，

然后在使用它們將新創建的selectedKeySet與selector綁定，并將新的selectedKeySet將原JDK中的selectedKeys替換。

上面就是多路復用器Selector的初始化過程，下面研究關鍵的run()方法。

    protected void run() {
        boolean oldWakenUp = this.wakenUp.getAndSet(false);

        try {
            if (this.hasTasks()) {
                this.selectNow();
            } else {
                this.select(oldWakenUp);
                if (this.wakenUp.get()) {
                    this.selector.wakeup();
                }
            }

            this.cancelledKeys = 0;
            this.needsToSelectAgain = false;
            int ioRatio = this.ioRatio;
            if (ioRatio == 100) {
                this.processSelectedKeys();
                this.runAllTasks();
            } else {
                long ioStartTime = System.nanoTime();
                this.processSelectedKeys();
                long ioTime = System.nanoTime() - ioStartTime;
                this.runAllTasks(ioTime * (long)(100 - ioRatio) / (long)ioRatio);
            }

            if (this.isShuttingDown()) {
                this.closeAll();
                if (this.confirmShutdown()) {
                    this.cleanupAndTerminate(true);
                    return;
                }
            }
        } catch (Throwable var8) {
            logger.warn("Unexpected exception in the selector loop.", var8);

            try {
                Thread.sleep(1000L);
            } catch (InterruptedException var7) {
                ;
            }
        }

        this.scheduleExecution();
    }

每次執行先將wakenUp還原為false，并將之前的wakeUp狀態保存到oldWakenUp變量中，這樣即使進入到后面的select(oldWakenUp)分支，如果有新任務到來，也能及時處理。

boolean oldWakenUp = this.wakenUp.getAndSet(false);

通過hasTasks（）方法判斷消息隊列當中是否有未處理的任務，如果有則調用selectNow()方法立即進行一次select操作，

看是否有準備就緒的Channel需要處理。

if (this.hasTasks()) {
                this.selectNow();
            }

Selector的selectNow()方法會立即觸發Selector的選擇操作，如果有準備就緒的Channel，則返回就緒Channel的集合，否則返回0。最后再判斷用戶是否調用了Selector的wakenUp()，如果有，則執行selector.wakeup()

    void selectNow() throws IOException {
        try {
            this.selector.selectNow();
        } finally {
            if (this.wakenUp.get()) {
                this.selector.wakeup();
            }

        }

    }

回到run()方法繼續分析，如果消息隊列中沒有待處理的消息，則執行select(oldWakenUp)方法

    private void select(boolean oldWakenUp) throws IOException {
        Selector selector = this.selector;

        try {
            int selectCnt = 0;
            long currentTimeNanos = System.nanoTime();
            long selectDeadLineNanos = currentTimeNanos + this.delayNanos(currentTimeNanos);

            while(true) {
                long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
                if (timeoutMillis <= 0L) {
                    if (selectCnt == 0) {
                        selector.selectNow();
                        selectCnt = 1;
                    }
                    break;
                }

                int selectedKeys = selector.select(timeoutMillis);
                ++selectCnt;
                if (selectedKeys != 0 || oldWakenUp || this.wakenUp.get() || this.hasTasks() || this.hasScheduledTasks()) {
                    break;
                }

                if (Thread.interrupted()) {
                    if (logger.isDebugEnabled()) {
                        logger.debug("Selector.select() returned prematurely because Thread.currentThread().interrupt() was called. Use NioEventLoop.shutdownGracefully() to shutdown the NioEventLoop.");
                    }

                    selectCnt = 1;
                    break;
                }

                long time = System.nanoTime();
                if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
                    selectCnt = 1;
                } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 && selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
                    logger.warn("Selector.select() returned prematurely {} times in a row; rebuilding selector.", selectCnt);
                    this.rebuildSelector();
                    selector = this.selector;
                    selector.selectNow();
                    selectCnt = 1;
                    break;
                }

                currentTimeNanos = time;
            }

            if (selectCnt > 3 && logger.isDebugEnabled()) {
                logger.debug("Selector.select() returned prematurely {} times in a row.", selectCnt - 1);
            }
        } catch (CancelledKeyException var13) {
            if (logger.isDebugEnabled()) {
                logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector - JDK bug?", var13);
            }
        }

    }

先取系統的納秒時間，調用delayNanos()方法計算獲得NioEventLoop中定時任務的觸發時間，計算下一個將要觸發的定時任務的剩余超時時間，將它轉換成毫秒，為超時時間增加0.5毫秒的調整值。對剩余的超時時間進行判斷，如果需要立即執行或者已經超時，則調用selector.selectNow()進行輪詢操作，將selectCnt設置為1，并退出當前循環。

            int selectCnt = 0;
            long currentTimeNanos = System.nanoTime();
            long selectDeadLineNanos = currentTimeNanos + this.delayNanos(currentTimeNanos);

            while(true) {
                long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
                if (timeoutMillis <= 0L) {
                    if (selectCnt == 0) {
                        selector.selectNow();
                        selectCnt = 1;
                    }
                    break;
                }

然后將定時操作剩余的超時時間作為參數進行select，每進行一次，就將計數器selectCnt加1，這個是為了下文解決JDK select的bug用的。

int selectedKeys = selector.select(timeoutMillis);
                ++selectCnt;

Select操作結束之后，需要對結果進行判斷，如果存在下列任意一種情況，則break操作

1.有Channel處于就緒狀態，即selectedKeys != 0 證明有讀寫操作需要jinxing

2.oldWakenUp為true

3.系統或者用戶調用了wakeup操作，喚醒當前的多路復用器

4.消息隊列當中有任務需要執行

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

如果本次Selector的輪詢結果為空，也沒有wakeup操作或是新的消息需要處理，則說明是個空輪詢，在JDK原生的NIO中，這可能觸發epoll的bug，它會導致Selector的空輪詢，使I/O線程一直處于100%狀態。這個問題在Netty中得到了修復，策略如下：

1.對Selector的select操作周期進行統計

2.每完成一次空的select操作進行一個計數

3.在某個周期內如果連續發生了N次（默認為512次）空輪詢，說明觸發了JDK NIO的epoll()死循環的bug.

                long time = System.nanoTime();
                if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
                    selectCnt = 1;
                } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 && selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
                    logger.warn("Selector.select() returned prematurely {} times in a row; rebuilding selector.", selectCnt);
                    this.rebuildSelector();
                    selector = this.selector;
                    selector.selectNow();
                    selectCnt = 1;
                    break;
                }

監測到Selector處于死循環的狀態下，會通過重建Selector來解決這個問題

    public void rebuildSelector() {
        if (!this.inEventLoop()) {
            this.execute(new Runnable() {
                public void run() {
                    NioEventLoop.this.rebuildSelector();
                }
            });
        } else {
            Selector oldSelector = this.selector;
            if (oldSelector != null) {
                Selector newSelector;
                try {
                    newSelector = this.openSelector();
                } catch (Exception var9) {
                    logger.warn("Failed to create a new Selector.", var9);
                    return;
                }

                int nChannels = 0;

                label69:
                while(true) {
                    try {
                        Iterator i$ = oldSelector.keys().iterator();

                        while(true) {
                            if (!i$.hasNext()) {
                                break label69;
                            }

                            SelectionKey key = (SelectionKey)i$.next();
                            Object a = key.attachment();

                            try {
                                if (key.isValid() && key.channel().keyFor(newSelector) == null) {
                                    int interestOps = key.interestOps();
                                    key.cancel();
                                    SelectionKey newKey = key.channel().register(newSelector, interestOps, a);
                                    if (a instanceof AbstractNioChannel) {
                                        ((AbstractNioChannel)a).selectionKey = newKey;
                                    }

                                    ++nChannels;
                                }
                            } catch (Exception var11) {
                                logger.warn("Failed to re-register a Channel to the new Selector.", var11);
                                if (a instanceof AbstractNioChannel) {
                                    AbstractNioChannel ch = (AbstractNioChannel)a;
                                    ch.unsafe().close(ch.unsafe().voidPromise());
                                } else {
                                    NioTask<SelectableChannel> task = (NioTask)a;
                                    invokeChannelUnregistered(task, key, var11);
                                }
                            }
                        }
                    } catch (ConcurrentModificationException var12) {
                        ;
                    }
                }

                this.selector = newSelector;

                try {
                    oldSelector.close();
                } catch (Throwable var10) {
                    if (logger.isWarnEnabled()) {
                        logger.warn("Failed to close the old Selector.", var10);
                    }
                }

                logger.info("Migrated " + nChannels + " channel(s) to the new Selector.");
            }
        }
    }

首先通過inEventLoop方法判斷是否是其它線程發起的rebuildSelector，如果是其它線程發起的，為了避免多個線程并發操作Selector和其它資源，則需要將rebuildSelector封裝成Task,放到NioEventLoop的消息隊列中，由NioEventLoop線程負責，這樣避免了線程安全問題。

if (!this.inEventLoop()) {
            this.execute(new Runnable() {
                public void run() {
                    NioEventLoop.this.rebuildSelector();
                }
            });
        }

接著通過openSelector新建并打開一個newSelector,通過循環，將原Selector上注冊時SocketChannel從舊的Selector上去除注冊，并重新注冊到新的Selector上，將newSelector賦個NioEventLoop,然后將老的Selector關閉。

通過銷毀舊的、有問題的多路復用器，使用新建的Selector，就可以解決空輪詢Selector導致的bug。

如果輪詢到了處于就緒狀態的SocketChannel,則需要處理網絡I/O事件

this.cancelledKeys = 0;
            this.needsToSelectAgain = false;
            int ioRatio = this.ioRatio;
            if (ioRatio == 100) {
                this.processSelectedKeys();
                this.runAllTasks();
            } else {
                long ioStartTime = System.nanoTime();
                this.processSelectedKeys();
                long ioTime = System.nanoTime() - ioStartTime;
                this.runAllTasks(ioTime * (long)(100 - ioRatio) / (long)ioRatio);
            }

其中processSelectedKeys（）代碼如下

    private void processSelectedKeys() {
        if (this.selectedKeys != null) {
            this.processSelectedKeysOptimized(this.selectedKeys.flip());
        } else {
            this.processSelectedKeysPlain(this.selector.selectedKeys());
        }

    }

由于默認沒有開啟selectedKeys優化，所以會調用processSelectedKeysPlain方法

    private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) {
        if (!selectedKeys.isEmpty()) {
            Iterator i = selectedKeys.iterator();

            while(true) {
                SelectionKey k = (SelectionKey)i.next();
                Object a = k.attachment();
                i.remove();
                if (a instanceof AbstractNioChannel) {
                    processSelectedKey(k, (AbstractNioChannel)a);
                } else {
                    NioTask<SelectableChannel> task = (NioTask)a;
                    processSelectedKey(k, task);
                }

                if (!i.hasNext()) {
                    break;
                }

                if (this.needsToSelectAgain) {
                    this.selectAgain();
                    selectedKeys = this.selector.selectedKeys();
                    if (selectedKeys.isEmpty()) {
                        break;
                    }

                    i = selectedKeys.iterator();
                }
            }

        }
    }

先對SelectedKeys進行保護性判斷，如果為空則返回。否則獲取SelectedKeys迭代器進行循環遍歷，獲取selectionKey和SocketChannel的附件對象，將已經選擇的選擇鍵從迭代器中刪除，防止下次被重復選擇和處理。

if (!selectedKeys.isEmpty()) {
            Iterator i = selectedKeys.iterator();

            while(true) {
                SelectionKey k = (SelectionKey)i.next();
                Object a = k.attachment();
                i.remove();

然后將SocketChannel附件對象進行判斷，如果包含AbstractNioChannel,則證明是SocketChannel或者是ServerSocketChannel,需要進行I/O讀寫相關的操作，否則就是NioTask,需要類型轉換為NioTask（由于Netty自身沒有實現NioTask）接口，所以通常系統不會執行該分支，除非用戶自行注冊該Task到多路復用器。

                if (a instanceof AbstractNioChannel) {
                    processSelectedKey(k, (AbstractNioChannel)a);
                } else {
                    NioTask<SelectableChannel> task = (NioTask)a;
                    processSelectedKey(k, task);
                }

從代碼中可以看到，接下來需要執行processSelectedKey。在該方法中進行I/O操作，首先從NioServerSocketChannel或者NioSocketChannel中獲取其內部類Unsafe，判斷選擇鍵是否可用，不可用則關閉unsafe，釋放連接資源

NioUnsafe unsafe = ch.unsafe();
        if (!k.isValid()) {
            unsafe.close(unsafe.voidPromise());
        }

如果選擇鍵可用，就獲取其值跟網絡操作位進行與運算。

 else {
            try {
                int readyOps = k.readyOps();
                if ((readyOps & 17) != 0 || readyOps == 0) {
                    unsafe.read();
                    if (!ch.isOpen()) {
                        return;
                    }
                }

                if ((readyOps & 4) != 0) {
                    ch.unsafe().forceFlush();
                }

                if ((readyOps & 8) != 0) {
                    int ops = k.interestOps();
                    ops &= -9;
                    k.interestOps(ops);
                    unsafe.finishConnect();
                }
            } catch (CancelledKeyException var5) {
                unsafe.close(unsafe.voidPromise());
            }

        }

如果是讀或者連接操作，則調用Unsafe的read方法。此處Unsafe的實現是個多態，對于NioServerSocketChannel,它的讀操作就是接受客戶端的TCP連接。

    protected int doReadMessages(List<Object> buf) throws Exception {
        SocketChannel ch = this.javaChannel().accept();

        try {
            if (ch != null) {
                buf.add(new NioSocketChannel(this, ch));
                return 1;
            }
        } catch (Throwable var6) {
            logger.warn("Failed to create a new channel from an accepted socket.", var6);

            try {
                ch.close();
            } catch (Throwable var5) {
                logger.warn("Failed to close a socket.", var5);
            }
        }

        return 0;
    }

對于NIOSocketChannel，它的讀操作就是從SocketChannel中讀取ByteBuffer

    protected int doReadBytes(ByteBuf byteBuf) throws Exception {
        return byteBuf.writeBytes(this.javaChannel(), byteBuf.writableBytes());
    }

如果網絡操作為寫，則證明有半包消息沒有發送完，通過調用forceFlush()使其繼續發送

                if ((readyOps & 4) != 0) {
                    ch.unsafe().forceFlush();
                }

如果網絡操作位為連接狀態，則需要對連接結果進行判讀

                if ((readyOps & 8) != 0) {
                    int ops = k.interestOps();
                    ops &= -9;
                    k.interestOps(ops);
                    unsafe.finishConnect();
                }

需要注意的是，在進行finishConnect判斷之前，需要將網絡操作位進行修改，注銷掉SelectionKey.OP_CONNECT。

處理完I/O事件之后，NioEventLoop需要執行非I/O操作的系統Task和定時任務,由于NioEventLoop需要同時處理I/O事件和

非I/O任務，為了保證兩者都能得到足夠的CPU時間被執行，Netty提供了I/O比例供用戶定制。如果I/O操作多于定時任務和Task，

則可以將I/O比例跳大，反之則調小，默認為50%

                long ioTime = System.nanoTime() - ioStartTime;
                this.runAllTasks(ioTime * (long)(100 - ioRatio) / (long)ioRatio);

即進行runAllTasks

    protected boolean runAllTasks(long timeoutNanos) {
        this.fetchFromScheduledTaskQueue();
        Runnable task = this.pollTask();
        if (task == null) {
            return false;
        } else {
            long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
            long runTasks = 0L;

            long lastExecutionTime;
            while(true) {
                try {
                    task.run();
                } catch (Throwable var11) {
                    logger.warn("A task raised an exception.", var11);
                }

                ++runTasks;
                if ((runTasks & 63L) == 0L) {
                    lastExecutionTime = ScheduledFutureTask.nanoTime();
                    if (lastExecutionTime >= deadline) {
                        break;
                    }
                }

                task = this.pollTask();
                if (task == null) {
                    lastExecutionTime = ScheduledFutureTask.nanoTime();
                    break;
                }
            }

            this.lastExecutionTime = lastExecutionTime;
            return true;
        }
    }

首先從定時任務消息隊列中彈出消息來處理，如果為空，則退出。

        this.fetchFromScheduledTaskQueue();
        Runnable task = this.pollTask();
        if (task == null) {
            return false;
        }

如果有，則循環執行定時任務，并且根據時間戳來判斷操作是否已經超過了分配給非I/O操作的超時時間，超過則退出，

默認每經過64次循環則進行一次上述判斷。防止由于非I/O任務過多導致I/O操作被長時間阻塞

            while(true) {
                try {
                    task.run();
                } catch (Throwable var11) {
                    logger.warn("A task raised an exception.", var11);
                }

                ++runTasks;
                if ((runTasks & 63L) == 0L) {
                    lastExecutionTime = ScheduledFutureTask.nanoTime();
                    if (lastExecutionTime >= deadline) {
                        break;
                    }
                }

                task = this.pollTask();
                if (task == null) {
                    lastExecutionTime = ScheduledFutureTask.nanoTime();
                    break;
                }
            }

runAllTasks方法執行完成之后，會判斷系統是否進入優雅停機狀態，如果處理關閉狀態，則需要調用closeAll方法，釋放資源，并放NioEventLoop線程退出循環，結束運行

            if (this.isShuttingDown()) {
                this.closeAll();
                if (this.confirmShutdown()) {
                    this.cleanupAndTerminate(true);
                    return;
                }
            }

closeAll（）方法里會遍歷所有的Channel，然后調用它的unsafe().close方法關閉所有鏈路，釋放線程池、ChannelPipeline和ChannelHandler等資源

NioEventLoop的源碼分析到此結束，歡迎大家一起討論。

本文鏈接：https://blog.csdn.net/xu_Melon/article/details/79229567

智能推薦

LiveData詳細分析

目錄介紹 01.LiveData是什么東西 02.使用LiveData的優勢 03.使用LiveData的步驟 04.簡單使用LiveData 05.observe()和observerForever() 06.LiveData原理介紹 07.observe訂閱源碼分析 08.setValue發送源碼分析 09.observeForever源碼 10.LiveData源碼總結 00.使用LiveD...

Lifecycle詳細分析

Lifecycle源碼分析目錄介紹 01.Lifecycle的作用是什么 02.Lifecycle的簡單使用 03.Lifecycle的使用場景 04.如何實現生命周期感知 05.注解方法如何被調用 06.addObserver調用分析 07.知識點梳理和總結一下 00.使用AAC實現bus事件總線利用LiveData實現事件總線，替代EventBus。充分利用了生命周期感知功能，可以在act...

ConcurrentHashmap 詳細分析

詳盡的分析 JDK8 后的ConcurrentHashmap，思路分析輔以源碼走讀，徹底讀懂 ConcurrentHashmap。簡介放入數據容器元素總數更新容器擴容協助擴容遍歷簡介在從 JDK8 開始，為了提高并發度，ConcurrentHashMap的源碼進行了很大的調整。在 JDK7 中，采用的是分段鎖的思路。簡單的說，就是ConcurrentHashMap是由多個HashM...

ION詳細分析

參考： http://blog.csdn.net/armwind/article/details/53454251?locationNum=2&fps=1 代碼路徑驅動代碼： kernel-3.18/drivers/staging/android/ion Native lib代碼: system\core\libion & vendor/mediatek/proprietary/...

CMA 詳細分析

關于CMA的config @LINUX/android/kernel/arch/arm/configs/msm8909_defconfig CONFIG_CMA=y 已經打開 # CONFIG_CMA_DEBUG is not set # # Default contiguous memory area size: # CONFIG_CMA_SIZE_MBYTES=8 //兩個配對定義 CONFI...

Netty5.0的NioEventLoop源碼詳細分析

智能推薦

猜你喜歡