Reactor 模型的核心思想&＃xff1a;

将关注的 I/O 事件注册到多路复用器上&＃xff0c;一旦有 I/O 事件触发&＃xff0c;将事件分发到事件处理器中&＃xff0c;执行就绪 I/O 事件对应的处理函数中。模型中有三个重要的组件&＃xff1a;

• 多路复用器&＃xff1a;由操作系统提供接口&＃xff0c;Linux 提供的 I/O 复用接口有select、poll、epoll 。
• 事件分离器&＃xff1a;将多路复用器返回的就绪事件分发到事件处理器中。
• 事件处理器&＃xff1a;处理就绪事件处理函数。

单 Reactor 单线程模型

 

示例代码: 

/**
* 等待事件到来&＃xff0c;分发事件处理
*/
class Reactor implements Runnable {
​private Reactor() throws Exception {SelectionKey sk &＃61; serverSocket.register(selector, SelectionKey.OP_ACCEPT);// attach Acceptor 处理新连接sk.attach(new Acceptor());}
​
​ &＃64;Overridepublic void run() {try {while (!Thread.interrupted()) {selector.select();Set selected &＃61; selector.selectedKeys();Iterator it &＃61; selected.iterator();while (it.hasNext()) {it.remove();//分发事件处理dispatch((SelectionKey) (it.next()));}}} catch (IOException ex) {//do something}}
​void dispatch(SelectionKey k) {// 若是连接事件获取是acceptor// 若是IO读写事件获取是handlerRunnable runnable &＃61; (Runnable) (k.attachment());if (runnable !&＃61; null) {runnable.run();}}
​
}

Reactor 线程&＃xff0c;负责多路分离套接字。

• 有新连接到来触发 OP_ACCEPT 事件之后&＃xff0c; 交由 Acceptor 进行处理。
• 有 IO 读写事件之后&＃xff0c;交给 Handler 处理。

Acceptor 主要任务是构造 Handler 。

• 在获取到 Client 相关的 SocketChannel 之后&＃xff0c;绑定到相应的 Handler 上。
• 对应的 SocketChannel 有读写事件之后&＃xff0c;基于 Reactor 分发&＃xff0c;Handler 就可以处理了。

注意&＃xff0c;所有的 IO 事件都绑定到 Selector 上&＃xff0c;由 Reactor 统一分发。

该模型适用于处理器链中业务处理组件能快速完成的场景。不过&＃xff0c;这种单线程模型不能充分利用多核资源&＃xff0c;所以实际使用的不多。

 

单 Reactor 多线程模型

相对于第一种单线程的模式来说&＃xff0c;在处理业务逻辑&＃xff0c;也就是获取到 IO 的读写事件之后&＃xff0c;交由线程池来处理&＃xff0c;这样可以减小主 Reactor 的性能开销&＃xff0c;从而更专注的做事件分发工作了&＃xff0c;从而提升整个应用的吞吐。

/**
* 多线程处理读写业务逻辑
*/
class MultiThreadHandler implements Runnable {public static final int READING &＃61; 0, WRITING &＃61; 1;int state;final SocketChannel socket;final SelectionKey sk;
​//多线程处理业务逻辑ExecutorService executorService &＃61; Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
​
​public MultiThreadHandler(SocketChannel socket, Selector sl) throws Exception {this.state &＃61; READING;this.socket &＃61; socket;sk &＃61; socket.register(selector, SelectionKey.OP_READ);sk.attach(this);socket.configureBlocking(false);}
​&＃64;Overridepublic void run() {if (state &＃61;&＃61; READING) {read();} else if (state &＃61;&＃61; WRITING) {write();}}
​private void read() {//任务异步处理executorService.submit(() -> process());
​//下一步处理写事件sk.interestOps(SelectionKey.OP_WRITE);this.state &＃61; WRITING;}
​private void write() {//任务异步处理executorService.submit(() -> process());
​//下一步处理读事件sk.interestOps(SelectionKey.OP_READ);this.state &＃61; READING;}
​/*** task 业务处理*/public void process() {//do IO ,task,queue something}
}

• 在 #read() 和 #write() 方法中&＃xff0c;提交 executorService 线程池&＃xff0c;进行处理。

 

多 Reactor 多线程模型

 

第三种模型比起第二种模型&＃xff0c;是将 Reactor 分成两部分&＃xff1a;

1. mainReactor 负责监听 ServerSocketChannel &＃xff0c;用来处理客户端新连接的建立&＃xff0c;并将建立的客户端的 SocketChannel 指定注册给 subReactor 。
2. subReactor 维护自己的 Selector &＃xff0c;基于 mainReactor 建立的客户端的 SocketChannel 多路分离 IO 读写事件&＃xff0c;读写网络数据。对于业务处理的功能&＃xff0c;另外扔给 worker 线程池来完成。

MultiWorkThreadAcceptor 示例代码如下&＃xff1a;

/**
* 多work 连接事件Acceptor,处理连接事件
*/
class MultiWorkThreadAcceptor implements Runnable {
​// cpu线程数相同多work线程int workCount &＃61; Runtime.getRuntime().availableProcessors();SubReactor[] workThreadHandlers &＃61; new SubReactor[workCount];volatile int nextHandler &＃61; 0;
​public MultiWorkThreadAcceptor() {this.init();}
​public void init() {nextHandler &＃61; 0;for (int i &＃61; 0; i ​&＃64;Overridepublic void run() {try {SocketChannel c &＃61; serverSocket.accept();if (c !&＃61; null) {// 注册读写synchronized (c) {// 顺序获取SubReactor&＃xff0c;然后注册channel SubReactor work &＃61; workThreadHandlers[nextHandler];work.registerChannel(c);nextHandler&＃43;&＃43;;if (nextHandler >&＃61; workThreadHandlers.length) {nextHandler &＃61; 0;}}}} catch (Exception e) {}}
}

SubReactor 示例代码如下&＃xff1a;

/**
* 多work线程处理读写业务逻辑
*/
class SubReactor implements Runnable {final Selector mySelector;
​//多线程处理业务逻辑int workCount &＃61;Runtime.getRuntime().availableProcessors();ExecutorService executorService &＃61; Executors.newFixedThreadPool(workCount);
​
​public SubReactor() throws Exception {// 每个SubReactor 一个selector this.mySelector &＃61; SelectorProvider.provider().openSelector();}
​/*** 注册chanel** &＃64;param sc* &＃64;throws Exception*/public void registerChannel(SocketChannel sc) throws Exception {sc.register(mySelector, SelectionKey.OP_READ | SelectionKey.OP_CONNECT);}
​&＃64;Overridepublic void run() {while (true) {try {//每个SubReactor 自己做事件分派处理读写事件selector.select();Set keys &＃61; selector.selectedKeys();Iterator iterator &＃61; keys.iterator();while (iterator.hasNext()) {SelectionKey key &＃61; iterator.next();iterator.remove();if (key.isReadable()) {read();} else if (key.isWritable()) {write();}}
​} catch (Exception e) {
​}}}
​private void read() {//任务异步处理executorService.submit(() -> process());}
​private void write() {//任务异步处理executorService.submit(() -> process());}
​/*** task 业务处理*/public void process() {//do IO ,task,queue something}
}

 

从代码中&＃xff0c;我们可以看到&＃xff1a;

1. mainReactor 主要用来处理网络 IO 连接建立操作&＃xff0c;通常&＃xff0c;mainReactor 只需要一个&＃xff0c;因为它一个线程就可以处理。
2. subReactor 主要和建立起来的客户端的 SocketChannel 做数据交互和事件业务处理操作。通常&＃xff0c;subReactor 的个数和 CPU 个数相等&＃xff0c;每个 subReactor 独占一个线程来处理。

---

此种模式中&＃xff0c;每个模块的工作更加专一&＃xff0c;耦合度更低&＃xff0c;性能和稳定性也大大的提升&＃xff0c;支持的可并发客户端数量可达到上百万级别。

关于此种模式的应用&＃xff0c;目前有很多优秀的框架已经在应用&＃xff0c;比如 Mina 和 Netty 等等。上述中去掉线程池的第三种形式的变种&＃xff0c;也是 Netty NIO 的默认模式。