04-netty基础-Reactor三种模型

1 基本概念

Reactor模型是一种事件驱动（Event-Driven）的设计模式，主要用于高效处理高并发、I/O密集型场景（如网络、服务器、分布式等）。其核心思想就是集中管理事件，将I/O操作与业务逻辑解耦，避免传统多线程模型中线程切换的开销，从而提升系统的吞吐量和响应速度。

核心目标：
        在高并发场景下，传统的 “一连接一线程” 模型会因线程创建 / 销毁、上下文切换的开销过大而效率低下。Reactor 模型通过以下方式解决这一问题：

• 用单个或少量线程监听多个 I/O 事件（如网络连接、数据读写），避免线程资源浪费；
• 仅当事件触发（如客户端发送数据）时才执行对应处理逻辑，实现 “事件就绪才处理” 的高效调度。

2 核心组件

 Reactor 模型的运行依赖四个关键组件，它们协同完成事件的检测、分发与处理：
1、事件源
产生事件的源头，通常是I/O相关的资源，例如：
        网络套接字(Socket)：客户端连接、数据发送/接收等事件的源头        
        文件描述符(FD)：文件读写、异常等事件的源头
2、事件多路分发器（Event Demultiplexer）
又称 “I/O 多路复用器”，是 Reactor 模型的 “感知器官”。
作用：持续监听多个事件源的事件（如 “可读”“可写”“异常”），当事件触发时标记为 “就绪”；
底层依赖：操作系统提供的 I/O 多路复用系统调用，如 Unix/Linux 的select/poll/epoll，或 BSD 的kqueue。

3、反应器（Reactor）
模型的 “核心调度者”，是事件处理的中枢。
作用：从事件多路分发器获取 “就绪事件”，根据事件类型和关联的事件源，分发给对应的事件处理器；
本质：通过 “事件注册 - 事件监听 - 事件分发” 的逻辑，实现对所有事件的集中管理。

4 事件处理器（Handler）
负责具体业务逻辑的 “执行者”。
作用：定义事件处理的回调方法（如handleRead处理可读事件、handleWrite处理可写事件），由 Reactor 触发执行；
特点：仅关注业务逻辑（如解析请求、生成响应），不关心事件的检测与分发。

3 单Reactor单线程模型

3.1 概念

        在单Reactor单线程模型中，他们的作用以及实现逻辑，首先客户端访问服务端，在服务端这边首先是使用Reactor监听accept事件和read事件，当有连接过来，就交给acceptor处理accept事件，当触发read事件，同时accept或把read事件交给handler处理。所有动作都是由一个线程完成的。

特点：单线程Reactor模型编程简单，比较适用于每个请求都可以快速完成的场景，但是不能发挥出多核CPU的优势，在一般情况下，不会使用单Reactor单线程模型。

3.2 原理图

3.3 代码实现

3.3.1 入口

入口： 启动Reactor线程

package com.bonnie.netty.reactor.single;import java.io.IOException;/** * 单Reactor单线程模型 */public class Main { public static void main(String[] args) throws IOException { new Thread(new Reactor(8080, \"Main-Thread\")).start(); }}

3.3.2 Reactor

1、启动服务端ServerSocketChannel
2、监听accept事件
3、监听read事件

package com.bonnie.netty.reactor.single;import java.io.IOException;import java.net.InetSocketAddress;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.ServerSocketChannel;import java.util.Iterator;import java.util.Set;/** * 模拟Reactor的单线程模型 */public class Reactor implements Runnable { Selector selector; ServerSocketChannel serverSocketChannel; public Reactor(int port, String threadName) throws IOException { selector = Selector.open(); serverSocketChannel = ServerSocketChannel.open(); // 绑定端口 serverSocketChannel.bind(new InetSocketAddress(port)); // 设置成非阻塞 serverSocketChannel.configureBlocking(Boolean.FALSE); // 注册OP_ACCEPT，事件，会调用Acceptor.run方法 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, new Acceptor(selector, serverSocketChannel)); } @Override public void run() { while (!Thread.interrupted()) { try { // 阻塞 selector.select(); Set selectionKeys = selector.selectedKeys(); Iterator iterator = selectionKeys.iterator(); while (iterator.hasNext()) {  // 我们之前说的分发事件就是这个地方分发了, 此处可能是accept事件，也可能是read事件  dispatcher(iterator.next());  // 分发完之后要删除key，防止重复key  iterator.remove(); } } catch (IOException e) { throw new RuntimeException(e); } } } private void dispatcher(SelectionKey key) { // 然后在这里通过key获取这个attachment，执行他的run方法，记住，这里并没有开启线程，所有叫做单线程Reactor单线程模型 Runnable runnable = (Runnable)key.attachment(); if (runnable!=null) { runnable.run(); } }}

 3.3.3 Acceptor

1、处理accept请求
2、把read事件转发给handler处理

package com.bonnie.netty.reactor.single;import java.io.IOException;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.ServerSocketChannel;import java.nio.channels.SocketChannel;/** * 接收连接请求，并注册handle到selector */public class Acceptor implements Runnable{ Selector selector; ServerSocketChannel serverSocketChannel; public Acceptor(Selector selector, ServerSocketChannel serverSocketChannel) { this.selector = selector; this.serverSocketChannel = serverSocketChannel; } @Override public void run() { try { SocketChannel socketChannel = serverSocketChannel.accept(); System.out.println(socketChannel.getRemoteAddress() + \" 收到连接！！！\"); // 设置成非阻塞 socketChannel.configureBlocking(Boolean.FALSE); // 注册事件，交由Handler处理 socketChannel.register(selector, SelectionKey.OP_READ, new Handler(socketChannel)); } catch (IOException e) { throw new RuntimeException(e); } }}

 3.3.4 Handler

处理read事件

package com.bonnie.netty.reactor.single;import java.io.IOException;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.ServerSocketChannel;import java.nio.channels.SocketChannel;/** * 接收连接请求，并注册handle到selector */public class Acceptor implements Runnable{ Selector selector; ServerSocketChannel serverSocketChannel; public Acceptor(Selector selector, ServerSocketChannel serverSocketChannel) { this.selector = selector; this.serverSocketChannel = serverSocketChannel; } @Override public void run() { try { SocketChannel socketChannel = serverSocketChannel.accept(); System.out.println(socketChannel.getRemoteAddress() + \" 收到连接！！！\"); // 设置成非阻塞 socketChannel.configureBlocking(Boolean.FALSE); // 注册事件，交由Handler处理 socketChannel.register(selector, SelectionKey.OP_READ, new Handler(socketChannel)); } catch (IOException e) { throw new RuntimeException(e); } }}

3.3.5 码云位置

git地址： https://gitee.com/huyanqiu6666/netty.git    分支： 250724-reactor

4 单Reactor多线程模型

4.1 概念

解决单Reactor单线程模型的不足，使用多线程处理handler提升处理能力，增加吞吐量。

4.2 原理图

4.3 代码实现

4.3.1 入口

package com.bonnie.netty.reactor.mult;import java.io.IOException;/** * 单reactor多线程模型：处理handle的时候是线程池 */public class MultMain { public static void main(String[] args) throws IOException { new Thread(new MultReactor(8080, \"Main-Thread\")).start(); }}

4.3.2 MultReactor

package com.bonnie.netty.reactor.mult;import java.io.IOException;import java.net.InetSocketAddress;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.ServerSocketChannel;import java.util.Iterator;import java.util.Set;/** * 模拟单Reactor多线程模型 * 1、监听accept事件 * 2、监听read事件 */public class MultReactor implements Runnable { Selector selector; ServerSocketChannel serverSocketChannel; public MultReactor(int port, String threadName) throws IOException { selector = Selector.open(); serverSocketChannel = ServerSocketChannel.open(); // 绑定端口 serverSocketChannel.bind(new InetSocketAddress(port)); // 设置成非阻塞 serverSocketChannel.configureBlocking(Boolean.FALSE); // 注册OP_ACCEPT，事件，会调用Acceptor.run方法 serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, new MultAcceptor(selector, serverSocketChannel)); } @Override public void run() { while (!Thread.interrupted()) { try { // 阻塞 selector.select(); Set selectionKeys = selector.selectedKeys(); Iterator iterator = selectionKeys.iterator(); while (iterator.hasNext()) {  // 我们之前说的分发事件就是这个地方分发了, 此处可能是accept事件，也可能是read事件  dispatcher(iterator.next());  // 分发完之后要删除key，防止重复key  iterator.remove(); } } catch (IOException e) { throw new RuntimeException(e); } } } private void dispatcher(SelectionKey key) { // 然后在这里通过key获取这个attachment，执行他的run方法，记住，这里并没有开启线程，所有叫做单线程Reactor单线程模型 Runnable runnable = (Runnable)key.attachment(); if (runnable!=null) { runnable.run(); } }}

4.3.3 MultAcceptor

package com.bonnie.netty.reactor.mult;import java.io.IOException;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.ServerSocketChannel;import java.nio.channels.SocketChannel;/** * 接收连接请求，并注册handle到selector * 1、处理accept事件 * 2、read事件转发给handler */public class MultAcceptor implements Runnable{ Selector selector; ServerSocketChannel serverSocketChannel; public MultAcceptor(Selector selector, ServerSocketChannel serverSocketChannel) { this.selector = selector; this.serverSocketChannel = serverSocketChannel; } @Override public void run() { try { SocketChannel socketChannel = serverSocketChannel.accept(); System.out.println(socketChannel.getRemoteAddress() + \" 收到连接！！！\"); // 设置成非阻塞 socketChannel.configureBlocking(Boolean.FALSE); // 注册事件，交由Handler处理 socketChannel.register(selector, SelectionKey.OP_READ, new MultHandler(socketChannel)); } catch (IOException e) { throw new RuntimeException(e); } }}

4.3.4 MultHandler

package com.bonnie.netty.reactor.mult;import org.apache.commons.lang3.StringUtils;import java.io.IOException;import java.nio.ByteBuffer;import java.nio.channels.SocketChannel;import java.util.concurrent.Executor;import java.util.concurrent.Executors;/** * Handler将read事件给线程池处理 */public class MultHandler implements Runnable { private SocketChannel socketChannel; public MultHandler(SocketChannel socketChannel) { this.socketChannel = socketChannel; } private Executor executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors() * 2); @Override public void run() { // 放到线程池中处理 executor.execute(new ReadHandle(socketChannel)); } private class ReadHandle implements Runnable{ private SocketChannel socketChannel; public ReadHandle(SocketChannel socketChannel) { this.socketChannel = socketChannel; } @Override public void run() { System.out.println(\"线程名称：\" + Thread.currentThread().getName()); // 定义一个ByteBuffer的数据结构 ByteBuffer byteBuffer = ByteBuffer.allocate(1024); int len=0, total=0; String msg = StringUtils.EMPTY; try { do {  len = socketChannel.read(byteBuffer);  if (len > 0) { total += len; msg += new String(byteBuffer.array());  }  System.out.println(socketChannel.getRemoteAddress() + \"客戶端的消息已收到，\" + msg); } while (len>byteBuffer.capacity()); } catch (IOException e) { throw new RuntimeException(e); } } }}

4.3.5 码云位置

git地址： https://gitee.com/huyanqiu6666/netty.git    分支： 250724-reactor

5 主从Reactor模型

5.1 概念

5.2 原理图

5.3 代码实现

5.3.1 入口

package com.bonnie.netty.reactor.main;import java.io.IOException;/** * 主从Reactor多线程模型 */public class MainMain { public static void main(String[] args) throws IOException { new Thread(new MainReactor(8080), \"Main-Thread\").start(); }}

5.3.2 MainReactor

package com.bonnie.netty.reactor.main;import java.io.IOException;import java.net.InetSocketAddress;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.ServerSocketChannel;import java.util.Iterator;import java.util.Set;/** * 构建Selector、ServerSocketChannel绑定端口，设置成非阻塞 * 注册accept事件 */public class MainReactor implements Runnable { private final Selector selector; private final ServerSocketChannel serverSocketChannel; public MainReactor(int port) throws IOException { // 主Reactor负责监听accept事件 selector = Selector.open(); serverSocketChannel = ServerSocketChannel.open(); serverSocketChannel.bind(new InetSocketAddress(port)); serverSocketChannel.configureBlocking(Boolean.FALSE); // 添加attachment为acceptor serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, new MainAcceptor(serverSocketChannel)); } @Override public void run() { while (!Thread.interrupted()) { try { // 等待客户端的连接到来 selector.select(); Set selectionKeys = selector.selectedKeys(); Iterator iterator = selectionKeys.iterator(); while (iterator.hasNext()) {  // 当有连接过来的时候就会转发任务  dispatch(iterator.next());  iterator.remove(); } } catch (IOException e) { throw new RuntimeException(e); } } } private void dispatch(SelectionKey key) { // 可能拿到的对象有两个 Acceptor Handler Runnable runnable = (Runnable)key.attachment(); if (runnable!=null) { runnable.run(); } }}

5.3.3 SubReactor

package com.bonnie.netty.reactor.main;import java.io.IOException;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.util.Iterator;import java.util.Set;/** * 子Reactor */public class SubReactor implements Runnable{ private Selector selector; public SubReactor(Selector selector) { this.selector = selector; } @Override public void run() { while (true) { try { // 所有的子Reactor阻塞 selector.select(); System.out.println(\"selector:\"+selector.toString()+\"thread:\"+Thread.currentThread().getName()); Set selectionKeys = selector.selectedKeys(); Iterator iterator = selectionKeys.iterator(); while (iterator.hasNext()) {  dispacher(iterator.next());  iterator.remove(); } } catch (IOException e) { throw new RuntimeException(e); } } } private void dispacher(SelectionKey selectionKey) { // 此处会调用workHandler里面的方法 Runnable runnable = (Runnable) selectionKey.attachment(); if (runnable!=null) { runnable.run(); } }}

5.3.4 MainAcceptor

package com.bonnie.netty.reactor.main;import java.io.IOException;import java.nio.channels.SelectionKey;import java.nio.channels.Selector;import java.nio.channels.ServerSocketChannel;import java.nio.channels.SocketChannel;/** * 处理MainAcceptor请求 */public class MainAcceptor implements Runnable{ private ServerSocketChannel serverSocketChannel; private final Integer core = Runtime.getRuntime().availableProcessors() * 2; private Integer index = 0; private Selector[] selectors = new Selector[core]; private SubReactor[] subReactors = new SubReactor[core]; private Thread[] threads = new Thread[core]; /** * 构造方法 * 1、初始化多个SubReactor * 2、初始化多个Selector * 3、每个SubReactor都有一个Selector * 4、创建线程包装SubReactor * 5、启动线程，也就是调用每一个SubReactor的run方法 */ public MainAcceptor(ServerSocketChannel serverSocketChannel) throws IOException { this.serverSocketChannel = serverSocketChannel; for (int i=0; i<core; i++) { selectors[i] = Selector.open(); subReactors[i] = new SubReactor(selectors[i]); threads[i] = new Thread(subReactors[i]); // 一初始化就工作起来 threads[i].start(); } } @Override public void run() { try { System.out.println(\"acceptor thread: \" + Thread.currentThread().getName()); // 此处就会接收连接的socketChannel SocketChannel socketChannel = serverSocketChannel.accept(); System.out.println(\"有客户端上来了：\"+socketChannel.getRemoteAddress()); socketChannel.configureBlocking(Boolean.FALSE); // 立即唤醒第一个阻塞的selector selectors[index].wakeup(); // 然后注册Read事件到该selector socketChannel.register(selectors[index], SelectionKey.OP_READ, new WorkHandler(socketChannel)); index = (++index) % core; } catch (IOException e) { throw new RuntimeException(e); } }}

5.3.5 WorkHandler

package com.bonnie.netty.reactor.main;import java.io.IOException;import java.nio.ByteBuffer;import java.nio.channels.SocketChannel;import java.nio.charset.StandardCharsets;/** * SubReactor把事件交给WorkHandler去执行 */public class WorkHandler implements Runnable{ private SocketChannel socketChannel; public WorkHandler(SocketChannel socketChannel) { this.socketChannel = socketChannel; } @Override public void run() { try { System.out.println(\"WorkHandler thread:\" + Thread.currentThread().getName()); ByteBuffer buffer = ByteBuffer.allocate(1024); // 数据读取到socketChannel中 socketChannel.read(buffer); String msg = new String(buffer.array(), StandardCharsets.UTF_8); System.out.println(socketChannel.getRemoteAddress() + \"发来了消息：\" + msg); // 给客户端会写消息 socketChannel.read(ByteBuffer.wrap(\"你的消息已收到\".getBytes(StandardCharsets.UTF_8))); } catch (IOException e) { throw new RuntimeException(e); } }}

5.3.6 码云位置

git地址： https://gitee.com/huyanqiu6666/netty.git    分支： 250724-reactor