1. NioEventLoopGroup
MultithreadEventExecutorGroup内部維護一個類型為EventExecutor的children數組, 其大小是 nThreads, 這樣就構成了一個線程池：
public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
private final EventExecutor[] children;
} 執行個體化NioEventLoopGroup時, 可以指定線程池的大小nThreads，否則nThreads預設為CPU * 2，這個nThreads就是MultithreadEventExecutorGroup内部children資料的大小：
public NioEventLoopGroup(int nThreads) {
this(nThreads, (Executor) null);
} 上面NioEventLoopGroup的構造函數中的nThreads最終會作為MultithreadEventExecutorGroup的children數組的容量，如下：
protected MultithreadEventExecutorGroup(int nThreads, Executor executor, EventExecutorChooserFactory chooserFactory, Object... args) {
...
children = new EventExecutor[nThreads];
for (int i = 0; i < nThreads; i ++) {
boolean success = false;
try {
children[i] = newChild(executor, args);
success = true;
} catch (Exception e) {...}
}
...
}
protected abstract EventExecutor newChild(Executor executor, Object... args) throws Exception;
MultithreadEventExecutorGroup中會調用newChild抽象方法來初始化children數組元素，抽象方法newChild是在NioEventLoopGroup中實作的，傳回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]);
} NioEventLoop通過調用openSelector()中的selector = provider.openSelector()來擷取一個selector對象：
public final class NioEventLoop extends SingleThreadEventLoop {
private Selector selector;
private Selector unwrappedSelector;
private final SelectorProvider provider;
NioEventLoop(NioEventLoopGroup parent, Executor executor, SelectorProvider selectorProvider, SelectStrategy strategy, RejectedExecutionHandler rejectedExecutionHandler) {
...
provider = selectorProvider;
final SelectorTuple selectorTuple = openSelector();
selector = selectorTuple.selector;
unwrappedSelector = selectorTuple.unwrappedSelector;
selectStrategy = strategy;
}
private SelectorTuple openSelector() {
final Selector unwrappedSelector;
try {
unwrappedSelector = provider.openSelector();
} catch (IOException e) {}
...
}
} 而這個provider正是在NioEventLoopGroup的newChild中被調用的：
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
return new NioEventLoop(this, executor, (SelectorProvider) args[0], ((SelectStrategyFactory) args[1]).newSelectStrategy(), (RejectedExecutionHandler) args[2]);
} 2. NioEventLoop
2.1 NioEventLoop體系
NioEventLoop間接繼承于SingleThreadEventExecutor，SingleThreadEventExecutor是Netty中對本地線程的抽象，它内部的thread屬性存儲了一個本地Java線程。是以，一個NioEventLoop其實和一個特定的線程綁定，在其生命周期内該綁定關系都不會改變。
AbstractScheduledEventExecutor實作了NioEventLoop的schedule功能，即我們可以通過調用一
個NioEventLoop執行個體的schedule方法來運作一些定時任務：
public abstract class AbstractScheduledEventExecutor extends AbstractEventExecutor {
@Override
public ScheduledFuture<?> schedule(Runnable command, long delay, TimeUnit unit) {
ObjectUtil.checkNotNull(command, "command");
ObjectUtil.checkNotNull(unit, "unit");
if (delay < 0) {
delay = 0;
}
validateScheduled(delay, unit);
return schedule(new ScheduledFutureTask<Void>(
this, command, null, ScheduledFutureTask.deadlineNanos(unit.toNanos(delay))));
}
<V> ScheduledFuture<V> schedule(final ScheduledFutureTask<V> task) {
if (inEventLoop()) {
scheduledTaskQueue().add(task);
} else {
execute(new Runnable() {
@Override
public void run() {
scheduledTaskQueue().add(task);
}
});
}
return task;
}
} 配合任務隊列的功能，可以調用一個NioEventLoop執行個體的execute方法來向任務隊列中添加一個task，并由NioEventLoop進行排程執行。
通常來說，NioEventLoop肩負着兩種任務，第一個是作為IO線程，執行與Channel相關的IO操作，包括調用select等待就緒的IO事件、讀寫資料與資料的處理等；而第二個任務是作為任務隊列，執行taskQueue中的任務。
2.2 NioEventLoop綁定線程
SingleThreadEventExecutor的thread屬性是與SingleThreadEventExecutor關聯的本地線程：
private void doStartThread() {
assert thread == null;
executor.execute(new Runnable() {
@Override
public void run() {
thread = Thread.currentThread();
if (interrupted) {
thread.interrupt();
}
try {
SingleThreadEventExecutor.this.run();
} catch (Throwable t) {}
...
}
});
} 在這個線程中所做的就是調用SingleThreadEventExecutor.this.run()方法, 而因為NioEventLoop實作了這個方法，是以根據多态性，其實調用的是NioEventLoop.run()方法。這個run()中做的就是從選擇器擷取就緒的事件：
protected void run() {
for (;;) {
try {
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.SELECT:
select(wakenUp.getAndSet(false));
if (wakenUp.get()) {
selector.wakeup();
}
default:
}
...
if (ioRatio == 100) {
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
runAllTasks();
}
} else {
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {}
}
}
private void select(boolean oldWakenUp) throws IOException {
Selector selector = this.selector;
try {
while(true) {
...
int selectedKeys = selector.select(timeoutMillis);
++selectCnt;
...
} catch (CancelledKeyException var13) {}
} 從選擇器選取就緒的事件後，會最終調用processSelectedKey具體處理每個事件：
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
...
try {
int readyOps = k.readyOps();
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
// Process OP_WRITE
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
// Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
ch.unsafe().forceFlush();
}
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
}
} catch (CancelledKeyException ignored) {}
} 2.3 EventLoop與Channel的關聯
channel會注冊到EventLoop的selector上，在SingleThreadEventLoop的register中：
public abstract class SingleThreadEventLoop extends SingleThreadEventExecutor implements EventLoop {
...
public ChannelFuture register(Channel channel) {
return this.register((ChannelPromise)(new DefaultChannelPromise(channel, this)));
}
public ChannelFuture register(ChannelPromise promise) {
ObjectUtil.checkNotNull(promise, "promise");
promise.channel().unsafe().register(this, promise);
return promise;
} 跟蹤代碼，最終會跟蹤到AbstractNioChannel的doRegister()：
public abstract class AbstractNioChannel extends AbstractChannel {
...
protected void doRegister() throws Exception {
boolean selected = false;
while(true) {
try {
this.selectionKey = this.javaChannel().register(this.eventLoop().unwrappedSelector(), 0, this);
return;
} catch (CancelledKeyException var3) {
if (selected) {
throw var3;
}
this.eventLoop().selectNow();
selected = true;
}
}
}
} 其中：
this.javaChannel().register(this.eventLoop().unwrappedSelector(), 0, this) 将NioSocketChannel執行個體注冊到了目前NioEventLoop的選擇器中，而unwrappedSelector()調用的則是NioEventLoop的unwrappedSelector屬性，即NioEventLoop所擁有的selector：
public final class NioEventLoop extends SingleThreadEventLoop {
...
Selector unwrappedSelector() {
return unwrappedSelector;
}
} 2.4 EventLoop運作
NioEventLoop是一個SingleThreadEventExecutor，SingleThreadEventExecutor是Netty中對本地線程的抽象，是以NioEventLoop的運作就是NioEventLoop所綁定的本地Java線程的運作，是以調用SingleThreadEventExecutor中thread屬性的start()方法就是運作這個線程的入口。
該線程運作入口在SingleThreadEventExecutor的doStartThread()中：
@Override
public void execute(Runnable task) {
...
boolean inEventLoop = inEventLoop();
if (inEventLoop) {
addTask(task);
} else {
startThread();
addTask(task);
...
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}
private void startThread() {
if (state == ST_NOT_STARTED) {
if (STATE_UPDATER.compareAndSet(this, ST_NOT_STARTED, ST_STARTED)) {
try {
doStartThread();
} catch (Throwable cause) {...}
}
}
}
private void doStartThread() {
assert thread == null;
executor.execute(new Runnable() {
@Override
public void run() {
thread = Thread.currentThread();
if (interrupted) {
thread.interrupt();
}
boolean success = false;
updateLastExecutionTime();
try {
SingleThreadEventExecutor.this.run();
success = true;
} catch (Throwable t) {}
...
}
});
} state是SingleThreadEventExecutor内部辨別目前thread狀态的屬性，初始化值為
ST_NOT_STARTED
，是以第一次調用startThread()時會執行if語句，之後執行doStartThread，startThread()是在本類的execute()中被調用。
這個execute()會在AbstractChannel的AbstractUnsafe的register()中被調用，注冊channel時被調用：
@Override
public final void register(EventLoop eventLoop, final ChannelPromise promise) {
...
if (eventLoop.inEventLoop()) {
register0(promise);
} else {
try {
eventLoop.execute(new Runnable() {
@Override
public void run() {
register0(promise);
}
});
} catch (Throwable t) {...}
}
} 在EventLoop中注冊channel的過程中，如果是第一次注冊，即從Bootstrap的bind()或connect()開始執行到AbstractUnsafe的register()時，整個代碼都是在主線程中運作的，是以上面的eventLoop.inEventLoop()就為false，于是進入到else分支，在這個分支中調用了eventLoop.execute，就會觸發startThread() 的調用，進而導緻了EventLoop所對應的Java線程的啟動。
AbstractUnsafe的register()中的eventLoop是一個NioEventLoop的執行個體，而NioEventLoop沒有實作execute方法, 是以調用的就是SingleThreadEventExecutor的execute。是以NioEventLoop啟動的源頭在這裡，是以當EventLoop.execute第一次被調用時。
Netty 的 IO 處理循環
netty中的EventLoop負責如下功能：
1. 作為 IO 線程，負責IO操作，将TCP資料從java nio Socket傳遞到handler中;
2. 作為任務線程，執行taskQueue中的任務。 java nio中，Selector角色會不斷的調用Java NIO的Selector.select()，用于查詢目前已經就緒的IO事件。netty中的Selector角色就是EventLoop。
繼續上面SingleThreadEventExecutor的doStartThread()，其中
SingleThreadEventExecutor.this.run()
實際調用的是NioEventLoop的run()，是以NioEventLoop啟動時實際開啟了一條線程去運作NioEventLoop的run()：
@Override
protected void run() {
for (;;) {
try {
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.SELECT:
select(wakenUp.getAndSet(false));
if (wakenUp.get()) {
selector.wakeup();
}
// fall through
default:
}
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
runAllTasks();
}
} else {
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {...}
...
}
}
private void select(boolean oldWakenUp) throws IOException {
Selector selector = this.selector;
try {
int selectCnt = 0;
for (;;) {
long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
if (timeoutMillis <= 0) {
if (selectCnt == 0) {
selector.selectNow();
selectCnt = 1;
}
break;
}
if (hasTasks() && wakenUp.compareAndSet(false, true)) {
selector.selectNow();
selectCnt = 1;
break;
}
int selectedKeys = selector.select(timeoutMillis);
selectCnt ++;
...
}
} catch (CancelledKeyException e) {...}
} run()中使用了for(;;)循環調用select，選取就緒的事件之後，最終調用processSelectedKey具體處理每個事件：
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
...
try {
int readyOps = k.readyOps();
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
// Process OP_WRITE
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
// Call forceFlush which will also take care of clear the OP_WRITE once there is nothing left to write
ch.unsafe().forceFlush();
}
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
}
} catch (CancelledKeyException ignored) {}
} 在select(boolean oldWakenUp)中，第一步：
long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
if (timeoutMillis <= 0) {
if (selectCnt == 0) {
selector.selectNow();
selectCnt = 1;
}
break;
} 這裡其實就是從一個定時任務隊列中取出定時任務，根據任務的截止時間計算：
protected long delayNanos(long currentTimeNanos) {
ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
if (scheduledTask == null) {
return SCHEDULE_PURGE_INTERVAL;
}
return scheduledTask.delayNanos(currentTimeNanos);
} 計算值為目前離目前定時任務下次執行時間之差，如果目前時間差不足0.5ms的話，即timeoutMillis<=0，那麼認為時間太短，終止本次循環；并且如果目前selectCnt值為0，執行執行一次selectNow。
然後，select(boolean oldWakenUp)調用hasTasks()方法來判斷目前任務隊列中是否有任務：
protected boolean hasTasks() {
assert inEventLoop();
return !taskQueue.isEmpty();
} 這個方法檢查了存放需要目前EventLoop執行的任務清單，即taskQueue是否為空，當taskQueue不為空時，就執行selectNow()，當taskQueue為空時，執行select(timeoutMillis)。
selectNow()和select(timeoutMillis)都用于檢測目前是否有就緒的IO事件，差別是如果目前沒有就緒的IO事件，selectNow()會立即傳回的；而select(timeoutMillis)會阻塞等待timeoutMillis時間。
當hasTasks()為true時，表示目前有任務需要執行，此時應當盡快執行任務，是以此時需要調用selectNow()，不能阻塞目前線程；當hasTasks()為false時，表示沒有需要執行的任務，那麼這時候可以調用select(timeoutMillis)阻塞等待IO就緒事件。
繼續看NioEventLoop的run()：
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
runAllTasks();
}
} else {
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
} processSelectedKeys()是處理分派就緒的IO事件，runAllTasks()是運作taskQueue中的任務。其中ioRatio表示配置給目前線程IO操作所占的時間比(即執行processSelectedKeys()在每次循環中所占用的時間)，ioRatio預設是50，表示IO操作和執行task的耗時為1 : 1。
根據IO操作耗時和ioRatio，可以計算執行task所需要的大概時間：
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio); ioRate = 50時，如果IO耗時100ms，那麼runAllTasks()大概耗時為100ms，當runAllTasks(long timeoutNanos)執行時間大于100ms時則結束循環：
/**
* Poll all tasks from the task queue and run them via {@link Runnable#run()} method. This method stops running
* the tasks in the task queue and returns if it ran longer than {@code timeoutNanos}.
*/
protected boolean runAllTasks(long timeoutNanos) {
...
final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
long runTasks = 0;
long lastExecutionTime;
for (;;) {
...
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.
if ((runTasks & 0x3F) == 0) {
lastExecutionTime = ScheduledFutureTask.nanoTime();
if (lastExecutionTime >= deadline) {
break;
}
}
}
...
} 對于processSelectedKeys()：
private void processSelectedKeys() {
if (selectedKeys != null) {
processSelectedKeysOptimized();
} else {
processSelectedKeysPlain(selector.selectedKeys());
}
} 調用openSelector()時，根據JVM平台的不同selectedKeys會有不同的值，根據selectedKeys是否為空分别調用processSelectedKeysOptimized()或processSelectedKeysPlain()。這兩個方法最終都會調用processSelectedKey()。