开坑！Netty作为主流的网络框架，活跃在微服务的各种流行组件里，是响应式微服务这座摩天大楼中最粗那根承重柱，现如今的Netty可谓不可不学，今天就把Netty最核心的组件EventLoop再撸一遍，面试的时候多几分底气。
一. SingleThreadEventExecutor
先看下NioEventLoop继承关系图，就会发现这货是个线程池，而且还是单线程的那种线程池，因此我们直接找到他父类中实现execute()地方。
在reactor模型中，我们知道每个selector会由一个线程去处理它上面的事件。在这里也是一样的，每个EventLoop会和一个工作线程绑定，inEventLoop的作用是看当前线程是否是EventLoop所绑定的线程，如果不是，则代表着该EventLoop的工作线程未启动，需要通过startThread()先启动，再将任务加入到EventLoop所维护的队列中。
public void execute(Runnable task) {
...
boolean inEventLoop = inEventLoop();
if (inEventLoop) {
addTask(task);//加入任务队列
} else {
startThread();//启动对应的工作线程
addTask(task);
if (isShutdown() && removeTask(task)) {
reject();
}
}
...
}
继续往下走，经历doStartThread() 到 SingleThreadEventExecutor.this.run()我们就来到了一个最核心的方法-----NioEventLoop的run方法。
二. Run
run方法中有三个核心步骤，分别是select(wakenUp.getAndSet(false))、 processSelectedKeys()、runAllTasks()。
select(wakenUp.getAndSet(false)) 主要是负责判断任务队列中有没有任务或者selector中有没有就绪的事件，顺便还解决了nio的selector上的空轮询问题。
processSelectedKeys() 主要是负责处理selector线程上就绪的事件，主EventLoop负责处理接收客户端socketChannel，子EventLoop负责处理socketChannel的读写等事件。
runAllTasks() 主要是负责任务的执行，比如客户端和服务端在建立连接后，会启动一个心跳任务检查对方服务是否还存活，像这种定时任务，NioEventLoop维护了两个队列一个定时任务队列，一个普通任务队列，当实行到这个方法时，将到达临界点的定时任务移到普通任务队列中，再从普通任务队列一个个拿出来去执行。
protected void run() {
for (;;) {
try {
switch (selectStrategy.calculateStrategy(selectNowSupplier, hasTasks())) {
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.SELECT:
select(wakenUp.getAndSet(false)); // 核心方法1，负责判断任务队列中有没有任务或者selector中有没有就绪的事件，顺便还解决了nio的selector上的空轮询问题
if (wakenUp.get()) {
selector.wakeup();
}
default:
}
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
if (ioRatio == 100) {
try {
processSelectedKeys();// 核心方法2，负责处理selector线程上就绪的事件
} finally {
runAllTasks();//// 核心方法3，负责处理任务队列或延时任务队列中的任务
}
} else {
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
final long ioTime = System.nanoTime() - ioStartTime;
runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
}
} catch (Throwable t) {
handleLoopException(t);
}
...
}
}
2.1 select()
第一部分，计算timeoutMillis 的作用是看定时任务队列中有没有即将到期的任务（到期时间<0.5ms）,如果有则跳出循环(无论什么时候跳出都会执行selector.selectNow()或者selector.select())。
第二部分， int selectedKeys = selector.select(timeoutMillis) 此时就是selector阻塞的监听事件的到来，不过设置了超时时间。
第三部分，解决空轮询Bug。每次for循环都会判断此次循环的时间，如果此时间过小且持续512次，则会重建selector（将之前注册到上面的channel重新移到新的selector上）
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;
}
if (hasTasks() && wakenUp.compareAndSet(false, true)) {
selector.selectNow();
selectCnt = 1;
break;
}
int selectedKeys = selector.select(timeoutMillis);
selectCnt ++;
if (selectedKeys != 0 || oldWakenUp || wakenUp.get() || hasTasks() || hasScheduledTasks()) {
break;
}
if (Thread.interrupted()) {
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;
}
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) {
logger.warn(
"Selector.select() returned prematurely {} times in a row; rebuilding Selector {}.",
selectCnt, selector);
rebuildSelector();
selector = this.selector;
selector.selectNow();
selectCnt = 1;
break;
}
currentTimeNanos = time;
}
...
}
}
2.2 processSelectedKeys()
这里面有两个路线
1.processSelectedKeysOptimized，操作优化过后的SelectedSelectionKeySet，这里是一个数组
2.processSelectedKeysPlain，操作原始的Selector，这里是一个HashSet
我们选择第一个进去看看。(这里关于SelectedSelectionKeySet以后我再研究下)
private void processSelectedKeys() {
if (selectedKeys != null) {
processSelectedKeysOptimized(selectedKeys.flip());
} else {
processSelectedKeysPlain(selector.selectedKeys());
}
}
在这里，就是一个循环拿事件去处理的一个过程。我们进到processSelectedKey里面看下到底是怎么处理的。
private void processSelectedKeysOptimized(SelectionKey[] selectedKeys) {
for (int i = 0;; i ++) {
final SelectionKey k = selectedKeys[i];
if (k == null) {
break;
}
selectedKeys[i] = null;
final Object a = k.attachment();
if (a instanceof AbstractNioChannel) {
processSelectedKey(k, (AbstractNioChannel) a);
} else {
NioTask<SelectableChannel> task = (NioTask<SelectableChannel>) a;
processSelectedKey(k, task);
}
if (needsToSelectAgain) {
for (;;) {
i++;
if (selectedKeys[i] == null) {
break;
}
selectedKeys[i] = null;
}
selectAgain();
selectedKeys = this.selectedKeys.flip();
i = -1;
}
}
}
看到下面这些OP开头的值，不就是NIO中的事件吗？
对于Boss EventLoopGroup来讲，就是处理ACCEPT事件接收socketChannel并且将其注册到Work EventLoopGroup的其中一个EventLoop监听读写事件。
对于Work EventLoopGroup来讲，就是响应读写事件，交由对应的channelPipeline来完成对应的处理。
private void processSelectedKey(SelectionKey k, AbstractNioChannel ch) {
final AbstractNioChannel.NioUnsafe unsafe = ch.unsafe();
if (!k.isValid()) {
final EventLoop eventLoop;
try {
eventLoop = ch.eventLoop();
}
....
}
try {
int readyOps = k.readyOps();
if ((readyOps & SelectionKey.OP_CONNECT) != 0) {
int ops = k.interestOps();
ops &= ~SelectionKey.OP_CONNECT;
k.interestOps(ops);
unsafe.finishConnect();
}
if ((readyOps & SelectionKey.OP_WRITE) != 0) {
ch.unsafe().forceFlush();
}
if ((readyOps & (SelectionKey.OP_READ | SelectionKey.OP_ACCEPT)) != 0 || readyOps == 0) {
unsafe.read();
if (!ch.isOpen()) {
return;
}
}
} catch (CancelledKeyException ignored) {
unsafe.close(unsafe.voidPromise());
}
}
2.3 runAllTasks()
NioEventLoop中有两个队列一个是存储定时任务的scheduledTaskQueue，另一个是执行普通任务的taskQueue。
这个fetchFromScheduledTaskQueue方法，会把到期的任务从scheduledTaskQueue移动到taskQueue中，之后调用runAllTasksFrom循环执行任务。
protected boolean runAllTasks() {
assert inEventLoop();
boolean fetchedAll;
boolean ranAtLeastOne = false;
do {
fetchedAll = fetchFromScheduledTaskQueue();
if (runAllTasksFrom(taskQueue)) {
ranAtLeastOne = true;
}
} while (!fetchedAll); // keep on processing until we fetched all scheduled tasks.
if (ranAtLeastOne) {
lastExecutionTime = ScheduledFutureTask.nanoTime();
}
afterRunningAllTasks();
return ranAtLeastOne;
}
