什么是Reactor模式
Reactor模式是一种设计模式,它是基于事件驱动的,可以并发的处理多个服务请求,当请求抵达后,依据多路复用策略,同步的派发这些请求至相关的请求处理程序。
Reactor模式角色构成
在早先的论文An Object Behavioral Pattern for
Demultiplexing and Dispatching Handles for Synchronous Events中Reactor模式主要有五大角色组成,分别如下:
Handle:操作系统提供的一种资源,用于表示一个个的事件,在网络编程中可以是一个连接事件,一个读取事件,一个写入事件,Handle是事件产生的发源地
Synchronous Event Demultiplexer:本质上是一个系统调用,用于等待事件的发生,调用方在调用它的时候会被阻塞,一直阻塞到同步事件分离器上有事件产生为止
Initiation Dispatcher:定义了一些用于控制事件的调度方式的规范,提供对事件管理。它本身是整个事件处理器的核心所在,Initiation Dispatcher会通过Synchronous Event Demultiplexer来等待事件的发生。一旦事件发生,Initiation Dispatcher首先会分离出每一个事件,然后调用事件处理器,最后调用相关的回调方法来处理这些事件
Event Handler:定义事件处理方法以供InitiationDispatcher回调使用
Concrete Event Handler:是事件处理器的实现。它本身实现了事件处理器所提供的各种回调方法,从而实现了特定于业务的逻辑。它本质上就是我们所编写的一个个的处理器实现。
Reactor模式实现流程
- 初始化 Initiation Dispatcher,然后将若干个Concrete Event Handler注册到 Initiation Dispatcher中,应用会标识出该事件处理器希望Initiation Dispatcher在某些事件发生时向其发出通知
- Initiation Dispatcher 会要求每个事件处理器向其传递内部的Handle,该Handle向操作系统标识了事件处理器
- 当所有的Concrete Event Handler都注册完毕后,就会启动 Initiation Dispatcher的事件循环,使用Synchronous Event Demultiplexer同步阻塞的等待事件的发生
- 当与某个事件源对应的Handle变为ready状态时,Synchronous Event Demultiplexer就会通知 Initiation Dispatcher
- Initiation Dispatcher会触发事件处理器的回调方法响应这个事件
Java NIO对Reactor的实现
在Java的NIO中,对Reactor模式有无缝的支持,即使用Selector类封装了操作系统提供的Synchronous Event Demultiplexer功能。Doug Lea(Java concurrent包的作者)在Scalable IO in Java中对此有非常详细的描述。概况来说其主要流程如下:
- 服务器端的Reactor线程对象会启动事件循环,并使用Selector来实现IO的多路复用
- 注册Acceptor事件处理器到Reactor中,Acceptor事件处理器所关注的事件是ACCEPT事件,这样Reactor会监听客户端向服务器端发起的连接请求事件
- 客户端向服务器端发起连接请求,Reactor监听到了该ACCEPT事件的发生并将该ACCEPT事件派发给相应的Acceptor处理器来进行处理。Acceptor处理器通过accept()方法得到与这个客户端对应的连接(SocketChannel),然后将该连接所关注的READ/WRITE事件以及对应的READ/WRITE事件处理器注册到Reactor中,这样一来Reactor就会监听该连接的READ/WRITE事件了。
- 当Reactor监听到有读或者写事件发生时,将相关的事件派发给对应的处理器进行处理
- 每当处理完所有就绪的感兴趣的I/O事件后,Reactor线程会再次执行select()阻塞等待新的事件就绪并将其分派给对应处理器进行处理
Doug Lea 在Scalable IO in Java中分别描述了单线程的Reactor,多线程模式的Reactor以及多Reactor线程模式。
单线程的Reactor,主要依赖Java NIO中的Channel,Buffer,Selector,SelectionKey。在单线程Reactor模式中,不仅I/O操作在该Reactor线程上,连非I/O的业务操作也在该线程上进行处理了,这可能会大大延迟I/O请求的响应
在多线程Reactor中添加了一个工作线程池,将非I/O操作从Reactor线程中移出转交给工作者线程池来执行。这样能够提高Reactor线程的I/O响应,不至于因为一些耗时的业务逻辑而延迟对后面I/O请求的处理,但是所有的I/O操作依旧由一个Reactor来完成,包括I/O的accept()、read()、write()以及connect()操作
多Reactor线程模式将“接受客户端的连接请求”和“与该客户端的通信”分在了两个Reactor线程来完成。mainReactor完成接收客户端连接请求的操作,它不负责与客户端的通信,而是将建立好的连接转交给subReactor线程来完成与客户端的通信,这样一来就不会因为read()数据量太大而导致后面的客户端连接请求得不到即时处理的情况。并且多Reactor线程模式在海量的客户端并发请求的情况下,还可以通过实现subReactor线程池来将海量的连接分发给多个subReactor线程,在多核的操作系统中这能大大提升应用的负载和吞吐量
代码示例:
// NIO selector 多路复用reactor线程模型
public class NIOReactor {
// 处理业务操作的线程池
private static ExecutorService workPool = Executors.newCachedThreadPool();
// 封装了selector.select()等事件轮询的代码
abstract class ReactorThread extends Thread {
Selector selector;
LinkedBlockingQueue<Runnable> taskQueue = new LinkedBlockingQueue<>();
volatile boolean running = false;
private ReactorThread() throws IOException {
selector = Selector.open();
}
// Selector监听到有事件后,调用这个方法
public abstract void handler(SelectableChannel channel) throws Exception;
@Override
public void run() {
// 轮询Selector