//io.netty.channel.nio.NioEventLoop#run
protected void run() {
// 永久循環
for (;;) {
try {
try {
// ------------------------- 1 selector選擇 -------------------
// 計算出選擇selector政策
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE: // NioEventLoop不支援
continue;
case SelectStrategy.BUSY_WAIT: // Nio不支援
// fall-through to SELECT since the busy-wait is not supported with NIO
case SelectStrategy.SELECT: // NioEventLoop支援的唯一政策
// 若執行這裡，說明目前任務隊列中沒有任務
select(wakenUp.getAndSet(false));
// 'wakenUp.compareAndSet(false, true)' is always evaluated
// before calling 'selector.wakeup()' to reduce the wake-up
// overhead. (Selector.wakeup() is an expensive operation.)
//
// However, there is a race condition in this approach.
// The race condition is triggered when 'wakenUp' is set to
// true too early.
//
// 'wakenUp' is set to true too early if:
// 1) Selector is waken up between 'wakenUp.set(false)' and
// 'selector.select(...)'. (BAD)
// 2) Selector is waken up between 'selector.select(...)' and
// 'if (wakenUp.get()) { ... }'. (OK)
//
// In the first case, 'wakenUp' is set to true and the
// following 'selector.select(...)' will wake up immediately.
// Until 'wakenUp' is set to false again in the next round,
// 'wakenUp.compareAndSet(false, true)' will fail, and therefore
// any attempt to wake up the Selector will fail, too, causing
// the following 'selector.select(...)' call to block
// unnecessarily.
//
// To fix this problem, we wake up the selector again if wakenUp
// is true immediately after selector.select(...).
// It is inefficient in that it wakes up the selector for both
// the first case (BAD - wake-up required) and the second case
// (OK - no wake-up required).
// 若目前線程剛被喚醒，selector立即将其選擇的結果傳回給我們
if (wakenUp.get()) {
selector.wakeup();
}
// fall through
default:
}
} catch (IOException e) {
// If we receive an IOException here its because the Selector is messed up. Let's rebuild
// the selector and retry. https://github.com/netty/netty/issues/8566
rebuildSelector0();
handleLoopException(e);
continue;
}
cancelledKeys = 0;
needsToSelectAgain = false;
// ioRatio用于控制IO處理與任務隊列中任務的處理所占時間比例
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
runAllTasks();
}
} else {
// ------------------------- 2 處理就緒的IO -------------------
// IO操作的開始時間
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
// ------------------------- 3 執行任務隊列中的任務 -------------------
// Ensure we always run tasks.
// IO操作總用時
final long ioTime = System.nanoTime() - ioStartTime;
// ioTime * [(100 - ioRatio) / ioRatio]
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {
handleLoopException(t);
}
// Always handle shutdown even if the loop processing threw an exception.
try {
if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
return;
}
}
} catch (Throwable t) {
handleLoopException(t);
}
}
}
以上代碼我們分三大流程：
1.selector選擇
2.處理就緒的IO
3.執行任務隊列中的任務
1.selector選擇
//io.netty.channel.nio.NioEventLoop#run
protected void run() {
// 永久循環
for (;;) {
try {
try {
// ------------------------- 1 selector選擇 -------------------
// 計算出選擇selector政策
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE: // NioEventLoop不支援
continue;
case SelectStrategy.BUSY_WAIT: // Nio不支援
// fall-through to SELECT since the busy-wait is not supported with NIO
case SelectStrategy.SELECT: // NioEventLoop支援的唯一政策
// 若執行這裡，說明目前任務隊列中沒有任務
select(wakenUp.getAndSet(false));
// 'wakenUp.compareAndSet(false, true)' is always evaluated
// before calling 'selector.wakeup()' to reduce the wake-up
// overhead. (Selector.wakeup() is an expensive operation.)
//
// However, there is a race condition in this approach.
// The race condition is triggered when 'wakenUp' is set to
// true too early.
//
// 'wakenUp' is set to true too early if:
// 1) Selector is waken up between 'wakenUp.set(false)' and
// 'selector.select(...)'. (BAD)
// 2) Selector is waken up between 'selector.select(...)' and
// 'if (wakenUp.get()) { ... }'. (OK)
//
// In the first case, 'wakenUp' is set to true and the
// following 'selector.select(...)' will wake up immediately.
// Until 'wakenUp' is set to false again in the next round,
// 'wakenUp.compareAndSet(false, true)' will fail, and therefore
// any attempt to wake up the Selector will fail, too, causing
// the following 'selector.select(...)' call to block
// unnecessarily.
//
// To fix this problem, we wake up the selector again if wakenUp
// is true immediately after selector.select(...).
// It is inefficient in that it wakes up the selector for both
// the first case (BAD - wake-up required) and the second case
// (OK - no wake-up required).
// 若目前線程剛被喚醒，selector立即将其選擇的結果傳回給我們
if (wakenUp.get()) {
selector.wakeup();
}
// fall through
default:
}
} catch (IOException e) {
// If we receive an IOException here its because the Selector is messed up. Let's rebuild
// the selector and retry. https://github.com/netty/netty/issues/8566
rebuildSelector0();
handleLoopException(e);
continue;
}
...
}
}
public final class NioEventLoop extends SingleThreadEventLoop {
...
private final IntSupplier selectNowSupplier = new IntSupplier() {
@Override
public int get() throws Exception {
// 非阻塞選擇
return selectNow();
}
};
/**
* Boolean that controls determines if a blocked Selector.select should
* break out of its selection process. In our case we use a timeout for
* the select method and the select method will block for that time unless
* waken up.
* 布爾值，用于控制選擇器是否被阻塞。select應該跳出它的選擇過程。
* 在我們的例子中，我們為選擇方法使用了一個逾時阻塞方式，而選擇方法将在那個時間内阻塞，除非被喚醒。
* 解釋：簡單來說就是在NioEventLoop的線程在執行SelectStrategy.SELECT政策對應
* 的select方法的時候，裡面會用selector.select(long)指定逾時時間阻塞的方式
* 進行選擇，如果我們希望跳出這個政策對應的整個select方法，可以将wakenUp變量
* 置為true，代表喚醒，内部會自己調用selector.wakeup(),喚醒線程并跳出
* 整個select選擇這一步，而去執行後面處理就緒IO事件和任務隊列中的任務邏輯。
*/
private final AtomicBoolean wakenUp = new AtomicBoolean();
int selectNow() throws IOException {
try {
// NIO底層的非阻塞選擇
return selector.selectNow();
} finally {
// restore wakeup state if needed
// wakenUp為true，表示目前eventLoop所綁定的線程剛剛被喚醒
// wakenUp為false，表示目前eventLoop所綁定的線程即将被阻塞
if (wakenUp.get()) {
// 立即将選擇的結果寫入到目前eventLoop的集合
// 喚醒selector.select方法
selector.wakeup();
}
}
}
...
}
/**
* //java.nio.channels.Selector#wakeup
*
* Causes the first selection operation that has not yet returned to return
* immediately.
* 使尚未傳回的第一個選擇操作立即傳回。
*
* <p> If another thread is currently blocked in an invocation of the
* {@link #select()} or {@link #select(long)} methods then that invocation
* will return immediately. If no selection operation is currently in
* progress then the next invocation of one of these methods will return
* immediately unless the {@link #selectNow()} method is invoked in the
* meantime. In any case the value returned by that invocation may be
* non-zero. Subsequent invocations of the {@link #select()} or {@link
* #select(long)} methods will block as usual unless this method is invoked
* again in the meantime.
*
* <p> Invoking this method more than once between two successive selection
* operations has the same effect as invoking it just once. </p>
*
* @return This selector
*/
public abstract Selector wakeup();
回到selectStrategy.calculateStrategy，再看這裡的意思就很清楚了：
若任務隊列有任務，則馬上進行非阻塞選擇
，此時傳回值肯定是大于等于0，直接走出swith case了，不會被任何一個case選中
//io.netty.channel.DefaultSelectStrategy
final class DefaultSelectStrategy implements SelectStrategy {
static final SelectStrategy INSTANCE = new DefaultSelectStrategy();
private DefaultSelectStrategy() { }
@Override
public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception {
// 若任務隊列有任務，則馬上進行非阻塞選擇
return hasTasks ? selectSupplier.get() : SelectStrategy.SELECT;
}
}
case SelectStrategy.SELECT: // NioEventLoop支援的唯一政策
// 若執行這裡，說明目前任務隊列中沒有任務
// 把wakenUp設定為false，意思是目前線程馬上即将要阻塞，阻塞去處理Selector.select進行選擇
select(wakenUp.getAndSet(false));
// 若目前線程剛被喚醒，selector立即将其選擇的結果傳回給我們
if (wakenUp.get()) {
selector.wakeup();
}
// fall through
default:
private void select(boolean oldWakenUp) throws IOException {
//每個NioEventLoop都有一個Selector，在構造裡面初始化的
Selector selector = this.selector;
try {
// 計數器，記錄目前選擇執行的輪數
int selectCnt = 0;
// 擷取目前select()開始的時間點
long currentTimeNanos = System.nanoTime();
// delayNanos():從定時任務隊列中取出一個定時任務，計算其還有多久就要執行了
// selectDeadLineNanos : 表示這個定時任務要開始執行的時間點
long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
for (;;) {
// --------------------- 1 處理定時任務 ------------------
// 對于馬上就要到執行時間的定時任務，立即進行選擇
long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
if (timeoutMillis <= 0) {
if (selectCnt == 0) {
// 非阻塞選擇
selector.selectNow();
selectCnt = 1;
}
break;
}
// --------------------- 2 在選擇期間，任務隊列中有新任務加入 ------------------
// If a task was submitted when wakenUp value was true, the task didn't get a chance to call
// Selector#wakeup. So we need to check task queue again before executing select operation.
// If we don't, the task might be pended until select operation was timed out.
// It might be pended until idle timeout if IdleStateHandler existed in pipeline.
if (hasTasks() && wakenUp.compareAndSet(false, true)) {
// 非阻塞選擇
selector.selectNow();
selectCnt = 1;
break;
}
// --------------------- 3 阻塞式選擇 ------------------
// select()方法結束的條件：
// 1)有channel被選擇
// 2)seleter.wakeup()被調用
// 3)目前線程被打斷
// 4)阻塞時間逾時
// 5)其實這裡還有一個結束的條件：
// 當長時間沒有就緒的channel時，輪詢會出現長時間空轉，進而會導緻CPU占用率飙升，
// 此時會使select()結束
// 注意，timeoutMillis 在這裡是作為select()的阻塞時長的
int selectedKeys = selector.select(timeoutMillis);
selectCnt ++;
if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
// - Selected something, 有channel被選擇
// - waken up by user, or 或 seleter.wakeup()被調用
// - the task queue has a pending task. // 任務隊列中有挂起的任務
// - a scheduled task is ready for processing // 有定時任務
break;
}
// --------------------- 4 處理一些特殊情況 ------------------
if (Thread.interrupted()) {
// Thread was interrupted so reset selected keys and break so we not run into a busy loop.
// As this is most likely a bug in the handler of the user or it's client library we will
// also log it.
//
// See https://github.com/netty/netty/issues/2426
if (logger.isDebugEnabled()) {
logger.debug("Selector.select() returned prematurely because " +
"Thread.currentThread().interrupt() was called. Use " +
"NioEventLoop.shutdownGracefully() to shutdown the NioEventLoop.");
}
selectCnt = 1;
break;
}
// 代碼走到這裡，說明select()結束的條件是4)或5)
// 記錄目前時間
long time = System.nanoTime();
// 下面的式子等價于：
// time - currentTimeNanos >= TimeUnit.MILLISECONDS.toNanos(timeoutMillis)
// 目前for循環已經執行的時長 >= 阻塞時長
// 若if的這個條件成立，說明前面的select()方法是通過條件4)結束的
if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
// timeoutMillis elapsed without anything selected.
selectCnt = 1;
// 執行else說明 目前for循環已經執行的時長 < 阻塞時長 ，說明前面的select()是通過
// 條件5)結束的。若空轉次數大于等于指定的門檻值512，則重新建構selector
} else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
// The code exists in an extra method to ensure the method is not too big to inline as this
// branch is not very likely to get hit very frequently.
selector = selectRebuildSelector(selectCnt);
selectCnt = 1;
break;
}
currentTimeNanos = time;
}
if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS) {
if (logger.isDebugEnabled()) {
logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
selectCnt - 1, selector);
}
}
} catch (CancelledKeyException e) {
if (logger.isDebugEnabled()) {
logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",
selector, e);
}
// Harmless exception - log anyway
}
}
1.1 處理定時任務
看下delayNanos：從定時任務隊列中取出一個定時任務，計算其還有多久就要執行了
public abstract class SingleThreadEventExecutor extends AbstractScheduledEventExecutor implements OrderedEventExecutor {
...
private static final long SCHEDULE_PURGE_INTERVAL = TimeUnit.SECONDS.toNanos(1);
//io.netty.util.concurrent.SingleThreadEventExecutor#delayNanos
// currentTimeNanos是目前select()開始的時間點
protected long delayNanos(long currentTimeNanos) {
// 從定時任務隊列中擷取一個任務
ScheduledFutureTask<?> scheduledTask = peekScheduledTask();
// 若沒有定時任務，則傳回1秒
if (scheduledTask == null) {
return SCHEDULE_PURGE_INTERVAL;
}
// 傳回任務還有多久就要開始執行
return scheduledTask.delayNanos(currentTimeNanos);
}
//擷取任務該方法在父類實作中
//io.netty.util.concurrent.AbstractScheduledEventExecutor#peekScheduledTask
final ScheduledFutureTask<?> peekScheduledTask() {
Queue<ScheduledFutureTask<?>> scheduledTaskQueue = this.scheduledTaskQueue;
if (scheduledTaskQueue == null) {
return null;
}
// 從定時任務隊列中取出一個任務
// 定時任務隊列中的任務是按照時間升序排列的
return scheduledTaskQueue.peek();
}
...
}
看下scheduledTask.delayNanos
final class ScheduledFutureTask<V> extends PromiseTask<V> implements ScheduledFuture<V>, PriorityQueueNode {
...
//傳回任務還有多久就要開始執行
//currentTimeNanos是目前select()開始的時間點
public long delayNanos(long currentTimeNanos) {
// currentTimeNanos - START_TIME : 目前定時任務系統已經啟動了多久
// deadlineNanos() - (currentTimeNanos - START_TIME) ：目前任務還需要等待多久就可以執行了
// 判斷取最大值，如果存活時間超過了定時任務指定的時間，會是負數，則傳回0，需要馬上執行
return Math.max(0, deadlineNanos() - (currentTimeNanos - START_TIME));
}
// 記錄目前類SingleThreadEventExecutor加載的時間點
// 可以了解為定時任務系統啟動的時間（類變量）
private static final long START_TIME = System.nanoTime();
public long deadlineNanos() {
// 目前定時任務的執行期限
// 這個期限一定是相對于START_TIME的
return deadlineNanos;
}
...
}
private void select(boolean oldWakenUp) throws IOException {
Selector selector = this.selector;
try {
...
for (;;) {
...
int selectedKeys = selector.select(timeoutMillis);
selectCnt ++;
if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
//下面幾種情況會退出循環：
// - Selected something, 有channel被選擇
// - waken up by user, or
// oldWakenUp為true，或者wakenUp.get()為ture
// 代表使用者主動将wakenUp置為true，同時會調用
// seleter.wakeup()方法來喚醒線程，是以這裡代表的
// 是seleter.wakeup()被調用的情況
// 前兩種情況考慮的是select()方法結束的1,2兩種情況
//
// - the task queue has a pending task. // 任務隊列中有挂起的任務
// - a scheduled task is ready for processing // 有定時任務要執行了
// 即最後兩種情況是因為select操作的時候又阻塞了一段時間，
// 是以再重新校驗一下任務隊列是否有任務要處理或者定時任務要執行了
break;
}
// 代碼走到這裡，說明select()結束不是1),2)兩種情況
// 這裡考慮的是select()方法結束原因是目前線程被打斷的情況
if (Thread.interrupted()) {
// Thread was interrupted so reset selected keys and break so we not run into a busy loop.
// As this is most likely a bug in the handler of the user or it's client library we will
// also log it.
//
// See https://github.com/netty/netty/issues/2426
if (logger.isDebugEnabled()) {
logger.debug("Selector.select() returned prematurely because " +
"Thread.currentThread().interrupt() was called. Use " +
"NioEventLoop.shutdownGracefully() to shutdown the NioEventLoop.");
}
selectCnt = 1;
break;
}
// 代碼走到這裡，說明select()結束的條件是4)或5)
// 記錄目前時間
long time = System.nanoTime();
// 下面的式子等價于：
// time - currentTimeNanos >= TimeUnit.MILLISECONDS.toNanos(timeoutMillis)
// 目前for循環已經執行的時長 >= 阻塞時長
// 若if的這個條件成立，說明前面的select()方法是通過條件4)結束的
if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
// timeoutMillis elapsed without anything selected.
selectCnt = 1;
// 執行else說明 目前for循環已經執行的時長 < 阻塞時長 ，說明前面的select()
// 是通過條件5)結束的。若空轉次數大于等于指定的門檻值512，則重新建構selector
} else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
// The code exists in an extra method to ensure the method is not too big to inline as this
// branch is not very likely to get hit very frequently.
//重新建構的邏輯就不關注了
selector = selectRebuildSelector(selectCnt);
//建構完成以後又重新置為1
selectCnt = 1;
break;
}
currentTimeNanos = time;
}
if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS) {
if (logger.isDebugEnabled()) {
logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
selectCnt - 1, selector);
}
}
} catch (CancelledKeyException e) {
if (logger.isDebugEnabled()) {
logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",
selector, e);
}
// Harmless exception - log anyway
}
}
final class SelectedSelectionKeySet extends AbstractSet<SelectionKey> {
//改成數組了
SelectionKey[] keys;
//數組的length
int size;
SelectedSelectionKeySet() {
keys = new SelectionKey[1024];
}
...
}
2.1 如果是優化過的Set就執行processSelectedKeysOptimized：
//io.netty.channel.nio.NioEventLoop#processSelectedKeysOptimized
private void processSelectedKeysOptimized() {
for (int i = 0; i < selectedKeys.size; ++i) {
final SelectionKey k = selectedKeys.keys[i];
// null out entry in the array to allow to have it GC'ed once the Channel close
// 設定為null以後channel關閉的時候可以馬上就行gc
// 其就相當于對set集合處理時，要将處理過的key從set集合中删除是一樣的，
// 為了避免對key的重複處理
selectedKeys.keys[i] = null;
// 對于NioEventLoop，key中的附件attachement中存放的是目前key所關聯的NioChannel
final Object a = k.attachment();
if (a instanceof AbstractNioChannel) {
// 處理目前周遊的key
processSelectedKey(k, (AbstractNioChannel) a);
} else {
//NioTask是在測試的時候放的，我們不用管
@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
selectedKeys.reset(i + 1);
selectAgain();
i = -1;
}
}
}
//io.netty.channel.nio.NioEventLoop#processSelectedKey
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
// 處理key失效的情況
if (!k.isValid()) {
final EventLoop eventLoop;
try {
eventLoop = ch.eventLoop();
} catch (Throwable ignored) {
// If the channel implementation throws an exception because there is no event loop, we ignore this
// because we are only trying to determine if ch is registered to this event loop and thus has authority
// to close ch.
return;
}
// Only close ch if ch is still registered to this EventLoop. ch could have deregistered from the event loop
// and thus the SelectionKey could be cancelled as part of the deregistration process, but the channel is
// still healthy and should not be closed.
// See https://github.com/netty/netty/issues/5125
if (eventLoop != this || eventLoop == null) {
return;
}
// close the channel if the key is not valid anymore
unsafe.close(unsafe.voidPromise());
return;
}
try {
// 擷取到目前key所有就緒的操作
int readyOps = k.readyOps();
// We first need to call finishConnect() before try to trigger a read(...) or write(...) as otherwise
// the NIO JDK channel implementation may throw a NotYetConnectedException.
// 若就緒操作中包含連接配接操作，處理連接配接就緒
// 這個一般在用戶端連接配接才會有這種情況
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// remove OP_CONNECT as otherwise Selector.select(..) will always return without blocking
// See https://github.com/netty/netty/issues/924
int ops = k.interestOps();
// 處理一次就會取消監聽連接配接事件
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
// 進行連接配接
unsafe.finishConnect();
}
// Process OP_WRITE first as we may be able to write some queued buffers and so free memory.
// 處理寫就緒
// 當将資料寫入到buffer，那麼目前channel就處于寫就緒
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();
}
// Also check for readOps of 0 to workaround possible JDK bug which may otherwise lead
// to a spin loop
// 還要檢查readOps是否為0，以解決可能出現的JDK bug，否則可能導緻自旋循環
// 如果沒有關注的事件，也unsafe.read，隻要read就會打斷高速循環
// 處理讀就緒 或 接收連接配接就緒
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
//io.netty.channel.nio.AbstractNioChannel.AbstractNioUnsafe#finishConnect
public final void finishConnect() {
// Note this method is invoked by the event loop only if the connection attempt was
// neither cancelled nor timed out.
// 注意，隻有在連接配接嘗試既沒有取消也沒有逾時的情況下，事件循環才會調用此方法。
assert eventLoop().inEventLoop();
try {
boolean wasActive = isActive();
doFinishConnect();
fulfillConnectPromise(connectPromise, wasActive);
} catch (Throwable t) {
fulfillConnectPromise(connectPromise, annotateConnectException(t, requestedRemoteAddress));
} finally {
// Check for null as the connectTimeoutFuture is only created if a connectTimeoutMillis > 0 is used
// See https://github.com/netty/netty/issues/1770
if (connectTimeoutFuture != null) {
connectTimeoutFuture.cancel(false);
}
connectPromise = null;
}
}
//說了隻有用戶端才有這種事件，是以doFinishConnect的實作我們看NioSocketChannel
//io.netty.channel.socket.nio.NioSocketChannel#doFinishConnect
protected void doFinishConnect() throws Exception {
//看到調用的是nio的finishConnect方法
//之前講nio的時候說過
if (!javaChannel().finishConnect()) {
throw new Error();
}
}
//io.netty.channel.nio.NioEventLoop#processSelectedKeysPlain
private void processSelectedKeysPlain(Set<SelectionKey> selectedKeys) {
// check if the set is empty and if so just return to not create garbage by
// creating a new Iterator every time even if there is nothing to process.
// See https://github.com/netty/netty/issues/597
if (selectedKeys.isEmpty()) {
return;
}
Iterator<SelectionKey> i = selectedKeys.iterator();
for (;;) {
final SelectionKey k = i.next();
final Object a = k.attachment();
i.remove();
if (a instanceof AbstractNioChannel) {
processSelectedKey(k, (AbstractNioChannel) a);
} else {
@SuppressWarnings("unchecked")
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
if (!i.hasNext()) {
break;
}
if (needsToSelectAgain) {
selectAgain();
selectedKeys = selector.selectedKeys();
// Create the iterator again to avoid ConcurrentModificationException
if (selectedKeys.isEmpty()) {
break;
} else {
i = selectedKeys.iterator();
}
}
}
}
io.netty.channel.nio.NioEventLoop#run
protected void run() {
// 永久循環
for (;;) {
try {
// ------------------------- 1 selector選擇 -------------------
...
cancelledKeys = 0;
needsToSelectAgain = false;
// ioRatio用于控制IO處理與任務隊列中任務的處理所占時間比例
// ioRatio初始值50
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
//如果是100，不用控制，全部處理完
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
runAllTasks();
}
} else {
// ------------------------- 2 處理就緒的IO -------------------
// IO操作的開始時間
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
// ------------------------- 3 執行任務隊列中的任務 -------------------
// Ensure we always run tasks.
// IO操作總用時
final long ioTime = System.nanoTime() - ioStartTime;
// ioTime * [(100 - ioRatio) / ioRatio]
// 如果ioRatio是20，則(100-20)/20 = 4 ，即ioTime * 4
// 如果ioRatio是50，則(100-50)50 = 1 ，即ioTime * 1
// ioRatio用于控制IO處理與任務隊列中任務的處理所占時間比例
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {
handleLoopException(t);
}
// Always handle shutdown even if the loop processing threw an exception.
try {
if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
return;
}
}
} catch (Throwable t) {
handleLoopException(t);
}
}
}
接下來看任務處理方法：
//io.netty.util.concurrent.SingleThreadEventExecutor#runAllTasks(long)
protected boolean runAllTasks(long timeoutNanos) {
// 将定時任務隊列中所有可以執行的任務（即已經超過定時時間的任務還沒處理的任務）添加到taskQueue
fetchFromScheduledTaskQueue();
// 從taskQueue中擷取一個任務
Runnable task = pollTask();
// 若該任務為null，說明目前任務隊列中沒有任務了，
// 此時執行tailTasks中的收尾任務
if (task == null) {
afterRunningAllTasks();
return false;
}
// ScheduledFutureTask.nanoTime()傳回的是ScheduledFutureTask類的存活時間
// 可以了解為異步任務系統啟動到現在的時間
// 計算runAllTasks方法最多執行到哪個時間點必須結束，控制處理任務的時間
final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
long runTasks = 0;
long lastExecutionTime;
// 周遊執行taskQueue中的所有任務
for (;;) {
// 執行目前周遊的任務
safeExecute(task);
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次任務檢查一次逾時
if ((runTasks & 0x3F) == 0) {
lastExecutionTime = ScheduledFutureTask.nanoTime();
if (lastExecutionTime >= deadline) {
break;
}
}
task = pollTask();
if (task == null) {
lastExecutionTime = ScheduledFutureTask.nanoTime();
break;
}
} // end-for
// 執行tailTasks中的收尾任務
afterRunningAllTasks();
this.lastExecutionTime = lastExecutionTime;
return true;
}
該方法三個核心流程：
将定時任務隊列中所有可以執行的任務（即已經超過定時時間還沒處理的任務）添加到taskQueue
執行目前周遊的任務
執行tailTasks中的收尾任務
3.1 将定時任務隊列中所有可以執行的任務（即已經超過定時還沒處理的任務）添加到taskQueue
//io.netty.util.concurrent.SingleThreadEventExecutor#fetchFromScheduledTaskQueue
// 将定時任務隊列中所有可以執行的任務添加到taskQueue
private boolean fetchFromScheduledTaskQueue() {
// 擷取目前時間相對于定時任務系統開始的時間的時長(即定時任務系統已經運作的時間)
long nanoTime = AbstractScheduledEventExecutor.nanoTime();
// 從定時任務隊列中取出一個最緊急的需要執行的定時任務
Runnable scheduledTask = pollScheduledTask(nanoTime);
while (scheduledTask != null) {
// 在定時任務不空的前提下，将任務添加到taskQueue
if (!taskQueue.offer(scheduledTask)) {
// No space left in the task queue add it back to the scheduledTaskQueue so we pick it up again.
// 若沒有添加成功，則重新放回到定時任務隊列
scheduledTaskQueue().add((ScheduledFutureTask<?>) scheduledTask);
return false;
}
// 從定時任務隊列中再取出一個最緊急的需要執行的定時任務
scheduledTask = pollScheduledTask(nanoTime);
}
return true;
}
