一篇學(xué)會(huì)NioEventLoopGroup源碼解析
本文轉(zhuǎn)載自微信公眾號(hào)「源碼學(xué)徒」，作者皇甫嗷嗷叫 。轉(zhuǎn)載本文請(qǐng)聯(lián)系源碼學(xué)徒公眾號(hào)。
NioEventLoopGroup的初始化源碼
一、尋找源碼的過程
我們前面說到過，NioEventLoopGroup我們可以近乎把它看作是一個(gè)線程池，該線程池會(huì)執(zhí)行一個(gè)一個(gè)的任務(wù)，我們常用的NioEventLoopGroup大概有兩種，NioEventLoopGroup(int nThreads),NioEventLoopGroup()，即一個(gè)是指定線程數(shù)量的，一個(gè)是默認(rèn)指定線程數(shù)量的!這里我們以無參構(gòu)造為入口進(jìn)行分析!
- EventLoopGroup work = new NioEventLoopGroup();
- public NioEventLoopGroup() {
- this(0);
- }
當(dāng)我們使用默認(rèn)的數(shù)量的時(shí)候，他會(huì)傳遞一個(gè)0，我們繼續(xù)往下跟!
- public NioEventLoopGroup(int nThreads) {
- this(nThreads, (Executor) null);
- }
注意這里傳遞的參數(shù)是:0,null
- public NioEventLoopGroup(int nThreads, Executor executor) {
- //每個(gè) group維護(hù)一個(gè) SelectorProvider 主要用它獲取selector選擇器
- this(nThreads, executor, SelectorProvider.provider());
- }
這里面多傳遞了一個(gè) SelectorProvider.provider()，該方法是JDK NIO提供的API主要可以獲取NIO選擇器或者Channel,如下圖：
我們回歸主線繼續(xù)跟：
- public NioEventLoopGroup(
- int nThreads, Executor executor, final SelectorProvider selectorProvider) {
- this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE);
- }
這里多傳遞了一個(gè) DefaultSelectStrategy選擇策略，這在后面講解NioEventLoop會(huì)具體講解，不做闡述!
- public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,
- final SelectStrategyFactory selectStrategyFactory) {
- super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject());
- }
我們會(huì)發(fā)現(xiàn)這里還會(huì)默認(rèn)傳遞一個(gè)拒絕策略RejectedExecutionHandlers.reject()，這個(gè)拒絕策略是干嘛的呢?
- @Override
- public void rejected(Runnable task, SingleThreadEventExecutor executor) {
- throw new RejectedExecutionException();
- }
我們得到一個(gè)結(jié)論，當(dāng)某些條件觸發(fā)這個(gè)拒絕策略，那么他會(huì)拋出一個(gè)RejectedExecutionException異常，具體什么時(shí)候觸發(fā)，后續(xù)也會(huì)詳細(xì)說明，這里只需要記住就OK了!
我們繼續(xù)回到主線, 這里我們開始調(diào)用父類，還記得上一節(jié)課我們分析的NioEventLoopGroup的父類是誰(shuí)嗎?沒錯(cuò)是：MultithreadEventLoopGroup, 我們會(huì)進(jìn)入到MultithreadEventLoopGroup里面：
- protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
- //線程數(shù)量為0時(shí) 使用默認(rèn)的cpu * 2
- super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
- }
nThreads還記得是幾嗎?是0對(duì)不對(duì)，這里有個(gè)判斷，當(dāng)你的線程數(shù)量為0的時(shí)候，會(huì)使用DEFAULT_EVENT_LOOP_THREADS當(dāng)作線程池的數(shù)量，DEFAULT_EVENT_LOOP_THREADS是多少呢?
- DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
默認(rèn)是CPU的兩倍，所以我們現(xiàn)在得到一個(gè)結(jié)論，當(dāng)我們使用默認(rèn)的NioEventLoopGroup的時(shí)候，系統(tǒng)會(huì)默認(rèn)使用系統(tǒng)CPU核數(shù)*2當(dāng)作線程池的數(shù)量!
我們上一步傳遞過來的selectorProvider、拒絕策略、selectStrategyFactory被封裝為數(shù)組，并放在args[0]，args[1], args[2]的位置!
我們繼續(xù)回到主線，這里又再次調(diào)用到父類，MultithreadEventExecutorGroup:
- protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
- this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
- }
注意，這里又再次多傳遞了一個(gè)參數(shù)：DefaultEventExecutorChooserFactory一個(gè)選擇器工廠，這里會(huì)返回一個(gè)選擇器，他是DefaultEventExecutorChooserFactory類型的，具體分析后面會(huì)分析!我們繼續(xù)回到主線：
- protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {
- ..........后續(xù)源碼補(bǔ)充..........
- }
到這里我們終于看到了一大段代碼，這里是EventLoopGroup的主要邏輯，我們逐行分析：
二、構(gòu)建線程執(zhí)行器
1. 源碼解析
- //newDefaultThreadFactory 構(gòu)建線程工廠
- if (executor == null) {
- //創(chuàng)建并保存線程執(zhí)行器 執(zhí)行器 執(zhí)行任務(wù)的 默認(rèn)是 DefaultThreadFactory 線程池
- executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
- }
這里會(huì)判斷我們傳入的執(zhí)行器是否為空，否則就新建一個(gè)，我們還記得executor是什么值嗎?是null，對(duì)不對(duì)，所以它會(huì)進(jìn)入到這里的邏輯我們進(jìn)入到newDefaultThreadFactory源碼里面看一下：
newDefaultThreadFactory()主要邏輯
- protected ThreadFactory newDefaultThreadFactory() {
- return new DefaultThreadFactory(getClass());
- }
可以看到，這里向執(zhí)行器里面?zhèn)魅肓艘粋€(gè) DefaultThreadFactory一個(gè)默認(rèn)的線程工廠!
ThreadPerTaskExecutor主要邏輯：
- /**
- * io.netty.util.concurrent.DefaultThreadFactory#newThread(java.lang.Runnable)
- *
- * 執(zhí)行任務(wù) 每次執(zhí)行任務(wù)都會(huì)創(chuàng)建一個(gè)線程實(shí)體對(duì)象
- * @param command 線程
- */
- @Override
- public void execute(Runnable command) {
- //執(zhí)行任務(wù)
- threadFactory.newThread(command).start();
- }
我們發(fā)現(xiàn)，這里調(diào)用了一個(gè)我們傳入的線程工廠，創(chuàng)建了一個(gè)新的線程并調(diào)用start方法啟動(dòng)了起來，那么他是如何創(chuàng)建的呢? 我們進(jìn)入到newThread源碼里面查看，由于我們默認(rèn)使用的線程工廠是 DefaultThreadFactory, 所以，我們會(huì)進(jìn)入到 DefaultThreadFactory#newThread
- @Override
- public Thread newThread(Runnable r) {
- //創(chuàng)建一個(gè)線程 每次執(zhí)行任務(wù)的時(shí)候都會(huì)創(chuàng)建一個(gè)線程實(shí)體
- Thread t = newThread(FastThreadLocalRunnable.wrap(r), prefix + nextId.incrementAndGet());
- try {
- if (t.isDaemon() != daemon) {
- t.setDaemon(daemon);
- }
- if (t.getPriority() != priority) {
- t.setPriority(priority);
- }
- } catch (Exception ignored) {
- // Doesn't matter even if failed to set.
- }
- return t;
- }
這里沒有太多的操作，只是會(huì)將一個(gè) Runnable封裝為一個(gè) Thread進(jìn)行返回，我們重點(diǎn)關(guān)注一下這個(gè)Thread，它和我們傳統(tǒng)使用的Thread是一樣的嗎? 我們跟進(jìn)到 newThread方法看一下：
- protected Thread newThread(Runnable r, String name) {
- //Netty自己封裝的線程
- return new FastThreadLocalThread(threadGroup, r, name);
- }
邏輯很簡(jiǎn)單，就是將一個(gè)Thread包裝為Netty自定義的 FastThreadLocalThread，至于為什么，我們暫時(shí)不往下多做解釋，后續(xù)章節(jié)會(huì)很詳細(xì)的解釋它!
2. 線程執(zhí)行器總結(jié)
這里我們會(huì)創(chuàng)建一個(gè)線程執(zhí)行器 ThreadPerTaskExecutor，使用默認(rèn)的線程工廠DefaultThreadFactory，線程執(zhí)行器會(huì)將一個(gè)任務(wù)包裝為一個(gè) FastThreadLocalThread對(duì)象，然后調(diào)用start方法開啟一個(gè)新的線程執(zhí)行任務(wù)!
三、創(chuàng)建對(duì)應(yīng)數(shù)量的執(zhí)行器
- //創(chuàng)建執(zhí)行器數(shù)組 數(shù)量和預(yù)設(shè)線程數(shù)量一致
- children = new EventExecutor[nThreads];
- for (int i = 0; i < nThreads; i ++) {
- boolean success = false;
- try {
- //創(chuàng)建執(zhí)行器 開始創(chuàng)建執(zhí)行器 這里的執(zhí)行機(jī)估計(jì)就會(huì)EventLoop 是NioEventLoop
- children[i] = newChild(executor, args);
- success = true;
- } catch (Exception e) {
- .....省略不必要代碼
- } finally {
- .....省略不必要代碼
- }
- }
1. 源碼解析
- children = new EventExecutor[nThreads];
首先他會(huì)創(chuàng)建一個(gè)空的EventExecutor執(zhí)行器數(shù)組，然后遍歷填充!
還記得 nThreads是幾嗎? 默認(rèn)是CPU*2的大小，所以這里會(huì)創(chuàng)建 CPU * 2數(shù)量的執(zhí)行器! 我們發(fā)現(xiàn)，for循環(huán)中填充的主要邏輯是newChild，所以，我們進(jìn)入到 newChild方法, 這里提示一點(diǎn)，我們創(chuàng)建的Group對(duì)象是一個(gè)什么對(duì)象? 是NioEventLoopGroup對(duì)象對(duì)不對(duì)，所以我們這里會(huì)進(jìn)入到 NioEventLoopGroup#newChild方法：
- @Override
- protected EventLoop newChild(Executor executor, Object... args) throws Exception {
- EventLoopTaskQueueFactory queueFactory = args.length == 4 ? (EventLoopTaskQueueFactory) args[3] : null;
- return new NioEventLoop(this, executor, (SelectorProvider) args[0],
- ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2], queueFactory);
- }
我們傳遞過來 的args長(zhǎng)度為3,前面做過解析 ：
args[0]為 selectorProvider、args[1]為拒絕策略、args[2]為selectStrategyFactory
所以 queueFactory為null， 然后我們?cè)僦攸c(diǎn)關(guān)注 NioEventLoop對(duì)象，可以看出，newChild方法返回的是 NioEventLoop，那么我們初步就可以確定，EventExecutor數(shù)組里面存在的是NioEventLoop對(duì)象!至此，我們就不深究了，NioEventLoop的初始化源碼分析我會(huì)放到下一節(jié)課分析，這里我們可以確定一件事， EventExecutor數(shù)組里面存在的是NioEventLoop對(duì)象!我們繼續(xù)回到主線：
2. 執(zhí)行器數(shù)組總結(jié)
for循環(huán)完畢之后，此時(shí)的EventExecutor[nThreads];數(shù)組就被填充滿了，里面的每一個(gè)元素都是NioEventLoop對(duì)象，每一個(gè)NioEventLoop對(duì)象都包含一個(gè) ThreadPerTaskExecutor線程執(zhí)行器對(duì)象!
四、創(chuàng)建一個(gè)執(zhí)行器選擇器
1. 源碼解析
- chooser = chooserFactory.newChooser(children);
還記得 chooserFactory是什么類型的嗎? 是DefaultEventExecutorChooserFactory類型的，忘了的可以往上翻一下尋找源碼的過程中的代碼或者調(diào)試一下!
我們進(jìn)入到 DefaultEventExecutorChooserFactory#newChooser 源碼邏輯中，并傳入剛剛我們循環(huán)填充好的數(shù)組：
- @Override
- public EventExecutorChooser newChooser(EventExecutor[] executors) {
- //判斷2的冪 isPowerOfTwo
- if (isPowerOfTwo(executors.length)) {
- return new PowerOfTwoEventExecutorChooser(executors);
- } else {
- //簡(jiǎn)單的
- return new GenericEventExecutorChooser(executors);
- }
- }
可以看到，這里似乎有兩種情況，返回的是不同的策略對(duì)象，當(dāng)你的數(shù)組長(zhǎng)度是2的冪等次方的時(shí)候，返回的是 PowerOfTwoEventExecutorChooser對(duì)象，否則返回 GenericEventExecutorChooser對(duì)象，我們就兩種情況全部分析一下：
I、PowerOfTwoEventExecutorChooser
- PowerOfTwoEventExecutorChooser(EventExecutor[] executors) {
- this.executors = executors;
- }
- @Override
- public EventExecutor next() {
- //2的冪等性 實(shí)現(xiàn)這個(gè) 也能實(shí)現(xiàn)循環(huán)取數(shù)的
- //executors 就是NioEventLoop數(shù)組 按照2次冪求本次獲取的EventLoop是個(gè)啥
- return executors[idx.getAndIncrement() & executors.length - 1];
- }
這段代碼的主要邏輯是，取一個(gè)自增的CAS類，與數(shù)組長(zhǎng)度做&運(yùn)算，最終會(huì)出現(xiàn)循環(huán)取數(shù)的結(jié)果：
從上面的圖片可以基本看出來, 該功能可以實(shí)現(xiàn)一個(gè)循環(huán)取數(shù)的功能，每次達(dá)到數(shù)組的尾部部都會(huì)重新回到頭部重新獲取!
代碼案例：
- public static void main(String[] args) {
- String[] strings = {"第一個(gè)", "第二個(gè)", "第三個(gè)", "第四個(gè)"};
- AtomicInteger idx = new AtomicInteger();
- for (int i = 0; i < 9; i++) {
- System.out.println(strings[idx.getAndIncrement() & strings.length -1]);
- }
- }
結(jié)果集
- 第一個(gè)
- 第二個(gè)
- 第三個(gè)
- 第四個(gè)
- 第一個(gè)
- 第二個(gè)
- 第三個(gè)
- 第四個(gè)
- 第一個(gè)
II、GenericEventExecutorChooser
當(dāng)你的線程數(shù)量不是2的冪次方的時(shí)候，會(huì)走一個(gè)通用的選擇器，具體實(shí)現(xiàn)源碼如下：
- GenericEventExecutorChooser(EventExecutor[] executors) {
- this.executors = executors;
- }
- @Override
- public EventExecutor next() {
- //自增 取模 以達(dá)到循環(huán)的目的
- //假設(shè)executors 長(zhǎng)度為5 那么 不斷的循環(huán)就會(huì)不斷的得到 0 1 2 3 4 0 1 2 3 4。。。
- return executors[Math.abs(idx.getAndIncrement() % executors.length)];
- }
這個(gè)代碼就不用了我做演示了吧，他的功能和上面那個(gè)功能是一樣的能 能夠達(dá)到一個(gè)循環(huán)取數(shù)的功能
思考
為什么Netty要分為兩個(gè)策略類來實(shí)現(xiàn)呢，直接用第二種不行嗎?
Netty官方對(duì)性能的要求達(dá)到了極致，大家要知道位運(yùn)算速度要高于直接取模運(yùn)算的，所以Netty官方即使是這一點(diǎn)也做了一個(gè)優(yōu)化!
2. 執(zhí)行器選擇器總結(jié)
我們通過上述可以了解到，這里會(huì)通過一個(gè)選擇器工廠創(chuàng)建一個(gè)選擇器，并保存在NioEvenetLoopGroup中，調(diào)用該選擇器的next方法會(huì)返回一個(gè)NioEventLoop對(duì)象，其中的獲取方式是不斷的循環(huán)，依次獲取NioEventLoop對(duì)象，這也是一個(gè)NioEventLoop對(duì)SocketChannel為一對(duì)多的基礎(chǔ)!這都是后話!
NioEventLoopGroup源碼總結(jié)
- 創(chuàng)建一個(gè)線程執(zhí)行器，當(dāng)調(diào)用該線程執(zhí)行器的execute方法的時(shí)候，會(huì)講一個(gè)Runable對(duì)象包裝為Thread對(duì)象，再將Thread對(duì)象包裝為FastThreadLocalThread對(duì)象，然后啟動(dòng)起來! 簡(jiǎn)單來說，每調(diào)用一次execute方法，都去創(chuàng)建并啟動(dòng)一條新線程執(zhí)行任務(wù)!
- 創(chuàng)建一個(gè)執(zhí)行器數(shù)組，數(shù)組長(zhǎng)度與我們傳遞的數(shù)量有關(guān)，默認(rèn)為CPU*2個(gè)數(shù)量，然后再循環(huán)填充這個(gè)空數(shù)組，數(shù)組里面的元素是一個(gè)NioEventLoop對(duì)象，每一個(gè)NioEventLoop對(duì)會(huì)持有一個(gè)線程執(zhí)行器的引用!
- 創(chuàng)建一個(gè)執(zhí)行器選擇器，調(diào)用該執(zhí)行器選擇器的next方法可以返回一個(gè)NioEventLoop對(duì)象，內(nèi)部是進(jìn)行循環(huán)取數(shù)的，每一個(gè)NioEventLoop都可能會(huì)被多次獲取!
