当前位置： 首页 > news >正文

04-netty基础-Reactor三种模型

news 2025/7/26 10:54:05

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));}@Overridepublic void run() {while (!Thread.interrupted()) {try {// 阻塞selector.select();Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()) {// 我们之前说的分发事件就是这个地方分发了, 此处可能是accept事件，也可能是read事件dispatcher(iterator.next());// 分发完之后要删除key，防止重复keyiterator.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;}@Overridepublic 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;}@Overridepublic 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));}@Overridepublic void run() {while (!Thread.interrupted()) {try {// 阻塞selector.select();Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()) {// 我们之前说的分发事件就是这个地方分发了, 此处可能是accept事件，也可能是read事件dispatcher(iterator.next());// 分发完之后要删除key，防止重复keyiterator.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;}@Overridepublic 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);@Overridepublic void run() {// 放到线程池中处理executor.execute(new ReadHandle(socketChannel));}private class ReadHandle implements Runnable{private SocketChannel socketChannel;public ReadHandle(SocketChannel socketChannel) {this.socketChannel = socketChannel;}@Overridepublic 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为acceptorserverSocketChannel.register(selector, SelectionKey.OP_ACCEPT, new MainAcceptor(serverSocketChannel));}@Overridepublic void run() {while (!Thread.interrupted()) {try {// 等待客户端的连接到来selector.select();Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> iterator = selectionKeys.iterator();while (iterator.hasNext()) {// 当有连接过来的时候就会转发任务dispatch(iterator.next());iterator.remove();}} catch (IOException e) {throw new RuntimeException(e);}}}private void dispatch(SelectionKey key) {// 可能拿到的对象有两个  Acceptor HandlerRunnable 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;}@Overridepublic void run() {while (true) {try {// 所有的子Reactor阻塞selector.select();System.out.println("selector:"+selector.toString()+"thread:"+Thread.currentThread().getName());Set<SelectionKey> selectionKeys = selector.selectedKeys();Iterator<SelectionKey> 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();}}@Overridepublic void run() {try {System.out.println("acceptor thread: " + Thread.currentThread().getName());// 此处就会接收连接的socketChannelSocketChannel socketChannel = serverSocketChannel.accept();System.out.println("有客户端上来了："+socketChannel.getRemoteAddress());socketChannel.configureBlocking(Boolean.FALSE);// 立即唤醒第一个阻塞的selectorselectors[index].wakeup();// 然后注册Read事件到该selectorsocketChannel.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;}@Overridepublic 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



http://www.lryc.cn/news/599788.html

相关文章：

  • 无需 Root 关闭联网验证 随意修改手机名称（适用于OPPO、一加、真我）
  • 【笔记】Handy Multi-Agent Tutorial 第四章: CAMEL框架下的RAG应用 （简介）
  • RocketMQ 5.3.0 ARM64 架构安装部署指南
  • 详解FreeRTOS开发过程（八）-- 时间标志
  • 【电赛学习笔记】MaxiCAM 项目实践——与单片机的串口通信
  • ESP32学习笔记_Components(1)——使用LED Strip组件点亮LED灯带
  • Yolov8/Yolov11实例分割训练自有数据集
  • AWS WebRTC：我们的业务模式
  • 壁纸管理 API 文档
  • MybatisPlus-17.扩展功能-JSON处理器
  • Asp.net core mvc中TagHelper的GetChildContentAsync和Content区别
  • 【04】C#入门到精通——C# 程序错误处理， try catch 捕获异常，避免程序崩溃
  • Android 的16 KB内存页设备需要硬件支持吗，还是只需要手机升级到Android15系统就可以
  • [python][基础]Flask 技术栈
  • c盘temp文件夹可以删除吗？C盘空间清理指南来了
  • epoll_event数据结构及使用案例详解
  • WPF高级学习（一）
  • 智能机器人的技术革命：从感知到决策的全栈架构解析
  • leetcode933最近的请求次数
  • Keepalived 深度技术解析与高可用实践指南
  • 三种深度学习模型（GRU、CNN-GRU、贝叶斯优化的CNN-GRU/BO-CNN-GRU）对北半球光伏数据进行时间序列预测
  • Python 爬虫（五）：PyQuery 框架
  • Gin 框架的中间件机制
  • 【世纪龙科技】新能源汽车电驱动总成装调与检修仿真教学软件
  • PHP：从入门到实践——构建现代Web应用的利器
  • 【STM32项目】有毒气体监测
  • VUE懒加载（4种方式）
  • 【Android】桌面小组件开发
  • Java设计模式-建造者模式
  • Tomcat线程池深度优化指南：高并发场景下的maxConnections计算与监控体系