1. 為什么是Reactor模式
寫多了代碼的兄弟們都知道，JAVA代碼由於到處面向接口及高度抽象，用到繼承多態和設計模式，程序的組織不是按照正常的理解順序來的，對代碼跟蹤很是個問題。所以，在閱讀別人的源碼時，如果不了解代碼的組織方式，往往是暈頭轉向，不知在何處。尤其是閱讀經典代碼的時候，更是如此。
反過來，如果先了解代碼的設計模式，再來去代碼，就會閱讀的很輕松，不會那么難懂。
像netty這樣的精品中的極品，肯定也是需要先從設計模式入手的。netty的整體架構，基於了一個著名的模式——Reactor模式。Reactor模式，是高性能網絡編程的必知必會模式。
首先熟悉Reactor模式，一定是磨刀不誤砍柴工。
2. Reactor模式簡介
Netty是典型的Reactor模型結構，關於Reactor的詳盡闡釋，本文站在巨人的肩膀上，借助 Doug Lea（就是那位讓人無限景仰的大爺）的“Scalable IO in Java”中講述的Reactor模式。
“Scalable IO in Java”的地址是：http://gee.cs.oswego.edu/dl/cpjslides/nio.pdf
Reactor模式也叫反應器模式，大多數IO相關組件如Netty、Redis在使用的IO模式，為什么需要這種模式，它是如何設計來解決高性能並發的呢？
3. 多線程IO的致命缺陷
最最原始的網絡編程思路就是服務器用一個while循環，不斷監聽端口是否有新的套接字連接，如果有，那么就調用一個處理函數處理，類似：
while(true){
socket = accept();
handle(socket)
}
這種方法的最大問題是無法並發，效率太低，如果當前的請求沒有處理完，那么后面的請求只能被阻塞，服務器的吞吐量太低。
之后，想到了使用多線程，也就是很經典的connection per thread，每一個連接用一個線程處理，類似：
package com.crazymakercircle.iodemo.base;
import com.crazymakercircle.config.SystemConfig;
import java.io.IOException;
import java.net.ServerSocket;
import java.net.Socket;
class BasicModel implements Runnable {
public void run() {
try {
ServerSocket ss =
new ServerSocket(SystemConfig.SOCKET_SERVER_PORT);
while (!Thread.interrupted())
new Thread(new Handler(ss.accept())).start();
//創建新線程來handle
// or, single-threaded, or a thread pool
} catch (IOException ex) { /* ... */ }
}
static class Handler implements Runnable {
final Socket socket;
Handler(Socket s) { socket = s; }
public void run() {
try {
byte[] input = new byte[SystemConfig.INPUT_SIZE];
socket.getInputStream().read(input);
byte[] output = process(input);
socket.getOutputStream().write(output);
} catch (IOException ex) { /* ... */ }
}
private byte[] process(byte[] input) {
byte[] output=null;
/* ... */
return output;
}
}
}
對於每一個請求都分發給一個線程，每個線程中都獨自處理上面的流程。
tomcat服務器的早期版本確實是這樣實現的。
多線程並發模式，一個連接一個線程的優點是：
一定程度上極大地提高了服務器的吞吐量，因為之前的請求在read阻塞以后，不會影響到后續的請求，因為他們在不同的線程中。這也是為什么通常會講“一個線程只能對應一個socket”的原因。另外有個問題，如果一個線程中對應多個socket連接不行嗎？語法上確實可以，但是實際上沒有用，每一個socket都是阻塞的，所以在一個線程里只能處理一個socket，就算accept了多個也沒用，前一個socket被阻塞了，后面的是無法被執行到的。
多線程並發模式，一個連接一個線程的缺點是：
缺點在於資源要求太高，系統中創建線程是需要比較高的系統資源的，如果連接數太高，系統無法承受，而且，線程的反復創建-銷毀也需要代價。
改進方法是：
采用基於事件驅動的設計，當有事件觸發時，才會調用處理器進行數據處理。使用Reactor模式，對線程的數量進行控制，一個線程處理大量的事件。
4. 單線程Reactor模型
Reactor模型的朴素原型
Java的NIO模式的Selector網絡通訊，其實就是一個簡單的Reactor模型。可以說是Reactor模型的朴素原型。
static class Server
{
public static void testServer() throws IOException
{
// 1、獲取Selector選擇器
Selector selector = Selector.open();
// 2、獲取通道
ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
// 3.設置為非阻塞
serverSocketChannel.configureBlocking(false);
// 4、綁定連接
serverSocketChannel.bind(new InetSocketAddress(SystemConfig.SOCKET_SERVER_PORT));
// 5、將通道注冊到選擇器上,並注冊的操作為：“接收”操作
serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
// 6、采用輪詢的方式，查詢獲取“准備就緒”的注冊過的操作
while (selector.select() > 0)
{
// 7、獲取當前選擇器中所有注冊的選擇鍵（“已經准備就緒的操作”）
Iterator<SelectionKey> selectedKeys = selector.selectedKeys().iterator();
while (selectedKeys.hasNext())
{
// 8、獲取“准備就緒”的時間
SelectionKey selectedKey = selectedKeys.next();
// 9、判斷key是具體的什么事件
if (selectedKey.isAcceptable())
{
// 10、若接受的事件是“接收就緒” 操作,就獲取客戶端連接
SocketChannel socketChannel = serverSocketChannel.accept();
// 11、切換為非阻塞模式
socketChannel.configureBlocking(false);
// 12、將該通道注冊到selector選擇器上
socketChannel.register(selector, SelectionKey.OP_READ);
}
else if (selectedKey.isReadable())
{
// 13、獲取該選擇器上的“讀就緒”狀態的通道
SocketChannel socketChannel = (SocketChannel) selectedKey.channel();
// 14、讀取數據
ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
int length = 0;
while ((length = socketChannel.read(byteBuffer)) != -1)
{
byteBuffer.flip();
System.out.println(new String(byteBuffer.array(), 0, length));
byteBuffer.clear();
}
socketChannel.close();
}
// 15、移除選擇鍵
selectedKeys.remove();
}
}
// 7、關閉連接
serverSocketChannel.close();
}
public static void main(String[] args) throws IOException
{
testServer();
}
}
實際上的Reactor模式，是基於Java NIO的，在他的基礎上，抽象出來兩個組件——Reactor和Handler兩個組件：
（1）Reactor：負責響應IO事件，當檢測到一個新的事件，將其發送給相應的Handler去處理；新的事件包含連接建立就緒、讀就緒、寫就緒等。
（2）Handler:將自身（handler）與事件綁定，負責事件的處理，完成channel的讀入，完成處理業務邏輯后，負責將結果寫出channel。
4.1. 什么是單線程Reactor呢？
如下圖所示：
這是最簡單的單Reactor單線程模型。Reactor線程是個多面手，負責多路分離套接字，Accept新連接，並分派請求到Handler處理器中。
下面的圖，來自於“Scalable IO in Java”，和上面的圖的意思，差不多。Reactor和Hander 處於一條線程執行。
順便說一下，可以將上圖的accepter，看做是一種特殊的handler。
4.2. 單線程Reactor的參考代碼
“Scalable IO in Java”，實現了一個單線程Reactor的參考代碼，Reactor的代碼如下：
Handler的代碼如下：
這兩段代碼，是建立在JAVA NIO的基礎上的，這兩段代碼建議一定要看懂。可以在IDE中去看源碼，這樣直觀感覺更佳。
如果對NIO的Seletor不完全了解，影響到上面的代碼閱讀，請閱讀瘋狂創客圈的Java NIO死磕 文章。
4.3. 單線程模式的缺點:
1、 當其中某個 handler 阻塞時， 會導致其他所有的 client 的 handler 都得不到執行， 並且更嚴重的是， handler 的阻塞也會導致整個服務不能接收新的 client 請求(因為 acceptor 也被阻塞了)。 因為有這么多的缺陷， 因此單線程Reactor 模型用的比較少。這種單線程模型不能充分利用多核資源，所以實際使用的不多。
2、因此，單線程模型僅僅適用於handler 中業務處理組件能快速完成的場景。
5. 多線程的Reactor
5.1. 基於線程池的改進
在線程Reactor模式基礎上，做如下改進：
（1）將Handler處理器的執行放入線程池，多線程進行業務處理。
（2）而對於Reactor而言，可以仍為單個線程。如果服務器為多核的CPU，為充分利用系統資源，可以將Reactor拆分為兩個線程。
一個簡單的圖如下：
5.2. 改進后的完整示意圖
下面的圖，來自於“Scalable IO in Java”，和上面的圖的意思，差不多，只是更加詳細。Reactor是一條獨立的線程，Hander 處於線程池中執行。
5.3. 多線程Reactor的參考代碼
“Scalable IO in Java”，的多線程Reactor的參考代碼，是基於單線程做一個線程池的改進，改進的Handler的代碼如下：
package com.crazymakercircle.ReactorModel;
import com.crazymakercircle.config.SystemConfig;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
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 now
selectionKey = 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 key
if (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 類沒有大的變化，參考前面的代碼。
6. Reactor持續改進
對於多個CPU的機器，為充分利用系統資源，將Reactor拆分為兩部分。代碼如下：
package com.crazymakercircle.ReactorModel;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;
class MthreadReactor implements Runnable
{
//subReactors集合, 一個selector代表一個subReactor
Selector[] 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, Acceptor
sk.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負責accept
if (connection != null)
{
new Handler(selectors[next], connection); //選個subReactor去負責接收到的connection
}
if (++next == selectors.length) next = 0;
}
}
}
7. Reactor編程的優點和缺點
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模式。