我要评论前言
周六在公司写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();
}
}
总结
reactor模型主要就是监听事件，分发事件和处理事件。其中reactor角色会负责监听事件 和分发事件，handler角色和acceptor角色会负责处理事件。
reactor模型虽然分为：单reactor单线程模型，单reactor多线程模型和主从reactor多线程模型，但是其本质就是nio的实现，是不过套了reactor设计模式的外壳。
在网络通信框架netty中，三种reactor模型都有使用到，所以想要学习netty的精髓，理解reactor模型是必不可少的。
以上就是一文详解reactor模型与实现示例的详细内容，更多关于reactor模型实现的资料请关注七九推其它相关文章！
微信扫一扫 
发表评论