文章目录
- EventLoop概述
- EventLoop 的概念
- EventLoop 的作用
- EventLoop 的生命周期
- EventLoopGroup
- EventLoop 的工作原理
- 总结
- 代码示例
- 💡 优雅关闭
- 演示 NioEventLoop 处理 io 事件
- 解决work中的channel读操作耗费时间过长,影响其他channel(客户端)的问题
- 💡 handler 执行中如何换人?
EventLoop概述
EventLoop 的概念
在 Netty 中,EventLoop
是一个非常核心的概念,它是 Netty 异步事件驱动模型的基础。EventLoop 负责处理 I/O 事件,包括读、写、连接和断开连接等操作。理解 EventLoop 的工作机制对于高效地使用 Netty 至关重要。
EventLoop 是 Netty 中的一个抽象类,它封装了一个线程和一组注册到该线程的 Channel。
EventLoop 的主要职责是处理与 Channel 相关的 I/O 事件,并且负责执行与 Channel 相关联的任务。
EventLoop 的作用
- 事件处理:EventLoop 负责
监听 Channel 上的 I/O 事件
,并在这些事件发生时调用相应的处理器。 - 任务执行:EventLoop 还负责执行与 Channel 关联的任务,这些任务可以是用户提交的业务逻辑或者 Netty 内部的一些操作。
EventLoop 的生命周期
EventLoop 的生命周期通常与它所绑定的 Channel 的生命周期相关。当一个 Channel 被创建时,它会被绑定到一个 EventLoop,并且这个 EventLoop 会负责处理该 Channel 的所有 I/O 事件。当 Channel 关闭时,EventLoop 仍然会存在,但它不再处理与该 Channel 相关的事件。
EventLoopGroup
EventLoopGroup
是 EventLoop 的容器,它管理一组 EventLoop 实例。EventLoopGroup 为 Channel 提供了一个或多个 EventLoop 供其使用。Netty 中常见的 EventLoopGroup 类型包括 NioEventLoopGroup 和 EpollEventLoopGroup。
NioEventLoopGroup:针对 NIO 模型的 EventLoopGroup,适用于大多数操作系统。
EpollEventLoopGroup:针对 Linux 操作系统的 EventLoopGroup,使用 epoll 代替 select,性能更高。
EventLoop 的工作原理
在一个典型的 Netty 服务器中,EventLoop 的工作原理如下:
- 初始化:当创建一个 Netty 服务器时,会创建一个或多个 EventLoopGroup。这些 EventLoopGroup 通常包括两个部分:一个 Boss Group 用于接收新的连接,一个 Worker Group 用于处理 I/O 操作。
Boss Group:接收客户端的连接请求,并为每个连接分配一个 Worker Group 中的 EventLoop。
Worker Group:处理 I/O 操作,包括读、写、连接和断开连接等。
- 事件循环:每个 EventLoop 都有一个事件循环,它不断地轮询注册在其上的 Channel,查找感兴趣的 I/O 事件。
- Select:EventLoop 会调用 select 方法来轮询 Channel,查找感兴趣的事件。
- 事件处理:当事件发生时,EventLoop 会调用相应的处理器来处理这些事件。
- 任务执行:除了处理 I/O 事件之外,EventLoop 还可以执行与 Channel 相关的任务,如用户提交的业务逻辑任务。
- 任务队列:每个 EventLoop 都有一个任务队列,用于存放待执行的任务。
- 事件传播:如果一个 Channel 需要从一个 EventLoop 转移到另一个 EventLoop,Netty 会确保事件传播的一致性。
总结
EventLoop (事件循环对象)本质是一个单线程执行器(同时维护了一个 Selector),里面有 run 方法处理 Channel 上源源不断的 io 事件。
它的继承关系比较复杂
- 一条线是继承自 j.u.c.ScheduledExecutorService 因此包含了线程池中所有的方法
- 另一条线是继承自 netty 自己的 OrderedEventExecutor,
- 提供了 boolean inEventLoop(Thread thread) 方法判断一个线程是否属于此 EventLoop
- 提供了 parent 方法来看看自己属于哪个 EventLoopGroup
EventLoopGroup (事件循环组)是一组 EventLoop,Channel 一般会调用 EventLoopGroup 的 register 方法来绑定其中一个 EventLoop,后续这个 Channel 上的 io 事件都由此 EventLoop 来处理(保证了 io 事件处理时的线程安全)
- 继承自 netty 自己的 EventExecutorGroup
- 实现了 Iterable 接口提供遍历 EventLoop 的能力
- 另有 next 方法获取集合中下一个 EventLoop
代码示例
@Slf4j
public class Test01EventLoop {
public static void main(String[] args) {
// 1. 创建事件循环组
EventLoopGroup group = new NioEventLoopGroup(2);
//NioEventLoopGroup是功能最全面的,既能处理io事件,也能向其提交普通任务,定时任务
// EventLoopGroup group = new DefaultEventLoopGroup();
// DefaultEventLoopGroup不能处理io事件,只能处理普通任务,定时任务
//创建NioEventLoopGroup时,如果不指定线程数,底层代码使用NettyRuntime.availableProcessors()创建线程数
// System.out.println(NettyRuntime.availableProcessors());//16
// 2. 获取下一个事件循环对象
//当创建的线程数为2时,使用group.next()依次获取对象(类似轮询的效果)
System.out.println(group.next());
System.out.println(group.next());
System.out.println(group.next());
System.out.println(group.next());
// 3. 执行普通任务,作用:可以用来执行耗时较长的任务
// group.next().submit(() ->{
// try {
// Thread.sleep(1000);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// log.debug("ok");
// });
// group.next().execute(() -> {
// try {
// Thread.sleep(1000);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// log.debug("ok");
// });
//submit()和execute()效果相同
// 4. 执行定时任务
group.next().scheduleAtFixedRate(() -> {
log.debug("ok");
}, 0, 1, TimeUnit.SECONDS);
log.debug("main");
}
}
💡 优雅关闭
优雅关闭 shutdownGracefully 方法。该方法会首先切换 EventLoopGroup 到关闭状态从而拒绝新的任务的加入,然后在任务队列的任务都处理完成后,停止线程的运行。从而确保整体应用是在正常有序的状态下退出的。
演示 NioEventLoop 处理 io 事件
服务器端两个 nio worker 工人
import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import java.nio.charset.Charset;
@Slf4j
public class EventLoopServer {
public static void main(String[] args) {
new ServerBootstrap()
// boss 和 worker
// 细分1:
// boss 只负责 ServerSocketChannel 上 accept 事件
// worker 只负责 socketChannel 上的读写
//group参数1:boss不需要指定线程数,因为ServerSocketChannel只会跟一个EventLoop进行绑定,
// 又因为服务器只有一个,所以只会占用一个线程,不用指定线程数。
//group参数1:work线程指定为两个
.group(new NioEventLoopGroup(), new NioEventLoopGroup(2))
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
/**
* @param ctx
* @param msg ByteBuf类型
* @throws Exception
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
log.debug(buf.toString(Charset.defaultCharset()));
}
});
}
})
.bind(8080);
}
}
细分1:
- boss 只负责 ServerSocketChannel 上 accept 事件
- worker 只负责 socketChannel 上的读写
客户端
@Slf4j
public class EventLoopClient {
public static void main(String[] args) throws InterruptedException {
// 2. 带有 Future,Promise 的类型都是和异步方法配套使用,用来处理结果
ChannelFuture channelFuture = new Bootstrap()
.group(new NioEventLoopGroup())
.channel(NioSocketChannel.class)
.handler(new ChannelInitializer<NioSocketChannel>() {
@Override // 在连接建立后被调用
protected void initChannel(NioSocketChannel ch) throws Exception {
ch.pipeline().addLast(new StringEncoder());
}
})
// 1. 连接到服务器
// 异步非阻塞, main 发起了调用,真正执行 connect 是 nio 线程
.connect(new InetSocketAddress("localhost", 8080)); // 1s 秒后
// 2.1 使用 sync 方法同步处理结果
channelFuture.sync(); // 阻塞住当前线程,直到nio线程连接建立完毕
Channel channel = channelFuture.channel();
log.debug("{}", channel);
channel.writeAndFlush("hello, world");
channel.writeAndFlush("hello, world");
channel.writeAndFlush("hello, world");
System.out.println();
}
}
从服务器端代码可以看到两个工人轮流处理 channel,但工人与 channel 之间进行了绑定
建立连接后,服务器会始终拿同一个EventLoop(线程)处理客户端所有的请求
解决work中的channel读操作耗费时间过长,影响其他channel(客户端)的问题
创建一个独立的 EventLoopGroup,将耗时的代码放到一个额外的组中线程处理
@Slf4j
public class EventLoopServer {
public static void main(String[] args) {
// 细分2:创建一个独立的 EventLoopGroup,将耗时的代码放到一个额外的组中线程处理
EventLoopGroup group = new DefaultEventLoopGroup();
new ServerBootstrap()
// boss 和 worker
// 细分1:
// boss 只负责 ServerSocketChannel 上 accept 事件
// worker 只负责 socketChannel 上的读写
//group参数1:boss不需要指定线程数,因为ServerSocketChannel只会跟一个EventLoop进行绑定,
// 又因为服务器只有一个,所以只会占用一个线程,不用指定线程数。
//group参数1:work线程指定为两个
.group(new NioEventLoopGroup(), new NioEventLoopGroup(2))
.channel(NioServerSocketChannel.class)
.childHandler(new ChannelInitializer<NioSocketChannel>() {
@Override
protected void initChannel(NioSocketChannel ch) throws Exception {
// ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {
// /**
// * @param ctx
// * @param msg ByteBuf类型
// * @throws Exception
// */
// @Override
// public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
// ByteBuf buf = (ByteBuf) msg;
// log.debug(buf.toString(Charset.defaultCharset()));
// }
// });
//如果读操作耗费的时间很上,会影响其他客户端的读写操作,
// 一个work管理多个channel,如果其中一个耗时过长则会影响其他channel的读写操作
ch.pipeline().addLast("handler1", new ChannelInboundHandlerAdapter() {
/**
* @param ctx
* @param msg ByteBuf类型
* @throws Exception
*/
@Override
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
log.debug(buf.toString(Charset.defaultCharset()));
ctx.fireChannelRead(msg); // 让消息传递给下一个handler,如果不添加则消息不会传递给handler2中
}
}).addLast(group, "handler2", new ChannelInboundHandlerAdapter() {
@Override // ByteBuf
public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
ByteBuf buf = (ByteBuf) msg;
log.debug(buf.toString(Charset.defaultCharset()));
}
});
}
})
.bind(8080);
}
}
细分2:创建一个独立的 EventLoopGroup,将耗时的代码放到一个额外的组中线程处理
以下两个客户端依次发送消息
可以看到,nio 工人和 非 nio 工人也分别绑定了 channel
💡 handler 执行中如何换人?
关键代码 io.netty.channel.AbstractChannelHandlerContext#invokeChannelRead()
static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {
final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);
// 下一个 handler 的事件循环是否与当前的事件循环是同一个线程
EventExecutor executor = next.executor();//返回下一个handler的eventLoop(eventLoop继承了EventExecutor)
//executor.inEventLoop():当前handler中的线程是否和executor(eventLoop)是同一个线程
// 是,直接调用
if (executor.inEventLoop()) {
next.invokeChannelRead(m);
}
// 不是,将要执行的代码作为任务提交给下一个事件循环处理(换人)
else {
//下一个handler线程创建一个线程执行next.invokeChannelRead(m);
executor.execute(new Runnable() {
@Override
public void run() {
next.invokeChannelRead(m);
}
});
//目的是切换线程
}
}
全部文章:
Netty笔记01-Netty的基本概念与用法
Netty笔记02-组件EventLoop
Netty笔记03-组件Channel
Netty笔记04-组件Future & Promise
Netty笔记05-组件Handler & Pipeline