Reactor模式
什么是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事件
while (running) {
try {
// 执行队列中的任务
runnable task;
while ((task = taskqueue.poll()) != null) {
task.run();
}
selector.select(1000);
// 获取查询结果
set<selectionkey> selectionkeys = selector.selectedkeys();
// 遍历查询结果
iterator<selectionkey> keyiterator = selectionkeys.iterator();
while (keyiterator.hasnext()) {
// 被封装的查询结果
selectionkey selectionkey = keyiterator.next();
keyiterator.remove();
int readyops = selectionkey.readyops();
// 关注 read 和 accept两个事件
if ((readyops & (selectionkey.op_read | selectionkey.op_accept)) != 0
|| readyops == 0) {
try {
selectablechannel channel = (selectablechannel) selectionkey.attachment();
channel.configureblocking(false);
handler(channel);
// 如果关闭了,就取消这个key的订阅
if (!channel.isopen()) {
selectionkey.cancel();
}
} catch (exception e) {
// 如果有异常,就取消这个key的订阅
selectionkey.cancel();
e.printstacktrace();
}
}
}
} catch (exception e) {
e.printstacktrace();
}
}
}
private selectionkey register(selectablechannel channel) throws exception {
// 为什么register要以任务提交的形式，让reactor线程去处理？
// 因为线程在执行channel注册到selector的过程中，会和调用selector.select()方法的线程争用同一把锁
// 而select()方法实在eventloop中通过while循环调用的，争抢的可能性很高，
// 为了让register能更快的执行，就放到同一个线程来处理
futuretask<selectionkey> futuretask =
new futuretask<>(() -> channel.register(selector, 0, channel));
taskqueue.add(futuretask);
return futuretask.get();
}
private void dostart() {
if (!running) {
running = true;
start();
}
}
}
private serversocketchannel serversocketchannel;
// 1、创建多个线程 - accept处理reactor线程 (accept线程)
private reactorthread[] mainreactorthreads = new reactorthread[1];
// 2、创建多个线程 - io处理reactor线程 (i/o线程)
private reactorthread[] subreactorthreads = new reactorthread[8];
// 初始化线程组
private void newgroup() throws ioexception {
// 创建mainreactor线程, 只负责处理serversocketchannel
for (int i = 0; i < mainreactorthreads.length; i++) {
mainreactorthreads[i] =
new reactorthread() {
atomicinteger incr = new atomicinteger(0);
@override
public void handler(selectablechannel channel) throws exception {
// 只做请求分发，不做具体的数据读取
serversocketchannel ch = (serversocketchannel) channel;
socketchannel socketchannel = ch.accept();
socketchannel.configureblocking(false);
// 收到连接建立的通知之后，分发给i/o线程继续去读取数据
int index = incr.getandincrement() % subreactorthreads.length;
reactorthread workeventloop = subreactorthreads[index];
workeventloop.dostart();
selectionkey selectionkey = workeventloop.register(socketchannel);
selectionkey.interestops(selectionkey.op_read);
system.out.println(
thread.currentthread().getname() + "收到新连接 : " + socketchannel.getremoteaddress());
}
};
}
// 创建io线程,负责处理客户端连接以后socketchannel的io读写
for (int i = 0; i < subreactorthreads.length; i++) {
subreactorthreads[i] =
new reactorthread() {
@override
public void handler(selectablechannel channel) throws exception {
// work线程只负责处理io处理，不处理accept事件
socketchannel ch = (socketchannel) channel;
bytebuffer requestbuffer = bytebuffer.allocate(1024);
while (ch.isopen() && ch.read(requestbuffer) != -1) {
// 长连接情况下,需要手动判断数据有没有读取结束 (此处做一个简单的判断: 超过0字节就认为请求结束了)
if (requestbuffer.position() > 0) break;
}
if (requestbuffer.position() == 0) return; // 如果没数据了, 则不继续后面的处理
requestbuffer.flip();
byte[] content = new byte[requestbuffer.limit()];
requestbuffer.get(content);
system.out.println(new string(content));
system.out.println(
thread.currentthread().getname() + "收到数据,来自：" + ch.getremoteaddress());
// todo 业务操作 数据库、接口...
workpool.submit(() -> {});
// 响应结果 200
string response =
"http/1.1 200 ok\r\n" + "content-length: 11\r\n\r\n" + "hello world";
bytebuffer buffer = bytebuffer.wrap(response.getbytes());
while (buffer.hasremaining()) {
ch.write(buffer);
}
}
};
}
}
// 始化channel,并且绑定一个eventloop线程
private void initandregister() throws exception {
// 1、 创建serversocketchannel
serversocketchannel = serversocketchannel.open();
serversocketchannel.configureblocking(false);
// 2、 将serversocketchannel注册到selector
int index = new random().nextint(mainreactorthreads.length);
mainreactorthreads[index].dostart();
selectionkey selectionkey = mainreactorthreads[index].register(serversocketchannel);
selectionkey.interestops(selectionkey.op_accept);
}
// 绑定端口
private void bind() throws ioexception {
// 1、 正式绑定端口，对外服务
serversocketchannel.bind(new inetsocketaddress(8080));
system.out.println("启动完成，端口8080");
}
public static void main(string[] args) throws exception {
nioreactor nioreactor = new nioreactor();
// 1、 创建main和sub两组线程
nioreactor.newgroup();
// 2、 创建serversocketchannel，注册到mainreactor线程上的selector上
nioreactor.initandregister();
// 3、 为serversocketchannel绑定端口
nioreactor.bind();
}
}
上一篇： jquery代码规范让代码越来越好看
下一篇： Linux基础之xargs命令的入门实例