Netty中NioEventLoopGroup的建立原始碼分析
NioEventLoopGroup的無參構造:
1 public NioEventLoopGroup() { 2 this(0); 3 }
呼叫了單參的構造:
1 public NioEventLoopGroup(int nThreads) { 2 this(nThreads, (Executor)null); 3 }
繼續看到雙參構造:
1 public NioEventLoopGroup(int nThreads, Executor executor) { 2 this(nThreads, executor, SelectorProvider.provider()); 3 }
在這裡是使用JDK中NIO的原生API:SelectorProvider的provider,產生了一個SelectorProvider物件呼叫,繼續呼叫三參構造。
關於SelectorProvider在我前面的部落格中有介紹過:【Java】NIO中Selector的建立原始碼分析,在Windows下預設建立了WindowsSelectorProvider物件。
繼續看三參構造:
1 public NioEventLoopGroup(int nThreads, ThreadFactory threadFactory, SelectorProvider selectorProvider) { 2 this(nThreads, threadFactory, selectorProvider, DefaultSelectStrategyFactory.INSTANCE); 3 }
在這裡建立了一個單例的DefaultSelectStrategyFactory 物件:
1 public final class DefaultSelectStrategyFactory implements SelectStrategyFactory { 2 public static final SelectStrategyFactory INSTANCE = new DefaultSelectStrategyFactory(); 3 4 private DefaultSelectStrategyFactory() { 5 } 6 7 public SelectStrategy newSelectStrategy() { 8 return DefaultSelectStrategy.INSTANCE; 9 } 10 }
DefaultSelectStrategyFactory實現的是SelectStrategyFactory 介面:
1 public interface SelectStrategyFactory { 2 SelectStrategy newSelectStrategy(); 3 }
該介面提供一個用來產生Select策略的方法,SelectStrategy介面如下:
1 public interface SelectStrategy { 2 int SELECT = -1; 3 int CONTINUE = -2; 4 5 int calculateStrategy(IntSupplier var1, boolean var2) throws Exception; 6 }
根據IntSupplier 和一個boolean值為Select策略提供了一個計算策略的方法。
在Netty中只提供了DefaultSelectStrategy這一種預設實現:
1 final class DefaultSelectStrategy implements SelectStrategy { 2 static final SelectStrategy INSTANCE = new DefaultSelectStrategy(); 3 4 private DefaultSelectStrategy() { 5 } 6 7 public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception { 8 return hasTasks ? selectSupplier.get() : -1; 9 } 10 }
其中IntSupplier :
1 public interface IntSupplier { 2 int get() throws Exception; 3 }
結合上面的來看,最終的選擇策略主要是根據IntSupplier的get值來得到的。
再回到構造:
1 public NioEventLoopGroup(int nThreads, ThreadFactory threadFactory, SelectorProvider selectorProvider, SelectStrategyFactory selectStrategyFactory) { 2 super(nThreads, threadFactory, new Object[]{selectorProvider, selectStrategyFactory, RejectedExecutionHandlers.reject()}); 3 }
這裡產生了一個拒絕策略:
1 public static RejectedExecutionHandler reject() { 2 return REJECT; 3 } 4 5 private static final RejectedExecutionHandler REJECT = new RejectedExecutionHandler() { 6 public void rejected(Runnable task, SingleThreadEventExecutor executor) { 7 throw new RejectedExecutionException(); 8 } 9 }; 10 11 public interface RejectedExecutionHandler { 12 void rejected(Runnable var1, SingleThreadEventExecutor var2); 13 }
將selectorProvider、selectStrategyFactory以及這個拒絕策略封裝在一個Object數組裡,再呼叫了父類MultithreadEventLoopGroup的構造:
1 protected MultithreadEventLoopGroup(int nThreads, ThreadFactory threadFactory, Object... args) { 2 super(nThreads == 0 ? DEFAULT_EVENT_LOOP_THREADS : nThreads, threadFactory, args); 3 }
在這裡對nThreads的大小進行了調整:
1 private static final int DEFAULT_EVENT_LOOP_THREADS = Math.max(1, SystemPropertyUtil.getInt("io.netty.eventLoopThreads", NettyRuntime.availableProcessors() * 2));
SystemPropertyUtil.getInt是根據key值"io.netty.eventLoopThreads",獲取系統配置值,在沒用設定時使用NettyRuntime.availableProcessors() * 2的值
NettyRuntime的availableProcessors實現如下:
1 synchronized int availableProcessors() { 2 if (this.availableProcessors == 0) { 3 int availableProcessors = SystemPropertyUtil.getInt("io.netty.availableProcessors", Runtime.getRuntime().availableProcessors()); 4 this.setAvailableProcessors(availableProcessors); 5 } 6 7 return this.availableProcessors; 8 }
還是一樣,根據key值"io.netty.availableProcessors",獲取系統配置值,在沒用設定時使用Runtime.getRuntime().availableProcessors(),是用來獲取處理器的個數。
這樣保證了在預設情況下nThreads的大小是總是cpu個數的2倍。
繼續回到構造,MultithreadEventLoopGroup繼續呼叫父類MultithreadEventExecutorGroup的構造:
1 protected MultithreadEventExecutorGroup(int nThreads, Executor executor, Object... args) { 2 this(nThreads, executor, DefaultEventExecutorChooserFactory.INSTANCE, args); 3 }
在這裡又初始化了一個單例的DefaultEventExecutorChooserFactory物件:
1 public static final DefaultEventExecutorChooserFactory INSTANCE = new DefaultEventExecutorChooserFactory();
DefaultEventExecutorChooserFactory 實現的是EventExecutorChooserFactory介面:
1 public interface EventExecutorChooserFactory { 2 EventExecutorChooserFactory.EventExecutorChooser newChooser(EventExecutor[] var1); 3 4 public interface EventExecutorChooser { 5 EventExecutor next(); 6 } 7 }
DefaultEventExecutorChooserFactory 的具體實現:
1 public EventExecutorChooser newChooser(EventExecutor[] executors) { 2 return (EventExecutorChooser)(isPowerOfTwo(executors.length) ? new DefaultEventExecutorChooserFactory.PowerOfTwoEventExecutorChooser(executors) : new DefaultEventExecutorChooserFactory.GenericEventExecutorChooser(executors)); 3 } 4 5 private static boolean isPowerOfTwo(int val) { 6 return (val & -val) == val; 7 }
isPowerOfTwo是用來檢查executors的大小是否是二的整數次方,若是二的整數次方,產生PowerOfTwoEventExecutorChooser,反之產生GenericEventExecutorChooser:
1 private static final class GenericEventExecutorChooser implements EventExecutorChooser { 2 private final AtomicInteger idx = new AtomicInteger(); 3 private final EventExecutor[] executors; 4 5 GenericEventExecutorChooser(EventExecutor[] executors) { 6 this.executors = executors; 7 } 8 9 public EventExecutor next() { 10 return this.executors[Math.abs(this.idx.getAndIncrement() % this.executors.length)]; 11 } 12 } 13 14 private static final class PowerOfTwoEventExecutorChooser implements EventExecutorChooser { 15 private final AtomicInteger idx = new AtomicInteger(); 16 private final EventExecutor[] executors; 17 18 PowerOfTwoEventExecutorChooser(EventExecutor[] executors) { 19 this.executors = executors; 20 } 21 22 public EventExecutor next() { 23 return this.executors[this.idx.getAndIncrement() & this.executors.length - 1]; 24 } 25 }
這兩種其實都是用了取模運算,只不過因為二的整數次方的特殊性而使用位運算。
回到構造,MultithreadEventExecutorGroup繼續呼叫本省的構造:
1 private final EventExecutor[] children; 2 private final Set<EventExecutor> readonlyChildren; 3 private final AtomicInteger terminatedChildren; 4 private final Promise<?> terminationFuture; 5 private final EventExecutorChooser chooser; 6 7 protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) { 8 this.terminatedChildren = new AtomicInteger(); 9 this.terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE); 10 if (nThreads <= 0) { 11 throw new IllegalArgumentException(String.format("nThreads: %d (expected: > 0)", nThreads)); 12 } else { 13 if (executor == null) { 14 executor = new ThreadPerTaskExecutor(this.newDefaultThreadFactory()); 15 } 16 17 this.children = new EventExecutor[nThreads]; 18 19 int j; 20 for(int i = 0; i < nThreads; ++i) { 21 boolean success = false; 22 boolean var18 = false; 23 24 try { 25 var18 = true; 26 this.children[i] = this.newChild((Executor)executor, args); 27 success = true; 28 var18 = false; 29 } catch (Exception var19) { 30 throw new IllegalStateException("failed to create a child event loop", var19); 31 } finally { 32 if (var18) { 33 if (!success) { 34 int j; 35 for(j = 0; j < i; ++j) { 36 this.children[j].shutdownGracefully(); 37 } 38 39 for(j = 0; j < i; ++j) { 40 EventExecutor e = this.children[j]; 41 42 try { 43 while(!e.isTerminated()) { 44 e.awaitTermination(2147483647L, TimeUnit.SECONDS); 45 } 46 } catch (InterruptedException var20) { 47 Thread.currentThread().interrupt(); 48 break; 49 } 50 } 51 } 52 53 } 54 } 55 56 if (!success) { 57 for(j = 0; j < i; ++j) { 58 this.children[j].shutdownGracefully(); 59 } 60 61 for(j = 0; j < i; ++j) { 62 EventExecutor e = this.children[j]; 63 64 try { 65 while(!e.isTerminated()) { 66 e.awaitTermination(2147483647L, TimeUnit.SECONDS); 67 } 68 } catch (InterruptedException var22) { 69 Thread.currentThread().interrupt(); 70 break; 71 } 72 } 73 } 74 } 75 76 this.chooser = chooserFactory.newChooser(this.children); 77 FutureListener<Object> terminationListener = new FutureListener<Object>() { 78 public void operationComplete(Future<Object> future) throws Exception { 79 if (MultithreadEventExecutorGroup.this.terminatedChildren.incrementAndGet() == MultithreadEventExecutorGroup.this.children.length) { 80 MultithreadEventExecutorGroup.this.terminationFuture.setSuccess((Object)null); 81 } 82 83 } 84 }; 85 EventExecutor[] var24 = this.children; 86 j = var24.length; 87 88 for(int var26 = 0; var26 < j; ++var26) { 89 EventExecutor e = var24[var26]; 90 e.terminationFuture().addListener(terminationListener); 91 } 92 93 Set<EventExecutor> childrenSet = new LinkedHashSet(this.children.length); 94 Collections.addAll(childrenSet, this.children); 95 this.readonlyChildren = Collections.unmodifiableSet(childrenSet); 96 } 97 }
首先是對terminatedChildren的初始化,沒什麼好說的,對terminationFuture的初始化使用DefaultPromise,用來非同步處理終止事件。executor初始化產生一個執行緒池。
接下來就是對children的操作,根據nThreads的大小,產生一個EventExecutor陣列,然後遍歷這個陣列,呼叫newChild給每一個元素初始化。
newChild是在NioEventLoopGroup中實現的:
1 protected EventLoop newChild(Executor executor, Object... args) throws Exception { 2 return new NioEventLoop(this, executor, (SelectorProvider)args[0], ((SelectStrategyFactory)args[1]).newSelectStrategy(), (RejectedExecutionHandler)args[2]); 3 }
在這裡直接使用executor,和之前放在args陣列中的SelectorProvider、SelectStrategyFactory(newSelectStrategy方法產生DefaultSelectStrategy)和RejectedExecutionHandler產生了一個NioEventLoop物件:
1 private Selector selector; 2 private Selector unwrappedSelector; 3 private SelectedSelectionKeySet selectedKeys; 4 private final SelectorProvider provider; 5 private final AtomicBoolean wakenUp = new AtomicBoolean(); 6 private final SelectStrategy selectStrategy; 7 8 NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider, SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) { 9 super(parent, executor, false, DEFAULT_MAX_PENDING_TASKS, rejectedExecutionHandler); 10 if (selectorProvider == null) { 11 throw new NullPointerException("selectorProvider"); 12 } else if (strategy == null) { 13 throw new NullPointerException("selectStrategy"); 14 } else { 15 this.provider = selectorProvider; 16 NioEventLoop.SelectorTuple selectorTuple = this.openSelector(); 17 this.selector = selectorTuple.selector; 18 this.unwrappedSelector = selectorTuple.unwrappedSelector; 19 this.selectStrategy = strategy; 20 } 21 }
NioEventLoop首先在繼承鏈上呼叫父類的構造,都是一些成員的賦值操作,簡單看一看:
1 protected SingleThreadEventLoop(EventLoopGroup parent, Executor executor, boolean addTaskWakesUp, int maxPendingTasks, RejectedExecutionHandler rejectedExecutionHandler) { 2 super(parent, executor, addTaskWakesUp, maxPendingTasks, rejectedExecutionHandler); 3 this.tailTasks = this.newTaskQueue(maxPendingTasks); 4 } 5 6 protected SingleThreadEventExecutor(EventExecutorGroup parent, Executor executor, boolean addTaskWakesUp, int maxPendingTasks, RejectedExecutionHandler rejectedHandler) { 7 super(parent); 8 this.threadLock = new Semaphore(0); 9 this.shutdownHooks = new LinkedHashSet(); 10 this.state = 1; 11 this.terminationFuture = new DefaultPromise(GlobalEventExecutor.INSTANCE); 12 this.addTaskWakesUp = addTaskWakesUp; 13 this.maxPendingTasks = Math.max(16, maxPendingTasks); 14 this.executor = (Executor)ObjectUtil.checkNotNull(executor, "executor"); 15 this.taskQueue = this.newTaskQueue(this.maxPendingTasks); 16 this.rejectedExecutionHandler = (RejectedExecutionHandler)ObjectUtil.checkNotNull(rejectedHandler, "rejectedHandler"); 17 } 18 19 protected AbstractScheduledEventExecutor(EventExecutorGroup parent) { 20 super(parent); 21 } 22 23 protected AbstractEventExecutor(EventExecutorGroup parent) { 24 this.selfCollection = Collections.singleton(this); 25 this.parent = parent; 26 }
在經過這繼承鏈上的一系列呼叫後,給provider成員賦值selectorProvider,就是之前建立好的WindowsSelectorProvider,然後使用openSelector方法,最終建立JDK原生的Selector:
1 private NioEventLoop.SelectorTuple openSelector() { 2 final AbstractSelector unwrappedSelector; 3 try { 4 unwrappedSelector = this.provider.openSelector(); 5 } catch (IOException var7) { 6 throw new ChannelException("failed to open a new selector", var7); 7 } 8 9 if (DISABLE_KEYSET_OPTIMIZATION) { 10 return new NioEventLoop.SelectorTuple(unwrappedSelector); 11 } else { 12 final SelectedSelectionKeySet selectedKeySet = new SelectedSelectionKeySet(); 13 Object maybeSelectorImplClass = AccessController.doPrivileged(new PrivilegedAction<Object>() { 14 public Object run() { 15 try { 16 return Class.forName("sun.nio.ch.SelectorImpl", false, PlatformDependent.getSystemClassLoader()); 17 } catch (Throwable var2) { 18 return var2; 19 } 20 } 21 }); 22 if (maybeSelectorImplClass instanceof Class && ((Class)maybeSelectorImplClass).isAssignableFrom(unwrappedSelector.getClass())) { 23 final Class<?> selectorImplClass = (Class)maybeSelectorImplClass; 24 Object maybeException = AccessController.doPrivileged(new PrivilegedAction<Object>() { 25 public Object run() { 26 try { 27 Field selectedKeysField = selectorImplClass.getDeclaredField("selectedKeys"); 28 Field publicSelectedKeysField = selectorImplClass.getDeclaredField("publicSelectedKeys"); 29 Throwable cause = ReflectionUtil.trySetAccessible(selectedKeysField, true); 30 if (cause != null) { 31 return cause; 32 } else { 33 cause = ReflectionUtil.trySetAccessible(publicSelectedKeysField, true); 34 if (cause != null) { 35 return cause; 36 } else { 37 selectedKeysField.set(unwrappedSelector, selectedKeySet); 38 publicSelectedKeysField.set(unwrappedSelector, selectedKeySet); 39 return null; 40 } 41 } 42 } catch (NoSuchFieldException var4) { 43 return var4; 44 } catch (IllegalAccessException var5) { 45 return var5; 46 } 47 } 48 }); 49 if (maybeException instanceof Exception) { 50 this.selectedKeys = null; 51 Exception e = (Exception)maybeException; 52 logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, e); 53 return new NioEventLoop.SelectorTuple(unwrappedSelector); 54 } else { 55 this.selectedKeys = selectedKeySet; 56 logger.trace("instrumented a special java.util.Set into: {}", unwrappedSelector); 57 return new NioEventLoop.SelectorTuple(unwrappedSelector, new SelectedSelectionKeySetSelector(unwrappedSelector, selectedKeySet)); 58 } 59 } else { 60 if (maybeSelectorImplClass instanceof Throwable) { 61 Throwable t = (Throwable)maybeSelectorImplClass; 62 logger.trace("failed to instrument a special java.util.Set into: {}", unwrappedSelector, t); 63 } 64 65 return new NioEventLoop.SelectorTuple(unwrappedSelector); 66 } 67 } 68 }
可以看到在一開始就使用provider的openSelector方法,即WindowsSelectorProvider的openSelector方法,建立了WindowsSelectorImpl物件(【Java】NIO中Selector的建立原始碼分析 )
然後根據DISABLE_KEYSET_OPTIMIZATION判斷:
1 private static final boolean DISABLE_KEYSET_OPTIMIZATION = SystemPropertyUtil.getBoolean("io.netty.noKeySetOptimization", false);
可以看到這個系統配置在沒有設定預設是false,如果設定了則直接建立一個SelectorTuple物件返回:
1 private static final class SelectorTuple { 2 final Selector unwrappedSelector; 3 final Selector selector; 4 5 SelectorTuple(Selector unwrappedSelector) { 6 this.unwrappedSelector = unwrappedSelector; 7 this.selector = unwrappedSelector; 8 } 9 10 SelectorTuple(Selector unwrappedSelector, Selector selector) { 11 this.unwrappedSelector = unwrappedSelector; 12 this.selector = selector; 13 } 14 }
可以看到僅僅是將unwrappedSelector和selector封裝了,unwrappedSelector對應的是JDK原生Selector沒有經過更改的,而selector對應的是經過更改修飾操作的。
在沒有系統配置下,就對Selector進行更改修飾操作:
首先建立SelectedSelectionKeySet物件,這個SelectedSelectionKeySet繼承自AbstractSet:
1 final class SelectedSelectionKeySet extends AbstractSet<SelectionKey> { 2 SelectionKey[] keys = new SelectionKey[1024]; 3 int size; 4 5 SelectedSelectionKeySet() { 6 } 7 ...... 8 }
後面是通過反射機制,使得WindowsSelectorImpl的selectedKeys和publicSelectedKeys成員直接賦值為SelectedSelectionKeySet物件。
WindowsSelectorImpl的這兩個成員是在SelectorImpl中定義的:
1 protected Set<SelectionKey> selectedKeys = new HashSet(); 2 private Set<SelectionKey> publicSelectedKeys;
從這裡就可以明白,在JDK原生的Selector中,selectedKeys和publicSelectedKeys這兩個Set的初始化大小都為0,而在這裡僅僅就是使其初始化大小變為1024。
後面就是對一些異常的處理,沒什麼好說的。
openSelector結束後,就可以分別對包裝過的Selector和未包裝過的Selector,即selector和unwrappedSelector成員賦值,再由selectStrategy儲存剛才產生的選擇策略,用於Selector的輪詢。
回到MultithreadEventExecutorGroup的構造,在呼叫newChild方法時即NioEventLoop建立的過程中可能出現異常情況,就需要遍歷children陣列,將之前建立好的NioEventLoop使用shutdownGracefully優雅地關閉掉:
shutdownGracefully在AbstractEventExecutor中實現:
1 public Future<?> shutdownGracefully() { 2 return this.shutdownGracefully(2L, 15L, TimeUnit.SECONDS); 3 }
這裡設定了超時時間,繼續呼叫SingleThreadEventExecutor的shutdownGracefully方法:
1 public Future<?> shutdownGracefully(long quietPeriod, long timeout, TimeUnit unit) { 2 if (quietPeriod < 0L) { 3 throw new IllegalArgumentException("quietPeriod: " + quietPeriod + " (expected >= 0)"); 4 } else if (timeout < quietPeriod) { 5 throw new IllegalArgumentException("timeout: " + timeout + " (expected >= quietPeriod (" + quietPeriod + "))"); 6 } else if (unit == null) { 7 throw new NullPointerException("unit"); 8 } else if (this.isShuttingDown()) { 9 return this.terminationFuture(); 10 } else { 11 boolean inEventLoop = this.inEventLoop(); 12 13 while(!this.isShuttingDown()) { 14 boolean wakeup = true; 15 int oldState = this.state; 16 int newState; 17 if (inEventLoop) { 18 newState = 3; 19 } else { 20 switch(oldState) { 21 case 1: 22 case 2: 23 newState = 3; 24 break; 25 default: 26 newState = oldState; 27 wakeup = false; 28 } 29 } 30 31 if (STATE_UPDATER.compareAndSet(this, oldState, newState)) { 32 this.gracefulShutdownQuietPeriod = unit.toNanos(quietPeriod); 33 this.gracefulShutdownTimeout = unit.toNanos(timeout); 34 if (oldState == 1) { 35 try { 36 this.doStartThread(); 37 } catch (Throwable var10) { 38 STATE_UPDATER.set(this, 5); 39 this.terminationFuture.tryFailure(var10); 40 if (!(var10 instanceof Exception)) { 41 PlatformDependent.throwException(var10); 42 } 43 44 return this.terminationFuture; 45 } 46 } 47 48 if (wakeup) { 49 this.wakeup(inEventLoop); 50 } 51 52 return this.terminationFuture(); 53 } 54 } 55 56 return this.terminationFuture(); 57 } 58 }
前三個判斷沒什麼好說的,isShuttingDown判斷:
1 public boolean isShuttingDown() { 2 return this.state >= 3; 3 }
在之前NioEventLoop建立的時候,呼叫了一系列的繼承鏈,其中state是在SingleThreadEventExecutor的構造方法中實現的,初始值是1,state有如下幾種狀態:
1 private static final int ST_NOT_STARTED = 1; 2 private static final int ST_STARTED = 2; 3 private static final int ST_SHUTTING_DOWN = 3; 4 private static final int ST_SHUTDOWN = 4; 5 private static final int ST_TERMINATED = 5;
可見在NioEventLoop初始化後處於尚未啟動狀態,並沒有Channel的註冊,也就不需要輪詢。
isShuttingDown就必然是false,就進入了else塊:
首先得到inEventLoop的返回值,該方法在AbstractEventExecutor中實現:
1 public boolean inEventLoop() { 2 return this.inEventLoop(Thread.currentThread()); 3 }
他傳入了一個當前執行緒,接著呼叫inEventLoop的過載,這個方法是在SingleThreadEventExecutor中實現:
1 public boolean inEventLoop(Thread thread) { 2 return thread == this.thread; 3 }
通過觀察之前的SingleThreadEventExecutor構造方法,發現並沒有對thread成員初始化,此時的this.thread就是null,那麼返回值就是false,即inEventLoop為false。
在while迴圈中又對isShuttingDown進行了判斷,shutdownGracefully當讓不僅僅使用在建立NioEventLoop物件失敗時才呼叫的,無論是在EventLoopGroup的關閉,還是Bootstrap的關閉,都會關閉繫結的NioEventLoop,所以在多執行緒環境中,有可能會被其他執行緒關閉。
在while迴圈中,結合上面可知滿足進入switch塊,在switch塊中令newState為3;
然後呼叫STATE_UPDATER的compareAndSet方法,STATE_UPDATER是用來原子化更新state成員的:
1 private static final AtomicIntegerFieldUpdater<SingleThreadEventExecutor> STATE_UPDATER = AtomicIntegerFieldUpdater.newUpdater(SingleThreadEventExecutor.class, "state");
所以這裡就是使用CAS操作,原子化更新state成員為3,也就是使當前狀態由ST_NOT_STARTED 變為了ST_SHUTTING_DOWN 狀態。
gracefulShutdownQuietPeriod和gracefulShutdownTimeout分別儲存quietPeriod和timeout的納秒級顆粒度。
前面可知oldState使1,呼叫doStartThread方法:
1 private void doStartThread() { 2 assert this.thread == null; 3 4 this.executor.execute(new Runnable() { 5 public void run() { 6 SingleThreadEventExecutor.this.thread = Thread.currentThread(); 7 if (SingleThreadEventExecutor.this.interrupted) { 8 SingleThreadEventExecutor.this.thread.interrupt(); 9 } 10 11 boolean success = false; 12 SingleThreadEventExecutor.this.updateLastExecutionTime(); 13 boolean var112 = false; 14 15 int oldState; 16 label1685: { 17 try { 18 var112 = true; 19 SingleThreadEventExecutor.this.run(); 20 success = true; 21 var112 = false; 22 break label1685; 23 } catch (Throwable var119) { 24 SingleThreadEventExecutor.logger.warn("Unexpected exception from an event executor: ", var119); 25 var112 = false; 26 } finally { 27 if (var112) { 28 int oldStatex; 29 do { 30 oldStatex = SingleThreadEventExecutor.this.state; 31 } while(oldStatex < 3 && !SingleThreadEventExecutor.STATE_UPDATER.compareAndSet(SingleThreadEventExecutor.this, oldStatex, 3)); 32 33 if (success && SingleThreadEventExecutor.this.gracefulShutdownStartTime == 0L) { 34 SingleThreadEventExecutor.logger.error("Buggy " + EventExecutor.class.getSimpleName() + " implementation; " + SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must be called before run() implementation terminates."); 35 } 36 37 try { 38 while(!SingleThreadEventExecutor.this.confirmShutdown()) { 39 ; 40 } 41 } finally { 42 try { 43 SingleThreadEventExecutor.this.cleanup(); 44 } finally { 45 SingleThreadEventExecutor.STATE_UPDATER.set(SingleThreadEventExecutor.this, 5); 46 SingleThreadEventExecutor.this.threadLock.release(); 47 if (!SingleThreadEventExecutor.this.taskQueue.isEmpty()) { 48 SingleThreadEventExecutor.logger.warn("An event executor terminated with non-empty task queue (" + SingleThreadEventExecutor.this.taskQueue.size() + ')'); 49 } 50 51 SingleThreadEventExecutor.this.terminationFuture.setSuccess((Object)null); 52 } 53 } 54 55 } 56 } 57 58 do { 59 oldState = SingleThreadEventExecutor.this.state; 60 } while(oldState < 3 && !SingleThreadEventExecutor.STATE_UPDATER.compareAndSet(SingleThreadEventExecutor.this, oldState, 3)); 61 62 if (success && SingleThreadEventExecutor.this.gracefulShutdownStartTime == 0L) { 63 SingleThreadEventExecutor.logger.error("Buggy " + EventExecutor.class.getSimpleName() + " implementation; " + SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must be called before run() implementation terminates."); 64 } 65 66 try { 67 while(!SingleThreadEventExecutor.this.confirmShutdown()) { 68 ; 69 } 70 71 return; 72 } finally { 73 try { 74 SingleThreadEventExecutor.this.cleanup(); 75 } finally { 76 SingleThreadEventExecutor.STATE_UPDATER.set(SingleThreadEventExecutor.this, 5); 77 SingleThreadEventExecutor.this.threadLock.release(); 78 if (!SingleThreadEventExecutor.this.taskQueue.isEmpty()) { 79 SingleThreadEventExecutor.logger.warn("An event executor terminated with non-empty task queue (" + SingleThreadEventExecutor.this.taskQueue.size() + ')'); 80 } 81 82 SingleThreadEventExecutor.this.terminationFuture.setSuccess((Object)null); 83 } 84 } 85 } 86 87 do { 88 oldState = SingleThreadEventExecutor.this.state; 89 } while(oldState < 3 && !SingleThreadEventExecutor.STATE_UPDATER.compareAndSet(SingleThreadEventExecutor.this, oldState, 3)); 90 91 if (success && SingleThreadEventExecutor.this.gracefulShutdownStartTime == 0L) { 92 SingleThreadEventExecutor.logger.error("Buggy " + EventExecutor.class.getSimpleName() + " implementation; " + SingleThreadEventExecutor.class.getSimpleName() + ".confirmShutdown() must be called before run() implementation terminates."); 93 } 94 95 try { 96 while(!SingleThreadEventExecutor.this.confirmShutdown()) { 97 ; 98 } 99 } finally { 100 try { 101 SingleThreadEventExecutor.this.cleanup(); 102 } finally { 103 SingleThreadEventExecutor.STATE_UPDATER.set(SingleThreadEventExecutor.this, 5); 104 SingleThreadEventExecutor.this.threadLock.release(); 105 if (!SingleThreadEventExecutor.this.taskQueue.isEmpty()) { 106 SingleThreadEventExecutor.logger.warn("An event executor terminated with non-empty task queue (" + SingleThreadEventExecutor.this.taskQueue.size() + ')'); 107 } 108 109 SingleThreadEventExecutor.this.terminationFuture.setSuccess((Object)null); 110 } 111 } 112 113 } 114 }); 115 }
剛才說過this.thread並沒有初始化,所以等於null成立,斷言可以繼續。
然後直接使executor運行了一個執行緒,這個executor其實就是在剛才的MultithreadEventExecutorGroup中產生的ThreadPerTaskExecutor物件。
線上程中,首先將SingleThreadEventExecutor的thread成員初始化為當前執行緒。
在這裡可能就有疑問了,為什麼會在關閉時會呼叫名為doStartThread的方法,這個方法不因該在啟動時呼叫嗎?
其實doStartThread在啟動時是會被呼叫的,當在啟動時被呼叫的話,每一個NioEventLoop都會被繫結一個執行緒用來處理真正的Selector操作,根據之前的說法就可以知道,每個EventLoopGroup在建立後都會被繫結cpu個數的二倍個NioEventLoop,而每個NioEventLoop都會繫結一個Selector物件,上面又說了在啟動時SingleThreadEventExecutor綁定了一個執行緒,即NioEventLoop綁定了一個執行緒來處理其繫結的Selector的輪詢。
至於關閉時還會啟動Selector的輪詢,就是為了解決註冊了的Channel沒有被處理的情況。
回到doStartThread方法,其實這個doStartThread方法的核心是SingleThreadEventExecutor.this.run(),這個方法就是正真的Selector的輪詢操作,在NioEventLoop中實現:
1 protected void run() { 2 while(true) { 3 while(true) { 4 try { 5 switch(this.selectStrategy.calculateStrategy(this.selectNowSupplier, this.hasTasks())) { 6 case -2: 7 continue; 8 case -1: 9 this.select(this.wakenUp.getAndSet(false)); 10 if (this.wakenUp.get()) { 11 this.selector.wakeup(); 12 } 13 default: 14 this.cancelledKeys = 0; 15 this.needsToSelectAgain = false; 16 int ioRatio = this.ioRatio; 17 if (ioRatio == 100) { 18 try { 19 this.processSelectedKeys(); 20 } finally { 21 this.runAllTasks(); 22 } 23 } else { 24 long ioStartTime = System.nanoTime(); 25 boolean var13 = false; 26 27 try { 28 var13 = true; 29 this.processSelectedKeys(); 30 var13 = false; 31 } finally { 32 if (var13) { 33 long ioTime = System.nanoTime() - ioStartTime; 34 this.runAllTasks(ioTime * (long)(100 - ioRatio) / (long)ioRatio); 35 } 36 } 37 38 long ioTime = System.nanoTime() - ioStartTime; 39 this.runAllTasks(ioTime * (long)(100 - ioRatio) / (long)ioRatio); 40 } 41 } 42 } catch (Throwable var21) { 43 handleLoopException(var21); 44 } 45 46 try { 47 if (this.isShuttingDown()) { 48 this.closeAll(); 49 if (this.confirmShutdown()) { 50 return; 51 } 52 } 53 } catch (Throwable var18) { 54 handleLoopException(var18); 55 } 56 } 57 } 58 }
進入switch塊,首先呼叫之前準備好的選擇策略,其中this.selectNowSupplier在NioEventLoop建立的時候就被建立了:
1 private final IntSupplier selectNowSupplier = new IntSupplier() { 2 public int get() throws Exception { 3 return NioEventLoop.this.selectNow(); 4 } 5 };
實際上呼叫了selectNow方法:
1 int selectNow() throws IOException { 2 int var1; 3 try { 4 var1 = this.selector.selectNow(); 5 } finally { 6 if (this.wakenUp.get()) { 7 this.selector.wakeup(); 8 } 9 10 } 11 12 return var1; 13 }
這裡就直接呼叫了JDK原生的selectNow方法。
之前說過的選擇策略:
1 public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception { 2 return hasTasks ? selectSupplier.get() : -1; 3 }
其中hasTasks是根據hasTasks方法來判斷,而hasTasks方法就是判斷任務佇列是否為空,那麼在一開始初始化,必然是空的,所以這裡calculateStrategy的返回值就是-1;
那麼case為-1條件成立,執行this.select(this.wakenUp.getAndSet(false)),其中wakenUp是一個原子化的Boolean,用來表示是需要喚醒Selector的輪詢阻塞,初始化是為true,這裡通過CAS操作設定為false代表不需要喚醒,後面在select執行完後,又判斷wakenUp是否需要喚醒,說明在select中對Selector的阻塞進行了檢查,若是需要喚醒,就通過Selector的原生API完成喚醒【Java】NIO中Selector的select方法原始碼分析
來看看這裡的select實現:
1 private void select(boolean oldWakenUp) throws IOException { 2 Selector selector = this.selector; 3 4 try { 5 int selectCnt = 0; 6 long currentTimeNanos = System.nanoTime(); 7 long selectDeadLineNanos = currentTimeNanos + this.delayNanos(currentTimeNanos); 8 9 while(true) { 10 long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L; 11 if (timeoutMillis <= 0L) { 12 if (selectCnt == 0) { 13 selector.selectNow(); 14 selectCnt = 1; 15 } 16 break; 17 } 18 19 if (this.hasTasks() && this.wakenUp.compareAndSet(false, true)) { 20 selector.selectNow(); 21 selectCnt = 1; 22 break; 23 } 24 25 int selectedKeys = selector.select(timeoutMillis); 26 ++selectCnt; 27 if (selectedKeys != 0 || oldWakenUp || this.wakenUp.get() || this.hasTasks() || this.hasScheduledTasks()) { 28 break; 29 } 30 31 if (Thread.interrupted()) { 32 if (logger.isDebugEnabled()) { 33 logger.debug("Selector.select() returned prematurely because Thread.currentThread().interrupt() was called. Use NioEventLoop.shutdownGracefully() to shutdown the NioEventLoop."); 34 } 35 36 selectCnt = 1; 37 break; 38 } 39 40 long time = System.nanoTime(); 41 if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) { 42 selectCnt = 1; 43 } else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 && selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) { 44 logger.warn("Selector.select() returned prematurely {} times in a row; rebuilding Selector {}.", selectCnt, selector); 45 this.rebuildSelector(); 46 selector = this.selector; 47 selector.selectNow(); 48 selectCnt = 1; 49 break; 50 } 51 52 currentTimeNanos = time; 53 } 54 55 if (selectCnt > 3 && logger.isDebugEnabled()) { 56 logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.", selectCnt - 1, selector); 57 } 58 } catch (CancelledKeyException var13) { 59 if (logger.isDebugEnabled()) { 60 logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?", selector, var13); 61 } 62 } 63 64 }
這個方法雖然看著很長,但核心就是判斷這個存放任務的阻塞佇列是否還有任務,若是有,就直接呼叫Selector的selectNow方法獲取就緒的檔案描述符,若是沒有就緒的檔案描述符該方法也會立即返回;若是阻塞佇列中沒有任務,就呼叫Selector的select(timeout)方法,嘗試在超時時間內取獲取就緒的檔案描述符。
因為現在是在執行NioEventLoopGroup的建立,並沒有Channel的註冊,也就沒有輪詢到任何檔案描述符就緒。
在輪詢結束後,回到run方法,進入default塊:
其中ioRatio是執行IO操作和執行任務佇列的任務用時比率,預設是50。若是ioRatio設定為100,就必須等到tasks阻塞佇列中的所有任務執行完畢才再次進行輪詢;若是小於100,那麼就根據(100 - ioRatio) / ioRatio的比值乘以ioTime計算出的超時時間讓所有任務嘗試在超時時間內執行完畢,若是到達超時時間還沒執行完畢,就在下一輪的輪詢中執行。
processSelectedKeys方法就是獲取Selector輪詢的SelectedKeys結果:
1 private void processSelectedKeys() { 2 if (this.selectedKeys != null) { 3 this.processSelectedKeysOptimized(); 4 } else { 5 this.processSelectedKeysPlain(this.selector.selectedKeys()); 6 } 7 8 }
selectedKeys 在openSelector時被初始化過了,若是在openSelector中出現異常selectedKeys才會為null。
processSelectedKeysOptimized方法:
1 private void processSelectedKeysOptimized() { 2 for(int i = 0; i < this.selectedKeys.size; ++i) { 3 SelectionKey k = this.selectedKeys.keys[i]; 4 this.selectedKeys.keys[i] = null; 5 Object a = k.attachment(); 6 if (a instanceof AbstractNioChannel) { 7 this.processSelectedKey(k, (AbstractNioChannel)a); 8 } else { 9 NioTask<SelectableChannel> task = (NioTask)a; 10 processSelectedKey(k, task); 11 } 12 13 if (this.needsToSelectAgain) { 14 this.selectedKeys.reset(i + 1); 15 this.selectAgain(); 16 i = -1; 17 } 18 } 19 20 }
這裡就通過遍歷在openSelector中注入進Selector的SelectedKeys,得到SelectionKey 物件。
在這裡可以看到Netty很巧妙地通過SelectionKey的attachment附件,將JDK中的Channel和Netty中的Channel聯絡了起來。
根據得到的附件Channel的型別,執行不同的processSelectedKey方法,去處理IO操作。
processSelectedKey(SelectionKey k, AbstractNioChannel ch)方法:
1 private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) { 2 NioUnsafe unsafe = ch.unsafe(); 3 if (!k.isValid()) { 4 NioEventLoop eventLoop; 5 try { 6 eventLoop = ch.eventLoop(); 7 } catch (Throwable var6) { 8 return; 9 } 10 11 if (eventLoop == this && eventLoop != null) { 12 unsafe.close(unsafe.voidPromise()); 13 } 14 } else { 15 try { 16 int readyOps = k.readyOps(); 17 if ((readyOps & 8) != 0) { 18 int ops = k.interestOps(); 19 ops &= -9; 20 k.interestOps(ops); 21 unsafe.finishConnect(); 22 } 23 24 if ((readyOps & 4) != 0) { 25 ch.unsafe().forceFlush(); 26 } 27 28 if ((readyOps & 17) != 0 || readyOps == 0) { 29 unsafe.read(); 30 } 31 } catch (CancelledKeyException var7) { 32 unsafe.close(unsafe.voidPromise()); 33 } 34 35 } 36 }
這裡的主要核心就是根據SelectedKey的readyOps值來判斷,處理不同的就緒事件,有如下幾種事件:
1 public static final int OP_READ = 1 << 0; 2 public static final int OP_WRITE = 1 << 2; 3 public static final int OP_CONNECT = 1 << 3; 4 public static final int OP_ACCEPT = 1 << 4;
結合來看上面的判斷就對應:連線就緒、寫就緒、偵聽或者讀就緒,交由Netty的AbstractNioChannel的NioUnsafe去處理不同事件的byte資料,NioUnsafe會將資料再交由ChannelPipeline雙向連結串列去處理。
關於ChannelPipeline會在後續的部落格中詳細介紹。
processSelectedKey(SelectionKey k, NioTask<SelectableChannel> task)這個方法的實現細節需要由使用者實現NioTask<SelectableChannel>介面,就不說了。
回到processSelectedKeys方法,在this.selectedKeys等於null的情況下:
1 private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) { 2 if (!selectedKeys.isEmpty()) { 3 Iterator i = selectedKeys.iterator(); 4 5 while(true) { 6 SelectionKey k = (SelectionKey)i.next(); 7 Object a = k.attachment(); 8 i.remove(); 9 if (a instanceof AbstractNioChannel) { 10 this.processSelectedKey(k, (AbstractNioChannel)a); 11 } else { 12 NioTask<SelectableChannel> task = (NioTask)a; 13 processSelectedKey(k, task); 14 } 15 16 if (!i.hasNext()) { 17 break; 18 } 19 20 if (this.needsToSelectAgain) { 21 this.selectAgain(); 22 selectedKeys = this.selector.selectedKeys(); 23 if (selectedKeys.isEmpty()) { 24 break; 25 } 26 27 i = selectedKeys.iterator(); 28 } 29 } 30 31 } 32 }
這是在openSelector中注入進Selector的SelectedKeys失敗的情況下,直接遍歷Selector本身的SelectedKeys,和processSelectedKeysOptimized沒有差別。
繼續回到run方法,在呼叫完processSelectedKeys方法後,就需要呼叫runAllTasks處理任務佇列中的任務:
runAllTasks()方法:
1 protected boolean runAllTasks() { 2 assert this.inEventLoop(); 3 4 boolean ranAtLeastOne = false; 5 6 boolean fetchedAll; 7 do { 8 fetchedAll = this.fetchFromScheduledTaskQueue(); 9 if (this.runAllTasksFrom(this.taskQueue)) { 10 ranAtLeastOne = true; 11 } 12 } while(!fetchedAll); 13 14 if (ranAtLeastOne) { 15 this.lastExecutionTime = ScheduledFutureTask.nanoTime(); 16 } 17 18 this.afterRunningAllTasks(); 19 return ranAtLeastOne; 20 }
首先呼叫fetchFromScheduledTaskQueue方法:
1 private boolean fetchFromScheduledTaskQueue() { 2 long nanoTime = AbstractScheduledEventExecutor.nanoTime(); 3 4 for(Runnable scheduledTask = this.pollScheduledTask(nanoTime); scheduledTask != null; scheduledTask = this.pollScheduledTask(nanoTime)) { 5 if (!this.taskQueue.offer(scheduledTask)) { 6 this.scheduledTaskQueue().add((ScheduledFutureTask)scheduledTask); 7 return false; 8 } 9 } 10 11 return true; 12 }
這裡就是通過pollScheduledTask不斷地從延時任務佇列獲取到期的任務,將到期的任務新增到taskQueue任務佇列中,為上面的runAllTasksFrom執行做準備;若是新增失敗,再把它放進延時任務佇列。
pollScheduledTask方法:
1 protected final Runnable pollScheduledTask(long nanoTime) { 2 assert this.inEventLoop(); 3 4 Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue; 5 ScheduledFutureTask<?> scheduledTask = scheduledTaskQueue == null ? null : (ScheduledFutureTask)scheduledTaskQueue.peek(); 6 if (scheduledTask == null) { 7 return null; 8 } else if (scheduledTask.deadlineNanos() <= nanoTime) { 9 scheduledTaskQueue.remove(); 10 return scheduledTask; 11 } else { 12 return null; 13 } 14 }
從延時任務佇列中獲取隊首的任務scheduledTask,若是scheduledTask的deadlineNanos小於等於nanoTime,說明該任務到期。
回到runAllTasks,將到期了的延時任務放在了任務佇列,由runAllTasksFrom執行這些任務:
1 protected final boolean runAllTasksFrom(Queue<Runnable> taskQueue) { 2 Runnable task = pollTaskFrom(taskQueue); 3 if (task == null) { 4 return false; 5 } else { 6 do { 7 safeExecute(task); 8 task = pollTaskFrom(taskQueue); 9 } while(task != null); 10 11 return true; 12 } 13 }
不斷地從任務佇列隊首獲取任務,然後執行,直到沒有任務。
pollTaskFrom是獲取隊首任務:
1 protected static Runnable pollTaskFrom(Queue<Runnable> taskQueue) { 2 Runnable task; 3 do { 4 task = (Runnable)taskQueue.poll(); 5 } while(task == WAKEUP_TASK); 6 7 return task; 8 }
其中WAKEUP_TASK,是用來巧妙地控制迴圈:
1 private static final Runnable WAKEUP_TASK = new Runnable() { 2 public void run() { 3 } 4 };
safeExecute是執行任務:
1 protected static void safeExecute(Runnable task) { 2 try { 3 task.run(); 4 } catch (Throwable var2) { 5 logger.warn("A task raised an exception. Task: {}", task, var2); 6 } 7 8 }
實際上就是執行Runnable 的run方法。
繼續回到runAllTasks方法,當所有到期任務執行完畢後,根據ranAtLeastOne判斷是否需要修改最後一次執行時間lastExecutionTime,最後呼叫afterRunningAllTasks方法,該方法是在SingleThreadEventLoop中實現的:
1 protected void afterRunningAllTasks() { 2 this.runAllTasksFrom(this.tailTasks); 3 }
這裡就僅僅執行了tailTasks佇列中的任務。runAllTasks到這裡執行完畢。
再來看看runAllTasks(timeoutNanos)方法:
1 protected boolean runAllTasks(long timeoutNanos) { 2 this.fetchFromScheduledTaskQueue(); 3 Runnable task = this.pollTask(); 4 if (task == null) { 5 this.afterRunningAllTasks(); 6 return false; 7 } else { 8 long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos; 9 long runTasks = 0L; 10 11 long lastExecutionTime; 12 while(true) { 13 safeExecute(task); 14 ++runTasks; 15 if ((runTasks & 63L) == 0L) { 16 lastExecutionTime = ScheduledFutureTask.nanoTime(); 17 if (lastExecutionTime >= deadline) { 18 break; 19 } 20 } 21 22 task = this.pollTask(); 23 if (task == null) { 24 lastExecutionTime = ScheduledFutureTask.nanoTime(); 25 break; 26 } 27 } 28 29 this.afterRunningAllTasks(); 30 this.lastExecutionTime = lastExecutionTime; 31 return true; 32 } 33 }
這個方法前面的runAllTasks方法類似,先通過fetchFromScheduledTaskQueue將所有到期了的延時任務放在taskQueue中,然後不斷從taskQueue隊首獲取任務,但是,若是執行到了到超過了63個任務,判斷是否達到了超時時間deadline,若是達到結束迴圈,留著下次執行,反之繼續迴圈執行任務。
回到run方法,在輪詢完畢,並且執行完任務後,通過isShuttingDown判斷當前狀態,在之前的CAS操作中,state已經變為了3,所以isShuttingDown成立,就需要呼叫closeAll方法
1 private void closeAll() { 2 this.selectAgain(); 3 Set<SelectionKey> keys = this.selector.keys(); 4 Collection<AbstractNioChannel> channels = new ArrayList(keys.size()); 5 Iterator var3 = keys.iterator(); 6 7 while(var3.hasNext()) { 8 SelectionKey k = (SelectionKey)var3.next(); 9 Object a = k.attachment(); 10 if (a instanceof AbstractNioChannel) { 11 channels.add((AbstractNioChannel)a); 12 } else { 13 k.cancel(); 14 NioTask<SelectableChannel> task = (NioTask)a; 15 invokeChannelUnregistered(task, k, (Throwable)null); 16 } 17 } 18 19 var3 = channels.iterator(); 20 21 while(var3.hasNext()) { 22 AbstractNioChannel ch = (AbstractNioChannel)var3.next(); 23 ch.unsafe().close(ch.unsafe().voidPromise()); 24 } 25 26 }
在這裡首先呼叫selectAgain進行一次輪詢:
1 private void selectAgain() { 2 this.needsToSelectAgain = false; 3 4 try { 5 this.selector.selectNow(); 6 } catch (Throwable var2) { 7 logger.warn("Failed to update SelectionKeys.", var2); 8 } 9 10 }
通過這次的輪詢,將當前仍有事件就緒的JDK的SelectionKey中繫結的Netty的Channel新增到channels集合中,遍歷這個集合,通過unsafe的close方法關閉Netty的Channel。
之後呼叫confirmShutdown方法:
1 protected boolean confirmShutdown() { 2 if (!this.isShuttingDown()) { 3 return false; 4 } else if (!this.inEventLoop()) { 5 throw new IllegalStateException("must be invoked from an event loop"); 6 } else { 7 this.cancelScheduledTasks(); 8 if (this.gracefulShutdownStartTime == 0L) { 9 this.gracefulShutdownStartTime = ScheduledFutureTask.nanoTime(); 10 } 11 12 if (!this.runAllTasks() && !this.runShutdownHooks()) { 13 long nanoTime = ScheduledFutureTask.nanoTime(); 14 if (!this.isShutdown() && nanoTime - this.gracefulShutdownStartTime <= this.gracefulShutdownTimeout) { 15 if (nanoTime - this.lastExecutionTime <= this.gracefulShutdownQuietPeriod) { 16 this.wakeup(true); 17 18 try { 19 Thread.sleep(100L); 20 } catch (InterruptedException var4) { 21 ; 22 } 23 24 return false; 25 } else { 26 return true; 27 } 28 } else { 29 return true; 30 } 31 } else if (this.isShutdown()) { 32 return true; 33 } else if (this.gracefulShutdownQuietPeriod == 0L) { 34 return true; 35 } else { 36 this.wakeup(true); 37 return false; 38 } 39 } 40 }
首先呼叫cancelScheduledTasks,取消所有的延時任務:
1 protected void cancelScheduledTasks() { 2 assert this.inEventLoop(); 3 4 PriorityQueue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue; 5 if (!isNullOrEmpty(scheduledTaskQueue)) { 6 ScheduledFutureTask<?>[] scheduledTasks = (ScheduledFutureTask[])scheduledTaskQueue.toArray(new ScheduledFutureTask[scheduledTaskQueue.size()]); 7 ScheduledFutureTask[] var3 = scheduledTasks; 8 int var4 = scheduledTasks.length; 9 10 for(int var5 = 0; var5 < var4; ++var5) { 11 ScheduledFutureTask<?> task = var3[var5]; 12 task.cancelWithoutRemove(false); 13 } 14 15 scheduledTaskQueue.clearIgnoringIndexes(); 16 } 17 }
遍歷scheduledTasks這個延時任務對立中所有的任務,通過cancelWithoutRemove將該任務取消。
至此輪詢的整個生命週期完成。
回到SingleThreadEventExecutor的doStartThread方法,在run方法執行完畢後,說明Selector輪詢結束,呼叫SingleThreadEventExecutor.this.cleanup()方法關閉Selector:
1 protected void cleanup() { 2 try { 3 this.selector.close(); 4 } catch (IOException var2) { 5 logger.warn("Failed to close a selector.", var2); 6 } 7 8 }
這次終於可以回到MultithreadEventExecutorGroup的構造,在children建立完畢後,用chooserFactory根據children的大小建立chooser,前面說過。
然後產生terminationListener非同步中斷監聽物件,給每個NioEventLoop設定中斷監聽,然後對children進行了備份處理,通過readonlyChildren儲存。
至此NioEventLoopGroup的建立全部結束。
&n