一文详解Reactor模型与实现示例
作者：半夏之沫 发布时间：2023-11-13 12:22:09
前言
周六在公司写Reactor模型，
一女同事问我为啥都2023年了还在学习Reactor模型呀，
我问她为啥快30的年纪了，周六还在公司看我写Reactor呀，
一时间办公室里，男的，女的，都沉默了。
在网络IO设计中，有两种高性能模型：Reactor模型和Proactor模型。Reactor基于同步IO模式，Proactor基于异步IO模式。
Netty网络框架，Redis等中间件中都有使用到Reactor模型。本文将对Reactor模型的如下三种分类进行学习和实现。
单Reactor单线程模型；
单Reactor多线程模型；
主从Reactor多线程模型。
如果不具备网络IO的相关知识，建议先阅读Java网络IO模型分析与实现。
一. Reactor设计模式
Reactor翻译过来的意思是：反应堆，所以Reactor设计模式本质是基于事件驱动的。在Reactor设计模式中，存在如下几个角色。
Handle（事件）。Reactor整体是基于Handle进行驱动，这里的Handle叫做事件，可以类比为BIO中的Socket，NIO中的Socket管道。比如当Socket管道有连接建立，或者有数据可读，那么此时就称作事件发生；
EventHandler(事件处理器)。有事件发生，就需要有相应的组件来处理事件，那么这里的组件就叫做事件处理器。EventHandler是一个抽象概念，其会有不同的具体实现，因为事件会有不同的类型，那么不同类型的事件，肯定都需要有相应的具体处理逻辑，这里的具体处理逻辑，就由EventHandler的具体实现来承载；
Concrete Event Handler（具体事件处理器）。是EventHandler的具体实现，用于处理不同类型的事件；
Synchronous Event Demultiplexer（事件多路分解器）。（这里将Synchronous Event Demultiplexer简称为Demultiplexer）Demultiplexer用于监听事件并得到所有发生事件的集合，在监听的状态下是阻塞的，直到有事件发生为止。Demultiplexer有一个很好的类比，就是NIO中的多路复用器Selector，当调用Selector的select() 方法后，会进入监听状态，当从select() 方法返回时，会得到SelectionKey的一个集合，而每一个SelectionKey中就保存着有事件发生的Socket管道；
Initiation Dispatcher（事件分发器）。现在已经有Concrete Event Handler（具体事件处理器）来处理不同的事件，也能通过Synchronous Event Demultiplexer（事件多路分解器）拿到发生的事件，那么最后需要做的事情，肯定就是将事件分发到正确的事件处理器上进行处理，而Initiation Dispatcher就是完成这个分发的事情。
Reactor设计模式的一个简单类图，如下所示。
通常，Reactor设计模式中的Reactor，可以理解为上述图中的Synchronous Event Demultiplexer + Initiation Dispatcher。
二. 单Reactor单线程模型
单Reactor单线程模型中，只有一个Reactor在监听事件和分发事件，并且监听事件，分发事件和处理事件都在一个线程中完成。示意图如下所示。
上述示意图中，一次完整的处理流程可以概括如下。
Reactor监听到ACCEPT事件发生，表示此时有客户端建立连接；
Reactor将ACCEPT事件分发给Acceptor处理；
Acceptor会在服务端创建与客户端通信的client-socket管道，然后注册到IO多路复用器selector上，并监听READ事件；
Reactor监听到READ事件发生，表示此时客户端数据可读；
Reactor将ACCEPT事件分发给Handler处理，Handler处理READ事件就会基于client-socket管道完成客户端数据的读取。
下面将基于Java语言，实现一个简单的单Reactor单线程模型的服务端，整体代码实现完全符合上述示意图，大家可以进行参照阅读。
首先实现Reactor，如下所示。
public class Reactor implements Runnable {
private final Selector selector;
public Reactor(int port) throws IOException {
// 开启多路复用
selector = Selector.open();
// 服务端创建listen-socket管道
ServerSocketChannel listenSocketChannel = ServerSocketChannel.open();
// 绑定端口
listenSocketChannel.socket().bind(new InetSocketAddress(port));
// 设置为非阻塞模式
listenSocketChannel.configureBlocking(false);
// ACCEPT事件的附加器是Acceptor
listenSocketChannel.register(selector, SelectionKey.OP_ACCEPT,
new Acceptor(selector, listenSocketChannel));
}
@Override
public void run() {
while (!Thread.interrupted()) {
try {
// 获取发生的事件
selector.select();
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterable = selectionKeys.iterator();
while (iterable.hasNext()) {
// 对事件进行分发
dispatch(iterable.next());
iterable.remove();
}
} catch (IOException e) {
e.printStackTrace();
}
LockSupport.parkNanos(1000 * 1000 * 1000);
}
}
private void dispatch(SelectionKey selectionKey) {
// 获取事件的附加器
// ACCEPT事件的附加器是Acceptor，故由Acceptor来处理ACCEPT事件
// READ事件的附加器是Handler，故由Handler来处理READ事件
Runnable attachment = (Runnable) selectionKey.attachment();
if (attachment != null) {
attachment.run();
}
}
}
已知Reactor会监听客户端连接的ACCEPT事件，还已知ACCEPT事件由Acceptor处理，所以在向多路复用器注册服务端用于监听客户端连接的listen-socket管道时，添加了一个Acceptor作为附加器，那么当发生ACCEPT事件时，就能够获取到作为ACCEPT事件附加器的Acceptor来处理ACCEPT事件。
下面看一下Acceptor的实现，如下所示。
public class Acceptor implements Runnable {
private final Selector selector;
private final ServerSocketChannel listenSocketChannel;
public Acceptor(Selector selector, ServerSocketChannel listenSocketChannel) {
this.selector = selector;
this.listenSocketChannel = listenSocketChannel;
}
@Override
public void run() {
try {
// 为连接的客户端创建client-socket管道
SocketChannel clientSocketChannel = listenSocketChannel.accept();
// 设置为非阻塞
clientSocketChannel.configureBlocking(false);
// READ事件的附加器是Handler
clientSocketChannel.register(selector, SelectionKey.OP_READ,
new Handler(clientSocketChannel));
} catch (IOException e) {
e.printStackTrace();
}
}
}
在Acceptor中就是在服务端创建与客户端通信的client-socket管道，然后注册到多路复用器上并指定监听READ事件，同时又因为READ事件由Handler处理，所以还添加了一个Handler作为附加器，当READ事件发生时可以获取到作为READ事件附加器的Handler来处理READ事件。
下面看一下Handler的实现，如下所示。
public class Handler implements Runnable {
private final SocketChannel clientSocketChannel;
public Handler(SocketChannel clientSocketChannel) {
this.clientSocketChannel = clientSocketChannel;
}
@Override
public void run() {
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
try {
// 读取数据
int read = clientSocketChannel.read(byteBuffer);
if (read <= 0) {
clientSocketChannel.close();
} else {
System.out.println(new String(byteBuffer.array()));
}
} catch (IOException e1) {
try {
clientSocketChannel.close();
} catch (IOException e2) {
e2.printStackTrace();
}
e1.printStackTrace();
}
}
}
在Handler中就是简单的读取数据并打印，当读取数据为空或者发生异常时，需要及时将管道关闭。
最后编写一个主程序将Reactor运行起来，如下所示。
public class MainServer {
public static void main(String[] args) throws IOException {
Thread reactorThread = new Thread(new Reactor(8080));
reactorThread.start();
}
}
现在来思考一下，单Reactor单线程模型有什么优点和缺点。优点其实就是模型简单，实现方便。缺点有两点，如下所示。
一个Reactor同时负责监听ACCEPT事件和READ事件；
只有一个线程在工作，处理效率低，无法利用多核CPU的优势。
但是尽管单Reactor单线程模型有上述的缺点，但是著名的缓存中间件Redis的服务端，就是使用的单Reactor单线程模型，示意图如下。
那为什么以性能著称的Redis会采取单Reactor单线程模型呢，其实就是因为Redis的操作都在内存中，读写都非常快速，所以单Reactor单线程模型也能运行得很流畅，同时还避免了多线程下的各种并发问题。
三. 单Reactor多线程模型
在理解了单Reactor单线程模型后，那么肯定就能想到，假如在Handler中处理READ事件的这个事情能够使用一个线程池来完成，从而就可以实现READ事件的处理不会阻塞主线程。而这样的一个模型，其实就是单Reactor多线程模型，示意图如下所示。
和单Reactor单线程模型唯一的不同，就是在Handler中多了一个线程池。
单Reactor多线程模型的代码实现，除了Handler以外，其余和单Reactor单线程模型一摸一样，所以下面就看一下单Reactor多线程模型中的Handler实现，如下所示。
public class Handler implements Runnable {
private static final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(16, 32,
60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(200));
private final SocketChannel clientSocketChannel;
public Handler(SocketChannel clientSocketChannel) {
this.clientSocketChannel = clientSocketChannel;
}
@Override
public void run() {
threadPool.execute(() -> {
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
try {
// 读取数据
int read = clientSocketChannel.read(byteBuffer);
if (read <= 0) {
clientSocketChannel.close();
} else {
System.out.println(new String(byteBuffer.array()));
}
// 睡眠10S，演示任务执行耗时长也不会阻塞处理其它客户端请求
LockSupport.parkNanos(1000 * 1000 * 1000 * 10L);
} catch (IOException e1) {
try {
clientSocketChannel.close();
} catch (IOException e2) {
e2.printStackTrace();
}
e1.printStackTrace();
}
});
}
}
其实就是每一个READ事件的处理会作为一个任务被扔到线程池中去处理。
单Reactor多线程模型虽然解决了只有一个线程的问题，但是可以发现，仍旧是只有一个Reactor在同时监听ACCEPT事件和READ事件。
那么现在思考一下，为什么一个Reactor同时监听ACCEPT事件和READ事件是不好的。其实就是因为通常客户端连接的建立是不频繁的，但是连接建立后数据的收发是频繁的，所以如果能够将监听READ事件这个动作拆分出来，让多个子Reactor来监听READ事件，而原来的主Reactor只监听ACCEPT事件，那么整体的效率，会进一步提升，而这，就是主从Reactor多线程模型。
四. 主从Reactor多线程模型
主从Reactor模型中，有一个主Reactor，专门监听ACCEPT事件，然后有多个从Reactor，专门监听READ事件，示意图如下所示。
上述示意图中，一次完整的处理流程可以概括如下。
主Reactor监听到ACCEPT事件发生，表示此时有客户端建立连接；
主Reactor将ACCEPT事件分发给Acceptor处理；
Acceptor会在服务端创建与客户端通信的client-socket管道，然后注册到从Reactor的IO多路复用器selector上，并监听READ事件；
从Reactor监听到READ事件发生，表示此时客户端数据可读；
从Reactor将ACCEPT事件分发给Handler处理，Handler处理READ事件就会基于client-socket管道完成客户端数据的读取。
下面将基于Java语言，实现一个简单的主从Reactor多线程模型的服务端，整体代码实现完全符合上述示意图，大家可以进行参照阅读。
首先是主Reactor的实现，如下所示。
public class MainReactor implements Runnable {
private final Selector selector;
public MainReactor(int port) throws IOException {
// 开多路复用器
selector = Selector.open();
// 服务端创建listen-socket管道
ServerSocketChannel listenSocketChannel = ServerSocketChannel.open();
// 设置为非阻塞
listenSocketChannel.configureBlocking(false);
// 绑定监听端口
listenSocketChannel.socket().bind(new InetSocketAddress(port));
// 将listen-socket管道绑定到主Reactor的多路复用器上
// 并且主Reactor上只会注册listen-socket管道，用于监听ACCEPT事件
listenSocketChannel.register(selector, SelectionKey.OP_ACCEPT,
new Acceptor(listenSocketChannel));
}
@Override
public void run() {
while (!Thread.interrupted()) {
try {
selector.select();
Set<SelectionKey> selectionKeys = selector.selectedKeys();
Iterator<SelectionKey> iterable = selectionKeys.iterator();
while (iterable.hasNext()) {
// 对事件进行分发
dispatch(iterable.next());
iterable.remove();
}
} catch (IOException e) {
e.printStackTrace();
}
LockSupport.parkNanos(1000 * 1000 * 1000);
}
}
private void dispatch(SelectionKey selectionKey) {
// 获取事件附加器，只会是Acceptor
Runnable attachment = (Runnable) selectionKey.attachment();
if (attachment != null) {
attachment.run();
}
}
}
主Reactor的实现中，还是先创建服务端监听客户端连接的listen-socket管道，然后注册到主Reactor的IO多路复用器上，并监听ACCEPT事件，同时我们现在知道，主Reactor的IO多路复用器上只会注册listen-socket管道且只会监听ACCEPT事件。同样，也添加了一个Acceptor作为附加器，那么当发生ACCEPT事件时，就能够获取到作为ACCEPT事件附加器的Acceptor来处理ACCEPT事件。
下面是Acceptor的实现，如下所示。
public class Acceptor implements Runnable {
// 指定从Reactor一共有16个
private static final int TOTAL_SUBREACTOR_NUM = 16;
// 服务端的listen-socket管道
private final ServerSocketChannel listenSocketChannel;
// 用于运行从Reactor
private final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(
TOTAL_SUBREACTOR_NUM, TOTAL_SUBREACTOR_NUM * 2,
60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(200));
// 从Reactor集合
private final List<SubReactor> subReactors = new ArrayList<>(TOTAL_SUBREACTOR_NUM);
public Acceptor(ServerSocketChannel listenSocketChannel) throws IOException {
this.listenSocketChannel = listenSocketChannel;
// 将从Reactor初始化出来并运行
for (int i = 0; i < TOTAL_SUBREACTOR_NUM; i++) {
SubReactor subReactor = new SubReactor(Selector.open());
subReactors.add(subReactor);
threadPool.execute(subReactor);
}
}
@Override
public void run() {
try {
// 为连接的客户端创建client-socket管道
SocketChannel clientSocketChannel = listenSocketChannel.accept();
// 设置为非阻塞
clientSocketChannel.configureBlocking(false);
// 任意选择一个从Reactor，让其监听连接的客户端的READ事件
Optional<SubReactor> anySubReactor = subReactors.stream().findAny();
if (anySubReactor.isPresent()) {
SubReactor subReactor = anySubReactor.get();
// 从Reactor的多路复用器会阻塞在select()方法上
// 这里需要先唤醒多路复用器，立即从select()方法返回
subReactor.getSelector().wakeup();
// 让从Reactor负责处理客户端的READ事件
clientSocketChannel.register(subReactor.getSelector(), SelectionKey.OP_READ,
new Handler(clientSocketChannel));
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
首先在Acceptor的构造函数中，会将所有从Reactor初始化出来，并且每一个从Reactor都会持有一个IO多路复用器。当一个从Reactor创建出来后就会立即运行，此时从Reactor的IO多路复用器就会开始监听，即阻塞在select() 方法上。
然后在Acceptor的主体逻辑中，会为连接的客户端创建client-socket管道，然后从所有从Reactor中基于某种策略（随机）选择一个从Reactor，并将client-socket管道注册在选择的从Reactor的IO多路复用器上，有一点需要注意，此时从Reactor的IO多路复用器可能会阻塞在select() 方法上，所以注册前需要先通过wakeup() 方法进行唤醒。
接下来继续看从Reactor的实现，如下所示。
public class SubReactor implements Runnable {
private final Selector selector;
public SubReactor(Selector selector) {
this.selector = selector;
}
@Override
public 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) {
e.printStackTrace();
}
LockSupport.parkNanos(1000 * 1000 * 1000);
}
}
private void dispatch(SelectionKey selectionKey) {
// 获取事件附加器，只会是Handler
Runnable runnable = (Runnable) selectionKey.attachment();
if (runnable != null) {
runnable.run();
}
}
public Selector getSelector() {
return selector;
}
}
从Reactor的实现中，会监听服务端为连接的客户端创建的client-socket管道上的READ事件，一旦有READ事件发生，就会使用作为附加器的Handler来处理READ事件。同样，从Reactor的IO多路复用器上只会注册client-socket管道且只会监听READ事件。
然后是Handler，因为是多线程模型，所以其实现和第三节中的Handler完全一样，下面再贴一下代码。
public class Handler implements Runnable {
private static final ThreadPoolExecutor threadPool = new ThreadPoolExecutor(16, 32,
60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(200));
private final SocketChannel clientSocketChannel;
public Handler(SocketChannel clientSocketChannel) {
this.clientSocketChannel = clientSocketChannel;
}
@Override
public void run() {
threadPool.execute(() -> {
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
try {
// 读取数据
int read = clientSocketChannel.read(byteBuffer);
if (read <= 0) {
clientSocketChannel.close();
} else {
System.out.println(new String(byteBuffer.array()));
}
// 睡眠10S，演示任务执行耗时长也不会阻塞处理其它客户端请求
LockSupport.parkNanos(1000 * 1000 * 1000 * 10L);
} catch (IOException e1) {
try {
clientSocketChannel.close();
} catch (IOException e2) {
e2.printStackTrace();
}
e1.printStackTrace();
}
});
}
}
最后编写一个主程序将主Reactor运行起来，如下所示。
public class MainServer {
public static void main(String[] args) throws IOException {
Thread mainReactorThread = new Thread(new MainReactor(8080));
mainReactorThread.start();
}
}
来源：https://juejin.cn/post/7210375522512666679
0
投稿猜你喜欢
需求某航空公司物流单信息查询，是一个post请求。通过后台模拟POST HTTP请求发现无法获取页面数据，通过查看航空公司网站后，发现网站使- MyBatis的前身叫iBatis，本是apache的一个开源项目, 2010年这个项目由apache software foundatio
- 实例如下所示：public class MainActivity {private static final String fileName
- 本文实例为大家分享了C#简单聊天室雏形的具体代码，供大家参考，具体内容如下程序使用的控制台的黑窗口模拟程序，第一次涉及网络编程，写出来方便以
- 本文实例讲述了C#采用OpenXml给Word文档添加表格的方法，是非常实用的操作技巧。分享给大家供大家参考。具体分析如下：这里将展示如何使
- @ConfigurationProperties注入创建一个新的模板此过程就不在这介绍了，在我SpringBoot专栏里有详细过程。⭐⭐⭐注
- 多继承指一个子类能同时继承于多个父类，从而同时拥有多个父类的特征，但缺点是显著的。1.若子类继承的父类中拥有相同的成员变量，子类在引用该变量
- HashMap 在不同的 JDK 版本下的实现是不同的，在 JDK 1.7 时，HashMap 底层是通过数组 + 链表实现的；而在 JDK
- 目前开发系统，附件文件一般都会使用第三方的存储空间来保存，一方面是为了开发者提供便利，另一方可以减轻系统的访问压力，下面介绍一下阿里云的OS
- 首先看一看什么是装箱和拆箱？简单的来说：装箱就是值类型转换为引用类型；拆箱就是引用类型转换为值类型。值类型,包括原类型（Sbyte、Byte
- 背景我在准备使用 JVM 的命令时候观察程序的动态，但是发现 Main 函数启动就退出了，所以也没办法直接观察，于是想到了如何让 Main
- 本文实例讲述了Java中的 * 、过滤器、 * 用法。分享给大家供大家参考，具体如下：一、 * ：是在面向切面编程的就是在你的servic
- 类注解@component 标注类，泛指各种组件，类不属于各种分类的时候，用它做标注。@Service 标注类，声明该类为业务层组件，用于处
- 一、为什么要编码不知道大家有没有想过一个问题，那就是为什么要编码？我们能不能不编码？要回答这个问题必须要回到计算机是如何表示我们人类能够理解
- 一、Hadoop的安装1. 下载地址：https://archive.apache.org/dist/hadoop/common/我下载的是
- 本文介绍在使用C#开发WinForm窗体程序时，如何设置窗体的大小不能被改变。我们在开发一个窗体（WinForm）程序时，所有的控件都部署在
- 我们日常的工作中都使用开发工具（IntelliJ IDEA 或 Eclipse 等）可以很方便的调试程序，或者是通过打包工具把项目打包成 j
- 本文实例讲述了java线程同步操作。分享给大家供大家参考，具体如下：java线程同步public class Hello { p
- Map的存储结构式Key/Value形式,Key 和 Value可以是普通类型,也可以是自己写的JavaBean(本文),还可以是带有泛型的
- 迪杰斯特拉算法迪杰斯特拉算法是由荷兰计算机科学家狄克斯特拉于1959 年提出的，因此又叫狄克斯特拉算法。是从一个顶点到其余各顶点的最短路径算
