腳本之家服務(wù)器常用軟件

快捷導(dǎo)航

分布式Netty源碼分析EventLoopGroup及介紹

更新時(shí)間：2022年03月24日 18:32:46 作者：乒乓狂魔

這篇文章主要介紹了分布式Netty源碼分析EventLoopGroup及介紹，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪

EventLoopGroup介紹

在前面一篇文章中提到了，EventLoopGroup主要負(fù)責(zé)2個(gè)事情，這里再重復(fù)下：

它主要包含2個(gè)方面的功能，注冊(cè)Channel和執(zhí)行一些Runnable任務(wù)。

功能1：先來(lái)看看注冊(cè)Channel

即將Channel注冊(cè)到Selector上，由Selector來(lái)調(diào)度Channel的相關(guān)事件，如讀、寫(xiě)、Accept等事件。

而EventLoopGroup的設(shè)計(jì)是，它包含多個(gè)EventLoop（每一個(gè)EventLoop通常內(nèi)部包含一個(gè)線(xiàn)程），在執(zhí)行上述注冊(cè)過(guò)程中是需要選擇其中的一個(gè)EventLoop來(lái)執(zhí)行上述注冊(cè)行為，這里就出現(xiàn)了一個(gè)選擇策略的問(wèn)題，該選擇策略接口是EventExecutorChooser，你也可以自定義一個(gè)實(shí)現(xiàn)。

從上面可以看到，EventLoopGroup做的工作大部分是一些總體性的工作如初始化上述多個(gè)EventLoop、EventExecutorChooser等，具體的注冊(cè)Channel還是交給它內(nèi)部的EventLoop來(lái)實(shí)現(xiàn)。

功能2：執(zhí)行一些Runnable任務(wù)

EventLoopGroup繼承了EventExecutorGroup，EventExecutorGroup也是EventExecutor的集合，EventExecutorGroup也是掌管著EventExecutor的初始化工作，EventExecutorGroup對(duì)于Runnable任務(wù)的執(zhí)行也是選擇內(nèi)部中的一個(gè)EventExecutor來(lái)做具體的執(zhí)行工作。

netty中很多任務(wù)都是異步執(zhí)行的，一旦當(dāng)前線(xiàn)程要對(duì)某個(gè)EventLoop執(zhí)行相關(guān)操作，如注冊(cè)Channel到某個(gè)EventLoop，如果當(dāng)前線(xiàn)程和所要操作的EventLoop內(nèi)部的線(xiàn)程不是同一個(gè)，則當(dāng)前線(xiàn)程就僅僅向EventLoop提交一個(gè)注冊(cè)任務(wù)，對(duì)外返回一個(gè)ChannelFuture。

總結(jié)：EventLoopGroup含有上述2種功能，它更多的是一個(gè)集合，但是具體的功能實(shí)現(xiàn)還是選擇內(nèi)部的一個(gè)item元素來(lái)執(zhí)行相關(guān)任務(wù)。這里的內(nèi)部item元素通常即實(shí)現(xiàn)了EventLoop，又實(shí)現(xiàn)了EventExecutor，如NioEventLoop等

繼續(xù)來(lái)看看EventLoopGroup的整體類(lèi)圖

從圖中可以看到有2路分支：

1 MultithreadEventLoopGroup：用于封裝多線(xiàn)程的初始化邏輯，指定線(xiàn)程數(shù)等，即初始化對(duì)應(yīng)數(shù)量的EventLoop，每個(gè)EventLoop分配到一個(gè)線(xiàn)程

上圖中的newChild方法，NioEventLoopGroup就采用NioEventLoop作為實(shí)現(xiàn)，EpollEventLoopGroup就采用EpollEventLoop作為實(shí)現(xiàn)

如NioEventLoopGroup的實(shí)現(xiàn)：

protected EventLoop newChild(Executor executor, Object... args) throws Exception {
    return new NioEventLoop(this, executor, (SelectorProvider) args[0],
        ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
}

2 EventLoop接口實(shí)現(xiàn)了EventLoopGroup接口，主要因?yàn)镋ventLoopGroup中的功能接口還是要靠?jī)?nèi)部的EventLoop來(lái)完成具體的操作

EventLoop介紹

EventLoop主要工作就是注冊(cè)Channel，并負(fù)責(zé)監(jiān)控管理Channel的讀寫(xiě)等事件，這就涉及到不同的監(jiān)控方式，linux下有3種方式來(lái)進(jìn)行事件監(jiān)聽(tīng)

select、poll、epoll

目前java的Selector接口的實(shí)現(xiàn)如下：

PollSelectorImpl：實(shí)現(xiàn)了poll方式

EPollSelectorImpl：實(shí)現(xiàn)了epoll方式

而Netty呢則使用如下：

NioEventLoop：采用的是jdk Selector接口（使用PollSelectorImpl的poll方式）來(lái)實(shí)現(xiàn)對(duì)Channel的事件檢測(cè)

EpollEventLoop：沒(méi)有采用jdk Selector的接口實(shí)現(xiàn)EPollSelectorImpl，而是Netty自己實(shí)現(xiàn)的epoll方式來(lái)實(shí)現(xiàn)對(duì)Channel的事件檢測(cè)，所以在EpollEventLoop中就不存在jdk的Selector。

NioEventLoop介紹

對(duì)于NioEventLoopGroup的功能，NioEventLoop都要做實(shí)際的實(shí)現(xiàn)，NioEventLoop既要實(shí)現(xiàn)注冊(cè)功能，又要實(shí)現(xiàn)運(yùn)行Runnable任務(wù)

對(duì)于注冊(cè)Channel：NioEventLoop將Channel注冊(cè)到NioEventLoop內(nèi)部的PollSelectorImpl上，來(lái)監(jiān)聽(tīng)該Channel的讀寫(xiě)事件

對(duì)于運(yùn)行Runnable任務(wù)：NioEventLoop的父類(lèi)的父類(lèi)SingleThreadEventExecutor實(shí)現(xiàn)了運(yùn)行Runnable任務(wù)，在SingleThreadEventExecutor中，有一個(gè)任務(wù)隊(duì)列還有一個(gè)分配的線(xiàn)程

private final Queue<Runnable> taskQueue;
private volatile Thread thread;

NioEventLoop在該線(xiàn)程中不僅要執(zhí)行Selector帶來(lái)的IO事件，還要不斷的從上述taskQueue中取出任務(wù)來(lái)執(zhí)行這些非IO事件。下面我們來(lái)詳細(xì)看下這個(gè)過(guò)程

protected void run() {
    for (;;) {
        try {
            switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
                case SelectStrategy.CONTINUE:
                    continue;
                case SelectStrategy.SELECT:
                    select(wakenUp.getAndSet(false));
                    if (wakenUp.get()) {
                        selector.wakeup();
                    }
                default:
                    // fallthrough
            }
            cancelledKeys = 0;
            needsToSelectAgain = false;
            final int ioRatio = this.ioRatio;
            if (ioRatio == 100) {
                processSelectedKeys();
                runAllTasks();
            } else {
                final long ioStartTime = System.nanoTime();

                processSelectedKeys();

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

            if (isShuttingDown()) {
                closeAll();
                if (confirmShutdown()) {
                    break;
                }
            }
        } catch (Throwable t) {
            ...
        }
    }
}

來(lái)詳細(xì)說(shuō)下這個(gè)過(guò)程：

1 計(jì)算當(dāng)前是否需要執(zhí)行select過(guò)程

如果當(dāng)前沒(méi)有Runnable任務(wù)，則執(zhí)行select（這個(gè)select過(guò)程稍后詳細(xì)來(lái)說(shuō)）。

如果當(dāng)前有Runnable任務(wù)，則要去執(zhí)行處理流程，此時(shí)順便執(zhí)行下selector.selectNow()，萬(wàn)一有事件發(fā)生那就賺了，沒(méi)有白走這次處理流程

2 根據(jù)IO任務(wù)的時(shí)間占比設(shè)置來(lái)執(zhí)行IO任務(wù)和非IO任務(wù)，即上面提到的Runnable任務(wù)

如果ioRatio=100則每次都是執(zhí)行全部的IO任務(wù)，執(zhí)行全部的非IO任務(wù) 默認(rèn)ioRatio=50，即一半時(shí)間用于處理IO任務(wù)，另一半時(shí)間用于處理非IO任務(wù)。怎么去控制非IO任務(wù)所占用時(shí)間呢？

這里是每執(zhí)行64個(gè)非IO任務(wù)（這里可能是每個(gè)非IO任務(wù)比較短暫，減少一些判斷帶來(lái)的消耗）就判斷下占用時(shí)間是否超過(guò)了上述時(shí)間限制

接下來(lái)詳細(xì)看下上述select過(guò)程

Selector selector = this.selector;
try {
    int selectCnt = 0;
    long currentTimeNanos = System.nanoTime();
    long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
    for (;;) {
        long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
        if (timeoutMillis <= 0) {
            if (selectCnt == 0) {
                selector.selectNow();
                selectCnt = 1;
            }
            break;
        }
        // If a task was submitted when wakenUp value was true, the task didn't get a chance to call
        // Selector#wakeup. So we need to check task queue again before executing select operation.
        // If we don't, the task might be pended until select operation was timed out.
        // It might be pended until idle timeout if IdleStateHandler existed in pipeline.
        if (hasTasks() && wakenUp.compareAndSet(false, true)) {
            selector.selectNow();
            selectCnt = 1;
            break;
        }
        int selectedKeys = selector.select(timeoutMillis);
        selectCnt ++;
        if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
            // - Selected something,
            // - waken up by user, or
            // - the task queue has a pending task.
            // - a scheduled task is ready for processing
            break;
        }
        if (Thread.interrupted()) {
            // Thread was interrupted so reset selected keys and break so we not run into a busy loop.
            // As this is most likely a bug in the handler of the user or it's client library we will
            // also log it.
            //
            // See https://github.com/netty/netty/issues/2426
            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) {
            // timeoutMillis elapsed without anything selected.
            selectCnt = 1;
        } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
                selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
            // The selector returned prematurely many times in a row.
            // Rebuild the selector to work around the problem.
            logger.warn(
                    "Selector.select() returned prematurely {} times in a row; rebuilding Selector {}.",
                    selectCnt, selector);
            rebuildSelector();
            selector = this.selector;
            // Select again to populate selectedKeys.
            selector.selectNow();
            selectCnt = 1;
            break;
        }
        currentTimeNanos = time;
    }
} catch (CancelledKeyException e) {
	...
}

1 首先計(jì)算此次select過(guò)程的截止時(shí)間

    protected long delayNanos(long currentTimeNanos) {
        ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
        if (scheduledTask == null) {
            return SCHEDULE_PURGE_INTERVAL;
        }
        return scheduledTask.delayNanos(currentTimeNanos);
    }

這里其實(shí)就是從一個(gè)定時(shí) 任務(wù)隊(duì)列中取出定時(shí)任務(wù)，如果有則計(jì)算出離當(dāng)前定時(shí)任務(wù)的下一次執(zhí)行時(shí)間之差，如果沒(méi)有則按照固定的1s作為select過(guò)程的時(shí)間

2 將當(dāng)前時(shí)間差轉(zhuǎn)化成ms

如果當(dāng)前時(shí)間差不足0.5ms的話(huà)，即timeoutMillis<=0，并且是第一次執(zhí)行，則認(rèn)為時(shí)間太短執(zhí)行執(zhí)行一次selectNow

3 如果有任務(wù)，則立即執(zhí)行一次selectNow，跳出for循環(huán)
4 然后就是普通的selector.select(timeoutMillis)

在這段時(shí)間內(nèi)如果有事件則跳出for循環(huán)，如果沒(méi)有事件則已經(jīng)花費(fèi)對(duì)應(yīng)的時(shí)間差了，再次執(zhí)行for循環(huán)，計(jì)算的timeoutMillis就會(huì)小于0，也會(huì)跳出for循環(huán)

在上述邏輯中，基本selectCnt都是1，不會(huì)出現(xiàn)很多次，而這里針對(duì)selectCnt有很多次的處理是基于一個(gè)情況：

 selector.select(timeoutMillis)

Selector的正常邏輯是一旦有事件就返回，沒(méi)有事件則最多等待timeoutMillis時(shí)間。然而底層操作系統(tǒng)實(shí)現(xiàn)可能有bug，會(huì)出現(xiàn)：即使沒(méi)有產(chǎn)生事件就直接返回了，并沒(méi)有按照要求等待timeoutMillis時(shí)間。

現(xiàn)在的解決辦法就是：記錄上述出現(xiàn)的次數(shù)，一旦超過(guò)512這個(gè)閾值（可設(shè)置），就重新建立新的Selector，并將之前的Channel也全部遷移到新的Selector上

至此，NioEventLoop的主邏輯流程就介紹完了，之后就該重點(diǎn)介紹其中對(duì)于IO事件的處理了。然后就會(huì)引出來(lái)ChannelPipeline的處理流程