Netty原始碼學習系列之5-NioEventLoop的run方法
前言
NioEventLoop的run方法，是netty中最核心的方法，沒有之一。在該方法中，完成了對已註冊的channel上來自底層作業系統的socket事件的處理（在服務端時事件包括客戶端的連線事件和讀寫事件，在客戶端時是讀寫事件）、單執行緒任務佇列的處理（服務端的註冊事件、客戶端的connect事件等），當然還包括對NIO空輪詢的規避、訊息的編解碼等。下面一起來探究一番，首先奉上run方法的原始碼：
protected void run() {
for (;;) {
try {
try {
// 1、確定處理策略
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.BUSY_WAIT:
case SelectStrategy.SELECT:
// 2、表示有socket事件，需要進行處理
if (wakenUp.get()) {
selector.wakeup();
}
default:
}
} catch (IOException e) {
// selector有異常，則重新建立一個
handleLoopException(e);
continue;
}
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
try {
// 3、處理來自客戶端或者服務端的socket事件
processSelectedKeys();
} finally {
// 4、處理佇列中的task任務
runAllTasks();
}
} else {
final long ioStartTime = System.nanoTime();
try {
// 3、處理來自客戶端或者服務端的socket事件
processSelectedKeys();
} finally {
final long ioTime = System.nanoTime() - ioStartTime;
// 4、處理佇列中的task任務
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {
handleLoopException(t);
}
// 執行shutdown後的善後邏輯
try {
if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
return;
}
}
} catch (Throwable t) {
handleLoopException(t);
}
}
}
run方法中有四個主要的方法，已在上面註釋中標出，主要邏輯概括起來就是：先通過select方法探知是否當前channel上有就緒的事件（方法1和方法2），然後處理這些事件（方法3），最後再處理佇列中的任務（方法4）。
一、selectStrategy.calculateStrategy方法
selectStrategy只有一個預設實現類DefaultSelectStrategy，實現方法如下，如果判斷有任務，則走selectSupplier.get()方法，否則直接返回SELECT -1，進入方法2-select方法。
public int calculateStrategy(IntSupplier selectSupplier, boolean hasTasks) throws Exception {
return hasTasks ? selectSupplier.get() : SelectStrategy.SELECT;
}
然後看一下匿名類selectSupplier.get方法中的邏輯，如下，可以看到它直接調的非阻塞select方法。
private final IntSupplier selectNowSupplier = new IntSupplier() {
@Override
public int get() throws Exception {
return selectNow();
}
};
總結一下calculateStrategy方法這麼做的用意。從run方法的整體順序中可以看到，每次迴圈中都是先執行方法3處理channel事件，再執行方法4處理佇列中的任務，即處理channel事件的優先順序更高。但如果佇列中有任務待處理，那麼為提高框架處理效能，就不允許執行阻塞的select方法，而是執行非阻塞的selectNow方法，這樣就能快速處理完channel事件後去處理佇列中的任務。
二、select(boolean)方法
要理解該方法，需先理解wakenUp變數和wakeup方法的作用。wakenUp是AtomicBoolean型別的變數，如果是true，則表示最近呼叫過wakeup方法，如果是false，則表示最近未呼叫wakeup方法，另外每次進入select(boolean)方法都會將wakenUp置為false。而wakeup方法是針對selector.select方法設計的，如果呼叫wakeup方法時處於selector.select阻塞方法中，則會直接喚醒處於selector.select阻塞中的執行緒，而如果呼叫wakeup方法時selector不處於selector.select阻塞方法中，則效果是在下一次調selector.select方法時不阻塞（有點像LockSupport.park/unpark的效果）。下面是select(boolean)方法邏輯：
private void select(boolean oldWakenUp) throws IOException {
Selector selector = this.selector;
try {
int selectCnt = 0;
long currentTimeNanos = System.nanoTime();
long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
for (;;) {
long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
if (timeoutMillis <= 0) {
if (selectCnt == 0) {
selector.selectNow();
selectCnt = 1;
}
break;
}
// 重點1：在呼叫阻塞的select方法前再判斷一遍是否有任務需要處理，此處邏輯雖然不多，但有深意 ***
if (hasTasks() && wakenUp.compareAndSet(false, true)) {
selector.selectNow();
selectCnt = 1;
break;
}
// 呼叫阻塞的select方法，但設定了超時時間
int selectedKeys = selector.select(timeoutMillis);
selectCnt ++;
if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
// 有事件；wakenUp之前是true(說明有新任務進入了佇列中)；wakenUp現在是true（說明有新任務在本方法執行的過程中進來了），有任務 滿足以上任意一個都退出迴圈
break;
}
if (Thread.interrupted()) {
// 省略異常日誌列印
selectCnt = 1;
break;
}
long time = System.nanoTime();
if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
// timeoutMillis elapsed without anything selected.
selectCnt = 1;
} else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
// 重點2： 說明觸發了空輪訓，需要做處理
selector = selectRebuildSelector(selectCnt);
selectCnt = 1;
break;
}
currentTimeNanos = time;
}
// catch 異常處理
}
該方法有兩處重點，均已標出。
重點1
該處邏輯需結合wakenUp變數和wakeup方法來理解。
首先，對wakenUp變數的操作除了run方法外，還有SingleThreadEventExecutor的execute方法。execute中新增完task後，會呼叫NioEventLoop中的重寫方法wakeup：
protected void wakeup(boolean inEventLoop) {
if (!inEventLoop && wakenUp.compareAndSet(false, true)) {
selector.wakeup();
}
}
注：selector.wakenUp方法用於喚醒被selector.select()或者selector.select(long time)阻塞的selector，讓其立馬返回key的數量。
它做了兩件事，1是通過cas將wakenUp由false變為true，2是呼叫selector.wakeup方法。
再來看select(boolean)方法的入口處，通過wakenUp.getAndSet(false)方法將wakenUp設為false，然後將原值作為入參傳入select(boolean)方法。
一切條件就緒，然後再回過頭看重點1（如下）。它想實現的功能就是如果佇列中有新的任務來了，能不調selector.select的阻塞方法，有任務等待執行時能不阻塞就不阻塞，提高效率。
if (hasTasks() && wakenUp.compareAndSet(false, true)) {
selector.selectNow();
selectCnt = 1;
break;
}
但細究一下會發現這個方法的兩個判斷邏輯存在一個矛盾，首先進入當前select(boolean)方法時，wakenUp被置為false，而在新增完任務後，NioEventLoop中的wakeup方法又會將wakenUp置為true，即如果hasTasks()方法返回true時，因為wakenUp已經被置為true了所以第二個條件肯定判斷為false，那if裡面的邏輯什麼場景下才會走到呢？
不知道各位園友們走到這裡的時候會不會有這樣的疑問，反正博主剛開始是被自己難倒了，後來又重新分析了下才找到原因。其實博主剛才對矛盾點的描述就未分清時間先後。因為有新任務來的時候，是先往佇列中新增任務，再將wakenUp置為true（selector.wakeup()方法可以認為與置為true是同時發生的），即如果添加了task但還沒來得及將wakenUp置為true時才會進入這個if中。
那麼新的問題來了，為什麼將wakenUp置為true了就不用進if中呢？是因為如果wakenUp已經是true了，那麼可以認為已經執行了selector.wakeup方法了，既然如此，selector.select雖然是阻塞方法也就不會再阻塞了，而是直接返回結果，所以沒必要再進if中。
此處還有一個容易讓人迷糊的地方就是下面的四個或的邏輯判斷：
if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
break;
}
即滿足這四個條件中的任意一個就退出迴圈，這四個條件各代表什麼意思？
第一個：channel中有socket事件需處理，這個肯定是要跳出迴圈處理的；
第二個：oldWakenUp為true，即進select(boolean)方法之前wakenUp為true，說明佇列中有新任務來了，所以也要跳出迴圈，出去處理；
第三個：wakenUp現在為true，說明在進入select(boolean)方法之後佇列中有新任務來了，需跳出迴圈處理；
第四/五個：兩個佇列中有任務，需出去處理。
其實就是說，如果當前沒有事件過來，佇列中又沒有任務處理，那麼就繼續走select(boolean)的無限for迴圈（反正沒事做），否則說明來菜了需要跳出迴圈出去處理。
重點2：
對於空輪訓的處理其實沒有太多花哨的地方，netty開發者設定了一個閾值512，如果selectCnt計數達到了512，說明觸發了空輪訓，此時selectRebuildSelector 方法會建立一個新的selector，將原selector上的全部事件重新註冊到新selector上。
注：空輪訓即調select(time)/select()阻塞方法的時候，由於出現了bug導致不阻塞而是直接返回空結果，並且後面每次都這樣，彷彿螺絲滑了絲一般順滑，，，
三、processSelectedKeys()方法
點進去看到裡面的邏輯，第一個方法是優化之後的處理，第二個是未優化的處理，一般都是走優化的邏輯。
private void processSelectedKeys() {
if (selectedKeys != null) {
processSelectedKeysOptimized();
} else {
processSelectedKeysPlain(selector.selectedKeys());
}
}
processSelectedKeysOptimized方法如下：
private void processSelectedKeysOptimized() {
for (int i = 0; i < selectedKeys.size; ++i) {
final SelectionKey k = selectedKeys.keys[i];
selectedKeys.keys[i] = null;
final Object a = k.attachment();
if (a instanceof AbstractNioChannel) {
processSelectedKey(k, (AbstractNioChannel) a); // 從attachment中取出之前放入的AbstractNioChannel物件，進行處理
} else {
@SuppressWarnings("unchecked")
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
if (needsToSelectAgain) {
selectedKeys.reset(i + 1);
selectAgain();
i = -1;
}
}
}
繼續跟進針對單個SelectionKey的處理：
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
if (!k.isValid()) {
// 針對無效key的處理
}
try {
int readyOps = k.readyOps(); // 獲取已經就緒的操作型別
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
// 1、針對連線事件的處理
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
// 2、針對寫事件的處理
ch.unsafe().forceFlush();
}
///3、針對讀事件/接受連線事件的處理
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
可以看到，在此方法中按不同的事件型別呼叫unsafe方法對其進行處理，再往後追溯就是pipeline的相關處理了，具體內容較多，有興趣可自行檢視，後面有機會博主也會繼續更新。
有一點需要著重提的是對ACCEPT事件的處理（服務端在接收到客戶端的連線請求時觸發該事件），因為是服務端，所以進入AbstractNioMessageChannel.NioMessageUnsafe#read方法，
可以看到有段do/while迴圈，如下:
do {
int localRead = doReadMessages(readBuf);
if (localRead == 0) {
break;
}
if (localRead < 0) {
closed = true;
break;
}
allocHandle.incMessagesRead(localRead);
} while (allocHandle.continueReading());
doReadMessages方法的實現位於NioServerSocketChannel中，可以看到第五行往buf中添加了一個NioSocketChannel物件。
protected int doReadMessages(List<Object> buf) throws Exception {
SocketChannel ch = SocketUtils.accept(javaChannel());
try {
if (ch != null) {
buf.add(new NioSocketChannel(this, 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;
}
再跳出來回到read方法，往下看有個for迴圈，開始了pipeline的呼叫，結合前面【https://www.cnblogs.com/zzq6032010/p/13034608.html】bind方法的博文可以知道，此時pipeline中除了頭尾兩個節點以外，還有一個ServerBootstrapAcceptor，此處最終就會調到ServerBootstrapAcceptor的channelRead方法，該方法很重要，最終將上面生成的NioSocketChannel中的pipeline、channelOption、attr初始化，然後註冊到childGroup上。至此，服務端具備了與客戶端通訊的能力，可正常處理read、write事件了。
int size = readBuf.size();
for (int i = 0; i < size; i ++) {
readPending = false;
pipeline.fireChannelRead(readBuf.get(i));
}
四、runAllTasks()
再貼上一下runAllTasks附近的程式碼：
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
首先說一下ioRatio變數，此變數控制的是當前執行緒中處理channel事件和處理任務佇列所用的時間比，如果為50（即50%），則二者用的時間相同，從上面程式碼中可以看出，ioTime即處理channel事件所用的時間，當ioRatio=50時，runAllTasks的入參就是ioTime；而如果ioRatio=10，則runAllTasks入參為9*ioTime，即處理任務佇列的最大時間是處理channel事件的9倍。
下面是runAllTasks方法程式碼：
protected boolean runAllTasks(long timeoutNanos) {
fetchFromScheduledTaskQueue();
Runnable task = pollTask();
if (task == null) {
afterRunningAllTasks();
return false;
}
final long deadline = ScheduledFutureTask.nanoTime() + timeoutNanos;
long runTasks = 0;
long lastExecutionTime;
for (;;) {
safeExecute(task);
runTasks ++;
if ((runTasks & 0x3F) == 0) { // 每隔64次計算一下超時時間
lastExecutionTime = ScheduledFutureTask.nanoTime();
if (lastExecutionTime >= deadline) {
break;
}
}
task = pollTask();
if (task == null) {
lastExecutionTime = ScheduledFutureTask.nanoTime();
break;
}
}
afterRunningAllTasks();
this.lastExecutionTime = lastExecutionTime;
return true;
}
整體邏輯不難，用一個for迴圈來依次取出任務處理，並且為了提高效率，每隔64次計算一下超時時間（對netty開發者來說，獲取系統納秒時間也是一筆效能開支，能少獲取就少獲取）。
總結
netty中最核心的run方法就介紹到這裡，至此，netty進行資料傳輸前的準備工作都已經過了一遍，但對於netty具體傳送、接收資料的流程還未涉及到。netty具體傳送、接收資料是藉助pipeline和在childHandler中新增的處理器完成的，這部分將不定期的在後面博文中講述，具體看緣分吧。