netty源码分析7-NioEventLoop-run方法疑难点

本文分享内容如下

select()和空轮询bug解决分析
EventLoop 中对selectKeys的改造
wakeup分析

select()和空轮询bug解决分析

当select空轮询（ selector.select(timeoutMillis); 未等待 timeoutMillis）执行次数达到SELECTOR_AUTO_REBUILD_THRESHOLD（默认512）时重新创建 selector, 并注册所有的channel和关注的事件。

private void select() throws IOException {

Selector selector = this.selector;

try {

int selectCnt = 0;

long currentTimeNanos = System.nanoTime();

//delayNanos（）获取即将执行的定时任务距离要执行的时间纳秒差值，没有获取到返回默认值1000ms

long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);

for (;;) {

// 因为EventLoop 要同时 select IO事件和执行任务，不能一直阻塞，当超出期限时间后，就跳出select（）,执行任务。

long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;

// 假设timeoutMillis 1000ms ,经过一次或多次循环后执行时间超出1000ms，则退出select循环。(注释A)

if (timeoutMillis <= 0) {

if (selectCnt == 0) {

selector.selectNow();

selectCnt = 1;

}

break;//code B }

//如果查询到IO事件会正常跳出循环，或者按照timeoutMillis时长阻塞后 code B 跳出循环，否则就是发生了空轮询。

int selectedKeys = selector.select(timeoutMillis);

selectCnt ++;

//有IO事件，被唤醒，有需要执行的任务都跳出循环

if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks()) {

// Selected something,

// waken up by user, or

// the task queue has a pending task.

break;

}

//解决NIO selector 空轮询的bug。注释A 中的处理，当selectCnt数量过大，一定是selector.select(timeoutMillis) 中阻塞功能失效，发生了空轮询，当空轮询数过多时，为了防止空轮询 CPU达到100%，重建selector

if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&

selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {

// The selector returned prematurely many times in a row.

// Rebuild the selector to work around the problem.

logger.warn(

"Selector.select() returned prematurely {} times in a row; rebuilding selector.",

selectCnt);

//重新创建 selector, 并注册所有的channel和关注的事件

rebuildSelector();

selector = this.selector;

// Select again to populate selectedKeys.

selector.selectNow();

selectCnt = 1;

break;

}

currentTimeNanos = System.nanoTime();

}

if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS) {

if (logger.isDebugEnabled()) {

logger.debug("Selector.select() returned prematurely {} times in a row.", selectCnt - 1);

}

} catch (CancelledKeyException e) {

if (logger.isDebugEnabled()) {

logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector - JDK bug?", e);

}

// Harmless exception - log anyway

}

rebuildSelector分析

public void rebuildSelector() {

if (!inEventLoop()) {

execute(new Runnable() {

@Override

public void run() {

rebuildSelector();

}

});

return;

}

final Selector oldSelector = selector;

final Selector newSelector;

if (oldSelector == null) {

return;

}

try {

newSelector = openSelector();

} catch (Exception e) {

logger.warn("Failed to create a new Selector.", e);

return;

}

// Register all channels to the new Selector.

int nChannels = 0;

for (;;) {

try {

for (SelectionKey key: oldSelector.keys()) {

Object a = key.attachment();

try {

if (key.channel().keyFor(newSelector) != null) {

continue;

}

int interestOps = key.interestOps();

key.cancel();

key.channel().register(newSelector, interestOps, a);

nChannels ++;

} catch (Exception e) {

logger.warn("Failed to re-register a Channel to the new Selector.", e);

if (a instanceof AbstractNioChannel) {

AbstractNioChannel ch = (AbstractNioChannel) a;

ch.unsafe().close(ch.unsafe().voidPromise());

} else {

@SuppressWarnings("unchecked")

NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;

invokeChannelUnregistered(task, key, e);

}

} catch (ConcurrentModificationException e) {

// Probably due to concurrent modification of the key set.

continue;

}

break;

}

selector = newSelector;

try {

// time to close the old selector as everything else is registered to the new one

oldSelector.close();

} catch (Throwable t) {

if (logger.isWarnEnabled()) {

logger.warn("Failed to close the old Selector.", t);

}

logger.info("Migrated " + nChannels + " channel(s) to the new Selector.");

}

rebuildSelector整体的逻辑比较清晰，

先创建Selector ，将原来的 channel,interestOps,attachment 注册到新的Selector 上，然后关闭旧的Selector。

EventLoop 中对selectKeys的改造

selectedKeys是一个 SelectedSelectionKeySet 类对象，

每次在轮询到nio事件的时候，netty只需要O(1)的时间复杂度就能将 SelectionKey 塞到 set中去，而jdk底层使用的hashSet需要O(lgn)的时间复杂度
优化过的 SelectedSelectionKeySet 的好处，遍历的时候遍历的是数组，相对jdk原生的HashSet效率有所提高

SelectedSelectionKeySet

当IO事件发生了一定是调用了add方法，这里只需要O(1)的时间复杂度。

public boolean add(SelectionKey o) {

if (o == null) {

return false;

}

if (isA) {

int size = keysASize;

keysA[size ++] = o;

keysASize = size;

if (size == keysA.length) {

doubleCapacityA();

}

} else {

int size = keysBSize;

keysB[size ++] = o;

keysBSize = size;

if (size == keysB.length) {

doubleCapacityB();

}

return true;

}

add 根据isA 判断使用哪个数组，实际上 keysA，keysB 这个两个数组是轮流使用的。

SelectionKey[] flip() {

if (isA) {

isA = false;

keysA[keysASize] = null;//因为数组存在复用，按照add的逻辑 keysASize位置应该是无效的

keysBSize = 0;//翻转前将另一个数组的添加位置赋值为0

return keysA;

} else {

isA = true;

keysB[keysBSize] = null;

keysASize = 0;

return keysB;

}

isA ,filp() 都是为使用2个数组而设计的。

filp这样设计原本是处于高并发，一致性的考虑，在高并发的情况下如果只有一个数组存储SelectKey, 这个数组会一直增长，假设数组没有并发问题，线程会一直处理IO事件，IO任务就一直得不到处理，而数组的修改是有并发问题的，添加进来的SelectKey有可能不会被及时的处理而跳过，而使用两个数组，一个用于添加SelectKey，一个用于SelectKey的分发执行。这样做是巧妙的办法，而新的版本中已经改为一个数组了，作者描述：一个数组虽然有一致性的问题，但是分发执行的时候小心使用可以解决这个问，如传递一个定长的size。

该问题官方描述：https://github.com/netty/netty/issues/6058#

wakeup分析

NioEventLoop run方法负责轮询IO事件和执行IO任务,这里简称为IO轮询方法.

IO轮询方法中有wakeup 的处理，还有wakeup好多的注释，花了我3个多小时，终于研究明白

通过分析原文注释和实验分析得知使用selelct.wakeup（）效果如下：

先执行selelct的还没返回的操作立即返回。

如果没有执行selelct，则下一次阻塞的 select() select(long timeout) 会立即返回

selectNow(), select() select(long timeout)都会清除 wakeup状态，不会影响下次 select() select(long timeout)的阻塞。

NioEventLoop向外暴露的wakeup方法

protected void wakeup(boolean inEventLoop) {

if (!inEventLoop && wakenUp.compareAndSet(false, true)) {

selector.wakeup();

}

这里根据inEventLoop进行判断，也就是说只有初次启动，或非EventLoop线程的才有可能修改wakenUp，并执行selector.wakeup();

调用场景

SingleThreadEventExecutor-execute()

public void execute(Runnable task) {

//...

boolean inEventLoop = inEventLoop();

if (inEventLoop) {

addTask(task);

} else {

startThread();

addTask(task);

if (isShutdown() && removeTask(task)) {

reject();

}

if (!addTaskWakesUp) {//addTaskWakesUp 默认是false

wakeup(inEventLoop);

}

熟悉吧？就是在启动EventLoop或提交IO任务时候会调用wakeup()。为啥要就这样搞这里先留个疑问设为问题1

结合IO轮询方法分析，如下

protected void run() {

for (;;) {

oldWakenUp = wakenUp.getAndSet(false);

try {

if (hasTasks()) {

selectNow();

} else {

select();

//源代码有很多注释，难以读懂设为问题2

if (wakenUp.get()) {

selector.wakeup();

}

//....

}

注意：selector的唤醒都是调用 NioEventLoop.wakeup()

问题1 IO轮询方法中 IO事件IO任务循环顺序执行，如果用户线程提交IO任务，而IO轮询方法所在线程由于没有IO事件，一直阻塞在select(long timeout)中，就影响了用户线程IO任务的执行, 所以需要执行selector.wakeup来停止阻塞，执行用户线程。

问题2 既然要执行selector.wakeup，那么 IO轮询过程中处于阻塞状态中执行是最有用的。分析IO轮询方法会出现2种不理想情况

selector.wakeup在 wakenUp.getAndSet(false) 和 select(long timeout)之间执行
selector.wakeup在 select(long timeout)和 if (wakenUp.get()){。。。}之间执行

情况2 下次执行select(long timeout)不会阻塞，算是尽量满足减少阻塞时间的需求。

情况1 由于执行了select(long timeout)后立即返回，导致selector 的wakeup状态复原，在这个期间，wakenUp.get()=false,后续执行 NioEventLoop.wakeup()不会调用成功,希望减少阻塞的目标没有达成，因此需要尽可能的完成目标。

if (wakenUp.get()) {

selector.wakeup();

}

这个尽可能的减少阻塞事件的处理有问题，如果情况2没有发生，会多执行了一次selector.wakeup();，猜测netty作者是经过权衡，才这么做的。

IO轮询方法中 IO事件IO任务按照配置好的时间比例执行，默认 50比50。selector.wakeup的运用是对此的优化。深究无用，理解到此就可以了。