文章目录
- 前言
- 类图
- 主要功能
- NioEventLoop如何实现事件循环
- NioEventLoop如何处理多路复用
- Netty如何管理Channel和Selector
- 管理Channel
- 管理Selector
- 注意事项
前言
Netty通过事件循环机制(EventLoop)处理IO事件和异步任务,简单来说,就是通过一个死循环,不断处理当前已发生的IO事件和待处理的异步任务。
① NioEventLoop是一个基于JDK NIO的异步事件循环类,它负责处理一个Channel的所有事件在这个Channel的生命周期期间。
② NioEventLoop的整个生命周期只会依赖于一个单一的线程来完成。一个NioEventLoop可以分配给多个Channel,NioEventLoop通过JDK Selector来实现I/O多路复用,以对多个Channel进行管理。
③ 如果调用Channel操作的线程是EventLoop所关联的线程,那么该操作会被立即执行。否则会将该操作封装成任务放入EventLoop的任务队列中。
④ 所有提交到NioEventLoop的任务都会先放入队列中,然后在线程中以有序(FIFO)/连续的方式执行所有提交的任务。
⑤ NioEventLoop的事件循环主要完成了:
- 已经注册到Selector的Channel的监控,并在感兴趣的事件可执行时对其进行处理;
- 完成任务队列(taskQueue)中的任务,以及对可执行的定时任务和周期性任务的处理(scheduledTaskQueue中的可执行的任务都会先放入taskQueue中后,再从taskQueue中依次取出执行)。
类图
主要功能
NioEventLoop
主要负责以下几个方面的功能:
- 事件循环:
NioEventLoop
是一个单线程的事件循环,它不断地轮询注册在其上的 Channel,看是否有就绪的 I/O 事件(如读、写、连接等),然后进行相应的处理。 - 任务执行:除了处理 I/O 事件外,
NioEventLoop
还负责执行定时任务和普通任务。这些任务可以是用户自定义的,也可以是 Netty 框架内部的。 - Channel 的注册与注销:Netty 中的 Channel 在创建后需要注册到一个
EventLoop
上,这样它才能开始处理 I/O 事件。同样地,当 Channel 不再需要时,也需要从EventLoop
上注销。 - Selector 的管理:
NioEventLoop
内部使用 Java NIO 的Selector
来轮询 Channel 的就绪状态。它负责Selector
的创建、销毁以及选择操作。
在 Netty 中,通常会有多个 NioEventLoop
实例,它们通常与多线程模型结合使用,以实现真正的并发处理。每个 NioEventLoop
负责处理一组 Channel 的 I/O 事件,这样可以将不同的 Channel 分布到不同的线程上,从而实现并发处理。
总的来说,NioEventLoop
是 Netty 中负责处理 I/O 事件和任务执行的核心组件,它使得 Netty 能够高效地处理大量的网络连接和 I/O 操作。
NioEventLoop如何实现事件循环
NioEventLoop
在Netty中实现事件循环的过程是Netty网络框架的核心机制之一,它结合了Java NIO的非阻塞特性和事件驱动模型,以高效地处理网络事件。以下是NioEventLoop
如何实现事件循环的详细过程:
-
轮询就绪事件:
在事件循环中,NioEventLoop
调用Selector
的select()
方法来等待Channel上就绪的事件。select()
方法会阻塞,直到至少有一个Channel的事件就绪,或者超时。 -
处理就绪事件:
一旦select()
方法返回,NioEventLoop
会获取就绪的Channel集合,并遍历这些Channel。对于每个就绪的Channel,NioEventLoop
会根据其感兴趣的事件类型调用相应的处理器(通常是ChannelPipeline
中的ChannelInboundHandler
)来处理这些事件。这可能包括读取数据、写入数据、处理连接等。 -
执行非I/O任务:
除了处理I/O事件外,NioEventLoop
还负责执行提交给它的非I/O任务。这些任务可能包括用户自定义的业务逻辑、回调方法等。NioEventLoop
通常有一个任务队列,用于存储待执行的任务。在事件循环中,它会检查任务队列,并依次执行其中的任务。 -
定时任务调度:
NioEventLoop
还支持定时任务的调度。用户可以将定时任务提交给NioEventLoop
,并指定任务的执行时间或执行间隔。NioEventLoop
内部会维护一个定时任务列表,并根据任务的调度信息在合适的时间点执行这些任务。
通过这个事件循环机制,NioEventLoop
能够高效地处理大量的网络连接和I/O事件,同时保持单线程的事件处理模型,简化了并发控制和线程安全的处理。此外,Netty还通过多线程模型和多个NioEventLoop
实例的配合使用,实现了真正的并发处理,从而能够处理更大规模的并发连接和事件。
@Override
protected void run() {
int selectCnt = 0;
// 必须使用死循环不断进行事件轮询,获取任务和通道的 IO 事件
for (;;) {
try {
int strategy;
try {
/**
* 返回处理策略,就分为两种:
* 有任务 hasTasks() == true,就不能等待IO事件了,先直接调用 selectNow() 方法,
* 获取当前准备好IO 的通道channel 的数量(0 表示一个都没有),处理 IO 事件 和任务。
*
* 没有任务 hasTasks() == false,返回 SelectStrategy.SELECT (是负数),
* 没有要及时处理的任务,先阻塞等待 IO 事件
*/
strategy = selectStrategy.calculateStrategy(selectNowSupplier, hasTasks());
switch (strategy) {
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) {
// 返回 -1,说明没有下一个计划任务,
// 将 curDeadlineNanos 设置为 NONE,
// 调用 selector.select 方法时,就没有超时,
// 要无限等待了,除非被唤醒或者有准备好的 IO 事件。
curDeadlineNanos = NONE;
}
// 设置 超时等待时间
nextWakeupNanos.set(curDeadlineNanos);
try {
if (!hasTasks()) {
// 当前没有任务,那么就通过 selector 查看有没有 IO 事件
// 并设置超时时间,超时时间到了那么就要执行计划任务了
// 如果 curDeadlineNanos 是 NONE,就没有超时,无限等待。
strategy = select(curDeadlineNanos);
}
} finally {
// 这个更新只是为了帮助阻止不必要的选择器唤醒,
// 所以使用lazySet是可以的(没有竞争条件)
nextWakeupNanos.lazySet(AWAKE);
}
// fall through
default:
}
} catch (IOException e) {
// 如果我们在这里接收到IOException,那是因为Selector搞错了。
// 让我们重新构建选择器并重试。
// https://github.com/netty/netty/issues/8566
rebuildSelector0();
selectCnt = 0;
handleLoopException(e);
continue;
}
/**
* 代码走到这里,
* 要么有 IO 事件,即 strategy >0
* 要么就是有任务要运行。
* 如果两个都不是,那么就有可能是 JDK 的 epoll 的空轮询 BUG
*/
selectCnt++;
cancelledKeys = 0;
needsToSelectAgain = false;
final int ioRatio = this.ioRatio;
boolean ranTasks;
if (ioRatio == 100) {
// 如果 ioRatio
try {
if (strategy > 0) {
processSelectedKeys();
}
} finally {
// 确保运行了所有待执行任务,包括当前时间已经过期的计划任务
ranTasks = runAllTasks();
}
} else if (strategy > 0) {
// strategy > 0 说明有 IO 事件,
// 那么需要调用 processSelectedKeys() 方法,执行 IO 时间
final long ioStartTime = System.nanoTime();
try {
processSelectedKeys();
} finally {
// 计算 IO 操作花费的时间
final long ioTime = System.nanoTime() - ioStartTime;
// 按照比例计算可以运行任务的超时时间 ioTime * (100 - ioRatio) / ioRatio,
// 超时时间到了,即使还有任务没有运行,也直接返回了,等下一个周期在运行这些任务
ranTasks = runAllTasks(ioTime * (100 - ioRatio) / ioRatio);
}
} else {
// strategy == 0 说明没有 IO 事件,不用处理 IO 了
// 调用 runAllTasks(0) 方法,超时时间为0,这将运行最小数量的任务
ranTasks = runAllTasks(0);
}
if (ranTasks || strategy > 0) {
// 要么有任务运行,要么有 IO 事件处理
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)) {
// 即没有任务运行,也没有IO 事件处理,就有可能是 JDK 的 epoll 的空轮询 BUG
// 调用 unexpectedSelectorWakeup(selectCnt) 方法处理。
// 可能会重新建立 Select
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 e;
} catch (Throwable t) {
handleLoopException(t);
} finally {
// Always handle shutdown even if the loop processing threw an exception.
try {
if (isShuttingDown()) {
// 如果事件轮询器开始 shutdown,就要关闭 IO 资源
closeAll();
if (confirmShutdown()) {
return;
}
}
} catch (Error e) {
throw e;
} catch (Throwable t) {
handleLoopException(t);
}
}
}
}
NioEventLoop如何处理多路复用
NioEventLoop
在 Netty 中处理多路复用的方式主要是基于 Java NIO 的非阻塞特性和选择器(Selector
)机制。多路复用允许单个线程同时处理多个通道(Channel
)的 I/O 事件,从而提高了系统的吞吐量和响应速度。
以下是 NioEventLoop
如何处理多路复用的具体步骤:
-
初始化与注册:
NioEventLoop
在初始化时创建一个Selector
对象,用于监控多个Channel
的状态。- 当一个
Channel
需要处理 I/O 事件时,它会被注册到NioEventLoop
的Selector
上,并指定感兴趣的事件类型(如读、写等)。
-
轮询就绪事件:
NioEventLoop
进入事件循环,调用Selector
的select()
方法等待通道上就绪的事件。select()
方法会阻塞,直到至少有一个Channel
的事件就绪,或者超时。
-
处理就绪事件:
- 当
select()
方法返回时,NioEventLoop
获取就绪的Channel
集合。 - 对于每个就绪的
Channel
,NioEventLoop
根据其感兴趣的事件类型调用相应的处理器(如ChannelInboundHandler
)来处理这些事件。
- 当
-
多路复用核心:
Selector
的核心作用在于它能够同时监控多个Channel
,并且只选择那些处于就绪状态的Channel
进行处理。- 通过这种方式,
NioEventLoop
可以使用单个线程高效地处理大量并发的Channel
,而无需为每个Channel
创建一个独立的线程。
-
异步非阻塞通信:
- 由于 Netty 采用了异步通信模式,
NioEventLoop
中的 IO 操作(如读、写)都是非阻塞的。 - 这意味着当一个 IO 操作不能立即完成时(例如,等待数据从网络读取),线程不会被阻塞,而是可以继续处理其他任务或事件。
- 由于 Netty 采用了异步通信模式,
-
任务调度与执行:
- 除了处理 I/O 事件外,
NioEventLoop
还负责执行提交给它的任务(如定时任务、用户自定义任务等)。 - 这些任务与 I/O 事件一起,在
NioEventLoop
的事件循环中得到调度和执行。
- 除了处理 I/O 事件外,
Netty如何管理Channel和Selector
管理Channel
-
注册Channel:
当一个新的连接建立时,Netty会创建一个新的Channel
实例(例如NioSocketChannel
或NioServerSocketChannel
),并将其注册到NioEventLoop
中的Selector
上。注册过程包括将Channel
添加到Selector
的监控列表中,并设置其感兴趣的事件类型(如读、写、连接等)。 -
事件处理:
一旦Selector
检测到某个Channel
上有事件就绪(例如数据可读或可写),它会通知NioEventLoop
。然后,NioEventLoop
会调用相应的ChannelPipeline
和ChannelHandler
来处理这些事件。 -
关闭Channel:
当连接关闭时,Netty会注销Channel
并从Selector
的监控列表中移除它。同时,相关的资源也会被释放。
管理Selector
-
创建Selector:
在NioEventLoop
初始化时,它会创建一个Selector
实例。这个Selector
用于监控所有注册到该NioEventLoop
的Channel
。 -
轮询就绪事件:
NioEventLoop
在事件循环中不断调用Selector
的select()
方法来等待Channel
上的事件就绪。一旦有事件就绪,Selector
会返回这些事件的集合。 -
处理就绪事件:
NioEventLoop
遍历Selector
返回的就绪事件集合,并为每个事件调用相应的处理器。这通常涉及到调用ChannelPipeline
中的ChannelHandler
来处理这些事件。 -
优化Selector的使用:
Netty可能会使用多个Selector
和NioEventLoop
实例来优化性能,特别是在处理大量并发连接时。通过分散负载到多个线程和Selector
上,Netty能够充分利用多核CPU的资源,提高吞吐量。
注意事项
- 通常情况下,你不需要直接管理
Selector
。Netty的NioEventLoop
和Channel
抽象为你提供了高级的API来处理I/O事件和连接。 - 在编写自定义的
ChannelHandler
时,你需要关注如何处理事件,而不是如何管理Channel
或Selector
。 - 如果你需要执行定时任务或自定义的后台任务,可以使用
NioEventLoop
的任务队列来提交这些任务。这些任务会在事件循环的适当时候得到执行。