Netty框架的主要線程就是I/O線程，線程模型設計的好壞，決定了系統的吞吐量、並發性和安全性等架構質量屬性。
一、Netty的線程模型
在討論Netty線程模型時候，一般首先會想到的是經典的Reactor線程模型，盡管不同的NIO框架對應Reactor模式的實現存在差異，但本質上還是遵循了Reactor的基礎線程模型。
Reactor的3中線程模型：
- Reactor單線程模型
- Reactor多線程模型
- 主從Reactor多線程模型
NioEventLoopGroup 與 Reactor 線程模型的對應關系 見《Netty中的三種Reactor（反應堆）》
Netty的線程模型並不是一成不變的，實際通過用戶的啟動配置參數來配置。
在創建ServerBootstrap類實例前，先創建兩個EventLoopGroup，它們實際上是兩個獨立的Reactor線程池，bossGroup負責接收客戶端的連接，workerGroup負責處理IO相關的讀寫操作，或者執行系統task、定時task等。
用於接收客戶端請求的線程池職責如下：
- 接收客戶端TCP連接，初始化Channel參數；
- 將鏈路狀態變更事件通知給ChannelPipeline；
處理IO操作的線程池職責如下：
- 異步讀取遠端數據，發送讀事件到ChannelPipeline；
- 異步發送數據到遠端，調用ChannelPipeline的發送消息接口；
- 執行系統調用Task；
- 執行定時任務Task，如空閑鏈路檢測等；
通過調整兩個EventLoopGroup的線程數、是否共享線程池等方式，Netty的Reactor線程模型可以在單線程、多線程和主從多線程間切換，用戶可以根據實際情況靈活配置。
為了提高性能，Netty在很多地方采用了無鎖化設計。例如在IO線程的內部進行串行操作，避免多線程競爭導致的性能下降。盡管串行化設計看上去CPU利用率不高，並發程度不夠，但是通過調整NIO線程池的線程參數，可以同時啟動多個串行化的線程並行運行，這種局部無鎖化的設計相比一個隊列——多個工作線程的模型性能更優。
它的設計原理如下
Netty的NioEventLoop讀取到消息之后，調用ChannelPipeline的fireChannelRead方法，只要用戶不主動切換線程，就一直由NioEventLoop調用用戶的Handler，期間不進行線程切換。這種串行化的處理方式避免了多線程操作導致的鎖競爭，從性能角度看是最優的。
Netty多線程編程的最佳實踐如下：
- 服務端創建兩個EventLoopGroup，用於邏輯隔離NIO acceptor和NIO IO線程；
- 盡量避免在用戶Handler里面啟動用戶線程（解碼后將POJO消息發送到后端業務線程除外）；
- 解碼要在NIO線程調用的解碼Handler中進行，不要切換到用戶線程中完成消息的解碼；
- 如果業務邏輯比較簡單，沒有復雜的業務邏輯計算，沒有可能阻塞線程的操作如磁盤操作、數據庫操作、網絡操作等，可以直接在NIO線程中進行業務邏輯操作，不用切換到用戶線程；
- 如果業務邏輯比較復雜，不要在NIO線程上操作，應將解碼后的POJO封裝成Task提交到業務線程池中執行，以保證NIO線程被盡快釋放，處理其他IO操作；
二、NioEventLoopGroup
NioEventLoopGroup 類層次結構
- 從類結構可知，NioEventLoopGroup是一個Schedule類型的線程池，線程池中的線程用數組存放，
EventLoopGroup(其實是MultithreadEventExecutorGroup) 內部維護一個類型為 EventExecutor children 數組, 其大小是 nThreads, 這樣就構成了一個線程池，線程池大小通過
在實例化 NioEventLoopGroup 時, 如果指定線程池大小, 則 nThreads 就是指定的值, 反之是處理器核心數 * 2; -
MultithreadEventExecutorGroup 中會調用 newChild 抽象方法(抽象方法 newChild 是在 NioEventLoopGroup 中實現的, 它返回一個 NioEventLoop 實例)來初始化 children 數組,也是在NioEventLoopGroup中調用NioEventLoop的構造函數來創建NioEventLoop；
@Override protected EventLoop newChild(Executor executor, Object... args) throws Exception { return new NioEventLoop(this, executor, (SelectorProvider) args[0]); }
-
NioEventLoop相關后面再講；
三、NioEventLoop
NioEventLoop 繼承於 SingleThreadEventLoop, 而 SingleThreadEventLoop 又繼承於 SingleThreadEventExecutor. SingleThreadEventExecutor 是 Netty 中對本地線程的抽象, 它內部有一個 Thread thread 屬性, 存儲了一個本地 Java 線程. 因此我們可以認為, 一個 NioEventLoop 其實和一個特定的線程綁定, 並且在其生命周期內, 綁定的線程都不會再改變。
3.1、NioEventLoop 類層次結構
NioEventLoop 的類層次結構圖還是比較復雜的, 不過我們只需要關注幾個重要的點即可. 首先 NioEventLoop 的繼承鏈如下:
NioEventLoop -> SingleThreadEventLoop -> SingleThreadEventExecutor -> AbstractScheduledEventExecutor
3.2、NioEventLoop實例化過程
從上圖可以看到, SingleThreadEventExecutor 有一個名為 thread 的 Thread 類型字段, 這個字段就代表了與 SingleThreadEventExecutor 關聯的本地線程。在 SingleThreadEventExecutor 構造器中, 通過 threadFactory.newThread 創建了一個新的 Java 線程. 在這個線程中所做的事情主要就是調用 SingleThreadEventExecutor.this.run() 方法, 而因為 NioEventLoop 實現了這個方法, 因此根據多態性, 其實調用的是 NioEventLoop.run() 方法。
protected SingleThreadEventExecutor( EventExecutorGroup parent, ThreadFactory threadFactory, boolean addTaskWakesUp) { this.parent = parent; this.addTaskWakesUp = addTaskWakesUp; thread = threadFactory.newThread(new Runnable() { @Override public void run() { boolean success = false; updateLastExecutionTime(); try { SingleThreadEventExecutor.this.run(); success = true; } catch (Throwable t) { logger.warn("Unexpected exception from an event executor: ", t); } finally { // 省略清理代碼 ... } } }); threadProperties = new DefaultThreadProperties(thread); taskQueue = newTaskQueue(); }
3.3、NioEventLoop的設計兩大功能：
1、是作為 IO 線程, 執行與 Channel 相關的 IO 操作, 包括 調用 select 等待就緒的 IO 事件、讀寫數據與數據的處理等；
2、第二個任務是作為任務隊列執行任務， 任務可以分為2類：
依次看這些功能的源碼吧：
對比2.1：上面2.1中的普通任務，NioEventLoop通過隊列來保存任務。在 SingleThreadEventLoop 中, 又實現了任務隊列的功能, 通過它, 我們可以調用一個 NioEventLoop 實例的 execute 方法來向任務隊列中添加一個 task, 並由 NioEventLoop 進行調度執行。
protected Queue<Runnable> newTaskQueue() { return new LinkedBlockingQueue<Runnable>(); }
添加任務的SingleThreadEventExecutor中的execute()方法：
@Override public void execute(Runnable task) { if (task == null) { throw new NullPointerException("task"); } boolean inEventLoop = inEventLoop(); if (inEventLoop) { addTask(task); } else { startThread(); addTask(task); if (isShutdown() && removeTask(task)) { reject(); } } if (!addTaskWakesUp) { wakeup(inEventLoop); } }
上面2.1中的普通task的執行：
protected boolean runAllTasks() { fetchFromScheduledTaskQueue(); Runnable task = pollTask(); if (task == null) { return false; } for (;;) { try { task.run(); } catch (Throwable t) { logger.warn("A task raised an exception.", t); } task = pollTask(); if (task == null) { lastExecutionTime = ScheduledFutureTask.nanoTime(); return true; } } }
對比2.2：schedule（定時） 任務處理
除了通過 execute 添加普通的 Runnable 任務外, 我們還可以通過調用 eventLoop.scheduleXXX 之類的方法來添加一個定時任務。
EventLoop 中實現任務隊列的功能在超類 SingleThreadEventExecutor 實現的, 而 schedule 功能的實現是在 SingleThreadEventExecutor 的父類, 即 AbstractScheduledEventExecutor 中實現的。
在 AbstractScheduledEventExecutor 中, 有以 scheduledTaskQueue 字段:
Queue<ScheduledFutureTask<?>> scheduledTaskQueue;
scheduledTaskQueue 是一個隊列(Queue), 其中存放的元素是 ScheduledFutureTask. 而 ScheduledFutureTask 我們很容易猜到, 它是對 Schedule 任務的一個抽象。我們來看一下 AbstractScheduledEventExecutor 所實現的 schedule 方法吧：
@Override public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) { ObjectUtil.checkNotNull(command, "command"); ObjectUtil.checkNotNull(unit, "unit"); if (delay < 0) { throw new IllegalArgumentException( String.format("delay: %d (expected: >= 0)", delay)); } return schedule(new ScheduledFutureTask<Void>( this, command, null, ScheduledFutureTask.deadlineNanos(unit.toNanos(delay)))); }
這是其中一個重載的 schedule, 當一個 Runnable 傳遞進來后, 會被封裝為一個 ScheduledFutureTask 對象, 這個對象會記錄下這個 Runnable 在何時運行、已何種頻率運行等信息。
當構建了 ScheduledFutureTask 后, 會繼續調用 另一個重載的 schedule 方法:
<V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) { if (inEventLoop()) { scheduledTaskQueue().add(task); } else { execute(new OneTimeTask() { @Override public void run() { scheduledTaskQueue().add(task); } }); } return task; }
在這個方法中, ScheduledFutureTask 對象就會被添加到 scheduledTaskQueue 中了。
任務的執行
當一個任務被添加到 taskQueue 后, 它是怎么被 EventLoop 執行的呢?
讓我們回到 NioEventLoop.run() 方法中, 在這個方法里, 會分別調用 processSelectedKeys() 和 runAllTasks() 方法, 來進行 IO 事件的處理和 task 的處理. processSelectedKeys() 方法我們已經分析過了, 下面我們來看一下 runAllTasks() 中到底有什么名堂吧。
runAllTasks 方法有兩個重載的方法, 一個是無參數的, 另一個有一個參數的. 首先來看一下無參數的 runAllTasks：
protected boolean runAllTasks() { fetchFromScheduledTaskQueue(); Runnable task = pollTask(); if (task == null) { return false; } for (;;) { try { task.run(); } catch (Throwable t) { logger.warn("A task raised an exception.", t); } task = pollTask(); if (task == null) { lastExecutionTime = ScheduledFutureTask.nanoTime(); return true; } } }
我們前面已經提到過, EventLoop 可以通過調用 EventLoop.execute 來將一個 Runnable 提交到 taskQueue 中, 也可以通過調用 EventLoop.schedule 來提交一個 schedule 任務到 scheduledTaskQueue 中。在此方法的一開始調用的 fetchFromScheduledTaskQueue() 其實就是將 scheduledTaskQueue 中已經可以執行的(即定時時間已到的 schedule 任務) 拿出來並添加到 taskQueue 中, 作為可執行的 task 等待被調度執行。
它的源碼如下:
private void fetchFromScheduledTaskQueue() { if (hasScheduledTasks()) { long nanoTime = AbstractScheduledEventExecutor.nanoTime(); for (;;) { Runnable scheduledTask = pollScheduledTask(nanoTime); if (scheduledTask == null) { break; } taskQueue.add(scheduledTask); } } }接下來 runAllTasks() 方法就會不斷調用 task = pollTask() 從 taskQueue 中獲取一個可執行的 task, 然后調用它的 run() 方法來運行此 task。
注意, 因為 EventLoop 既需要執行 IO 操作, 又需要執行 task, 因此我們在調用 EventLoop.execute 方法提交任務時, 不要提交耗時任務, 更不能提交一些會造成阻塞的任務, 不然會導致我們的 IO 線程得不到調度, 影響整個程序的並發量.
=======================
3.4、Netty中的IO處理循環 （上面1中的作為io線程，用於IO操作的源碼解析如下：）
3.4.1、回顧下Nio的Selector 的基本使用流程見《Java NIO系列教程（六） 多路復用器Selector》
-
通過 Selector.open() 打開一個 Selector.
-
將 Channel 注冊到 Selector 中, 並設置需要監聽的事件(interest set)
-
不斷重復:
-
調用 select() 方法
-
調用 selector.selectedKeys() 獲取 selected keys
-
迭代每個 selected key:
-
1) 從 selected key 中獲取 對應的 Channel 和附加信息(如果有的話)
-
2) 判斷是哪些 IO 事件已經就緒了, 然后處理它們. 如果是 OP_ACCEPT 事件, 則調用 "SocketChannel clientChannel = ((ServerSocketChannel) key.channel()).accept()" 獲取 SocketChannel, 並將它設置為 非阻塞的, 然后將這個 Channel 注冊到 Selector 中.
-
3) 根據需要更改 selected key 的監聽事件.
-
4) 將已經處理過的 key 從 selected keys 集合中刪除.
-
-
3.4.2、依次看NIO中的關鍵幾步分別在Netty中如何實現的：
private static SocketChannel newSocket(SelectorProvider provider) { ... return provider.openSocketChannel(); }
對比2：將 Channel 注冊到 Selector 中, 並設置需要監聽的事件(interest set) 的操作我們在《？》, 我們在來回顧一下, 在客戶端的 Channel 注冊過程中, 會有如下調用鏈:
Bootstrap.initAndRegister -> AbstractBootstrap.initAndRegister -> MultithreadEventLoopGroup.register -> SingleThreadEventLoop.register -> AbstractUnsafe.register -> AbstractUnsafe.register0 -> AbstractNioChannel.doRegister
在 AbstractUnsafe.register 方法中調用了 register0 方法:
@Override public final void register(EventLoop eventLoop, final ChannelPromise promise) { // 省略條件判斷和錯誤處理 AbstractChannel.this.eventLoop = eventLoop; register0(promise); }
register0 方法代碼如下:
private void register0(ChannelPromise promise) { boolean firstRegistration = neverRegistered; doRegister(); neverRegistered = false; registered = true; safeSetSuccess(promise); pipeline.fireChannelRegistered(); // Only fire a channelActive if the channel has never been registered. This prevents firing // multiple channel actives if the channel is deregistered and re-registered. if (firstRegistration && isActive()) { pipeline.fireChannelActive(); } }
register0 又調用了 AbstractNioChannel.doRegister:
@Override protected void doRegister() throws Exception { // 省略錯誤處理 selectionKey = javaChannel().register(eventLoop().selector, 0, this); }
在這里 javaChannel() 返回的是一個 Java NIO SocketChannel 對象, 我們將此 SocketChannel 注冊到前面第一步獲取的 Selector 中。
對比3：thread 的 run 循環
當 EventLoop.execute 第一次被調用時, 就會觸發 startThread() 的調用（3.2、NioEventLoop實例化過程）, 進而導致了 EventLoop 所對應的 Java 線程的啟動。
run()中的selector相關的用於IO操作的，網絡的讀取操作是在run方法中去執行的，首先看有沒有未執行的任務，有的話直接執行，否則就去輪訓看是否有就緒的Channel，如下：
@Override protected void run() {
//死循環，NioEventLoop的事件循環就在這里 for (;;) { oldWakenUp = wakenUp.getAndSet(false); try { if (hasTasks()) {//父類SingleThreadEventExecutor中定義的taskQueue是否有任務，如果有立刻執行（父類SingleThreadEventExecutor中還定義了delayedTaskQueue） selectNow(); } else {//如果沒有select則進行輪詢 select(oldWakenUp); if (wakenUp.get()) { selector.wakeup(); } } //當輪訓到有就緒的Channel時，就進行網絡的讀寫操作 cancelledKeys = 0; final long ioStartTime = System.nanoTime(); needsToSelectAgain = false; if (selectedKeys != null) { processSelectedKeysOptimized(selectedKeys.flip()); } else { processSelectedKeysPlain(selector.selectedKeys()); } final long ioTime = System.nanoTime() - ioStartTime; final int ioRatio = this.ioRatio; runAllTasks(ioTime * (100 - ioRatio) / ioRatio); if (isShuttingDown()) { closeAll(); if (confirmShutdown()) { break; } } } catch (Throwable t) { logger.warn("Unexpected exception in the selector loop.", t); // Prevent possible consecutive immediate failures that lead to // excessive CPU consumption. try { Thread.sleep(1000); } catch (InterruptedException e) { // Ignore. } } } }
對比3-IO 事件的輪詢
分支1：selectNow() 和 select(oldWakenUp) 之間有什么區別：
void selectNow() throws IOException { try { selector.selectNow(); } finally { // restore wakup state if needed if (wakenUp.get()) { selector.wakeup(); } } }
selector 字段正是 Java NIO 中的多路復用器 Selector. 那么這里 selector.selectNow() 就很好理解了, selectNow() 方法會檢查當前是否有就緒的 IO 事件, 如果有, 則返回就緒 IO 事件的個數; 如果沒有, 則返回0. 注意, selectNow() 是立即返回的, 不會阻塞當前線程. 當 selectNow() 調用后, finally 語句塊中會檢查 wakenUp 變量是否為 true, 當為 true 時, 調用 selector.wakeup() 喚醒 select() 的阻塞調用。
再來看一下 else 分支的 select(oldWakenUp) 方法:
select(oldWakenUp)中輪詢時，可能為空，也沒有wakeup操作或是最新的消息處理，則說明本次輪訓是一個空輪訓，此時會觸發jdk的epoll bug，它會導致Selector進行空輪訓，使i/o線程處於100%。為了避免這個bug。需要對selector進行統計：
private void select(boolean oldWakenUp) throws IOException { Selector selector = this.selector; try { int selectCnt = 0; long currentTimeNanos = System.nanoTime(); long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos); //當前時間延時1分鍾 for (;;) { long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L; if (timeoutMillis <= 0) { if (selectCnt == 0) { selector.selectNow(); selectCnt = 1; } break; } int selectedKeys = selector.select(timeoutMillis); selectCnt ++; //當輪詢到新的事件或者wakenUp或者有任務時跳出輪詢的循環 if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks()) { // Selected something, // waken up by user, or // the task queue has a pending task. break; } //否則，可能已經觸發jdk的epoll bug，通過下面各種策略退出 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); rebuildSelector(); selector = this.selector; // Select again to populate selectedKeys. selector.selectNow(); selectCnt = 1; break; } currentTimeNanos = System.nanoTime(); } if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS) { if (logger.isDebugEnabled()) { logger.debug("Selector.select() returned prematurely {} times in a row.", selectCnt - 1); } } } catch (CancelledKeyException e) { if (logger.isDebugEnabled()) { logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector - JDK bug?", e); } // Harmless exception - log anyway } }
對比3--IO 事件的處理
在 NioEventLoop.run() 方法中, 第一步是通過 select/selectNow 調用查詢當前是否有就緒的 IO 事件. 那么當有 IO 事件就緒時, 第二步自然就是處理這些 IO 事件啦.
首先讓我們來看一下 NioEventLoop.run 中循環的剩余部分:
//run()方法的中間一段 //... 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); } //...
上面列出的代碼中, 有兩個關鍵的調用, 第一個是 processSelectedKeys() 調用, 根據字面意思, 我們可以猜出這個方法肯定是查詢就緒的 IO 事件, 然后處理它; 第二個調用是 runAllTasks(), 這個方法我們也可以一眼就看出來它的功能就是運行 taskQueue 中的任務。
這里的代碼還有一個十分有意思的地方, 即 ioRatio. 那什么是 ioRatio呢? 它表示的是此線程分配給 IO 操作所占的時間比(即運行 processSelectedKeys 耗時在整個循環中所占用的時間). 例如 ioRatio 默認是 50, 則表示 IO 操作和執行 task 的所占用的線程執行時間比是 1 : 1. 當知道了 IO 操作耗時和它所占用的時間比, 那么執行 task 的時間就可以很方便的計算出來了。
設 IO 操作耗時為 ioTime, ioTime 占的時間比例為 ioRatio, 則: ioTime / ioRatio = taskTime / taskRatio taskRatio = 100 - ioRatio => taskTime = ioTime * (100 - ioRatio) / ioRatio
根據上面的公式, 當我們設置 ioRate = 70 時, 則表示 IO 運行耗時占比為70%, 即假設某次循環一共耗時為 100ms, 那么根據公式, 我們知道 processSelectedKeys() 方法調用所耗時大概為70ms(即 IO 耗時), 而 runAllTasks() 耗時大概為 30ms(即執行 task 耗時).
當 ioRatio 為 100 時, Netty 就不考慮 IO 耗時的占比, 而是分別調用 processSelectedKeys()、runAllTasks(); 而當 ioRatio 不為 100時, 則執行到 else 分支, 在這個分支中, 首先記錄下 processSelectedKeys() 所執行的時間(即 IO 操作的耗時), 然后根據公式, 計算出執行 task 所占用的時間, 然后以此為參數, 調用 runAllTasks().
我們這里先分析一下 processSelectedKeys() 方法調用, runAllTasks() 我們留到下一節再分析。
processSelectedKeys() 方法的源碼如下:
private void processSelectedKeys() { if (selectedKeys != null) { processSelectedKeysOptimized(selectedKeys.flip()); } else { processSelectedKeysPlain(selector.selectedKeys()); } }
這個方法中, 會根據 selectedKeys 字段是否為空, 而分別調用 processSelectedKeysOptimized 或 processSelectedKeysPlain. selectedKeys 字段是在調用 openSelector() 方法時, 根據 JVM 平台的不同, 而有設置不同的值, 在我所調試這個值是不為 null 的. 其實 processSelectedKeysOptimized 方法 processSelectedKeysPlain 沒有太大的區別,先看processSelectedKeysPlain():
processSelectedKey 中處理了三個事件, 分別是:
-
OP_READ, 可讀事件, 即 Channel 中收到了新數據可供上層讀取。
-
OP_WRITE, 可寫事件, 即上層可以向 Channel 寫入數據。
-
OP_CONNECT, 連接建立事件, 即 TCP 連接已經建立, Channel 處於 active 狀態。
簡單的說就是對網絡位判斷，當網絡位為寫的時候，則說明有半包消息沒有發送完成，需要繼續調用flush方法進行發送。后面的如果網絡操作位為連接狀態，則需要對連接結果進行判斷。
對於服務端處理連接的請求如下：
由於NioMessageUnsafe是AbstractNioMessageChannel的內部類，調用外部類doReadMessages()方法
下面是服務端NioServerSocketChannel中的該方法
@Override protected int doReadMessages(List<Object> buf) throws Exception { SocketChannel ch = javaChannel().accept(); try { if (ch != null) { buf.add(new NioSocketChannel(this, childEventLoopGroup().next(), ch)); return 1; } } catch (Throwable t) { logger.warn("Failed to create a new channel from an accepted socket.", t); try { ch.close(); } catch (Throwable t2) { logger.warn("Failed to close a socket.", t2); } } return 0; }
@Override protected int doReadBytes(ByteBuf byteBuf) throws Exception { return byteBuf.writeBytes(javaChannel(), byteBuf.writableBytes()); }
private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) { for (int i = 0;; i ++) { final SelectionKey k = selectedKeys[i]; if (k == null) { break; } // null out entry in the array to allow to have it GC'ed once the Channel close // See https://github.com/netty/netty/issues/2363 selectedKeys[i] = null; final Object a = k.attachment(); if (a instanceof AbstractNioChannel) { processSelectedKey(k, (AbstractNioChannel) a); } else { @SuppressWarnings("unchecked") NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a; processSelectedKey(k, task); } if (needsToSelectAgain) { // null out entries in the array to allow to have it GC'ed once the Channel close // See https://github.com/netty/netty/issues/2363 for (;;) { if (selectedKeys[i] == null) { break; } selectedKeys[i] = null; i++; } selectAgain(); // Need to flip the optimized selectedKeys to get the right reference to the array // and reset the index to -1 which will then set to 0 on the for loop // to start over again. // // See https://github.com/netty/netty/issues/1523 selectedKeys = this.selectedKeys.flip(); i = -1; } } }
上面代碼關鍵的點就兩個：迭代 selectedKeys 獲取就緒的 IO 事件, 然后為每個事件都調用 processSelectedKey 來處理它。
這里正好完美對應上了我們提到的 Selector 的使用流程中的第三步里操作。
還有一點需要注意的是, 我們可以調用 selectionKey.attach(object) 給一個 selectionKey 設置一個附加的字段, 然后可以通過 Object attachedObj = selectionKey.attachment() 獲取它. 上面代代碼正是通過了 k.attachment() 來獲取一個附加在 selectionKey 中的對象, 那么這個對象是什么呢? 它又是在哪里設置的呢? 我們再來回憶一下 SocketChannel 是如何注冊到 Selector 中的:
在客戶端的 Channel 注冊過程中, 會有如下調用鏈:
Bootstrap.initAndRegister -> AbstractBootstrap.initAndRegister -> MultithreadEventLoopGroup.register -> SingleThreadEventLoop.register -> AbstractUnsafe.register -> AbstractUnsafe.register0 -> AbstractNioChannel.doRegister
最后的 AbstractNioChannel.doRegister 方法會調用 SocketChannel.register 方法注冊一個 SocketChannel 到指定的 Selector:
@Override protected void doRegister() throws Exception { // 省略錯誤處理 selectionKey = javaChannel().register(eventLoop().selector, 0, this); }
特別注意一下 register 的第三個參數, 這個參數是設置 selectionKey 的附加對象的, 和調用 selectionKey.attach(object) 的效果一樣. 而調用 register 所傳遞的第三個參數是 this, 它其實就是一個 NioSocketChannel 的實例. 那么這里就很清楚了, 我們在將 SocketChannel 注冊到 Selector 中時, 將 SocketChannel 所對應的 NioSocketChannel 以附加字段的方式添加到了selectionKey 中.
再回到 processSelectedKeysOptimized 方法中, 當我們獲取到附加的對象后, 我們就調用 processSelectedKey 來處理這個 IO 事件:
final Object a = k.attachment(); if (a instanceof AbstractNioChannel) { processSelectedKey(k, (AbstractNioChannel) a); } else { @SuppressWarnings("unchecked") NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a; processSelectedKey(k, task); }
final long ioTime = System.nanoTime() - ioStartTime; final int ioRatio = this.ioRatio; runAllTasks(ioTime * (100 - ioRatio) / ioRatio); if (isShuttingDown()) { closeAll(); if (confirmShutdown()) { break; } }
protected boolean runAllTasks(long timeoutNanos) { fetchFromDelayedQueue(); //獲取SingleThreadEventExecutor中的taskQueue Runnable task = pollTask(); if (task == null) { return false; } final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos; long runTasks = 0; long lastExecutionTime; for (;;) { try { task.run(); } catch (Throwable t) { logger.warn("A task raised an exception.", t); } runTasks ++; // Check timeout every 64 tasks because nanoTime() is relatively expensive. // XXX: Hard-coded value - will make it configurable if it is really a problem. //當循環次數為64時，這時候會去比較上次執行時間和延時的關系，如果大於延時那么就退出，這里獲取naotime是每64次進行獲取一次。 //這么做的目的是一來是獲取naotime比較耗時，另外也不能長時間執行task，讓io阻塞，所以一般每64個任務就會返回 if ((runTasks & 0x3F) == 0) { lastExecutionTime = ScheduledFutureTask.nanoTime(); if (lastExecutionTime >= deadline) { break; } } task = pollTask(); if (task == null) { lastExecutionTime = ScheduledFutureTask.nanoTime(); break; } } this.lastExecutionTime = lastExecutionTime; return true; }
1、獲取SingleThreadEventExecutor中的taskQueue，如果沒有任務就退出；
2、當循環次數為64時，這時候會去比較上次執行時間和延時的關系，如果大於延時那么就退出，這里獲取naotime是每64次進行獲取一次。這么做的目的是一來是獲取naotime比較耗時，另外也不能長時間執行task，讓io阻塞，所以一般每64個任務就會返回；