【Netty】(3)—源碼NioEventLoopGroup
阿新 • • 發佈:2018-12-06
自己 父類 同步問題 sso 例如 回避 log lean dal
netty(3)—源碼NioEventLoopGroup
一、概念
NioEventLoopGroup對象可以理解為一個線程池,內部維護了一組線程,每個線程負責處理多個Channel上的事件,而一個Channel只對應於一個線程,這樣可以回避多線程下的數據同步問題。
我們先回顧下 上篇博客的服務器代碼
// 定義一對線程組 // 主線程組, 用於接受客戶端的連接,但是不做任何處理,跟老板一樣,不做事 EventLoopGroup bossGroup = new NioEventLoopGroup(); // 從線程組, 老板線程組會把任務丟給他,讓手下線程組去做任務 EventLoopGroup workerGroup = new NioEventLoopGroup(); // netty服務器的創建, 輔助工具類,用於服務器通道的一系列配置 ServerBootstrap serverBootstrap = new ServerBootstrap(); serverBootstrap.group(bossGroup, workerGroup) //綁定兩個線程組 //省略......
職責
:
- 作為服務端
Acceptor 線程
,負責處理客戶端的請求接入。 - 作為客戶端
Connector 線程
,負責註冊監聽連接操作位,用於判斷異步連接結果。 - 作為
IO 線程
,監聽網絡讀操作位,負責從 SocketChannel 中讀取報文。 - 作為 IO 線程,負責向 SocketChannel 寫入報文發送給對方,如果發生寫半包,會自動註冊監聽寫事件,用 於後續繼續發送半包數據,直到數據全部發送完成。
- 作為
定時任務線程
,可以執行定時任務,例如鏈路空閑檢測和發送心跳消息等。 - 作為線程執行器可以執行普通的任務線程(Runnable)。
二、NioEventLoopGroup源碼分析
上面的代碼 創建bossGroup及workerGroup時,使用了NioEventLoopGroup的無參構造方法,本篇將從此無參構造入手,詳細分析NioEventLoopGroup的初始化過程。
/** * 1、首先我們看看NioEventLoopGroup的無參構造方法: * 作用:線程數為0 */ public NioEventLoopGroup() { this(0); } /** * 2、繼續調用構造函數。 * 作用:指定線程為0,且Executor為null */ public NioEventLoopGroup(int nThreads) { this(nThreads, (Executor) null); } /** * 3、繼續調用構造函數 * 作用:此構造方法它會指定selector的輔助類 "SelectorProvider.provider()" */ public NioEventLoopGroup(int nThreads, Executor executor) { this(nThreads, executor, SelectorProvider.provider()); } /** * 4、繼續調用構造函數 * 作用:初始化了一個默認的選擇策略工廠,用於生成select策略 */ public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider) { this(nThreads, executor, selectorProvider, DefaultSelectStrategyFactory.INSTANCE); } /** * 5、繼續調用構造函數 * 作用:指定拒絕策略:RejectedExecutionHandlers.reject() */ public NioEventLoopGroup(int nThreads, Executor executor, final SelectorProvider selectorProvider,final SelectStrategyFactory selectStrategyFactory) { super(nThreads, executor, selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject()); }
經過上面一系列的構造方法調用,此時參數值對應如下:
- ? nThreads: 0
- ? executor: null
- ? selectorProvider: SelectorProvider.provider()
- ? selectStrategyFactory: DefaultSelectStrategyFactory.INSTANCE
- ? 以及指定了拒絕策略: RejectedExecutionHandlers.reject()
/**
* 6、從這裏開始 調用父類 MultithreadEventLoopGroup 的構造函數
* 作用: 就是當指定的線程數為0時,使用默認的線程數DEFAULT_EVENT_LOOP_THREADS,
* 而DEFAULT_EVENT_LOOP_THREAD是在靜態代碼塊中就被執行。
*/
protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {
super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, executor, args);
}
/**
* 6.1 我們看下靜態代碼塊
* 作用:到這一步得出關鍵的一點:`如果初始化NioEventLoopGroup未指定線程數,默認是CPU核心數*2`。
*/
private static final int DEFAULT_EVENT_LOOP_THREADS;
static {
DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt(
"io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2))
}
/**
* 7、繼續調用父類 MultithreadEventLoopGroup 構造函數
* 作用:指定了一個EventExecutor的選擇工廠DefaultEventExecutorChooserFactory,
* 此工廠主要是用於選擇下一個可用的EventExecutor
*/
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) {
this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args);
}
/**
* 8、繼續調用父類 MultithreadEventLoopGroup 構造函數 這裏就是核心代碼 刪除部分非核心代碼
* 作用單獨分析
*/
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {
//1、
//executor校驗非空, 如果為空就創建ThreadPerTaskExecutor, 該類實現了 Executor接口
// 這個executor 是用來執行線程池中的所有的線程,也就是所有的NioEventLoop,其實從
//NioEventLoop構造器中也可以知道,NioEventLoop構造器中都傳入了executor這個參數。
if (executor == null) {
executor = new ThreadPerTaskExecutor(newDefaultThreadFactory());
}
//2、
//這裏的children數組, 其實就是線程池的核心實現,線程池中就是通過指定的線程數組來實現線程池;
//數組中每個元素其實就是一個EventLoop,EventLoop是EventExecutor的子接口。
children = new EventExecutor[nThreads];
//for循環實例化children數組,NioEventLoop對象
for (int i = 0; i < nThreads; i++) {
boolean success = false;
//3、
//newChild(executor, args) 函數在NioEventLoopGroup類中實現了,
// 實質就是就是存入了一個 NIOEventLoop類實例
children[i] = newChild(executor, args);
success = true;
}
//4、實例化線程工廠執行器選擇器: 根據children獲取選擇器
chooser = chooserFactory.newChooser(children);
//5、為每個EventLoop線程添加 線程終止監聽器
final FutureListener<Object> terminationListener = new FutureListener<Object>() {
@Override
public void operationComplete(Future<Object> future) throws Exception {
if (terminatedChildren.incrementAndGet() == children.length) {
terminationFuture.setSuccess(null);
}
}
};
//6、將children 添加到對應的set集合中去重, 表示只可讀。
Set<EventExecutor> childrenSet = new LinkedHashSet<EventExecutor>(children.length);
Collections.addAll(childrenSet, children);
readonlyChildren = Collections.unmodifiableSet(childrenSet);
}
}
/**
* 8.3.1 我們再來看下 newChild(executor, args) 裏的方法
* 我們可以看到 返回的就是一個 NioEventLoop
*/
@Override
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
return new NioEventLoop(this, executor, (SelectorProvider) args[0],
((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
}
我們再回顧總結一下:
- NioEventLoopGroup初始化時未指定線程數,那麽會使用默認線程數,即
線程數 = CPU核心數 * 2
; - 每個NioEventLoopGroup對象內部都有一組可執行的
NioEventLoop數組
,一個NIOEventLoop可以理解成就是一個線程。 - 所有的NIOEventLoop線程是使用相同的 executor、SelectorProvider、SelectStrategyFactory、RejectedExecutionHandler以及是屬於某一個NIOEventLoopGroup的。
這一點從 newChild(executor, args); 方法就可以看出:newChild()的實現是在NIOEventLoopGroup中實現的。 - 當有IO事件來時,需要從線程池中選擇一個線程出來執行,這時候的NioEventLoop選擇策略是由GenericEventExecutorChooser實現的, 並調用該類的next()方法。
- 每個NioEventLoopGroup對象都有一個NioEventLoop選擇器與之對應,其會根據NioEventLoop的個數,動態選擇chooser(如果是2的冪次方,則按位運算,否則使用普通的輪詢)
所以通過上面的分析,我們得出NioEventLoopGroup主要功能就是為了創建一定數量的NioEventLoop,而真正的重點就在NioEventLoop中,它是整個netty線程執行的關鍵。
有關NioEventLoop可以參考文章: [netty源碼分析]--EventLoopGroup與EventLoop
如果一個人充滿快樂,正面的思想,那麽好的人事物就會和他共鳴,而且被他吸引過來。同樣,一個人老帶悲傷,倒黴的事情也會跟過來。
——在自己心情低落的時候,告誡自己不要把負能量帶給別人。(大校12)
【Netty】(3)—源碼NioEventLoopGroup