🔭 嗨，您好 👋 我是 vnjohn，在互联网企业担任 Java 开发，CSDN 优质创作者
📖 推荐专栏：Spring、MySQL、Nacos、Java，后续其他专栏会持续优化更新迭代
🌲文章所在专栏：网络 I/O
🤔 我当前正在学习微服务领域、云原生领域、消息中间件等架构、原理知识
💬 向我询问任何您想要的东西，ID：vnjohn
🔥觉得博主文章写的还 OK，能够帮助到您的，感谢三连支持博客🙏
😄 代词: vnjohn
⚡ 有趣的事实：音乐、跑步、电影、游戏
目录
- 前言
- 单 Group 混杂模式
- SelectorThread
- SelectorThreadSingletonGroup
- SelectorSingletonGroupMainThread
- 测试单 Group 混杂
- 小结
- 多 Group 主从模式
- SelectorThread
- SelectorGroup
- SelectorMultiGroupMainThread
- 测试多 Group 主从
- 小结
- IO Threads
- 总结
前言
在之前的文章中，从阻塞 I/O：BIO、非阻塞 I/O：NIO、多路复用 select/poll、多路复用 epoll
重要的 I/O 模型也是现在市场上大部分中间件运用的模型也就是基于 I/O 多路复用：epoll，比如：Redis、RocketMQ、Nginx 等，这些地方都运用了 epoll，只不过在 RocketMQ 的实现采用了 Netty，而 Netty 也基于 epoll 这套多路复用模型进行实现的，上篇文章详细介绍了单 Selector 多线程模型、单 Selector 单线程模型，在此文章会才是真正意义上 Netty 的变相实现，看它是如何一步步从单 Selector 非线性模型 —> 单 Selector 线性模型 —> 单 Selector Group 混杂模式 —> 多 Selector Group 主从模式一步步演练过来的，本篇博文主要围绕单 Group 混杂模式 —> 多 Group 主从模式进行具体的展开.
Group 组的含义，它可以是充当 Boss 组角色也可以是充当 Worker 组角色
Boss：负责接收服务端 channel 以及客户端连接的 channel
Worker：负责的是 R/W 工作，由于读写操作较于频繁，它一般在分配资源上会多于优于 Boss
单 Group 混杂模式
单个 Group 混杂模式，代表的就是一个 Group 中，它既可以是 Boss 也可以 Worker，而在单个 Group 中，它就是混杂的存在，既可以做 Boss 的工作，也可以做 Worker 的工作
单 Group 中存在多个 Selector
其实在单 Group 混杂模式中，它的代码与单 Selector 单线程模型区别不大，只是在单 Group 采用了多线程的方式来进行了处理，充分利用了多核 CPU 的资源，而在单 Group 我只负责让其中一个线程进行服务端 channel，而其他的线程负责处理来自客户端 channel 的 accept 工作和客户端的读与服务端的写工作处理，接下来就看代码吧！！！
SelectorThread
package org.vnjohn.group.singleton; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.*; import java.util.Iterator; import java.util.Set; import java.util.concurrent.LinkedBlockingQueue; /** * @author vnjohn * @since 2023/12/15 */ public class SelectorThread implements Runnable { Selector selector = null; LinkedBlockingQueue lbq = new LinkedBlockingQueue(); SelectorThreadSingletonGroup selectorThreadGroup = null; public SelectorThread(SelectorThreadSingletonGroup selectorThreadGroup) { try { this.selectorThreadGroup = selectorThreadGroup; selector = Selector.open(); } catch (IOException e) { e.printStackTrace(); } } @Override public void run() { // loop while (true) { try { // 1.select:如果一直没有 fd，该方法会阻塞，一直没有返回，通过调用 wakeup() 唤醒 System.out.println(Thread.currentThread().getName() + " ： before select ......" + selector.keys().size()); int num = selector.select(); System.out.println(Thread.currentThread().getName() + " ： after select ......" + selector.keys().size()); // 2.处理 selectKeys if (num > 0) { Set selectionKeys = selector.selectedKeys(); Iterator iterator = selectionKeys.iterator(); while (iterator.hasNext()) { // 每一个 fd 是线性处理的过程 SelectionKey key = iterator.next(); iterator.remove(); if (key.isAcceptable()) { // 接受客户端的过程 acceptHandler(key); } else if (key.isReadable()) { readHandler(key); } else if (key.isWritable()) { } } } // 3.处理 queue runTask,队列是堆里的对象，线程的栈是独立的，堆是共享的，只有方法的逻辑，本地变量是线程隔离的 if (!lbq.isEmpty()) { Channel channel = lbq.take(); // accept 使用的是 ServerSocketChannel if (channel instanceof ServerSocketChannel) { ServerSocketChannel server = (ServerSocketChannel) channel; server.register(selector, SelectionKey.OP_ACCEPT); System.out.println(Thread.currentThread().getName() + " register server"); // read / write 使用的是 SocketChannel } else if (channel instanceof SocketChannel) { SocketChannel client = (SocketChannel) channel; ByteBuffer buffer = ByteBuffer.allocateDirect(4096); client.register(selector, SelectionKey.OP_READ, buffer); System.out.println(Thread.currentThread().getName() + " register client:" + client.getRemoteAddress()); } } } catch (IOException | InterruptedException e) { e.printStackTrace(); } } } private void readHandler(SelectionKey key) { System.out.println(Thread.currentThread().getName() + " readHandler......."); ByteBuffer buffer = (ByteBuffer) key.attachment(); SocketChannel client = (SocketChannel) key.channel(); buffer.clear(); while (true) { try { int num = client.read(buffer); if (num > 0) { // 将读到的内容翻转,然后直接写出 buffer.flip(); while (buffer.hasRemaining()) { client.write(buffer); } buffer.clear(); } else if (num == 0) { break; } else { // 有可能客户端断开了-异常情况 System.out.println("client:" + client.getRemoteAddress() + " closed...."); key.cancel(); client.close(); break; } } catch (IOException e) { e.printStackTrace(); } } } private void acceptHandler(SelectionKey key) { System.out.println(Thread.currentThread().getName() + " acceptHandler......."); ServerSocketChannel server = (ServerSocketChannel) key.channel(); try { SocketChannel client = server.accept(); client.configureBlocking(false); // choose a selector and register !! selectorThreadGroup.nextSelectorV2(client); } catch (IOException e) { e.printStackTrace(); } } }Selector 线程仍然是做这么几件事情，Selector#select 阻塞调用等待事件的到来，当有新的客户端连接时，通过 SelectorThreadGroup#nextSelectorV2 去选择对应的 accept 线程进行处理往阻塞队列中塞入元素；当有事件需要读则处理读事件具体的业务方法，模拟业务场景，将接收到的数据再回写给客户端
在阻塞队列等待事件到来时，通过 channel 来区分是服务端的 channel 还是客户端的 channel，服务端的 channel 则注册一个 accept 事件，如果是客户端 channel 那么就将它注册一个 accept 后再注册 read 事件后返回.
在这里并没有区分事件的类型 accept、R/W，也没有区分客户端和服务端的事件怎么处理
SelectorThreadSingletonGroup
package org.vnjohn.group.singleton; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.channels.Channel; import java.nio.channels.ServerSocketChannel; import java.util.concurrent.atomic.AtomicInteger; /** * @author vnjohn * @since 2023/12/16 */ public class SelectorThreadSingletonGroup { SelectorThread[] selectorThreads; ServerSocketChannel server = null; AtomicInteger xid = new AtomicInteger(0); public SelectorThreadSingletonGroup(int num) { // num 是线程数 selectorThreads = new SelectorThread[num]; // 启动多个线程，一个线程对应一个 selector for (int i = 0; i对比单 Selector 单线程模式，在选择 Selector 时做了一下区分，当属于服务端 channel 时，以写死的方式让第一个线程来进行处理，自然而然它会走到属于第一个线程的 Selector 进行处理；若是客户端 channel，让其他的线程去进行处理，无论是客户端的 accept 还是 R/W 事件
在这里能够看出问题，第一个线程资源一直浪费在哪里，它只做了服务端 channel 的 accept 工作，而其他的工作都交给了客户端去进行处理了，而客户端处理虽然有多个线程，通过 nextV2 方法分配的方式必然会造成资源的倾斜，可能有的客户端处理的事件过多，有的客户端处理的事件过少，这也就是为什么会提出负载均衡这个概念了，轮询、权重？
SelectorSingletonGroupMainThread
package org.vnjohn.group.singleton; /** * @author vnjohn * @since 2023/12/16 */ public class SelectorSingletonGroupMainThread { public static void main(String[] args) { // 1、创建 IO Thread（一个或多个） // 单个 group 混杂模式：即当 BOSS 又当 worker // 混杂模式：只有一个线程负责 accept，每个线程都会被分配 client 进行 R/W（包括负责客户端 accept） SelectorThreadSingletonGroup singletonGroup = new SelectorThreadSingletonGroup(3); // 2、应该把监听（9999）的 server 注册到某一个 selector 上 singletonGroup.bind(9999); } }分配一个 Group 组，该组分配三个线程，一个线程负责服务单 accept，其他线程负责客户端连接以及客户端的读写，既充当了 Boss 工作也充当了 Worker 工作
测试单 Group 混杂
1、启动主线程 Main 方法，控制台输出内容如下：
SelectorSingletonGroupThread-1 ： before select ......0 SelectorSingletonGroupThread-0 ： before select ......0 SelectorSingletonGroupThread-2 ： before select ......0 SelectorSingletonGroupThread-0 ： after select ......0 SelectorSingletonGroupThread-0 register server SelectorSingletonGroupThread-0 ： before select ......1
一个 Group 中三个 Selector 都启动成功，由第一个线程负责接收服务端 channel accept 事件
2、nc localhost 9999 模拟客户端来连接服务端进行读、写操作，它会由其中一个线程（并非第一个线程）进行 accept、R/W，但是客户端的 accept 会交给服务端所在的 SelectorSingletonGroupThread-0 去进行建立绑定 TCP 关系
SelectorSingletonGroupThread-0 ： after select ......1 SelectorSingletonGroupThread-0 acceptHandler....... SelectorSingletonGroupThread-0 ： before select ......1 SelectorSingletonGroupThread-2 ： after select ......0 SelectorSingletonGroupThread-2 register client:/0:0:0:0:0:0:0:1:55759 SelectorSingletonGroupThread-2 ： before select ......1
3、当我在客户端触发写数据的操作时，它会由 SelectorSingletonGroupThread-2 去进行读、写操作
SelectorSingletonGroupThread-2 ： after select ......1 SelectorSingletonGroupThread-2 readHandler....... SelectorSingletonGroupThread-2 ： before select ......1
从以上返回的内容来看，SelectorSingletonGroupThread-0 线程它只负责接收服务端的 accept 以及与客户端之间建立的 TCP 关系，而其他的线程负责接收客户端的 accept 以及客户端的 R/W 操作
小结
由单 Group 混杂模式来看，虽然使用了多个 Selector 多线程来利用好多核多 CPU 资源，但从实际运行的角度来看，它不是最优解，它仍然会有一些资源倾斜以及浪费问题
1、比如：SelectorSingletonGroupThread-0，它只是接收了服务端 channel accept 以及建立好 TCP 关系，它就不做任何的操作了，这部分工作对于网络而言只是一小部分的工作，大部分的工作基本上都是围绕在客户端与服务单之间发生的 R/W 操作的，从这点来看，SelectorSingletonGroupThread-0 它并完全的能够利用好这个线程能做的事情
2、其他的 Selector 线程，通过 nextV2 方法来分配某个线程来进行处理客户端 accept、R/W 事件，在多个客户端同时进来时，肯定会造成资源倾斜的问题，有的线程很忙碌，有的线程停滞不前
基于设计理念来看，需要将业务进行解耦，比如说读写分离，而在这里应该考虑的是 accept 工作，无论是服务端 channel 还是客户端 channel 应该都交由给一个线程去进行处理，而其他线程只是专注于处理客户端的读写事件，这样的好处在于，能够让我们更加理解，分水岭分界开各自要做的事情
在下面要介绍的就是「服务端 channel、客户端 channel」交由给一个 Group 去做：Boss 组
客户端与服务端之间的读写工作交给其他 Group 去做：Worker 组
多 Group 主从模式
如上图是 Netty 中工作的架构图，类比下面要介绍多 Group 主从模式
Boss 组：主要是用于接收客户端连接进来，然后分配到 workerGroup 线程，交由 Worker 组中的线程去 accept 注册到自己的 selector 中，后续该客户端的 fd 都交由分配到的 worker 线程去线性处理，而 Boss 只是负责与客户端连接，但不负责对客户端进行 event 处理，event 交由 worker 线程去处理.
SelectorThread
package org.vnjohn.group.multi; import java.io.IOException; import java.nio.ByteBuffer; import java.nio.channels.*; import java.util.Iterator; import java.util.Set; import java.util.concurrent.LinkedBlockingQueue; /** 1. @author vnjohn 2. @since 2023/12/16 */ public class SelectorThread extends ThreadLocal implements Runnable { /** * 每线程对应一个 selector，多线程情况下，该主机，该程序并发进来客户端会被分配到多个 selector 上 * 注意：每个客户端只绑定到其中一个 selector，其实不会有交互问题：每个客户端都是线性执行的 */ Selector selector = null; /** * 每个线程持有自己独立的 LinkedBlockingQueue 对象 */ LinkedBlockingQueue linkedBlockingQueue = get(); SelectorThreadGroup selectorThreadGroup = null; @Override protected LinkedBlockingQueue initialValue() { return new LinkedBlockingQueue(); } public SelectorThread(SelectorThreadGroup selectorThreadGroup) { try { this.selectorThreadGroup = selectorThreadGroup; selector = Selector.open(); } catch (IOException e) { e.printStackTrace(); } } @Override public void run() { // loop while (true) { try { // 1.select:如果一直没有 fd，该方法会阻塞，一直没有返回，通过调用 wakeup() 唤醒 int num = selector.select(); // 2.处理 selectKeys if (num > 0) { Set selectionKeys = selector.selectedKeys(); Iterator iterator = selectionKeys.iterator(); // // 每一个 fd 是线性处理的过程 while (iterator.hasNext()) { SelectionKey key = iterator.next(); iterator.remove(); if (key.isAcceptable()) { // 复杂:接受客户端的过程（接收之后,要注册,新的客户端注册到那个 selector 上呢?） acceptHandler(key); } else if (key.isReadable()) { readHandler(key); } else if (key.isWritable()) { } } } // 3.处理 queue runTask,队列是堆里的对象，线程的栈是独立的，堆是共享的 if (!linkedBlockingQueue.isEmpty()) { Channel c = linkedBlockingQueue.take(); // accept 使用的是 ServerSocketChannel if (c instanceof ServerSocketChannel) { ServerSocketChannel server = (ServerSocketChannel) c; server.register(selector, SelectionKey.OP_ACCEPT); System.out.println(Thread.currentThread().getName() + " register server " + server.getLocalAddress()); // read、write 使用的是 SocketChannel } else if (c instanceof SocketChannel) { SocketChannel client = (SocketChannel) c; ByteBuffer buffer = ByteBuffer.allocateDirect(4096); client.register(selector, SelectionKey.OP_READ, buffer); System.out.println(Thread.currentThread().getName() + " register client:" + client.getRemoteAddress()); } } } catch (IOException | InterruptedException e) { e.printStackTrace(); } } } private void readHandler(SelectionKey key) { System.out.println(Thread.currentThread().getName() + " readHandler......."); ByteBuffer buffer = (ByteBuffer) key.attachment(); SocketChannel client = (SocketChannel) key.channel(); buffer.clear(); while (true) { try { int num = client.read(buffer); if (num > 0) { // 将读到的内容翻转,然后直接写出 buffer.flip(); while (buffer.hasRemaining()) { client.write(buffer); } buffer.clear(); } else if (num == 0) { break; } else { // 有可能客户端断开了-异常情况 System.out.println("client:" + client.getRemoteAddress() + " closed...."); key.cancel(); client.close(); break; } } catch (IOException e) { e.printStackTrace(); } } } private void acceptHandler(SelectionKey key) { System.out.println(Thread.currentThread().getName() + " acceptHandler......."); ServerSocketChannel server = (ServerSocketChannel) key.channel(); try { SocketChannel client = server.accept(); client.configureBlocking(false); // 此时的 stg 已经是 worker 组了,然后让 worker 组去分发客户端的 accept selectorThreadGroup.nextSelectorV3(client); } catch (IOException e) { e.printStackTrace(); } } /** * 此时如果是 boss 组调用的话，会将当前类的 SelectorThreadGroup 设置为 workerGroup 来执行操作 * * @param stgWorker */ public void setWorker(SelectorThreadGroup stgWorker) { this.selectorThreadGroup = stgWorker; } }在 SelectorThread 中，本地线程独有 LinkedBlockingQueue，线程在执行时分为以下三步：
1、Selector#select：阻塞直到拿到有状态的 FDS
2、 若 FDS 数量返回大于 0，接下来就是处理 SelectKeys 里面每个 SelectKey 对象
若当前 SelectKey 状态为 accept，就交由给 WorkerGroup 里的线程去注册
若当前 SelectKey 状态为 read，说明当前的 FD 已经在 WorkerGroup 里的线程了，直接线性处理与该客户端的数据交互即可
通过 SelectKey#attachment 方法接收客户端发送过来的数据，通过 SelectKey#channel 拿到客户端信息
通过客户端 channel > SocketChannel#read 方法读取来自客户端的数据，通过 SocketChannel#write 方法写回到客户端中
3、处理 LinkedBlockingQueue 队列中的元素，也就是 Task，队列是属于堆里的对象，而线程栈是独享的，堆是共享的，再去队列中取出对应的 channel
若 channel 为 ServerSocketChannel，则给它注册到 BossGroup 中，给它绑定上 Accept Event
若 channel 为 SocketChannel，则给它注册到 WorkerGroup 中，给它绑定上 Read Event
SelectorGroup
package org.vnjohn.group.multi; import java.io.IOException; import java.net.InetSocketAddress; import java.nio.channels.Channel; import java.nio.channels.ServerSocketChannel; import java.util.concurrent.atomic.AtomicInteger; /** * @author vnjohn * @since 2023/12/16 */ public class SelectorThreadGroup { SelectorThread[] selectorThreads; ServerSocketChannel server = null; AtomicInteger xid = new AtomicInteger(0); // // 如果 /** * 若 workerThreadGroup 是当前 group,说明就是 worker 组 * 若不是当前 group，说明是 BOSS 下的 worker 组，Boss 要持有对 Worker 组的引用 */ SelectorThreadGroup workerThreadGroup = this; public void setWorker(SelectorThreadGroup selectorThreadGroup) { this.workerThreadGroup = selectorThreadGroup; } public SelectorThreadGroup(int num, Boolean bossGroup) { // num 是线程数 selectorThreads = new SelectorThread[num]; // 启动多个线程，一个线程对应一个 selector for (int i = 0; iSelectorGroup#next 方法是取出 Boss Group 中的线程，SelectorGroup#nextV3 方法是取出 Worker Group 中的线程，若对应的组进行了多次调用，两种方式都是通过轮询分配工作给线程的匹配规则
BoosGroup、WorkerGroup 在处理的过程有不一样区别
若是 ServerSocket 则分配到当前 Boss Group 中的 BlockingQueue 中，若客户端连接进来了，通过 Boss 分配给到 WorkerGroup 组其中一个线程去处理客户端操作 event
SO，BossGroup 要持有 WorkerGroup 引用，才能在客户端到来时，Boss 能够通过 WorkerGroup 对象去分配一个线程处理这个客户端的操作
SelectorMultiGroupMainThread
package org.vnjohn.group.multi; /** * @author vnjohn * @since 2023/12/16 */ public class SelectorMultiGroupMainThread { public static void main(String[] args) { // Boss 有自己的线程组 SelectorThreadGroup boss = new SelectorThreadGroup(3, Boolean.TRUE); // worker 有自己的线程组 SelectorThreadGroup worker = new SelectorThreadGroup(3, Boolean.FALSE); boss.setWorker(worker); boss.bind(6666); boss.bind(7777); boss.bind(8888); } }BossGroup 每分配一个线程，都需要去进行 accept，触发服务端的 bind 操作，然后这个被选中的 Boss 线程必须持有对 WorkerGroup 引用
测试多 Group 主从
1、启动主线程 main 方法，控制台输出内容，如下：
SelectorBossGroupThread-0 register server /0:0:0:0:0:0:0:0:8888 SelectorBossGroupThread-2 register server /0:0:0:0:0:0:0:0:7777 SelectorBossGroupThread-1 register server /0:0:0:0:0:0:0:0:6666
每个服务端 channel 都注册到对应的 Boss Selector 线程
2、nc localhost 8888、nc localhost 6666、nc localhost 7777 模拟客户端连接，如此，每个客户端的连接都会交由给对应端口所在 Boss 线程进行 accept，并会轮询分配给到 WorkerGroup 中的线程中去，控制台输出内容如下：
SelectorBossGroupThread-0 acceptHandler....... SelectorWorkerGroupThread-1 register client:/0:0:0:0:0:0:0:1:58175 SelectorBossGroupThread-1 acceptHandler....... SelectorWorkerGroupThread-2 register client:/0:0:0:0:0:0:0:1:58288 SelectorBossGroupThread-2 acceptHandler....... SelectorWorkerGroupThread-0 register client:/0:0:0:0:0:0:0:1:58397
BossGroupThread-0：8888 > WorkerGroupThread-1
BossGroupThread-1：6666 > WorkerGroupThread-2
BossGroupThread-2：7777 > WorkerGroupThread-0
小结
BossGroup 负责接收客户端，由它轮询分配 WorkerGroup 线程先 accept，然后交由给 WorkerGroup 线程去处理客户端的 R/W 操作！
核心：BossGroup 持有 WorkerGroup 中引用，各个线程持有对应 TaskQueue，Boss 处理 accept 以及分配客户端的工作，Worker 处理客户端的读写 R/W 工作
IO Threads
io threads 是为了更好发挥硬件以及 CPU 多核的处理能力，在对客户端有状态的 FD 进行 R/W 操作时，拿到数据以后，防止对当前的 FD 读取/写入时一直阻塞，将其他 FD 交由给其他业务线程去处理，io threads 就是为了解决 IO R/W 问题而存在的，也就是提高性能而存在的一种机制.
总结
该篇博文主要介绍多路复用模型 Epoll 下「单 Group 混杂模式与多 Group 主从模式」之间的区别，先是说明了在单 Group 混杂模式中由于 Event 未划分清晰造成资源倾斜问题，后者介绍了多 Group 主从模式，解决资源倾斜存在的问题，结合 BossGroup + WorkerGroup + 链接阻塞队列的方式来完成，Netty Reactor 它的工作架构图类比于此模式，只是它在此基础上做了很多的优化工作，也就是为什么大多数中间价会使用 Netty 原因，最重要的就是为了充分发挥我们硬件以及多核 CPU 资源，希望您能够喜欢，感谢三连支持！
参考文献：
学习帮助文档：
- man pages：yum install man
- pthread man pages：yum -y install man-pages
🌟🌟🌟愿你我都能够在寒冬中相互取暖，互相成长，只有不断积累、沉淀自己，后面有机会自然能破冰而行！
博文放在 网络 I/O 专栏里，欢迎订阅，会持续更新！
如果觉得博文不错，关注我 vnjohn，后续会有更多实战、源码、架构干货分享！
推荐专栏：Spring、MySQL，订阅一波不再迷路
大家的「关注❤️ + 点赞👍 + 收藏⭐」就是我创作的最大动力！谢谢大家的支持，我们下文见！
