前言

通過上一章的學(xué)習(xí), 我們了解了Server啟動的大致流程, 有很多組件與模塊并沒有細(xì)講, 從這個章開始, 我們開始詳細(xì)剖析netty的各個組件, 并結(jié)合啟動流程, 將這些組件的使用場景及流程進(jìn)行一個詳細(xì)的說明

這一章主要學(xué)習(xí)NioEventLoop相關(guān)的知識,何為NioEventLoop? NioEventLoop是netty的一個線程, 在上一節(jié)我們創(chuàng)建兩個NioEventLoopGroup:

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

這里創(chuàng)建了兩個group, 我們提過這是boss線程組和worker線程組, 其實這兩個線程組就相當(dāng)于兩個NioEventLoop的集合, 默認(rèn)每個NioEventLoopGroup創(chuàng)建時, 如果不傳入線程數(shù), 會創(chuàng)建cpu核數(shù)*2個NioEventLoop線程, 其中boss線程通過輪詢處理Server的accept事件, 而完成accept事件之后, 就會創(chuàng)建客戶端channel, 通過一定的策略, 分發(fā)到worker線程進(jìn)行處理, 而worker線程, 則主要用于處理客戶端的讀寫事件

除了輪詢事件, EventLoop線程還維護(hù)了兩個隊列, 一個是延遲任務(wù)隊列, 另一個是普通任務(wù)隊列, 在進(jìn)行事件輪詢的同時, 如果隊列中有任務(wù)需要執(zhí)行則會去執(zhí)行隊列中的任務(wù)

一個NioEventLoop綁定一個selector用于處理多個客戶端channel, 但是一個客戶端channel只能被一個NioEventLoop處理, 具體關(guān)系如圖2-0-1所示:

圖中我們看到, 一個NioEventLoopGroup下有多個NioEventLoop線程, 而一個線程可以處理多個channel, 其中有個叫pipeline和handler的東西, 同學(xué)們可能比較陌生, 這是netty的事件傳輸機(jī)制, 每個pipeline和channel唯一綁定, 這里只需要稍作了解, 之后章節(jié)會帶大家詳細(xì)剖析

了解了這些概念, 我們繼續(xù)以小節(jié)的形式對NioEventLoop進(jìn)行剖析

第一節(jié):  NioEventLoopGroup之創(chuàng)建線程執(zhí)行器

首先回到第一章最開始的demo, 我們最初創(chuàng)建了兩個線程組:

EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup();

創(chuàng)建EventLoopGroup的構(gòu)造方法

這里, 我們跟隨創(chuàng)建EventLoopGroup的構(gòu)造方法, 來繼續(xù)學(xué)習(xí)NioEventLoopGroup的創(chuàng)建過程

以workerGroup為例我們跟進(jìn)其構(gòu)造方法:

public NioEventLoopGroup() {
    this(0);
}

繼續(xù)跟進(jìn)this(0):

public NioEventLoopGroup(int nThreads) {
    this(nThreads, (Executor) null);
}

這里的nThreads就是剛傳入的0, 繼續(xù)跟進(jìn):

public NioEventLoopGroup(int nThreads, Executor executor) {
    this(nThreads, executor, SelectorProvider.provider());
}

這里nThreads仍然為0, executor為null, 這個execute是用于開啟NioEventLoop線程所需要的線程執(zhí)行器, SelectorProvider.provider()是用于創(chuàng)建selector, 這個之后會講到

我們一直跟到構(gòu)造方法最后:

public NioEventLoopGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, 
                         final SelectorProvider selectorProvider, 
                         final SelectStrategyFactory selectStrategyFactory, 
                         final RejectedExecutionHandler rejectedExecutionHandler) {
    super(nThreads, executor, chooserFactory, selectorProvider, selectStrategyFactory, rejectedExecutionHandler);
}

這里調(diào)用了父類的構(gòu)造方法

跟進(jìn)super, 進(jìn)入了其父類MultithreadEventExecutorGroup的構(gòu)造方法中:

protected MultithreadEventLoopGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, 
                                 Object... args) {
    super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, chooserFactory, args);
}

這里我們看到, 如果傳入的線程數(shù)量參數(shù)為0, 則會給一個默認(rèn)值, 這個默認(rèn)值就是兩倍的CPU核數(shù), chooserFactory是用于創(chuàng)建線程選擇器, 之后會講到, 繼續(xù)跟代碼之后, 我們就看到了創(chuàng)建NioEventLoop的真正邏輯, 在MultithreadEventExecutorGroup類的構(gòu)造方法中

跟到MultithreadEventExecutorGroup類的構(gòu)造方法

protected MultithreadEventExecutorGroup(int nThreads, Executor executor, 
                                        EventExecutorChooserFactory chooserFactory, Object... args) {
    //代碼省略
    if (executor == null) {
        //創(chuàng)建一個新的線程執(zhí)行器(1)
        executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
    }
    //構(gòu)造NioEventLoop(2)
    children = new EventExecutor[nThreads];
    for (int i = 0; i < nThreads; i ++) {
        boolean success = false;
        try {
            children[i] = newChild(executor, args);
            success = true;
        } catch (Exception e) {
            throw new IllegalStateException("failed to create a child event loop", e);
        } finally {
           //代碼省略
        }
    }
    //創(chuàng)建線程選擇器(3)
    chooser = chooserFactory.newChooser(children);
    //代碼省略
}

這邊將代碼主要分為三個步驟:

1.創(chuàng)建線程執(zhí)行器

2.創(chuàng)建EventLoop

3.創(chuàng)建線程選擇器

這一小節(jié)我們主要剖析第1步, 創(chuàng)建線程執(zhí)行器

這里有個new DefaultThreadFactory()創(chuàng)建一個DefaultThreadFactory對象, 這個對象作為參數(shù)傳入ThreadPerTaskExecutor的構(gòu)造函數(shù),  DefaultThreadFactory顧名思義, 是一個線程工廠, 用于創(chuàng)建線程的, 簡單看下這個類的繼承關(guān)系:

public class DefaultThreadFactory implements ThreadFactory{//類體
}

我們繼續(xù)跟進(jìn)該類的構(gòu)造方法:

protected ThreadFactory newDefaultThreadFactory() {
    return new DefaultThreadFactory(getClass());
}

其中g(shù)etClass()就是當(dāng)前類的class對象, 而當(dāng)前類是NioEventLoopGroup

繼續(xù)跟進(jìn)到DefaultThreadFactory的構(gòu)造方法中:

public DefaultThreadFactory(Class<?> poolType) {
    this(poolType, false, Thread.NORM_PRIORITY);
}

poolType是NioEventLoop的class對象, Thread.NORM_PRIORITY是設(shè)置默認(rèn)的優(yōu)先級為5

繼續(xù)跟構(gòu)造方法:

public DefaultThreadFactory(Class<?> poolType, boolean daemon, int priority) {
    this(toPoolName(poolType), daemon, priority);
}

這里的toPoolName(poolType)是將線程組命名, 這里返回后結(jié)果是"nioEventLoopGroup"(開n頭小寫), daemon為false, priority為5

繼續(xù)跟構(gòu)造方法:

public DefaultThreadFactory(String poolName, boolean daemon, int priority) {
    this(poolName, daemon, priority, System.getSecurityManager() == null ?
            Thread.currentThread().getThreadGroup() : System.getSecurityManager().getThreadGroup());
}

 System.getSecurityManager() == null ? Thread.currentThread().getThreadGroup() : System.getSecurityManager().getThreadGroup() 

這段代碼是通過三目運算創(chuàng)建jdk底層的線程組

繼續(xù)跟this():

public DefaultThreadFactory(String poolName, boolean daemon, int priority, ThreadGroup threadGroup) {
    //省略驗證代碼
    //線程名字前綴
    prefix = poolName + '-' + poolId.incrementAndGet() + '-';
    this.daemon = daemon;
    //優(yōu)先級
    this.priority = priority;
    //初始化線程組
    this.threadGroup = threadGroup;
}

這里初始化了DefaultThreadFactory的屬性, prefix為poolName(也就是nioEventLoopGroup)+'-'+線程組id(原子自增)+'-'

以及初始化了優(yōu)先級和jdk底層的線程組等屬性

回到最初MultithreadEventExecutorGroup類的構(gòu)造方法中, 我們看繼續(xù)看第一步:

//創(chuàng)建一個新的線程執(zhí)行器(1)
executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());

我們繼續(xù)跟進(jìn)ThreadPerTaskExecutor的類中

public final class ThreadPerTaskExecutor implements Executor {
    private final ThreadFactory threadFactory;
    public ThreadPerTaskExecutor(ThreadFactory threadFactory) {
        if (threadFactory == null) {
            throw new NullPointerException("threadFactory");
        }
        this.threadFactory = threadFactory;
    }
    @Override
    public void execute(Runnable command) {
        //起一個線程
        threadFactory.newThread(command).start();
    }
}

我們發(fā)現(xiàn)這個類非常簡單, 繼承了jdk的Executor類, 從繼承關(guān)系中我就能猜想到, 而這個類就是用于開啟線程的線程執(zhí)行器

構(gòu)造方法傳入ThreadFactory類型的參數(shù), 這個ThreadFactory就是我們剛才剖析的DefaultThreadFactory, 這個類繼承了ThreadFactory, 所以在構(gòu)造方法中初始化了ThreadFactory類型的屬性

我們再看重寫的 execute(Runnable command) 方法, 傳入一個任務(wù), 然后由threadFactory對象創(chuàng)建一個線程執(zhí)行該任務(wù)

這個execute(Runnable command)方法, 其實就是用開開啟NioEventLoop線程用的, 那么NioEventLoop什么時候開啟的, 后面章節(jié)會進(jìn)行剖析

這樣, 通過 executor = new ThreadPerTaskExecutor(newDefaultThreadFactory()) 這種方式就返回了一個線程執(zhí)行器Executor, 用于開啟NioEventLoop線程

以上就是Netty組件NioEventLoopGroup創(chuàng)建線程執(zhí)行器源碼解析的詳細(xì)內(nèi)容，更多關(guān)于Netty NioEventLoopGroup線程執(zhí)行器的資料請關(guān)注腳本之家其它相關(guān)文章！

最新評論