单线程Reactor反应器模式
在事件驱动模式中,当有事件触发时,事件源会将事件dispatch分发到handler处理器进行事件处理。反应器模式中的反应器角色,类似于事件驱动模式中的dispatcher事件分发器角色。
在反应器模式中,有Reactor反应器和Handler处理器两个重要的组件:
(1)Reactor反应器:负责查询IO事件,当检测到一个IO事件,将其发送给相应的Handler处理器去处理。这里的IO事件,就是NIO中选择器监控的通道IO事件。(2)Handler处理器:与IO事件(或者选择键)绑定,负责IO事件的处理。完成真正的连接建立、通道的读取、处理业务逻辑、负责将结果写出到通道等。
2.1 什么是单线程Reactor反应器
什么是单线程版本的Reactor反应器模式呢?简单地说,Reactor反应器和Handers处理器处于一个线程中执行。它是最简单的反应器模型,如图4-1所示。
基于Java NIO,如何实现简单的单线程版本的反应器模式呢?需要用到SelectionKey选择键的几个重要的成员方法:
方法一:void attach(Object o)
此方法可以将任何的Java POJO对象,作为附件添加到SelectionKey实例,相当于附件属性的setter方法。这方法非常重要,因为在单线程版本的反应器模式中,需要将Handler处理器实例,作为附件添加到SelectionKey实例。
方法二:Object attachment()
此方法的作用是取出之前通过attach(Object o)添加到SelectionKey选择键实例的附件,相当于附件属性的getter方法,与attach(Object o)配套使用。
这个方法同样非常重要,当IO事件发生,选择键被select方法选到,可以直接将事件的附件取出,也就是之前绑定的Handler处理器实例,通过该Handler,完成相应的处理。
总之,在反应器模式中,需要进行attach和attachment结合使用:在选择键注册完成之后,调用attach方法,将Handler处理器绑定到选择键;当事件发生时,调用attachment方法,可以从选择键取出Handler处理器,将事件分发到Handler处理器中,完成业务处理。
2.2 单线程Reactor反应器的参考代码
Doug Lea在《Scalable IO in Java》中,实现了一个单线程Reactor反应器模式的参考代码。这里,我们站在巨人的肩膀上,借鉴Doug Lea的实现,对其进行介绍。为了方便说明,对Doug Lea的参考代码进行一些适当的修改。具体的参考代码如下:
package com.crazymakercircle.ReactorModel;
//...
class Reactor implements Runnable {
Selector selector;
ServerSocketChannelserverSocket;
EchoServerReactor() throws IOException {
//....省略:打开选择器、serverSocket连接监听通道
//注册serverSocket的accept事件
SelectionKeysk =
serverSocket.register(selector, SelectionKey.OP_ACCEPT);
//将新连接处理器作为附件,绑定到sk选择键
sk.attach(new AcceptorHandler());
}
public void run() {
//选择器轮询
try {
while (! Thread.interrupted()) {
selector.select();
Set selected = selector.selectedKeys();
Iterator it = selected.iterator();
while (it.hasNext()) {
//反应器负责dispatch收到的事件
SelectionKeysk=it.next();
dispatch(sk);
}
selected.clear();
}
} catch (IOException ex) { ex.printStackTrace(); }
}
//反应器的分发方法
void dispatch(SelectionKey k) {
Runnable handler = (Runnable) (k.attachment());
//调用之前绑定到选择键的handler处理器对象
if (handler ! = null) {
handler.run();
}
}
// 新连接处理器
class AcceptorHandler implements Runnable {
public void run() {
//接受新连接
//需要为新连接,创建一个输入输出的handler处理器
}
}
//….
}
在上面的代码中,设计了一个Handler处理器,叫作AcceptorHandler处理器,它是一个内部类。在注册serverSocket服务监听连接的接受事件之后,创建一个AcceptorHandler新连接处理器的实例,作为附件,被设置(attach)到了SelectionKey中。
//注册serverSocket的接受(accept)事件
SelectionKeysk =
serverSocket.register(selector, SelectionKey.OP_ACCEPT);
//将新连接处理器作为附件,绑定到sk选择键
sk.attach(new AcceptorHandler());
当新连接事件发生后,取出了之前attach到SelectionKey中的Handler业务处理器,进行socket的各种IO处理。
void dispatch(SelectionKey k) {
Runnable r = (Runnable) (k.attachment());
//调用之前绑定到选择键的处理器对象
if (r ! = null) {
r.run();
}
}
AcceptorHandler处理器的两大职责:一是接受新连接,二是在为新连接创建一个输入输出的Handler处理器,称之为IOHandler。
// 新连接处理器
class AcceptorHandler implements Runnable {
public void run() {
// 接受新连接
// 需要为新连接创建一个输入输出的handler处理器
}
}
IOHandler,顾名思义,就是负责socket的数据输入、业务处理、结果输出。示例代码如下:
package com.crazymakercircle.ReactorModel;
//...
class IOHandler implements Runnable {
final SocketChannel channel;
final SelectionKeysk;
IOHandler (Selector selector, SocketChannel c) throws IOException {
channel = c;
c.configureBlocking(false);
//仅仅取得选择键,稍候设置感兴趣的IO事件
sk = channel.register(selector, 0);
//将Handler处理器作为选择键的附件
sk.attach(this);
//注册读写就绪事件
sk.interestOps(SelectionKey.OP_READ|SelectionKey.OP_WRITE);
}
public void run() {
//...处理输入和输出
}
}
在IOHandler的构造器中,有两点比较重要:
(1)将新的SocketChannel传输通道,注册到了反应器Reactor类的同一个选择器中。这样保证了Reactor类和Handler类在同一个线程中执行。
(2)Channel传输通道注册完成后,将IOHandler自身作为附件,attach到了选择键中。这样,在Reactor类分发事件(选择键)时,能执行到IOHandler的run方法。
如果上面的示例代码比较绕口,不要紧。为了彻底地理解个中妙处,自己动手开发一个可以执行的实例。下面基于反应器模式,实现了一个EchoServer回显服务器实例。仔细阅读和运行这个实例,就可以明白上面这段绕口的程序代码的真正含义了。
3 一个Reactor反应器版本的EchoServer实践案例
EchoServer回显服务器的功能很简单:读取客户端的输入,回显到客户端,所以也叫回显服务器。基于Reactor反应器模式来实现,设计3个重要的类:
(1)设计一个反应器类:EchoServerReactor类。
(2)设计两个处理器类:AcceptorHandler新连接处理器、EchoHandler回显处理器。
反应器类EchoServerReactor的实现思路和前面的示例代码基本上相同,具体如下:
package com.crazymakercircle.ReactorModel;
//.....
//反应器
class EchoServerReactor implements Runnable {
Selector selector;
ServerSocketChannelserverSocket;
EchoServerReactor() throws IOException {
//...获取选择器、开启serverSocket服务监听通道
//...绑定AcceptorHandler新连接处理器到selectKey
}
//轮询和分发事件
public void run() {
try {
while (! Thread.interrupted()) {
selector.select();
Set<SelectionKey> selected = selector.selectedKeys();
Iterator<SelectionKey> it = selected.iterator();
while (it.hasNext()) {
//反应器负责dispatch收到的事件
SelectionKey sk = it.next();
dispatch(sk);
}
selected.clear();
}
} catch (IOException ex) {
ex.printStackTrace();
}
}
void dispatch(SelectionKeysk) {
Runnable handler = (Runnable) sk.attachment();
//调用之前attach绑定到选择键的handler处理器对象
if (handler ! = null) {
handler.run();
}
}
// Handler:新连接处理器
class AcceptorHandler implements Runnable {
public void run() {
try {
SocketChannel channel = serverSocket.accept();
if (channel ! = null)
new EchoHandler(selector, channel);
} catch (IOException e) {
e.printStackTrace();
}
}
}
public static void main(String[] args) throws IOException {
new Thread(new EchoServerReactor()).start();
}
}
EchoHandler回显处理器,主要是完成客户端的内容读取和回显,具体如下:
import com.crazymakercircle.util.Logger;
//...
class EchoHandler implements Runnable {
final SocketChannel channel;
final SelectionKeysk;
final ByteBufferbyteBuffer = ByteBuffer.allocate(1024);
static final int RECIEVING = 0, SENDING = 1;
int state = RECIEVING;
EchoHandler(Selector selector, SocketChannel c) throws IOException {
channel = c;
c.configureBlocking(false);
//取得选择键,再设置感兴趣的IO事件
sk = channel.register(selector, 0);
//将Handler自身作为选择键的附件
sk.attach(this);
//注册Read就绪事件
sk.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
public void run() {
try {
if (state == SENDING) {
//写入通道
channel.write(byteBuffer);
//写完后,准备开始从通道读,byteBuffer切换成写入模式
byteBuffer.clear();
//写完后,注册read就绪事件
sk.interestOps(SelectionKey.OP_READ);
//写完后,进入接收的状态
state = RECIEVING;
} else if (state == RECIEVING) {
//从通道读
int length = 0;
while ((length = channel.read(byteBuffer)) > 0) {
Logger.info(new String(byteBuffer.array(), 0, length));
}
//读完后,准备开始写入通道,byteBuffer切换成读取模式
byteBuffer.flip();
//读完后,注册write就绪事件
sk.interestOps(SelectionKey.OP_WRITE);
//读完后,进入发送的状态
state = SENDING;
}
//处理结束了,这里不能关闭select key,需要重复使用
//sk.cancel();
} catch (IOException ex) {
ex.printStackTrace();
}
}
}
以上两个类,是一个基于反应器模式的EchoServer回显服务器的完整实现。它是一个单线程版本的反应器模式,Reactor反应器和所有的Handler处理器,都执行在同一条线程中。
运行EchoServerReactor类中的main方法,可以启动回显服务器。如果要看到回显输出,还需要启动客户端。客户端的代码,在同一个包下,类名为EchoClient,负责数据的发送。打开源代码工程,直接运行即可。由于篇幅原因,这里不再贴出客户端的代码。
4 单线程Reactor反应器模式的缺点
单线程Reactor反应器模式,是基于Java的NIO实现的。相对于传统的多线程OIO,反应器模式不再需要启动成千上万条线程,效率自然是大大提升了。
在单线程反应器模式中,Reactor反应器和Handler处理器,都执行在同一条线程上。这样,带来了一个问题:当其中某个Handler阻塞时,会导致其他所有的Handler都得不到执行。在这种场景下,如果被阻塞的Handler不仅仅负责输入和输出处理的业务,还包括负责连接监听的AcceptorHandler处理器。这个是非常严重的问题。为什么?一旦AcceptorHandler处理器阻塞,会导致整个服务不能接收新的连接,使得服务器变得不可用。因为这个缺陷,因此单线程反应器模型用得比较少。
另外,目前的服务器都是多核的,单线程反应器模式模型不能充分利用多核资源。总之,在高性能服务器应用场景中,单线程反应器模式实际使用的很少。
多线程的Reactor反应器模式
既然Reactor反应器和Handler处理器,挤在一个线程会造成非常严重的性能缺陷。那么,可以使用多线程,对基础的反应器模式进行改造和演进。
3.1 多线程池Reactor反应器演进
多线程池Reactor反应器的演进,分为两个方面:
(1)首先是升级Handler处理器。既要使用多线程,又要尽可能的高效率,则可以考虑使用线程池。
(2)其次是升级Reactor反应器。可以考虑引入多个Selector选择器,提升选择大量通道的能力。
总体来说,多线程池反应器的模式,大致如下:
(1)将负责输入输出处理的IOHandler处理器的执行,放入独立的线程池中。这样,业务处理线程与负责服务监听和IO事件查询的反应器线程相隔离,避免服务器的连接监听受到阻塞。
(2)如果服务器为多核的CPU,可以将反应器线程拆分为多个子反应器(SubReactor)线程;同时,引入多个选择器,每一个SubReactor子线程负责一个选择器。这样,充分释放了系统资源的能力;也提高了反应器管理大量连接,提升选择大量通道的能力。
3.2 多线程Reactor反应器的实践案例
在前面的“回显服务器”(EchoServer)的基础上,完成多线程Reactor反应器的升级。多线程反应器的实践案例设计如下:
(1)引入多个选择器。
(2)设计一个新的子反应器(SubReactor)类,一个子反应器负责查询一个选择器。(3)开启多个反应器的处理线程,一个线程负责执行一个子反应器(SubReactor)。
为了提升效率,建议SubReactor的数量和选择器的数量一致。避免多个线程负责一个选择器,导致需要进行线程同步,引起的效率降低。这个实践案例的代码如下:
package com.crazymakercircle.ReactorModel;
//....反应器
class MultiThreadEchoServerReactor {
ServerSocketChannelserverSocket;
AtomicInteger next = new AtomicInteger(0);
//选择器集合,引入多个选择器
Selector[] selectors = new Selector[2];
//引入多个子反应器
SubReactor[] subReactors = null;
MultiThreadEchoServerReactor() throws IOException {
//初始化多个选择器
selectors[0] = Selector.open();
selectors[1] = Selector.open();
serverSocket = ServerSocketChannel.open();
InetSocketAddress address =
new InetSocketAddress(NioDemoConfig.SOCKET_SERVER_IP,
NioDemoConfig.SOCKET_SERVER_PORT);
serverSocket.socket().bind(address);
//非阻塞
serverSocket.configureBlocking(false);
//第一个选择器,负责监控新连接事件
SelectionKeysk =
serverSocket.register(selectors[0], SelectionKey.OP_ACCEPT);
//绑定Handler:attach新连接监控handler处理器到SelectionKey(选择键)
sk.attach(new AcceptorHandler());
//第一个子反应器,一子反应器负责一个选择器
SubReactor subReactor1 = new SubReactor(selectors[0]);
//第二个子反应器,一子反应器负责一个选择器
SubReactor subReactor2 = new SubReactor(selectors[1]);
subReactors = new SubReactor[]{subReactor1, subReactor2};
}
private void startService() {
// 一子反应器对应一个线程
new Thread(subReactors[0]).start();
new Thread(subReactors[1]).start();
}
//子反应器
class SubReactor implements Runnable {
//每个线程负责一个选择器的查询和选择
final Selector selector;
public SubReactor(Selector selector) {
this.selector = selector;
}
public void run() {
try {
while (! Thread.interrupted()) {
selector.select();
Set<SelectionKey>keySet = selector.selectedKeys();
Iterator<SelectionKey> it = keySet.iterator();
while (it.hasNext()) {
//反应器负责dispatch收到的事件
SelectionKeysk = it.next();
dispatch(sk);
}
keySet.clear();
}
} catch (IOException ex) {
ex.printStackTrace();
}
}
void dispatch(SelectionKeysk) {
Runnable handler = (Runnable) sk.attachment();
//调用之前attach绑定到选择键的handler处理器对象
if (handler ! = null) {
handler.run();
}
}
}
// Handler:新连接处理器
class AcceptorHandler implements Runnable {
public void run() {
try {
SocketChannel channel = serverSocket.accept();
if (channel ! = null)
new MultiThreadEchoHandler(selectors[next.get()], channel);
} catch (IOException e) {
e.printStackTrace();
}
if (next.incrementAndGet() == selectors.length) {
next.set(0);
}
}
}
public static void main(String[] args) throws IOException {
MultiThreadEchoServerReactor server =
new MultiThreadEchoServerReactor();
server.startService();
}
}
上面是反应器的演进代码,再来看看Handler处理器的多线程演进实践。
3.3 多线程Handler处理器的实践案例
基于前面的单线程反应器的EchoHandler回显处理器的程序代码,予以改进,新的回显处理器为:MultiThreadEchoHandler。主要的升级是引入了一个线程池(ThreadPool),业务处理的代码执行在自己的线程池中,彻底地做到业务处理线程和反应器IO事件线程的完全隔离。这个实践案例的代码如下:
package com.crazymakercircle.ReactorModel;
//...
class MultiThreadEchoHandler implements Runnable {
final SocketChannel channel;
final SelectionKeysk;
final ByteBufferbyteBuffer = ByteBuffer.allocate(1024);
static final int RECIEVING = 0, SENDING = 1;
int state = RECIEVING;
//引入线程池
static ExecutorService pool = Executors.newFixedThreadPool(4);
MultiThreadEchoHandler(Selector selector, SocketChannel c) throws
IOException {
channel = c;
c.configureBlocking(false);
//取得选择键,、再设置感兴趣的IO事件
sk = channel.register(selector, 0);
//将本Handler作为sk选择键的附件,方便事件分发(dispatch)
sk.attach(this);
//向sk选择键注册Read就绪事件
sk.interestOps(SelectionKey.OP_READ);
selector.wakeup();
}
public void run() {
//异步任务,在独立的线程池中执行
pool.execute(new AsyncTask());
}
//业务处理,不在反应器线程中执行
public synchronized void asyncRun() {
try {
if (state == SENDING) {
//写入通道
channel.write(byteBuffer);
//写完后,准备开始从通道读,byteBuffer切换成写入模式
byteBuffer.clear();
//写完后,注册read就绪事件
sk.interestOps(SelectionKey.OP_READ);
//写完后,进入接收的状态
state = RECIEVING;
} else if (state == RECIEVING) {
//从通道读
int length = 0;
while ((length = channel.read(byteBuffer)) > 0) {
Logger.info(new String(byteBuffer.array(), 0, length));
}
//读完后,准备开始写入通道,byteBuffer切换成读取模式
byteBuffer.flip();
//读完后,注册write就绪事件
sk.interestOps(SelectionKey.OP_WRITE);
//读完后,进入发送的状态
state = SENDING;
}
//处理结束了,这里不能关闭select key,需要重复使用
//sk.cancel();
} catch (IOException ex) {
ex.printStackTrace();
}
}
//异步任务的内部类
class AsyncTask implements Runnable {
public void run() {
MultiThreadEchoHandler.this.asyncRun();
}
}
}
代码中设计了一个内部类AsyncTask,是一个简单的异步任务的提交类。它使得异步业务asyncRun方法,可以独立地提交到线程池中。另外,既然业务处理异步执行,需要在asyncRun方法的前面加上synchronized同步修饰符。
至此,多线程版本的反应器模式,实践案例的代码就演示完了。执行新版本的多线程MultiThreadEchoServerReactor服务器,可以使用之前的EchoClient客户端与之配置,完成整个回显(echo)的通信演示。
演示的输出和之前单线程版本的EchoServer回显服务器示例,是一模一样的。
反应器模式和生产者消费者模式对比
相似之处:在一定程度上,反应器模式有点类似生产者消费者模式。在生产者消费者模式中,一个或多个生产者将事件加入到一个队列中,一个或多个消费者主动地从这个队列中提取(Pull)事件来处理。
不同之处在于:反应器模式是基于查询的,没有专门的队列去缓冲存储IO事件,查询到IO事件之后,反应器会根据不同IO选择键(事件)将其分发给对应的Handler处理器来处理。
反应器模式和观察者模式(Observer Pattern)对比
相似之处在于:在反应器模式中,当查询到IO事件后,服务处理程序使用单路/多路分发(Dispatch)策略,同步地分发这些IO事件。观察者模式(Observer Pattern)也被称作发布/订阅模式,它定义了一种依赖关系,让多个观察者同时监听某一个主题(Topic)。这个主题对象在状态发生变化时,会通知所有观察者,它们能够执行相应的处理。
不同之处在于:在反应器模式中,Handler处理器实例和IO事件(选择键)的订阅关系,基本上是一个事件绑定到一个Handler处理器;每一个IO事件(选择键)被查询后,反应器会将事件分发给所绑定的Handler处理器;而在观察者模式中,同一个时刻,同一个主题可以被订阅过的多个观察者处理。
最后,总结一下反应器模式的优点和缺点。作为高性能的IO模式,反应器模式的优点如下:
· 响应快,虽然同一反应器线程本身是同步的,但不会被单个连接的同步IO所阻塞;· 编程相对简单,最大程度避免了复杂的多线程同步,也避免了多线程的各个进程之间切换的开销;
· 可扩展,可以方便地通过增加反应器线程的个数来充分利用CPU资源。反应器模式的缺点如下:
· 反应器模式增加了一定的复杂性,因而有一定的门槛,并且不易于调试。
· 反应器模式需要操作系统底层的IO多路复用的支持,如Linux中的epoll。如果操作系统的底层不支持IO多路复用,反应器模式不会有那么高效。
· 同一个Handler业务线程中,如果出现一个长时间的数据读写,会影响这个反应器中其他通道的IO处理。例如在大文件传输时,IO操作就会影响其他客户端(Client)的响应时间。因而对于这种操作,还需要进一步对反应器模式进行改进。