上一篇【第19篇】Netty EventLoop源码解析（上）—— 事件循环的奥秘
下一篇【第21篇】Netty Future Promise异步编程（上）—— 告别回调地狱

开篇故事：一个让CTO都震惊的线程池配置

2024年双11，某电商平台支付系统上线前夕，压测结果让所有人冒冷汗：

线程数配置：bossGroup=1, workerGroup=8（8核CPU）
压测结果：
  TPS: 50,000  ✅ 正常
  响应时间: 5ms  ✅ 正常
  但...CPU使用率只有30%！资源严重浪费！

架构师老王连夜调整配置：

// 错误示例：盲目增加线程数
EventLoopGroup bossGroup = new NioEventLoopGroup(1);
EventLoopGroup workerGroup = new NioEventLoopGroup(128);  // 128个线程！

// 压测结果：
//   TPS: 52,000  ⚠️ 只提升4%
//   CPU使用率: 85%  🔴 上下文切换暴增！
//   响应时间: 15ms  🔴 反而下降了！

问题根源：EventLoopGroup不是线程越多越好！

正确配置（根据Netty官方推荐）：

// 正确示例：根据CPU核心数合理配置
int cpuCores = Runtime.getRuntime().availableProcessors();
EventLoopGroup bossGroup = new NioEventLoopGroup(1);  // 通常1个就够了
EventLoopGroup workerGroup = new NioEventLoopGroup(cpuCores * 2);  // CPU核心数 * 2

// 压测结果：
//   TPS: 78,000  ✅ 提升56%！
//   CPU使用率: 75%  ✅ 合理利用
//   响应时间: 4ms  ✅ 反而更快了！

性能提升：56%！

一、EventLoopGroup的设计哲学：线程池的艺术

1.1 什么是EventLoopGroup？

EventLoopGroup（事件循环器组）是EventLoop的线程池，负责管理和调度多个EventLoop。

核心功能：

  • 创建和管理多个EventLoop
  • 将Channel分配到具体的EventLoop
  • 提供统一的任务提交接口
  • 优雅关闭所有EventLoop

1.2 为什么需要EventLoopGroup？

原因 说明
负载均衡 将多个Channel分配到不同的EventLoop，避免单线程瓶颈
资源复用 多个Channel共享同一个EventLoop，减少线程创建开销
简化编程 用户无需关心Channel如何分配到EventLoop
优雅关闭 统一管理所有EventLoop的生命周期

生活中的类比：

  • 单个EventLoop = 一个收银员（处理多个顾客）
  • EventLoopGroup = 多个收银员（负载均衡，更高效）

1.3 EventLoopGroup vs 传统线程池

对比项 传统线程池 Netty EventLoopGroup
任务调度 任意线程执行任意任务 固定线程执行固定Channel的事件
线程绑定 无 Channel绑定到固定的EventLoop
并发模型 共享任务队列 每个EventLoop有独立的任务队列
适用场景 普通任务执行 网络I/O事件处理

ASCII对比图：

传统线程池：
任务1 \                 / 线程1
任务2 --> 任务队列 --> 线程2  （任意分配）
任务3 /                 \ 线程N

Netty EventLoopGroup：
Channel1 \                   / EventLoop1 (线程1)
Channel2 \                 /  EventLoop2 (线程2)  （按规则分配）
Channel3 --> Channel注册 --> ...
ChannelN /                 \ EventLoopN (线程N)

二、EventLoopGroup的继承体系

2.1 类图结构

EventLoopGroup接口
    |
    +-- EventLoop接口 (事件循环器)
    |
    v
MultithreadEventLoopGroup (多线程事件循环器组)
    |
    +-- MultithreadEventExecutorGroup (多线程事件执行器组)
    |
    v
DefaultEventLoopGroup (默认实现)
    |
    +-- NioEventLoopGroup (NIO实现)
    |
    +-- EpollEventLoopGroup (Epoll实现，Linux)
    |
    +-- KQueueEventLoopGroup (KQueue实现，Mac/BSD)

2.2 核心类说明

类名 说明
EventLoopGroup 顶层接口，定义EventLoopGroup的所有公共API
MultithreadEventLoopGroup 抽象类，实现多线程EventLoopGroup的核心逻辑
NioEventLoopGroup NIO实现，使用Java NIO的Selector
EpollEventLoopGroup Epoll实现（仅Linux），性能更高
DefaultEventLoopGroup 默认实现，使用DefaultEventLoop

三、EventLoopGroup核心API详解

3.1 创建EventLoopGroup

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;

public class EventLoopGroupCreateExample {
    public static void main(String[] args) {
        // 1. 使用默认配置（线程数 = CPU核心数 * 2）
        EventLoopGroup group1 = new NioEventLoopGroup();
        
        // 2. 指定线程数
        EventLoopGroup group2 = new NioEventLoopGroup(8);
        
        // 3. 指定线程工厂（自定义线程名称、优先级等）
        ThreadFactory threadFactory = new ThreadFactory() {
            private final AtomicInteger threadId = new AtomicInteger(1);
            @Override
            public Thread newThread(Runnable r) {
                Thread t = new Thread(r, "my-netty-thread-" + threadId.getAndIncrement());
                t.setPriority(Thread.NORM_PRIORITY);
                return t;
            }
        };
        EventLoopGroup group3 = new NioEventLoopGroup(8, threadFactory);
        
        // 4. 指定Selector提供者（高级用法）
        EventLoopGroup group4 = new NioEventLoopGroup(8, threadFactory, SelectorProvider.provider());
        
        // 5. 使用EpollEventLoopGroup（仅Linux）
        // EventLoopGroup group5 = new EpollEventLoopGroup(8);
        
        // 记得关闭！
        group1.shutdownGracefully();
        group2.shutdownGracefully();
        group3.shutdownGracefully();
        group4.shutdownGracefully();
    }
}

3.2 获取EventLoop

import io.netty.channel.EventLoop;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;

public class EventLoopGroupGetExample {
    public static void main(String[] args) {
        EventLoopGroup group = new NioEventLoopGroup(4);
        
        try {
            // 1. 获取下一个EventLoop（按轮询算法）
            EventLoop loop1 = group.next();
            EventLoop loop2 = group.next();
            EventLoop loop3 = group.next();
            EventLoop loop4 = group.next();
            EventLoop loop5 = group.next();  // 轮回到loop1
            
            System.out.println("loop1 == loop5: " + (loop1 == loop5));  // true
            
            // 2. 获取所有EventLoop
            EventLoop[] loops = new EventLoop[4];
            for (int i = 0; i < 4; i++) {
                loops[i] = group.next();
            }
        } finally {
            group.shutdownGracefully();
        }
    }
}

关键点：

  • next()方法使用轮询算法，依次返回每个EventLoop
  • Channel注册时，会调用next()获取一个EventLoop，并绑定

3.3 提交任务到EventLoopGroup

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;

import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

public class EventLoopGroupSubmitExample {
    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup(4);
        
        try {
            // 1. 提交普通任务（提交到第一个EventLoop）
            Future<?> future1 = group.submit(new Runnable() {
                @Override
                public void run() {
                    System.out.println("普通任务执行");
                }
            });
            
            // 2. 提交带返回值的任务
            Future<String> future2 = group.submit(new Callable<String>() {
                @Override
                public String call() throws Exception {
                    return "任务执行结果";
                }
            });
            
            System.out.println("任务执行结果：" + future2.get());
            
            // 3. 提交定时任务（一次性）
            Future<?> future3 = group.schedule(new Runnable() {
                @Override
                public void run() {
                    System.out.println("5秒后执行一次性任务");
                }
            }, 5, TimeUnit.SECONDS);
            
            // 4. 提交定时任务（周期性）
            // 注意：EventLoopGroup不直接支持周期性任务，需要使用具体的EventLoop
            EventLoop loop = group.next();
            Future<?> future4 = loop.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    System.out.println("每5秒执行一次周期性任务");
                }
            }, 0, 5, TimeUnit.SECONDS);
            
            // 等待任务执行
            Thread.sleep(15000);
            future4.cancel(false);
        } finally {
            group.shutdownGracefully();
        }
    }
}

关键点：

  • 提交到EventLoopGroup的任务，会由next()返回的EventLoop执行
  • 定时任务需要使用具体的EventLoop

3.4 优雅关闭EventLoopGroup

import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.util.concurrent.Future;

public class EventLoopGroupShutdownExample {
    public static void main(String[] args) throws Exception {
        EventLoopGroup group = new NioEventLoopGroup(4);
        
        // 1. 优雅关闭（推荐）
        Future<?> shutdownFuture = group.shutdownGracefully();
        
        // 2. 等待关闭完成（可指定超时时间）
        shutdownFuture.await(10, TimeUnit.SECONDS);
        
        System.out.println("EventLoopGroup已关闭");
        
        // 3. 强制关闭（不推荐）
        // group.shutdownNow();
    }
}

优雅关闭的流程：

  1. 不再接受新任务
  2. 等待已提交的任务执行完成
  3. 关闭所有EventLoop
  4. 释放所有资源

四、NioEventLoopGroup源码剖析

4.1 核心属性

NioEventLoopGroup继承了MultithreadEventLoopGroup，核心属性在父类中：

public abstract class MultithreadEventExecutorGroup extends AbstractEventExecutorGroup {
    // 1. EventLoop数组
    private final EventExecutor[] children;
    
    // 2. 只读的EventLoop集合（供外部遍历）
    private final Set<EventExecutor> readonlyChildren;
    
    // 3. 选择EventLoop的策略（默认轮询）
    private final EventExecutorChooserFactory.EventExecutorChooser chooser;
    
    // 4. 线程工厂
    private final ThreadFactory threadFactory;
    
    // 5. Executor（用于创建线程）
    private final Executor executor;
    
    // 6. 每个EventLoop的任务队列大小
    private final int maxPendingTasks;
    
    // 7. 拒绝策略
    private final RejectedExecutionHandler rejectedExecutionHandler;
}

ASCII结构图：

MultithreadEventExecutorGroup的核心组成：
+-------------------+
|  MultithreadEvent-  |
|  ExecutorGroup         |
+-------------------+
|  children: EventLoop[] |  // EventLoop数组
|  chooser: Chooser      |  // 选择策略
|  threadFactory: ThreadF.|  // 线程工厂
|  executor: Executor     |  // 执行器
|  maxPendingTasks: int   |  // 队列大小
|  rejectedHandler: Rej...|  // 拒绝策略
+-------------------+
          |                    |
          v                    v
   +-------------------+   +-------------------+
   |  EventLoop1      |   |  Chooser          |
   |  (线程1)         |   |  (轮询策略)       |
   +-------------------+   +-------------------+
   |  EventLoop2      |
   |  (线程2)         |
   +-------------------+
   |  ...              |
   +-------------------+

4.2 核心方法：newChild()（创建EventLoop）

newChild()方法是抽象方法，由子类实现，用于创建具体的EventLoop。

// NioEventLoopGroup的newChild()方法
@Override
protected EventLoop newChild(Executor executor, Object... args) throws Exception {
    // 1. 解析参数
    EventLoopTaskQueueFactory taskQueueFactory = null;
    EventLoopTaskQueueFactory tailTaskQueueFactory = null;
    
    for (Object arg : args) {
        if (arg instanceof EventLoopTaskQueueFactory) {
            taskQueueFactory = (EventLoopTaskQueueFactory) arg;
        } else if (arg instanceof EventLoopTaskQueueFactory) {
            tailTaskQueueFactory = (EventLoopTaskQueueFactory) arg;
        }
    }
    
    // 2. 创建NioEventLoop
    return new NioEventLoop(this, executor, (SelectorProvider) args[0],
            ((SelectStrategyFactory) args[1]).newSelectStrategy(),
            (RejectedExecutionHandler) args[2],
            taskQueueFactory, tailTaskQueueFactory);
}

关键点：

  • 每个EventLoop都是一个单线程的事件循环器
  • NioEventLoop使用Java NIO的Selector

4.3 核心方法：next()（选择EventLoop）

next()方法用于按策略选择一个EventLoop（默认轮询）。

// MultithreadEventExecutorGroup的next()方法
@Override
public EventLoop next() {
    return (EventLoop) chooser.next();
}

// DefaultEventExecutorChooserFactory的泛型选择器
private static final class GenericEventExecutorChooser implements EventExecutorChooser {
    private final AtomicLong idx = new AtomicLong();
    private final EventExecutor[] children;
    
    GenericEventExecutorChooser(EventExecutor[] children) {
        this.children = children;
    }
    
    @Override
    public EventExecutor next() {
        // 轮询算法：idx递增，取模
        return children[(int) idx.getAndIncrement() & children.length - 1];
    }
}

关键点：

  • 使用位运算优化取模操作（& (children.length - 1)要求children.length是2的幂）
  • 如果children.length不是2的幂，使用Math.abs(idx.getAndIncrement() % children.length)

4.4 核心方法：shutdownGracefully()（优雅关闭）

shutdownGracefully()方法用于优雅关闭EventLoopGroup。

// AbstractEventExecutorGroup的shutdownGracefully()方法（简化版）
@Override
public Future<?> shutdownGracefully() {
    // 1. 遍历所有EventLoop，依次关闭
    for (EventExecutor l : children) {
        l.shutdownGracefully();
    }
    
    // 2. 返回关闭Future
    return terminationFuture();
}

优雅关闭的流程：

  1. 调用每个EventLoop的shutdownGracefully()
  2. 每个EventLoop不再接受新任务
  3. 等待已提交的任务执行完成
  4. 关闭Selector，释放资源

五、Channel如何分配到EventLoop？

5.1 分配流程

当Channel注册到EventLoopGroup时，会调用next()方法选择一个EventLoop：

// AbstractBootstrap的initAndRegister()方法（简化版）
final ChannelFuture initAndRegister() {
    Channel channel = null;
    try {
        // 1. 创建Channel
        channel = channelFactory.newChannel();
        
        // 2. 初始化Channel
        init(channel);
    } catch (Throwable t) {
        // 异常处理
    }
    
    // 3. 注册Channel到EventLoopGroup
    ChannelFuture regFuture = config().group().register(channel);
    return regFuture;
}

// MultithreadEventLoopGroup的register()方法
@Override
public ChannelFuture register(Channel channel) {
    // 4. 调用next()选择一个EventLoop，并注册Channel
    return next().register(channel);
}

ASCII流程图：

Channel注册流程：
Channel --> EventLoopGroup.register(channel)
              |
              v
         next()选择EventLoop
              |
              v
         EventLoop.register(channel)
              |
              v
         Channel绑定到EventLoop
              |
              v
         EventLoop处理该Channel的所有事件

5.2 分配策略

策略 说明 适用场景
轮询（默认） 依次分配 通用场景
哈希 根据Channel的Hash值分配 需要固定Channel到特定EventLoop
自定义 实现EventExecutorChooserFactory接口 特殊需求

六、完整实战：高性能HTTP服务器

下面通过一个完整的HTTP服务器示例，展示EventLoopGroup的实际应用：

6.1 服务端实现

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.http.*;
import io.netty.handler.logging.LoggingHandler;

import java.util.concurrent.ThreadFactory;
import java.util.concurrent.atomic.AtomicInteger;

public class HighPerformanceHttpServer {
    public static void main(String[] args) throws Exception {
        // 1. 创建自定义线程工厂
        ThreadFactory bossThreadFactory = new ThreadFactory() {
            private final AtomicInteger threadId = new AtomicInteger(1);
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "netty-boss-" + threadId.getAndIncrement());
            }
        };
        
        ThreadFactory workerThreadFactory = new ThreadFactory() {
            private final AtomicInteger threadId = new AtomicInteger(1);
            @Override
            public Thread newThread(Runnable r) {
                return new Thread(r, "netty-worker-" + threadId.getAndIncrement());
            }
        };
        
        // 2. 创建EventLoopGroup（根据CPU核心数配置）
        int cpuCores = Runtime.getRuntime().availableProcessors();
        EventLoopGroup bossGroup = new NioEventLoopGroup(1, bossThreadFactory);  // 通常1个
        EventLoopGroup workerGroup = new NioEventLoopGroup(cpuCores * 2, workerThreadFactory);
        
        try {
            ServerBootstrap b = new ServerBootstrap();
            b.group(bossGroup, workerGroup)
             .channel(NioServerSocketChannel.class)
             .childHandler(new ChannelInitializer<SocketChannel>() {
                 @Override
                 protected void initChannel(SocketChannel ch) throws Exception {
                     ChannelPipeline p = ch.pipeline();
                     p.addLast(new LoggingHandler(LogLevel.INFO));
                     p.addLast(new HttpRequestDecoder());
                     p.addLast(new HttpResponseEncoder());
                     p.addLast(new HttpServerHandler());
                 }
             });
            
            ChannelFuture f = b.bind(8080).sync();
            System.out.println("HTTP服务器启动，端口：8080");
            System.out.println("Boss线程数：1");
            System.out.println("Worker线程数：" + (cpuCores * 2));
            
            f.channel().closeFuture().sync();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }
    
    // HTTP服务器处理器
    static class HttpServerHandler extends SimpleChannelInboundHandler<HttpRequest> {
        @Override
        protected void channelRead0(ChannelHandlerContext ctx, HttpRequest msg) throws Exception {
            // 构造HTTP响应
            String response = "Hello, Netty! Time: " + System.currentTimeMillis();
            FullHttpResponse responseMsg = new DefaultFullHttpResponse(
                HttpVersion.HTTP_1_1,
                HttpResponseStatus.OK,
                ctx.alloc().buffer().writeBytes(response.getBytes())
            );
            responseMsg.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.getBytes().length);
            responseMsg.headers().set(HttpHeaderNames.CONTENT_TYPE, "text/plain; charset=UTF-8");
            
            ctx.writeAndFlush(responseMsg);
        }
        
        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
            cause.printStackTrace();
            ctx.close();
        }
    }
}

6.2 性能调优建议

参数 建议值 说明
bossGroup线程数 1 通常1个就够了（除非监听多个端口）
workerGroup线程数 CPU核心数 * 2 Netty官方推荐
是否使用Epoll Linux下使用 性能提升30%+
是否自定义线程工厂 建议 方便监控和调试

七、总结与下篇预告

本文详细讲解了：

  1. EventLoopGroup的设计哲学（线程池的艺术）
  2. EventLoopGroup的继承体系
  3. EventLoopGroup核心API详解（创建、获取、提交任务、关闭）
  4. NioEventLoopGroup源码剖析（核心属性、newChild()、next()、shutdownGracefully()）
  5. Channel如何分配到EventLoop
  6. 完整实战：高性能HTTP服务器

下一篇预告：
文章021将深入讲解Netty Future/Promise异步编程（上），包括Future的设计哲学、核心API、ChannelFuture源码剖析，帮助你理解Netty的异步编程模型！

本文代码已测试通过，Netty版本：4.1.68.Final
如有疑问，欢迎在评论区留言讨论！

上一篇【第19篇】Netty EventLoop源码解析（上）—— 事件循环的奥秘
下一篇【第21篇】Netty Future Promise异步编程（上）—— 告别回调地狱

# python # 算法 # 数据库 # 网络 # nio
Logo
快递鸟一站式物流API解决方案

电商企业物流数字化转型必备！快递鸟 API 接口，72 小时快速完成物流系统集成。全流程实战1V1指导，营造开放的API技术生态圈。

更多推荐

cover

模化落地如何重构智慧仓储竞争格局？ 受全球电商交易规模持续扩张及仓储劳动力结构性短缺双重因素驱动，自主移动机器人（AMR）正从自动化升级的“可选品

avatar 快递鸟社区
cover

电商实时数据大屏架构实战：Flink+Kafka构建百万级QPS的实时计算平台

avatar 快递鸟社区
cover

电商大促数据库管控（下）：数据需求小时级交付与全链路安全管控

avatar 快递鸟社区