bossGroup接受连接
workerGroup处理读写请求
NioEventLoop的run方法
我们先要知道,当netty启动的时候,有几种类型的NioEventLoop实例.第一种是boss端启动的时候,会有一个NioEventLoop来监听客户端的连接,第二种就是workerGroup启动的时候,每个线程都会有一个NioEventLoop实例,这个用来处理客户端的读,写请求.
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
//默认16个线程
EventLoopGroup workerGroup = new NioEventLoopGroup();
然后就是这个线程要做哪些工作?我们知道netty有三种事件类型,第一个是IO事件,第二个是普通的任务,存放到tailTask里的,第三种就是定时任务,存放在scheduledTaskQueue里的.
问题
1.如果一个线程在阻塞中等待io事件,那么普通任务和定时任务怎么处理?
2.如果一个线程忙于普通的任务,此时io事件怎么处理?
3.如果控制普通任务和io任务的比例呢?
超时的select和selectNow
netty只有一种情况会调用select().而selectNow并不是阻塞的,select(long timeout)会阻塞timeout的时间.
//NODE:Long.MAX_VALUE,0x7fffffffffffffffL
if (deadlineNanos == NONE) {
//即便这样,也可以被wakeup唤醒.
return selector.select();
}
wakeup()方法
Causes the first selection operation that has not yet returned to return immediately.
现在可以看具体的源码了,上面的问题想透了,run方法就很简单了.
//run方法被哪调用的?doStartThread()
//NioEventLoop 每次循环的处理流程都包含事件轮询 select、事件处理 processSelectedKeys、任务处理 runAllTasks 几个步骤，
//是典型的 Reactor 线程模型的运行机制。而且 Netty 提供了一个参数 ioRatio，可以调整 I/O 事件处理和任务处理的时间比例。
//下面我们将着重从事件处理和任务处理两个核心部分出发，详细介绍 Netty EventLoop 的实现原理。
@Override
protected void run() {
//
int selectCnt = 0;
//死循环.
for (; ; ) {
try {
//
int strategy;
try {
//策略是什么意思呢?线程什么时候处理io事件,什么时候处理异步队列事件.
//select方法返回值.(The number of keys, possibly zero
//whose ready-operation sets were updated by the selection operation)
strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
switch (strategy) {
//-2
case SelectStrategy.CONTINUE:
continue;
case SelectStrategy.BUSY_WAIT:
// fall-through to SELECT since the busy-wait is not supported with NIO
case SelectStrategy.SELECT:
//下一次定时任务的截止日期.
long curDeadlineNanos = nextScheduledTaskDeadlineNanos();
if (curDeadlineNanos == -1L) {
curDeadlineNanos = NONE; // nothing on the calendar
}
//
nextWakeupNanos.set(curDeadlineNanos);
try {
//如果当前 NioEventLoop 线程存在异步任务，会通过 selectSupplier.get() 最终调用到 selectNow() 方法，selectNow() 是非阻塞，执行后立即返回。
//如果存在就绪的 I/O 事件，那么会走到 default 分支后直接跳出，然后执行 I/O 事件处理 processSelectedKeys 和异步任务队列处理 runAllTasks 的逻辑。
//所以在存在异步任务的场景，NioEventLoop 会优先保证 CPU 能够及时处理异步任务。
//如果没有任务.why?因为select是阻塞的,如果线程阻塞了,异步线程的任务如何处理呢?
//高啊,高啊!!
if (!hasTasks()) {
System.out.println("监听线程启动threadname is:" + Thread.currentThread().getName());
//最终会在这里等待连接.有连接之后,会接着下面的处理.
strategy = select(curDeadlineNanos);
System.out.println("监听到客户端流了...");
}
} finally {
// This update is just to help block unnecessary selector wakeups
// so use of lazySet is ok (no race condition)
nextWakeupNanos.lazySet(AWAKE);
}
// 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();
selectCnt = 0;
handleLoopException(e);
continue;
}
//
selectCnt++;
//
cancelledKeys = 0;
//
needsToSelectAgain = false;
//
final int ioRatio = this.ioRatio;
boolean ranTasks;
if (ioRatio == 100) {
try {
//有就绪的事件,再处理.
if (strategy > 0) {
processSelectedKeys();
}
} finally {
// Ensure we always run tasks.执行是什么任务呢?
ranTasks = runAllTasks();
}
} else if (strategy > 0) {//
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
// Ensure we always run tasks.
final long ioTime = System.nanoTime() - ioStartTime;
//runAllTasks有哪些任务.入参:处理任务的时间.
ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
} else {
//第一次:strategy = 0.
ranTasks = runAllTasks(0); // This will run the minimum number of tasks
}
//
if (ranTasks || strategy > 0) {
if (selectCnt > MIN_PREMATURE_SELECTOR_RETURNS && logger.isDebugEnabled()) {
logger.debug("Selector.select() returned prematurely {} times in a row for Selector {}.",
selectCnt - 1, selector);
}
selectCnt = 0;
} else if (unexpectedSelectorWakeup(selectCnt)) { // Unexpected wakeup (unusual case)
selectCnt = 0;
}
} catch (CancelledKeyException e) {
// Harmless exception - log anyway
if (logger.isDebugEnabled()) {
logger.debug(CancelledKeyException.class.getSimpleName() + " raised by a Selector {} - JDK bug?",
selector, e);
}
} catch (Error e) {
throw (Error) e;
} catch (Throwable t) {
handleLoopException(t);
} finally {
// Always handle shutdown even if the loop processing threw an exception.
try {
if (isShuttingDown()) {
closeAll();
if (confirmShutdown()) {
return;
}
}
} catch (Error e) {
throw (Error) e;
} catch (Throwable t) {
handleLoopException(t);
}
}
}
}
