介面定義 io.netty.channel.EventLoopGroup extends EventExecutorGroup

方法	說明
ChannelFuture register(Channel channel)	把一個channel註冊到一個EventLoop
ChannelFuture register(Channel channel, ChannelPromise promise);	同上

io.netty.channel.EventLoop extends OrderedEventExecutor, EventLoopGroup

方法	說明
EventLoopGroup parent()	得到建立這個eventLoop的EventLoopGroup

EventLoopGroup定義的主要方法是register, 這個方法的語義是把channel和eventLoop繫結在一起。一個channel對應一個eventLoop, 一個eventLoop會持有多個channel。 I/O執行緒EventLoopGroup的抽象實現 io.netty.channel.MultithreadEventLoopGroup extends MultithreadEventExecutorGroup implements EventLoopGroup io.netty.channel.SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop 兩個類主功能都是實現了EventLoopGroup定義的register方法 MultithreadEventLoopGroup public ChannelFuture register(Channel channel) { return next().register(channel); } public ChannelFuture register(Channel channel, ChannelPromise promise) { return next().register(channel, promise); } SingleThreadEventLoop public ChannelFuture register(Channel channel) { return register(channel, new DefaultChannelPromise(channel, this)); } public ChannelFuture register(final Channel channel, final ChannelPromise promise) { channel.unsafe().register(this, promise); return promise; } register的實現主要是為了呼叫Channel.Unsafe例項的register方法。 NIO實現 io.netty.channel.nio.NioEventLoopGroup extends MultithreadEventLoopGroup io.netty.channel.nio.NioEventLoop extends SingleThreadEventLoop NioEventLoopGroup是在MultithreadEventLoopGroup基礎上實現了對JDK NIO Selector的封裝, 它實現以下幾個功能:

• 建立selector
• 在selector上註冊channel感興趣的NIO事件
• 實現EventExecutor的run方法，定義NIO事件和Executor任務的處理流程。
• 把NIO事件轉換成對channel unsafe的呼叫或NioTask的呼叫
• 控制執行緒執行I/O操作和排隊任務的用時比例
• 處理epoll selector cpu 100%的bug

下面來具體分析這幾個功能的實現。 建立Selector NioEventLoop#openSelector()實現了建立selector的功能，預設情況下，使用SelectorProvider#openSelector()方法建立一個新個selector: final Selector unwrappedSelector = provider.openSelector(); 如果設定環境變數io.netty.noKeySetOptimization=true, 會建立一個selectedKeySet = new SelectedSelectionKeySet(), 然後使用java的反射機制把selector的selectedKeys和publicSelectedKeys替換成selectedKeySet，具體步驟是: 1.得到selector的真正型別: sun.nio.ch.SelectorImpl Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() { @Override public Object run() { try { return Class.forName( " sun.nio.ch.SelectorImpl", false, PlatformDependent.getSystemClassLoader()); } catch (Throwable cause) { return cause; } } }); final Class<?> selectorImplClass = (Class<?>) maybeSelectorImplClass; 2.替換selector是屬性unwrappedSelector Field selectedKeysField = selectorImplClass.getDeclaredField(" selectedKeys"); Field publicSelectedKeysField = selectorImplClass.getDeclaredField(" publicSelectedKeys"); selectedKeysField.set(unwrappedSelector, selectedKeySet); publicSelectedKeysField.set(unwrappedSelector, selectedKeySet); 之所以會設計一個這樣的優化選項，是因為一般情況下呼叫完selector的select或selectNow方法後需要呼叫Selector#selectedKeys()得到觸發NIO事件的的SelectableChannel，這樣優化之後，可以直接從selectedKeySet中得到已經觸發了NIO事件的SelectableChannel。 在selector上註冊channel感興趣的NIO事件 NioEventLoop提供了unwrappedSelector方法，這個方法返回了它建立好的Selector例項。這樣任何的外部類都可以把任意的SelectableChannel註冊到這selector上。在AbstractNioChannel中， doRegister方法的實現就是使用了這個方法: selectionKey = javaChannel().register(eventLoop().unwrappedSelector(), 0, this); 另外，它還提供了一個register方法: public void register(final SelectableChannel ch, final int interestOps, final NioTask<?> task) 這個方法會把task當成SelectableChannel的附件註冊到selector上: ch.register(selector, interestOps, task); 實現EventExecutor的run方法，定義NIO事件和Executor任務的處理流程 在NioEventLoop的run方法中實現NIO事件和EventExecutor的任務處理邏輯，這個run方法在io.netty.util.concurrent.SingleThreadEventExecutor中定義。在上一章中，我們看到了DefaultEventExecutor中是如何實現這個run方法的，這裡我們將要看到這run方法的另一個實現。和SingleThreadEventExecutor中的run方法相比，NioEventLoop的run方法不僅要及時地執行taskQueue中的任務，還要能及時地處理NIO事件，因此它會同時檢查selector中的NIO事件和和taskQueue佇列，任何一箇中有事件需要處理或有任務需要執行，它不會阻塞執行緒。同時它也保證了在沒有NIO事件和任務的情況下執行緒不會無謂的空轉浪費CUP資源。 run主要實現如下，為了更清晰的說明它的主要功能，我對原來的程式碼進行了一些刪減。 for(;;){ try{ / /phase1: 同時檢查NIO事件和任務 switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) { case SelectStrategy.CONTINUE: continue; case SelectStrategy.SELECT: select(wakenUp.getAndSet(false)); // 在taskQueue中沒有任務的時候執行select } // phase2: 進入處理NIO事件，執行executor任務 try{ // 處理NIO事件 processSelectedKeys(); }finally{ // 處理taskQueu中的任務 runAllTasks(); } }catch(Throwable t){ handleLoopException(t); } } run方法有兩個階段構成： phase1: 檢查NIO事件或executor任務，如果有任何的NIO事件或executor任務進入phase2。 這樣階段的主要工作在selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())和select中完成。 selectStrategy.calculateStrategy實現 selectStrategy.calculateStrategy(selectNowSupplier, hasTasks()) 這行程式碼的含義是: 如果hasTasks() == true, 呼叫以下selector#selectNow, 然後進入phase2。 否則呼叫select。這裡使用了strategy模式，預設的strategy實現是io.netty.channe.DefaultSelectStrategy implements SelectStrategy @Override public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception { return hasTasks ? selectSupplier.get() : SelectStrategy.SELECT; } DefaultSelectStrategy實現了SelectStrategy介面，這介面定義了兩個常量: int SELECT = -1; int CONTINUE = -2; 執行時selectSuppler引數傳入的是selectNowSupplier, 它的實現如下： private final IntSupplier selectNowSupplier = new IntSupplier() { @Override public int get() throws Exception { return selectNow(); } }; 這裡的get方法呼叫了selectNow, selectNow呼叫的是Selector#selectNew方法，這個方法的返回值是>=0。 hashTasks的傳入的引數是hasTask()的返回值: return !taskQueue.isEmpty(); 程式碼讀到這裡就會發現，使用預設的的SelectStrategy實現，calculateStrategy在hasTasks()==true時返回值>=0, hasTasks() == false時返回值是SelectStrategy.SELECT，不會返回SelectStrategy.CONTINUE。 select實現 select的執行邏輯是: 1. 計算超select方法的結束時間selectDeadLineNanos long currentTimeNanos = System.nanoTime(); long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos); 2. 進入迴圈，檢查超時--超時跳出迴圈。 long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L; if (timeoutMillis <= 0) { if (selectCnt == 0) { selector.selectNow(); selectCnt = 1; } break; } 3. 如果在select執行過程中有executor任務提交或可以當前的wakeUp由false變成true, 跳出迴圈 if (hasTasks() && wakenUp.compareAndSet(false, true)) { selector.selectNow(); selectCnt = 1; break; } 4. 呼叫selector#select等待NIO事件。 int selectedKeys = selector.select(timeoutMillis); selectCnt ++; 5. 如果滿足這些條件的任何一個，跳出迴圈: 有NIO事件、wakeUp的新舊值都是true、taskQueue中有任務、有定時任務到期。 if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) { break; } 6. 如果執行緒被中斷，跳出迴圈。 if (Thread.interrupted()) { break; } 7. 如果selector.select超時，沒有檢查到任何NIO事件, 會在下次迴圈開始時跳出迴圈。 如果每次超時，跳到第2步繼續下一次迴圈。 long time = System.nanoTime(); if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) { selectCnt = 1; } currentTimeNanos = time; select 最遲會在當前時間>= selectDeadLineNanos時返回，這個時間是最近一個到期的定時任務執行的時間，換言之如果沒有任何的NIO事件或executor任務，select會在定時任務到期時返回。如果沒有定時任務，delayNanos(currentTimeNanos)返回的值是 TimeUnit.SECONDS.toNanos(1)，即1秒。 select會在檢查到任何NIO事件或executor任務時返回，為了保證這點，在selector.select(timeoutMillis)前後都會呼叫hasTasks檢查executor任務，為了能在呼叫executet提交任務時喚醒selector.select，NioEventLoop覆蓋了SingleThreadEventExecutor的wake方法: protected void wakeup(boolean inEventLoop) { if (!inEventLoop && wakenUp.compareAndSet(false, true)) { selector.wakeup(); } } 這個方法會及時的喚醒selector.select, 保證新提交的任務可以得到及時的執行。 phase2: 進入處理NIO事件，執行executor任務 這個階段是先呼叫processSelectedKeys()處理NIO事件，然後掉用 runAllTasks()處理所有已經到期的定時任務和已經在排隊的任務。這個階段還實現了NIO事件和executor任務的用時比例管理，這個特性稍後會詳細分析。