一文详解Reactor模型与实现示例
作者:半夏之沫
前言
周六在公司写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模型实现的资料请关注脚本之家其它相关文章!