最新国产好看的视频,伊人天堂AV在线,国产Aaaaaa视频,蜜臀视频在线观看一区,人妻av色图,密臀久久久精品影片,青青视频免费观看毛片,久草在线观看视,国产三级精品色情在线

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

 更新時間:2022年03月24日 18:32:46   作者:乒乓狂魔  
這篇文章主要介紹了分布式Netty源碼分析EventLoopGroup及介紹,有需要的朋友可以借鑒參考下,希望能夠有所幫助,祝大家多多進步,早日升職加薪

EventLoopGroup介紹

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

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

功能1:先來看看注冊Channel

即將Channel注冊到Selector上,由Selector來調(diào)度Channel的相關(guān)事件,如讀、寫、Accept等事件。

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

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

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

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

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

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

繼續(xù)來看看EventLoopGroup的整體類圖

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

  • 1 MultithreadEventLoopGroup:用于封裝多線程的初始化邏輯,指定線程數(shù)等,即初始化對應(yīng)數(shù)量的EventLoop,每個EventLoop分配到一個線程

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

如NioEventLoopGroup的實現(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接口實現(xiàn)了EventLoopGroup接口,主要因為EventLoopGroup中的功能接口還是要靠內(nèi)部的EventLoop來完成具體的操作

EventLoop介紹

EventLoop主要工作就是注冊Channel,并負責(zé)監(jiān)控管理Channel的讀寫等事件,這就涉及到不同的監(jiān)控方式,linux下有3種方式來進行事件監(jiān)聽

select、poll、epoll

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

PollSelectorImpl:實現(xiàn)了poll方式

EPollSelectorImpl:實現(xiàn)了epoll方式

而Netty呢則使用如下:

NioEventLoop:采用的是jdk Selector接口(使用PollSelectorImpl的poll方式)來實現(xiàn)對Channel的事件檢測

EpollEventLoop:沒有采用jdk Selector的接口實現(xiàn)EPollSelectorImpl,而是Netty自己實現(xiàn)的epoll方式來實現(xiàn)對Channel的事件檢測,所以在EpollEventLoop中就不存在jdk的Selector。

NioEventLoop介紹

對于NioEventLoopGroup的功能,NioEventLoop都要做實際的實現(xiàn),NioEventLoop既要實現(xiàn)注冊功能,又要實現(xiàn)運行Runnable任務(wù)

對于注冊Channel:NioEventLoop將Channel注冊到NioEventLoop內(nèi)部的PollSelectorImpl上,來監(jiān)聽該Channel的讀寫事件

對于運行Runnable任務(wù):NioEventLoop的父類的父類SingleThreadEventExecutor實現(xiàn)了運行Runnable任務(wù),在SingleThreadEventExecutor中,有一個任務(wù)隊列還有一個分配的線程

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

NioEventLoop在該線程中不僅要執(zhí)行Selector帶來的IO事件,還要不斷的從上述taskQueue中取出任務(wù)來執(zhí)行這些非IO事件。下面我們來詳細看下這個過程

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) {
            ...
        }
    }
}

來詳細說下這個過程:

  • 1 計算當前是否需要執(zhí)行select過程

如果當前沒有Runnable任務(wù),則執(zhí)行select(這個select過程稍后詳細來說)。

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

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

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

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

接下來詳細看下上述select過程

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 首先計算此次select過程的截止時間
    protected long delayNanos(long currentTimeNanos) {
        ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
        if (scheduledTask == null) {
            return SCHEDULE_PURGE_INTERVAL;
        }
        return scheduledTask.delayNanos(currentTimeNanos);
    }

這里其實就是從一個定時 任務(wù)隊列中取出定時任務(wù),如果有則計算出離當前定時任務(wù)的下一次執(zhí)行時間之差,如果沒有則按照固定的1s作為select過程的時間

  • 2 將當前時間差轉(zhuǎn)化成ms

如果當前時間差不足0.5ms的話,即timeoutMillis<=0,并且是第一次執(zhí)行,則認為時間太短執(zhí)行執(zhí)行一次selectNow

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

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

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

 selector.select(timeoutMillis)

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

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

至此,NioEventLoop的主邏輯流程就介紹完了,之后就該重點介紹其中對于IO事件的處理了。然后就會引出來ChannelPipeline的處理流程

EpollEventLoop介紹

EpollEventLoop和NioEventLoop的主流程邏輯基本上是差不多的,不同之處就在于EpollEventLoop用epoll方式替換NioEventLoop中的PollSelectorImpl的poll方式。

這里不再詳細說明了,之后會詳細的說明Netty的epoll方式和jdk中的epoll方式的區(qū)別。

后續(xù)

下一篇就要詳細描述下NioEventLoop對于IO事件的處理,即ChannelPipeline的處理流程。

以上就是分布式Netty源碼分析EventLoopGroup及介紹的詳細內(nèi)容,更多關(guān)于分布式Netty EventLoopGroup源碼分析的資料請關(guān)注腳本之家其它相關(guān)文章!

相關(guān)文章

  • mybatis嵌套循環(huán)map方式(高級用法)

    mybatis嵌套循環(huán)map方式(高級用法)

    這篇文章主要介紹了mybatis嵌套循環(huán)map方式(高級用法),具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2022-01-01
  • java控制臺實現(xiàn)學(xué)生信息管理系統(tǒng)(IO版)

    java控制臺實現(xiàn)學(xué)生信息管理系統(tǒng)(IO版)

    這篇文章主要為大家詳細介紹了java控制臺實現(xiàn)學(xué)生信息管理系統(tǒng)(IO版),文中示例代碼介紹的非常詳細,具有一定的參考價值,感興趣的小伙伴們可以參考一下
    2021-04-04
  • Springmvc @PathVariable的用法解析

    Springmvc @PathVariable的用法解析

    這篇文章主要介紹了Springmvc @PathVariable的用法解析,具有很好的參考價值,希望對大家有所幫助。如有錯誤或未考慮完全的地方,望不吝賜教
    2021-10-10
  • MyBatis框架簡介及入門案例詳解

    MyBatis框架簡介及入門案例詳解

    MyBatis是一個優(yōu)秀的持久層框架,它對jdbc的操作數(shù)據(jù)庫的過程進行封裝,使開發(fā)者只需要關(guān)注SQL本身,而不需要花費精力去處理例如注冊驅(qū)動、創(chuàng)建connection、創(chuàng)建statement、手動設(shè)置參數(shù)、結(jié)果集檢索等jdbc繁雜的過程代碼,本文將作為最終篇為大家介紹MyBatis的使用
    2022-08-08
  • IDEA如何將Java項目打包成可執(zhí)行的Jar包

    IDEA如何將Java項目打包成可執(zhí)行的Jar包

    在Java開發(fā)中,我們通常會將我們的項目打包成可執(zhí)行的Jar包,以便于在其他環(huán)境中部署和運行,本文將介紹如何使用IDEA集成開發(fā)環(huán)境將Java項目打包成可執(zhí)行的Jar包,感興趣的朋友一起看看吧
    2023-07-07
  • Java封裝數(shù)組實現(xiàn)在數(shù)組中查詢元素和修改元素操作示例

    Java封裝數(shù)組實現(xiàn)在數(shù)組中查詢元素和修改元素操作示例

    這篇文章主要介紹了Java封裝數(shù)組實現(xiàn)在數(shù)組中查詢元素和修改元素操作,結(jié)合實例形式分析了java針對數(shù)組元素查詢、修改的封裝操作實現(xiàn)技巧,需要的朋友可以參考下
    2020-03-03
  • JavaApi實現(xiàn)更新刪除及讀取節(jié)點

    JavaApi實現(xiàn)更新刪除及讀取節(jié)點

    這篇文章主要介紹了JavaApi實現(xiàn)更新刪除及讀取節(jié)點,文中通過示例代碼介紹的非常詳細,對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價值,需要的朋友可以參考下
    2020-05-05
  • Java中的SimpleDateFormat使用詳解

    Java中的SimpleDateFormat使用詳解

    SimpleDateFormat 是一個以國別敏感的方式格式化和分析數(shù)據(jù)的具體類。這篇文章主要介紹了Java中的SimpleDateFormat使用詳解,需要的朋友可以參考下
    2017-03-03
  • SpringBoot集成Tess4J實現(xiàn)OCR的示例代碼

    SpringBoot集成Tess4J實現(xiàn)OCR的示例代碼

    Tess4J是一個基于Tesseract OCR引擎的Java接口,可以用來識別圖像中的文本,說白了,就是封裝了它的API,讓Java可以直接調(diào)用,本文給大家介紹了SpringBoot集成Tess4J實現(xiàn)OCR的示例,需要的朋友可以參考下
    2024-12-12
  • Java提示解析時已到達文件結(jié)尾的解決方法

    Java提示解析時已到達文件結(jié)尾的解決方法

    在本篇文章中小編給大家分享了關(guān)于Java提示解析時已到達文件結(jié)尾的解決方法,需要的朋友們學(xué)習(xí)下。
    2019-07-07

最新評論

宁乡县| 广东省| 寿宁县| 静宁县| 洱源县| 措美县| 沽源县| 台山市| 黑水县| 乡城县| 揭西县| 大港区| 繁昌县| 曲周县| 龙胜| 濮阳市| 如东县| 宁海县| 云霄县| 都昌县| 文成县| 红河县| 保靖县| 汶上县| 和静县| 桓台县| 温宿县| 开阳县| 安徽省| 富民县| 成安县| 福清市| 哈巴河县| 澜沧| 桐城市| 班戈县| 阿勒泰市| 潞城市| 连江县| 祁门县| 明光市|