Netty线程模型 Reactor模型

线程模型

线程模型有传统阻塞I/O服务器模型和Reactor模型(网络主流模型)

根据Reactor的数量和处理资源池线程的数量不同，有3种典型的实现：

单Reactor单线程
单Reactor多线程
主从Reactor多线程

Netty基于主从Reactor多线程模型，并做了一些改进，其中主从Reactor多线程有多个Reactor

单线程Reactor模型

Java的NIO模式的Selector网络通讯，其实就是一个简单的Reactor模型。可以说是Reactor模型的朴素原型。

实际上的Reactor模式，是基于Java NIO的，在他的基础上，抽象出来两个组件——Reactor和Handler两个组件：

（1）Reactor：负责响应IO事件，当检测到一个新的事件，将其发送给相应的Handler去处理；新的事件包含连接建立就绪、读就绪、写就绪等。

（2）Handler:将自身（handler）与事件绑定，负责事件的处理，完成channel的读入，完成处理业务逻辑后，负责将结果写出channel。

可以将上图的accepter，看做是一种特殊的handler。

class Reactor implements Runnable
{final Selector selector;final ServerSocketChannel serverSocket;Reactor(int port) throws IOException{ //Reactor初始化selector = Selector.open();serverSocket = ServerSocketChannel.open();serverSocket.socket().bind(new InetSocketAddress(port));//非阻塞serverSocket.configureBlocking(false);//分步处理,第一步,接收accept事件SelectionKey sk =serverSocket.register(selector, SelectionKey.OP_ACCEPT);//attach callback object, Acceptorsk.attach(new Acceptor());}public void run(){try{while (!Thread.interrupted()){selector.select();Set selected = selector.selectedKeys();Iterator it = selected.iterator();while (it.hasNext()){//Reactor负责dispatch收到的事件dispatch((SelectionKey) (it.next()));}selected.clear();}} catch (IOException ex){ /* ... */ }}void dispatch(SelectionKey k){Runnable r = (Runnable) (k.attachment());//调用之前注册的callback对象if (r != null){r.run();}}// inner classclass Acceptor implements Runnable{public void run(){try{SocketChannel channel = serverSocket.accept();if (channel != null)new Handler(selector, channel);} catch (IOException ex){ /* ... */ }}}
}

class Handler implements Runnable
{final SocketChannel channel;final SelectionKey sk;ByteBuffer input = ByteBuffer.allocate(SystemConfig.INPUT_SIZE);ByteBuffer output = ByteBuffer.allocate(SystemConfig.SEND_SIZE);static final int READING = 0, SENDING = 1;int state = READING;Handler(Selector selector, SocketChannel c) throws IOException{channel = c;c.configureBlocking(false);// Optionally try first read nowsk = channel.register(selector, 0);//将Handler作为callback对象sk.attach(this);//第二步,注册Read就绪事件sk.interestOps(SelectionKey.OP_READ);selector.wakeup();}boolean inputIsComplete(){/* ... */return false;}boolean outputIsComplete(){/* ... */return false;}void process(){/* ... */return;}public void run(){try{if (state == READING){read();}else if (state == SENDING){send();}} catch (IOException ex){ /* ... */ }}void read() throws IOException{channel.read(input);if (inputIsComplete()){process();state = SENDING;// Normally also do first write now//第三步,接收write就绪事件sk.interestOps(SelectionKey.OP_WRITE);}}void send() throws IOException{channel.write(output);//write完就结束了, 关闭select keyif (outputIsComplete()){sk.cancel();}}
}

单线程模式的缺点:

1、当其中某个 handler 阻塞时，会导致其他所有的 client 的 handler 都得不到执行，并且更严重的是， handler 的阻塞也会导致整个服务不能接收新的 client 请求(因为 acceptor 也被阻塞了)。因为有这么多的缺陷，因此单线程Reactor 模型用的比较少。这种单线程模型不能充分利用多核资源，所以实际使用的不多。

2、因此，单线程模型仅仅适用于handler 中业务处理组件能快速完成的场景。

多线程线程Reactor模型

基于线程池的改进

在线程Reactor模式基础上，做如下改进：

（1）将Handler处理器的具体执行放入线程池，多线程进行业务处理。

（2）而对于Reactor而言，可以仍为单个线程。如果服务器为多核的CPU，为充分利用系统资源，可以将Reactor拆分为两个线程。

class MthreadHandler implements Runnable
{final SocketChannel channel;final SelectionKey selectionKey;ByteBuffer input = ByteBuffer.allocate(SystemConfig.INPUT_SIZE);ByteBuffer output = ByteBuffer.allocate(SystemConfig.SEND_SIZE);static final int READING = 0, SENDING = 1;int state = READING;ExecutorService pool = Executors.newFixedThreadPool(2);static final int PROCESSING = 3;MthreadHandler(Selector selector, SocketChannel c) throws IOException{channel = c;c.configureBlocking(false);// Optionally try first read nowselectionKey = channel.register(selector, 0);//将Handler作为callback对象selectionKey.attach(this);//第二步,注册Read就绪事件selectionKey.interestOps(SelectionKey.OP_READ);selector.wakeup();}boolean inputIsComplete(){/* ... */return false;}boolean outputIsComplete(){/* ... */return false;}void process(){/* ... */return;}public void run(){try{if (state == READING){read();}else if (state == SENDING){send();}} catch (IOException ex){ /* ... */ }}synchronized void read() throws IOException{// ...channel.read(input);if (inputIsComplete()){state = PROCESSING;//使用线程pool异步执行pool.execute(new Processer());}}void send() throws IOException{channel.write(output);//write完就结束了, 关闭select keyif (outputIsComplete()){selectionKey.cancel();}}synchronized void processAndHandOff(){process();state = SENDING;// or rebind attachment//process完,开始等待write就绪selectionKey.interestOps(SelectionKey.OP_WRITE);}class Processer implements Runnable{public void run(){processAndHandOff();}}}

Reactor 类没有大的变化，参考前面的代码。

优点：

多线程可以充分利用多核CPU的处理能力
采用线程池复用线程，减少创建和销毁线程带来的性能开销

缺点：

reactor处理所有事件的监听和响应，在单线程运行，高并发场景下容易出现性能瓶颈
多线程数据共享和访问比较复杂

多Reactor多线程模型

方案说明：

Reactor主线程MainReactor对象通过select监听连接事件，收到事件后，通过Acceptor处理连接事件
当Acceptor处理连接事件后，MainReactor将连接分配给SubReactor
SubReactor将连接加入到连接队列进行监听，并创建handler进行各种事件处理
当有新事件发生时，SubReactor就会调用对应的handler进行处理
handler通过Read读取数据，分发给后面的worker线程处理
worker线程池会分配独立的worker线程进行业务处理，并返回结果
handler收到响应的结果后，再通过send将结果返回给client
MainReactor主线程可以关联多个SubReactor子线程

mainReactor负责监听server socket，用来处理新连接的建立，将建立的socketChannel指定注册给subReactor。

subReactor维护自己的selector, 基于mainReactor 注册的socketChannel多路分离IO读写事件，读写网络数据，对业务处理的功能，另其扔给worker线程池来完成。

就是上一篇的Boss和Worker

Netty基础 NIO SocketChannel 网络编程自制小型服务器,多线程优化*_清风拂来水波不兴的博客-CSDN博客

class MthreadReactor implements Runnable
{//subReactors集合, 一个selector代表一个subReactorSelector[] selectors=new Selector[2];int next = 0;final ServerSocketChannel serverSocket;MthreadReactor(int port) throws IOException{ //Reactor初始化selectors[0]=Selector.open();selectors[1]= Selector.open();serverSocket = ServerSocketChannel.open();serverSocket.socket().bind(new InetSocketAddress(port));//非阻塞serverSocket.configureBlocking(false);//分步处理,第一步,接收accept事件SelectionKey sk =serverSocket.register( selectors[0], SelectionKey.OP_ACCEPT);//attach callback object, Acceptorsk.attach(new Acceptor());}public void run(){try{while (!Thread.interrupted()){for (int i = 0; i <2 ; i++){selectors[i].select();Set selected =  selectors[i].selectedKeys();Iterator it = selected.iterator();while (it.hasNext()){//Reactor负责dispatch收到的事件dispatch((SelectionKey) (it.next()));}selected.clear();}}} catch (IOException ex){ /* ... */ }}void dispatch(SelectionKey k){Runnable r = (Runnable) (k.attachment());//调用之前注册的callback对象if (r != null){r.run();}}class Acceptor { // ...public synchronized void run() throws IOException{SocketChannel connection =serverSocket.accept(); //主selector负责acceptif (connection != null){new Handler(selectors[next], connection); //选个subReactor去负责接收到的connection}if (++next == selectors.length) next = 0;}}
}

6.1. 优点

1）响应快，不必为单个同步时间所阻塞，虽然Reactor本身依然是同步的；

2）编程相对简单，可以最大程度的避免复杂的多线程及同步问题，并且避免了多线程/进程的切换开销；

3）可扩展性，可以方便的通过增加Reactor实例个数来充分利用CPU资源；

4）可复用性，reactor框架本身与具体事件处理逻辑无关，具有很高的复用性；

6.2. 缺点

1）相比传统的简单模型，Reactor增加了一定的复杂性，因而有一定的门槛，并且不易于调试。

2）Reactor模式需要底层的Synchronous Event Demultiplexer支持，比如Java中的Selector支持，操作系统的select系统调用支持，如果要自己实现Synchronous Event Demultiplexer可能不会有那么高效。

3） Reactor模式在IO读写数据时还是在同一个线程中实现的，即使使用多个Reactor机制的情况下，那些共享一个Reactor的Channel如果出现一个长时间的数据读写，会影响这个Reactor中其他Channel的相应时间，比如在大文件传输时，IO操作就会影响其他Client的相应时间，因而对这种操作，使用传统的Thread-Per-Connection或许是一个更好的选择，或则此时使用改进版的Reactor模式如Proactor模式。