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

04-netty基础-Reactor三种模型

news 2025/8/2 22:41: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.dtcms.com/a/297205.html

相关文章：

  • docker compose xtify-music-web
  • 华为OpenStack架构学习9篇 连载—— 02 OpenStack界面管理【附全文阅读】
  • VR 三维重建：重塑建筑工程全生命周期的数字化革命
  • [NLP]多电源域设计的仿真验证方法
  • Redis 5.0.14安装教程
  • Android 10.0 sts CtsSecurityBulletinHostTestCases的相关异常分析
  • 关于自定义域和 GitHub Pages（Windows）
  • OpenCV（04）梯度处理，边缘检测，绘制轮廓，凸包特征检测，轮廓特征查找
  • [python][flask]Flask-Login 使用详解
  • uniapp小程序上传图片并压缩
  • 吊汤：厨房的鲜味密码
  • 若依框架 ---一套快速开发平台
  • STM32-中断配置教程（寄存器版）
  • 【应急响应】进程隐藏技术与检测方式（二）
  • Gin 框架的中间件机制
  • 三种深度学习模型（GRU、CNN-GRU、贝叶斯优化的CNN-GRU/BO-CNN-GRU）对北半球光伏数据进行时间序列预测
  • win11 使用adb 获取安卓系统日志
  • ESP32学习笔记_Peripherals(4)——MCPWM基础使用
  • C++ : list的模拟
  • Kafka——多线程开发消费者实例
  • 使用OpenCV做个图片校正工具
  • 技术演进中的开发沉思-45 DELPHI VCL系列：6种方法
  • 关于新学C++编程Visual Studio 2022开始，使用Cmake工具构建Opencv和SDK在VS里编译项目开发简介笔记
  • RocketMQ常见问题梳理
  • 三、Spark 运行环境部署：全面掌握四种核心模式
  • 【内网穿透】使用FRP实现内网与公网Linux/Ubuntu服务器穿透项目部署多项目穿透方案
  • vue使用xlsx库导出excel
  • 编程语言Java——核心技术篇（三）异常处理详解
  • 字符串 “asdasjkfkasgfgshaahsfaf” 经过哈夫曼编码之后存储比特数是多少？
  • [实战] 用1 PPS 驯服本地恒温晶振（OCXO/TCXO）