NioEventLoop启动和执行
NioEventLoop启动
在服务端启动的代码中，我们看到netty在注册和绑定时，判断了当前线程是否是NioEventLoop线程。如果不是，
则将这些操作包装成一个任务丢到EventExecutor中来完成。
// 调用SingleThreadEventExecutor对象的execute方法
eventLoop.execute(() -> register0(promise));
// SingleThreadEventExecutor对象的execute方法
@Override
public void execute(Runnable task) {
boolean inEventLoop = inEventLoop();
addTask(task);
if (!inEventLoop) {
startThread();
}
if (!addTaskWakesUp && wakesUpForTask(task)) {
wakeup(inEventLoop);
}
}
在execute方法中，再次判断是否是NioEventLoop线程，若不是则执行startThread方法。startThread方法通过CAS
将线程的state修改为已启动，成功后进入doStartThread方法。这个方法包装了一个任务，交由在创建NioEventLoop
时设置的Executor执行。默认情况下，它是ThreadPerTaskExecutor，也因此，它会启动一个新的线程执行任务。包装
任务的主要逻辑有3个：
- 将当前线程与nioEventLoop绑定；
- 更新上次执行的时长为当前时间-上个任务启动时间；
- 执行NioEventLoop的run方法；
private void doStartThread() {
executor.execute(() -> {
thread = Thread.currentThread();
updateLastExecutionTime();
SingleThreadEventExecutor.this.run();
});
}
至此，NioEvnetLoop就启动了。
NioEventLoop执行
当NioEventLoop启动后，就开始执行SingleThreadEventExecutor的run方法。此方法是一个死循环，也可以分为3个步骤
- 轮询channel中就绪的IO事件
- 处理轮询出的IO事件
- 处理所有任务，也包括定时任务
轮询事件
整个轮询IO事件的流程如下
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.BUSY_WAIT:
case SelectStrategy.SELECT:
select(wakenUp.getAndSet(false));
if (wakenUp.get()) {
selector.wakeup();
}
default:
}
在循环的开始阶段，调用选择策略器选择select策略，默认策略下，先判断是否有任务，若没有任务，调用selectNow(),否则进入SelectStrategy.SELECT，也即调用select(wakeUp.getAndSet(false))。
selectNow()方法
int selectNow() throws IOException {
try {
return selector.selectNow();
} finally {
if (wakenUp.get()) {
selector.wakeup();
}
}
}
nioEventLoop的selectNow方法会调用持有的Selector对象的selectNow方法。此方法轮询后，即使没有事件也会立即返回,而selector.select方法则会阻塞。
finally操作保证当wakenUp字段为true时，调用一次selector.wakeup方法，此方法会使阻塞的select方法唤醒，若当前没有select阻塞，则下一次select会立即返回。
select(boolean oldWakenUp)方法
首先看到入参为wakeup.getAndSet(false)。wakeup的作用稍后分析，这里简单提一下它的作用是控制将阻塞的selector唤醒。
详细代码如下
private void select(boolean oldWakenUp) throws IOException {
Selector selector = this.selector;
int selectCnt = 0;
// 步骤1
long currentTimeNanos = System.nanoTime();
long selectDeadLineNanos = currentTimeNanos + delayNanos(currentTimeNanos);
long normalizedDeadlineNanos = selectDeadLineNanos - initialNanoTime();
if (nextWakeupTime != normalizedDeadlineNanos) {
nextWakeupTime = normalizedDeadlineNanos;
}
for (;;) {
// 步骤2
long timeoutMillis = (selectDeadLineNanos - currentTimeNanos + 500000L) / 1000000L;
if (timeoutMillis <= 0) {
if (selectCnt == 0) {
selector.selectNow();
selectCnt = 1;
}
break;
}
// 步骤3
if (hasTasks() && wakenUp.compareAndSet(false, true)) {
selector.selectNow();
selectCnt = 1;
break;
}
// 步骤4
int selectedKeys = selector.select(timeoutMillis);
selectCnt++;
if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
break;
}
// 步骤5
if (Thread.interrupted()) {
selectCnt = 1;
break;
}
// 步骤6
long time = System.nanoTime();
if (time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos) {
selectCnt = 1;
} else if (SELECTOR_AUTO_REBUILD_THRESHOLD > 0 &&
selectCnt >= SELECTOR_AUTO_REBUILD_THRESHOLD) {
selector = selectRebuildSelector(selectCnt);
selectCnt = 1;
break;
}
currentTimeNanos = time;
}
代码较长，可以分为6个步骤
- 步骤1计算了多种精确到纳秒级别的时间，⑴当前时间；⑵select阻塞截止时间，这里又会根据是否有定时任务来计算，若有到时间的定时任务，则取最近一个定时
任务的截止时间，若没有定时任务或定时任务还没到时间,则取1秒后;⑶规整化截止时间与下次唤醒时间
从步骤2开始，又进入一个死循环内：
- 四舍五入计算阻塞超时时间。若超时时间小于0且空轮询次数为0，执行一次selectNow后返回。
- 轮询前先判断有没有任务，若有任务，且wakeup由false设置为true了，则执行selectNow。否则会因为无法唤醒selector耽误这个任务的执行。执行完后，结束本次循环。
- 阻塞式select。阻塞结束后，发生下列条件之一时，结束本次循环：⑴轮询到了IO事件；⑵进入select(boolean wakeup)之前，参数oldWakeup为true，也即之前有过wakeup的动作;⑶当前需要唤醒,可能是用户主动调用wakeup方法唤醒的;⑷队列里有任务了，可能是外部线程添加的;⑸有定时任务到期了
- 若线程被打断，设置空轮询次数为1，结束此次循环
- 根据当前时间与进入方法时计算的时间判断阻塞式select是否超时，若time - TimeUnit.MILLISECONDS.toNanos(timeoutMillis) >= currentTimeNanos成立，则有time-currentTimeNanos>=timeoutMillis,说明这次select执行的时间不够，可能触发了空轮询，将空轮询次数为1，计算累计空轮询次数是否大于阈值(阈值SELECTOR_AUTO_REBUILD_THRESHOLD默认为512)，当大于阈值时，重建selector，以规避JDK空轮询bug。反之，则进行了一次有效的select，将累计空轮询次数置为1，结束本次循环。
规避空轮询bug
其实netty规避空轮询bug的方式也很巧妙，就是通过新建selector，并将旧selector上的key和attchment复制过去
private Selector selectRebuildSelector(int selectCnt) throws IOException {
rebuildSelector();
Selector selector = this.select
// Select again to populate selectedKeys.
selector.selectNow();
return selector;
}
private void rebuildSelector0() {
final Selector oldSelector = selector;
final SelectorTuple newSelectorTuple
newSelectorTuple = openSelector();
// 将老selector的key和attchment传递给新selector
for (SelectionKey key : oldSelector.keys()) {
Object a = key.attachment();
int interestOps = key.interestOps();
key.cancel();
SelectionKey newKey = key.channel().register(newSelectorTuple.unwrappedSelector, interestOps, a);
if (a instanceof AbstractNioChannel) {
// Update SelectionKey
((AbstractNioChannel) a).selectionKey = newKey;
}
selector = newSelectorTuple.selector;
unwrappedSelector = newSelectorTuple.unwrappedSelector;
// 关闭老selector
oldSelector.close();
}
}
代码足够详细，就不多加解释了
select(wakeup.getAndSet(false))执行完后，还有这样几行代码①
if (wakenUp.get()) {
selector.wakeup();
}
之前提到wakeup的作用是控制将阻塞的selector唤醒。这里先详细说下。
回顾上文，SingleThreadEventExecutor对象的execute方法有一个添加任务后调用wakeup的动作，nioEventLoop重写了wakeup方法如下
protected void wakeup(boolean inEventLoop) {
if (!inEventLoop && wakenUp.compareAndSet(false, true)) {
selector.wakeup();
}
}
这里进行了2个判断，!inEventLoop表明这是外部线程，selector.wakeup()使阻塞的select操作立即唤醒，以便及时处理此时添加的这个任务。
这个方法让外部线程在加入任务时，能及时唤醒selector处理任务
根据netty的解释，wakeup.compareAndSet(false, true)总是在selector.wakeup之前调用，以便在同时多个任务时减少selector.wakeup的性能消耗。
代码①的注释进一步提到存在两种竞态条件使wakeup太早被设置为true。
- 如果Selector在wakeup.set(false)和selector.select(timeout)之间被唤醒。这里发生在步骤4之前。
- 如果Selector在selector.select()和if(wakeup.get())之间被唤醒。这里发生在步骤4之后。
在第一种情况，接下来的一次selector.select(timeout)(注：称为select1)将立即唤醒。之后由于wakeup为true，wakeup.compareAndSet(false, true)将失败，从而导致无法调用selector.wakeup，
假如这期间(从步骤4到下一次selector.select(timeout)(注：称为select2))加入一个任务，那么得等到下一次select超时，任务才能得到处理。
所以查询完任务后，如果发现wakeup为true，再调用一次selector.wakup()。
不过细心的读者会留意到，在步骤3的几个条件里，netty会调用hasTask查看任务队列是否有任务，且在进入select方法前，会把wakeup设置为false，所以wakenUp.compareAndSet(false, true)会成功，因此会调用selectNow，而不必等到select2超时才处理任务。
第二种情况下，select2会立即返回，没有问题。
那这段代码有何意义？
实际上笔者个人认为这段代码属于遗留代码，理由是笔者在52im找到了netty3的代码，在netty3中的AbstractNioSelector类中，wakeup设置为false后，直接调用了selector.select(timeout)。在当时看来，这不失为一种解决方案。
到了这里，NioEventLoop完成了启动，并查询出了selectionKey，下一步就是处理selectionKey。
so····未完待续
网友评论