Netty从0到1系列之EventLoopGroup

---

推荐阅读:

【01】Netty从0到1系列之I/O模型
【02】Netty从0到1系列之NIO
【03】Netty从0到1系列之Selector
【04】Netty从0到1系列之Channel
【05】Netty从0到1系列之Buffer(上)
【06】Netty从0到1系列之Buffer(下)
【07】Netty从0到1系列之零拷贝技术
【08】Netty从0到1系列之整体架构、入门程序
【09】Netty从0到1系列之EventLoop

---

一、EventLoopGroup

1.1 核心概念

EventLoopGroup 是 Netty 框架的核心线程调度与管理组件，它本质上是多个 EventLoop 的容器和调度器。Netty 基于 Reactor 线程模型，使用一个或多个 EventLoopGroup 来高效处理海量的网络连接和 I/O 操作，实现了卓越的性能和可伸缩性。

事件循环组: EventLoopGroup 是一组 EventLoop，Channel 一般会调用 EventLoopGroup 的 register 方法来绑定其中一个 EventLoop，后续这个 Channel 上的 io 事件都由此 EventLoop 来处理（保证了 io 事件处理时的线程安全）

🌟 核心职责：

• 管理多个 EventLoop 线程
• 轮询分配 Channel 到 EventLoop
• 统一启动与关闭所有 EventLoop
• 实现主从 Reactor 模型

1.2 工作原理

【注意: EventLoop核心】

1.3 核心原理与架构设计

EventLoopGrouop的继承体系

io.netty.util.concurrent-> EventExecutorGroup-> EventLoopGroup-> MultithreadEventLoopGroup-> NioEventLoopGroup / EpollEventLoopGroup / ...

✅ 关键实现：

• NioEventLoopGroup：基于 JDK NIO
• EpollEventLoopGroup：Linux 专属，基于 epoll 系统调用，性能更高

• EventExecutorGroup：定义了线程池的基本行为，如 shutdownGracefully(), submit()
• EventLoopGroup：扩展了 EventExecutorGroup，增加了注册 Channel 的能力
• MultithreadEventLoopGroup：提供了多线程实现的基础
• NioEventLoopGroup：基于 Java NIO 的具体实现

public interface EventLoopGroup extends EventExecutorGroup {// 从组中获取一个 EventLoop（轮询）@OverrideEventLoop next();// 创建新连接时，选择一个 EventLoop 绑定// 向此EventLoop注册一个 Channel .注册完成后，退回ChannelFuture者将收到通知ChannelFuture register(Channel channel);// Channel使用 ChannelFuture.EventLoop注册完成后，通过ChannelFuture者将收到通知，并将被退回。ChannelFuture register(ChannelPromise promise);/*** 向此EventLoop注册一个 Channel .注册完成后，通过ChannelFuture者将收到通知，并将被退回。* 荒废的* 请改用 register(ChannelPromise) 。** @deprecated Use {@link #register(ChannelPromise)} instead.*/@DeprecatedChannelFuture register(Channel channel, ChannelPromise promise);
}

继承自 netty 自己的 EventExecutorGroup

• 实现了 Iterable 接口提供遍历 EventLoop 的能力
• 另有 next 方法获取集合中下一个 EventLoop

1.4 示例

1.4.1 EventLoopGroup基本示例

@Slf4j
public class EventLoopGroupDemo01 {public static void main(String[] args) {try(EventLoopGroup eventLoopGroup = new NioEventLoopGroup(2);){EventLoop next1 = eventLoopGroup.next();EventLoop next2 = eventLoopGroup.next();EventLoop next3 = eventLoopGroup.next();log.debug("next1 : {}", next1);log.debug("next2 : {}", next2);log.debug("next3 : {}", next3);}catch (Exception _){}}
}

1.4.2 优雅的关闭EventLoopGroup

优雅关闭 shutdownGracefully 方法。该方法会首先切换 EventLoopGroup 到关闭状态从而拒绝新的任务的加入，然后在任务队列的任务都处理完成后，停止线程的运行。从而确保整体应用是在正常有序的状态下退出的.

public static void main(String[] args) {EventLoopGroup eventLoopGroup = new NioEventLoopGroup(3);for (EventExecutor eventExecutor : eventLoopGroup) {System.out.println(eventExecutor);}// 优雅的关闭事件循环组eventLoopGroup.shutdownGracefully();
}

1.4.3 自定义配置EventLoopGroup

/*** @author: laoren* @date: 2025/9/1 14:56* @description: 自定义配置EventLoopGroup* @version: 1.0.0*/
public class CustomEventLoopGroupExample {public static void main(String[] args) {// 自定义线程工厂ThreadFactory bossThreadFactory = new DefaultThreadFactory("tc-boss-thread");ThreadFactory workerThreadFactory = new DefaultThreadFactory("tc-worker-thread");// 创建自定义配置的EventLoopGroupNioEventLoopGroup bossGroup = new NioEventLoopGroup(1, bossThreadFactory);// 使用更多自定义配置项NioEventLoopGroup workGroup = new NioEventLoopGroup(8,  // 线程数workerThreadFactory, // 线程工厂SelectorProvider.provider() // Selector提供者);System.out.println("bossGroup 线程数: " + bossGroup.executorCount());System.out.println("workGroup 线程数: " + workGroup.executorCount());}
}

1.4.4 性能优化配置

package cn.tcmeta.demo02;import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;public class OptimizedServer {private static final boolean USE_EPOLL = true; // 在Linux上使用epollprivate static final int PORT = 8080;public static void main(String[] args) throws InterruptedException {// 根据操作系统选择最佳实现EventLoopGroup bossGroup = USE_EPOLL ?new EpollEventLoopGroup(1) : new NioEventLoopGroup(1);EventLoopGroup workerGroup = USE_EPOLL ?new EpollEventLoopGroup() : new NioEventLoopGroup();try {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup).channel(USE_EPOLL ?EpollServerSocketChannel.class : NioServerSocketChannel.class).childHandler(new SimpleChannelInboundHandler<String>() {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, String msg) {ctx.writeAndFlush("Echo: " + msg + "\n");}})// 性能优化选项.option(ChannelOption.SO_BACKLOG, 1024) // 等待连接队列大小.option(ChannelOption.SO_REUSEADDR, true) // 重用地址.childOption(ChannelOption.TCP_NODELAY, true) // 禁用Nagle算法.childOption(ChannelOption.SO_KEEPALIVE, true) // 保持连接.childOption(ChannelOption.ALLOCATOR,PooledByteBufAllocator.DEFAULT); // 使用池化内存分配器ChannelFuture future = bootstrap.bind(PORT).sync();System.out.println("优化服务器启动在端口: " + PORT);future.channel().closeFuture().sync();} finally {bossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

1.4.5 基本使用

package cn.tcmeta.demo02;import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;/*** 演示 EventLoopGroup 的基本使用*/
public class EventLoopGroupBasicExample {public static void main(String[] args) throws Exception {// 创建包含 3 个线程的 EventLoopGroupNioEventLoopGroup group = new NioEventLoopGroup(3);try {System.out.println("✅ EventLoopGroup created with " + group.executorCount() + " threads");// 演示轮询分配for (int i = 0; i < 6; i++) {// next() 返回一个 EventLoopio.netty.channel.EventLoop eventLoop = group.next();System.out.println("Iteration " + i + " -> " + "EventLoop thread: " + Thread.currentThread().getName());// 提交任务到该 EventLoopeventLoop.execute(() -> {System.out.println("   🔧 Task executed by: " + Thread.currentThread().getName());});}// 模拟运行Thread.sleep(2000);} finally {// 优雅关闭Future<?> future = group.shutdownGracefully();future.sync(); // 等待关闭完成System.out.println("🔚 EventLoopGroup shutdown completed.");}}
}

1.4.6 主从 Reactor 模型（生产环境标准配置）

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;/*** 演示 Boss + Worker EventLoopGroup 的主从模型*/
public class EventLoopGroupReactorModel {public static void main(String[] args) throws Exception {// Boss Group：负责接收连接，通常 1 个线程EventLoopGroup bossGroup = new NioEventLoopGroup(1);// Worker Group：负责 I/O 读写，CPU 核心数 × 2EventLoopGroup workerGroup = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors() * 2);try {ServerBootstrap bootstrap = new ServerBootstrap();bootstrap.group(bossGroup, workerGroup)  // 设置主从 Group.channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) {ChannelPipeline pipeline = ch.pipeline();// 获取当前 Channel 的 EventLoopEventLoop eventLoop = ch.eventLoop();System.out.println("🔗 Channel " + ch.id() + " assigned to: " + eventLoop.thread().getName());// 添加处理器pipeline.addLast(new SimpleChannelInboundHandler<Object>() {@Overrideprotected void channelRead0(ChannelHandlerContext ctx, Object msg) {System.out.println("📩 Received: " + msg + " on thread: " + Thread.currentThread().getName());// 回写响应ctx.writeAndFlush(Unpooled.copiedBuffer("Echo: " + msg.toString() + "\n",java.nio.charset.StandardCharsets.UTF_8));}});}});// 绑定端口ChannelFuture future = bootstrap.bind(8080).sync();System.out.println("🚀 Server started on port 8080");// 等待服务器关闭future.channel().closeFuture().sync();} finally {// 优雅关闭两个 GroupbossGroup.shutdownGracefully();workerGroup.shutdownGracefully();}}
}

telnet localhost 8080
# 输入任意消息，观察日志中不同的 EventLoop 线程处理

1.4.7 NioEventLoop处理I/O事件

package cn.tcmeta.demo02;/*** 服务端端示例代码*/
@Slf4j
public class MyServer {public static void main(String[] args) throws InterruptedException {EventLoopGroup boss = new NioEventLoopGroup(1); // 管理连接EventLoopGroup worker = new NioEventLoopGroup(2); // 处理io读写等事件new ServerBootstrap().group(boss, worker).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new ChannelInboundHandlerAdapter() {// Object msg: ByteBuf类型的数据.@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {ByteBuf byteBuf = msg instanceof ByteBuf ? ((ByteBuf) msg) : null;// 读取数据操作.if (byteBuf != null) {byte[] buf = new byte[16];ByteBuf len = byteBuf.readBytes(buf, 0, byteBuf.readableBytes());log.debug("{} -- {}", Thread.currentThread().getName(), new String(buf));}super.channelRead(ctx, msg);}});}}).bind(8888).sync();log.debug("服务器已经启动了.......");}
}

开启两个客户端进行测试:

• nc ip port, 之后发送消息

当前代码当中有2个worker的EventLoopGroup, 这两个「轮询执行」. 并且每个worker与自己要处理的channel进行绑定操作.

1.4.8 添加一个ChannelInboundHandlerAdapter

package cn.tcmeta.demo02;import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.logging.LoggingHandler;
import lombok.extern.slf4j.Slf4j;/*** @author laoren* 创建时间:  2023/12/13 21:34*/
@Slf4j
public class MyServer02 {public static void main(String[] args) throws InterruptedException {// 工作线程,处理连接EventLoopGroup boss = new NioEventLoopGroup(1);// 业务线程,具体负责:// io事件, 普通的任务, 定时任务.EventLoopGroup worker = new NioEventLoopGroup(2);// 添加一个业务处理EventLoopGroup defaultEventLoopGroup = new DefaultEventLoopGroup(2);// 当然这里还有其它实现, 譬如说: DefaultNioEventLoopGroup, 但是这个只能处理: 普通任务或者定时任务.new ServerBootstrap().group(boss, worker).channel(NioServerSocketChannel.class).childHandler(new ChannelInitializer<SocketChannel>() {@Overrideprotected void initChannel(SocketChannel ch) throws Exception {ch.pipeline().addLast(new LoggingHandler()).addLast(new StringDecoder()) // 处理解码// 添加业务处理器.addLast("handler", new ChannelInboundHandlerAdapter() { // handler, 处理客户端返回来的消息@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.info(Thread.currentThread().getName() + "  读取的内容是: " + msg);ctx.fireChannelRead(msg);// 这个必须得调用一下, 传递给下一个handler进行处理}// 添加第二个handler处理器, 处理消息}).addLast(defaultEventLoopGroup, "handler2", new ChannelInboundHandlerAdapter() {@Overridepublic void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {log.info(Thread.currentThread().getName() + " 读取的消息内容是: " + msg);}});}}).bind(8888).sync();}
}

重点说明

ctx.fireChannelRead(msg);将消息传递给下一个handler进行处理.

static void invokeChannelRead(final AbstractChannelHandlerContext next, Object msg) {final Object m = next.pipeline.touch(ObjectUtil.checkNotNull(msg, "msg"), next);// 下一个handler的事件循环是否与当前的事件循环是【同一个线程】// EventExecutor其实就是一个eventLoop// next.executor(), 返回下一个handler的eventLoopEventExecutor executor = next.executor();// 如果是, 则直接调用// 当前handler中的线程,是否和eventLoop是同一个线程if (executor.inEventLoop()) {next.invokeChannelRead(m);} else {// 如果不是, 将要执行的代码作为任务提交给下一个事件循环处理【handler】// exetutor, 表示: 下一个handler线程.executor.execute(new Runnable() {@Overridepublic void run() {// 封装任务,去执行了.next.invokeChannelRead(m);}});}
}

• 如果两个handler绑定的是同一个线程,那么就直接调用
• 反之,把要调用的代码封装成一个任务对象,由【下一个handler的线程来处理】

过观察可以发现, 消息会经过两个handler处理.

1.5 底层实现原理分析

1.5.1 轮询分配策略【Round-Robin】

• EventLoopGroup 维护一个 EventLoop 数组
• 每次调用 next() 返回下一个 EventLoop（循环）
• 确保 Channel 均匀分布到各个线程

// MultithreadEventLoopGroup.next() 伪代码
public EventLoop next() {return (EventLoop) children[Math.abs((int) (childIndex.getAndIncrement() % children.length))];
}

🔑 关键点：childIndex 是原子递增的，保证线程安全的轮询。

1.5.2 主从 Reactor 模型的实现

Netty 通过 两个 EventLoopGroup 实现主从架构：

• Boss Group：负责 accept 新连接，通常 1 个线程
• Worker Group：负责 I/O 读写，通常 CPU 核心数 × 2 个线程

1.5.3 内部结构：children 数组

• EventLoopGroup 在初始化时创建指定数量的 EventLoop
• 存储在 children 数组中
• next() 方法从数组中轮询获取

public abstract class MultithreadEventLoopGroup extends MultithreadEventExecutorGroupimplements EventLoopGroup {private final EventLoop[] children;protected MultithreadEventLoopGroup(int nThreads, Executor executor, Object... args) {super(nThreads, executor, args);// 创建 nThreads 个 EventLoopchildren = new EventLoop[nThreads];for (int i = 0; i < nThreads; i++) {children[i] = newChild(executor, args);}}@Overridepublic EventLoop next() {return (EventLoop) super.next();}
}

1.5.4 EventLoop线程模型

每个 EventLoop 都绑定一个专用线程，遵循以下工作模式：

// 伪代码：EventLoop 的核心执行逻辑
while (!isTerminated()) {// 1. 检查是否有定时任务需要执行long timeout = checkScheduledTasks();// 2. 轮询I/O事件（Selector操作）if (hasTasks()) {// 有任务时使用非阻塞selectselector.selectNow();} else {// 无任务时使用带超时的阻塞selectselector.select(timeout);}// 3. 处理就绪的I/O事件processSelectedKeys();// 4. 处理所有任务（包括普通任务和定时任务）runAllTasks();
}

1.5.5 任务调度机制

EventLoopGroup 实现了高效的任务调度：

import io.netty.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;public class TaskSchedulingExample {public static void main(String[] args) {EventLoopGroup group = new NioEventLoopGroup(2);// 1. 立即执行任务group.execute(() -> {System.out.println("立即执行的任务");});// 2. 定时任务：延迟执行ScheduledFuture<?> future = group.schedule(() -> {System.out.println("延迟5秒执行的任务");}, 5, TimeUnit.SECONDS);// 3. 固定速率定时任务group.scheduleAtFixedRate(() -> {System.out.println("每秒执行一次的定时任务");}, 1, 1, TimeUnit.SECONDS);// 4. 带固定延迟的定时任务group.scheduleWithFixedDelay(() -> {System.out.println("任务完成后延迟2秒再执行下一次");}, 1, 2, TimeUnit.SECONDS);// 优雅关闭group.shutdownGracefully();}
}

任务执行优先级：

• I/O 任务优先于普通任务
• 普通任务按提交顺序执行
• 定时任务在到达指定时间后加入普通任务队列

1.6 实践经验与最佳实践

1.6.1 线程数配置策略

public class ThreadConfigurationBestPractice {public static void main(String[] args) {int cpuCores = Runtime.getRuntime().availableProcessors();// 最佳实践配置int bossThreads = 1; // 通常1个足够，除非需要绑定多个端口int workerThreads = cpuCores * 2; // 通用经验公式// 根据应用类型调整String appType = "ioIntensive"; // 或 "cpuIntensive"if ("cpuIntensive".equals(appType)) {workerThreads = cpuCores; // CPU密集型任务，线程数不宜过多} else if ("ioIntensive".equals(appType)) {workerThreads = cpuCores * 2; // IO密集型任务，可以多一些线程}EventLoopGroup bossGroup = new NioEventLoopGroup(bossThreads);EventLoopGroup workerGroup = new NioEventLoopGroup(workerThreads);System.out.println("CPU核心数: " + cpuCores);System.out.println("Boss线程数: " + bossThreads);System.out.println("Worker线程数: " + workerThreads);}
}

1.6.2 资源管理与优雅关闭

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import java.util.concurrent.TimeUnit;public class GracefulShutdownExample {public static void main(String[] args) {EventLoopGroup group = new NioEventLoopGroup();// 添加JVM关闭钩子Runtime.getRuntime().addShutdownHook(new Thread(() -> {System.out.println("JVM关闭中，优雅关闭EventLoopGroup...");// 优雅关闭：先停止接受新任务，然后等待已有任务完成group.shutdownGracefully();try {// 等待最多5秒完成关闭if (!group.awaitTermination(5, TimeUnit.SECONDS)) {System.err.println("EventLoopGroup未能在5秒内完全关闭");}} catch (InterruptedException e) {Thread.currentThread().interrupt();}}));// 模拟工作负载for (int i = 0; i < 10; i++) {group.execute(() -> {try {Thread.sleep(1000);System.out.println("任务执行完成: " + Thread.currentThread().getName());} catch (InterruptedException e) {Thread.currentThread().interrupt();}});}// 主线程等待一段时间后触发关闭try {Thread.sleep(3000);System.exit(0);} catch (InterruptedException e) {Thread.currentThread().interrupt();}}
}

1.7 优缺点总结

✅ 优点为

1. 高性能：基于Reactor模式和高效的线程调度，支持海量连接
2. 低资源消耗：少量线程处理大量连接，减少上下文切换
3. 灵活性：支持多种传输协议（NIO、Epoll、OIO等）
4. 易用性：简化了并发编程的复杂性
5. 健壮性：经过大规模生产环境验证

优点	说明
高并发支持	多线程处理 I/O，支持百万级连接
资源可控	线程数可配置，避免资源耗尽
主从分离	职责分明，性能优化
跨平台	支持 NIO、epoll、kqueue
易于扩展	可自定义 EventLoop 实现

❌ 缺点

1. 学习曲线：需要理解异步编程和Netty特有的概念
2. 调试复杂性：异步模式使得调试和问题排查更加困难
3. 内存管理：需要手动管理ByteBuf的生命周期，容易导致内存泄漏

缺点	说明
配置复杂	需理解主从模型、线程数设置
调试困难	多线程异步，日志分散
内存开销	每个 EventLoop 维护 Selector、任务队列
阻塞风险	任一线程阻塞影响其负责的所有 Channel

1.8 EventLoopGroup核心价值

维度	说明
核心思想	线程资源池化 + 轮询分配 + 主从分离
关键技术	Reactor 模式、线程池、事件循环
性能优势	高吞吐、低延迟、高并发
设计精髓	“让专业的人做专业的事” —— Boss 接收连接，Worker 处理 I/O
适用场景	所有 Netty 网络服务（RPC、网关、消息系统）

1.9 细节问题

1.9.1 ✅ 线程命名与调试友好

• Netty 为每个线程生成清晰的名称：
  • nioEventLoopGroup-2-1
  • nioEventLoopGroup-2-2
• 格式：[type]EventLoopGroup-[groupId]-[threadId]

1.9.2 ✅ 资源自动释放

• shutdownGracefully()

  方法：

  • 停止接收新任务
  • 处理完队列中的任务
  • 关闭所有 EventLoop
  • 释放 Selector、Channel 等资源

💡 建议：Linux 环境使用 EpollEventLoopGroup 可获得更高性能。

1.9.3 ✅ 支持多传输类

EventLoopGroup 实现	传输类型	平台	性能
NioEventLoopGroup	NIO	跨平台	高
EpollEventLoopGroup	epoll	Linux	更高
KQueueEventLoopGroup	kqueue	macOS/BSD	高

1.10 一句话总结

EventLoopGroup 是 Netty 的“调度中枢” —— 它通过 主从架构 和 轮询分配，实现了 连接接收 与 I/O 处理 的分离，是构建高性能网络服务的标准范式。

1.11 EventLoopGroup优缺点

✅ 优点

优点	说明
高并发支持	多线程处理 I/O，支持百万级连接
资源可控	线程数可配置，避免资源耗尽
主从分离	职责分明，性能优化
跨平台	支持 NIO、epoll、kqueue
易于扩展	可自定义 EventLoop 实现

❌ 缺点

缺点	说明
配置复杂	需理解主从模型、线程数设置
调试困难	多线程异步，日志分散
内存开销	每个 EventLoop 维护 Selector、任务队列
阻塞风险	任一线程阻塞影响其负责的所有 Channel