注：本篇文章例子基於上一篇進行：Java NIO學習與記錄（七）： Reactor單線程模型的實現
前言：單線程Reactor模型的缺點
緊接着上篇Reactor單線程模型的例子來，假設Handler的read那里的處理方式延遲5s，當做是業務性能瓶頸，改變下原來的Handler，讓其read方法在處理時延遲5s：
private void read() throws IOException {
if (selectionKey.isValid()) {
System.out.println("服務端讀取數據前");
readBuffer.clear();
int count = socketChannel.read(readBuffer);
if (count > 0) {
try {
Thread.sleep(5000L); //讀取信息后睡眠5s當做業務處理瓶頸
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(String.format("收到來自 %s 的消息: %s",
socketChannel.getRemoteAddress(),
new String(readBuffer.array())));
status = SEND;
selectionKey.interestOps(SelectionKey.OP_WRITE);
} else {
selectionKey.cancel();
socketChannel.close();
System.out.println("read時-------連接關閉");
}
}
}
現在同樣開啟兩個客戶端同時連接到該服務端，然后請求-->收到響應-->再次請求的流程走10次，會發現，客戶端每收到一次響應需要10s，同樣的如果開啟3個客戶端，則需要15s，因為單線程的Reactor模型是串行的，業務處理的瓶頸可以影響到全局的事件分發，這種模型下，如果存在類似例子中的瓶頸點是致命的（例子的5s是誇張處理），因為新進來的連接也會排隊，整個select都會被Handler的處理給阻塞掉，舉個實際點的例子，redis在使用時大部分時候會避免使用類似keys這種重操作，為什么呢？就是因為redis是單線程，這里說的單線程其實並不是說redis服務端就一個線程，而是說redis采用的NIO Reactor模型就是單線程的Reactor模型，跟上面代碼里做的改動一樣，5s可以理解成重操作，影響整個模型的正常運作，redis之所以采用單線程模式，是因為redis大部分操作實在是太快了，快到使用這種模式也可以提供近十萬/秒的並發能力，單線程模型實現起來簡單且可控性強，所以redis很自然的選擇了這種模式。回到問題本身，我們自己的業務可能並沒有redis那樣高的處理能力，搞不好幾個網絡請求就可以造成性能瓶頸，拖慢甚至拖垮整個處理模型，所以大部分RPC框架和web容器並不會采用單線程的Reactor模型實現，那么有沒有什么方法可以優化這種模型呢？比如，把這個瓶頸點利用獨立線程異步出去處理，這樣可以保證不影響select的執行，也就很好的避免了上面的問題了，下面介紹兩種多線程異步的Reactor模型。
一、單Reactor多線程模型
模型圖：
圖1
上圖與單線程Reactor模型對比可以看出，讀入數據后，對數據的業務處理部分被線程池做了異步處理，也就是說，上述5s的那段瓶頸被放到了子線程去處理，select的執行不會受到任何影響，因此對新的連接處理、多個客戶端的響應速度都應該可以得到保障。
現在來改寫下前篇文章里的單線程處理模式的Handler，更名為AsyncHandler：
public class AsyncHandler implements Runnable {
private final Selector selector;
private final SelectionKey selectionKey;
private final SocketChannel socketChannel;
private ByteBuffer readBuffer = ByteBuffer.allocate(1024);
private ByteBuffer sendBuffer = ByteBuffer.allocate(2048);
private final static int READ = 0; //讀取就緒
private final static int SEND = 1; //響應就緒
private final static int PROCESSING = 2; //處理中
private int status = READ; //所有連接完成后都是從一個讀取動作開始的
//開啟線程數為5的異步處理線程池
private static final ExecutorService workers = Executors.newFixedThreadPool(5);
AsyncHandler(SocketChannel socketChannel, Selector selector) throws IOException {
this.socketChannel = socketChannel;
this.socketChannel.configureBlocking(false);
selectionKey = socketChannel.register(selector, 0);
selectionKey.attach(this);
selectionKey.interestOps(SelectionKey.OP_READ);
this.selector = selector;
this.selector.wakeup();
}
@Override
public void run() { //如果一個任務正在異步處理，那么這個run是直接不觸發任何處理的，read和send只負責簡單的數據讀取和響應，業務處理完全不阻塞這里的處理
switch (status) {
case READ:
read();
break;
case SEND:
send();
break;
default:
}
}
private void read() {
if (selectionKey.isValid()) {
try {
readBuffer.clear();
int count = socketChannel.read(readBuffer);
if (count > 0) {
status = PROCESSING; //置為處理中，處理完成后該狀態為響應，表示讀入處理完成，接下來可以響應客戶端了
workers.execute(this::readWorker); //異步處理
} else {
selectionKey.cancel();
socketChannel.close();
System.out.println("read時-------連接關閉");
}
} catch (IOException e) {
System.err.println("處理read業務時發生異常！異常信息：" + e.getMessage());
selectionKey.cancel();
try {
socketChannel.close();
} catch (IOException e1) {
System.err.println("處理read業務關閉通道時發生異常！異常信息：" + e.getMessage());
}
}
}
}
void send() {
if (selectionKey.isValid()) {
status = PROCESSING; //置為執行中
workers.execute(this::sendWorker); //異步處理
selectionKey.interestOps(SelectionKey.OP_READ); //重新設置為讀
}
}
//讀入信息后的業務處理
private void readWorker() {
try {
Thread.sleep(5000L);
} catch (InterruptedException e) {
e.printStackTrace();
}
System.out.println(String.format("收到來自客戶端的消息: %s",
new String(readBuffer.array())));
status = SEND;
selectionKey.interestOps(SelectionKey.OP_WRITE); //注冊寫事件
this.selector.wakeup(); //喚醒阻塞在select的線程，因為該interestOps寫事件是放到子線程的，select在該channel還是對read事件感興趣時又被調用，因此如果不主動喚醒，select可能並不會立刻select該讀就緒事件（在該例中，可能永遠不會被select到）
}
private void sendWorker() {
try {
sendBuffer.clear();
sendBuffer.put(String.format("我收到來自%s的信息辣：%s, 200ok;",
socketChannel.getRemoteAddress(),
new String(readBuffer.array())).getBytes());
sendBuffer.flip();
int count = socketChannel.write(sendBuffer);
if (count < 0) {
selectionKey.cancel();
socketChannel.close();
System.out.println("send時-------連接關閉");
} else {
//再次切換到讀
status = READ;
}
} catch (IOException e) {
System.err.println("異步處理send業務時發生異常！異常信息：" + e.getMessage());
selectionKey.cancel();
try {
socketChannel.close();
} catch (IOException e1) {
System.err.println("異步處理send業務關閉通道時發生異常！異常信息：" + e.getMessage());
}
}
}
}
可以看到，read里、send里的邏輯處理被異步出去執行，新增了中間狀態“執行中”，主要用來防止事件重復觸發，重復執行異步邏輯，當異步邏輯處理完畢才會更改狀態值，這時候可以繼續處理接下來的事件（讀或寫）。
把Accptor類里的實現換成AsyncHandler，運行服務端和客戶端會發現，兩個客戶端的響應均為5s，也不會阻塞新增的連接，新增至三個或者更多的客戶端基本可以保持客戶端響應均為5s（說明：這里5s是誇張比喻，正常瓶頸沒這么誇張，若開了n多客戶端，每個都阻塞5s，那么線程池也會發生排隊，因為子線程個數有限，處理不過來，最后還是阻塞，一定會遠超過5s）。
通過多線程Reactor模型，降低了業務代碼瓶頸導致影響整個Reactor執行鏈路的風險，但是即便如此，read、send操作仍然和接收請求（accept）處於同一個線程，這就意味着read、send的處理可能會影響到對客戶端連接的接收能力，那么有沒有一種辦法，可以把讀寫流程徹底異步出去，負責連接的線程就只負責接收連接？於是多Reactor多線程模型就產生了，這種模型也叫主從Reactor模型，該模型下可以分為一個主Reactor專門處理連接事件，而多個從Reactor負責讀寫、業務處理等，這樣服務端可以接收並處理更多的請求，提升服務端的吞吐能力（該模型或者說所有基於NIO的Reactor模型，都是以提升服務端處理能力為基礎的，NIO在某些情況下不一定會比BIO處理速度快，但一定比BIO穩，就像NIO可以利用很少的線程處理大量的客戶端請求，而BIO在大量客戶端請求過來的情況下，由於各種操作均會阻塞線程，會處理不過來）。
二、主從Reactor模型
還是把之前文章的圖拿來展示下這種模型的流程，可以與上面圖1進行對比，看看發生了哪些變化：
圖2
上圖就是主從Reactor模型的一個流程，看下與圖1的不同之處，多了SubReactor這樣一個角色，這個角色就是用來處理讀寫操作的Reactor，現在仍然基於之前的例子，進行改寫，明確需要改寫的點：
①新增SubReactor
②Acceptor那里進行初始化一批SubReactor，進行分發處理
③為了區分客戶端分別是被哪個SubReactor處理的讀寫操作，還需要改寫下AsyncHandler，在里面加上SubReactor的序號，打印信息時進行區分。
ok，總結完改動點，現在基於上面的代碼（代碼初代目版本：Reactor單線程模型的實現）改寫一下這幾個類：
step1.首先新增SubReactor類
public class SubReactor implements Runnable {
private final Selector selector;
private boolean register = false; //注冊開關表示，為什么要加這么個東西，可以參考Acceptor設置這個值那里的描述
private int num; //序號，也就是Acceptor初始化SubReactor時的下標
SubReactor(Selector selector, int num) {
this.selector = selector;
this.num = num;
}
@Override
public void run() {
while (!Thread.interrupted()) {
System.out.println(String.format("%d號SubReactor等待注冊中...", num));
while (!Thread.interrupted() && !register) {
try {
if (selector.select() == 0) {
continue;
}
} catch (IOException e) {
e.printStackTrace();
}
Set selectedKeys = selector.selectedKeys();
Iterator it = selectedKeys.iterator();
while (it.hasNext()) {
dispatch(it.next());
it.remove();
}
}
}
}
private void dispatch(SelectionKey key) {
Runnable r = (Runnable) (key.attachment());
if (r != null) {
r.run();
}
}
void registering(boolean register) {
this.register = register;
}
}
這個類負責Acceptor交給自己的事件select（例子中實際上就是read、send）。
step2.Acceptor類的更改
public class Acceptor implements Runnable {
private final ServerSocketChannel serverSocketChannel;
private final int coreNum = Runtime.getRuntime().availableProcessors(); // 獲取CPU核心數
private final Selector[] selectors = new Selector[coreNum]; // 創建selector給SubReactor使用，個數為CPU核心數（如果不需要那么多可以自定義，畢竟這里會吞掉一個線程）
private int next = 0; // 輪詢使用subReactor的下標索引
private SubReactor[] reactors = new SubReactor[coreNum]; // subReactor
private Thread[] threads = new Thread[coreNum]; // subReactor的處理線程
Acceptor(ServerSocketChannel serverSocketChannel) throws IOException {
this.serverSocketChannel = serverSocketChannel;
// 初始化
for (int i = 0; i < coreNum; i++) {
selectors[i] = Selector.open();
reactors[i] = new SubReactor(selectors[i], i); //初始化sub reactor
threads[i] = new Thread(reactors[i]); //初始化運行sub reactor的線程
threads[i].start(); //啟動（啟動后的執行參考SubReactor里的run方法）
}
}
@Override
public void run() {
SocketChannel socketChannel;
try {
socketChannel = serverSocketChannel.accept(); // 連接
if (socketChannel != null) {
System.out.println(String.format("收到來自 %s 的連接",
socketChannel.getRemoteAddress()));
socketChannel.configureBlocking(false); //
reactors[next].registering(true); // 注意一個selector在select時是無法注冊新事件的，因此這里要先暫停下select方法觸發的程序段，下面的weakup和這里的setRestart都是做這個事情的，具體參考SubReactor里的run方法
selectors[next].wakeup(); // 使一個阻塞住的selector操作立即返回
SelectionKey selectionKey = socketChannel.register(selectors[next],
SelectionKey.OP_READ); // 當前客戶端通道SocketChannel向selector[next]注冊一個讀事件，返回key
selectors[next].wakeup(); // 使一個阻塞住的selector操作立即返回
reactors[next].registering(false); // 本次事件注冊完成后，需要再次觸發select的執行，因此這里Restart要在設置回false（具體參考SubReactor里的run方法）
selectionKey.attach(new AsyncHandler(socketChannel, selectors[next], next)); // 綁定Handler
if (++next == selectors.length) {
next = 0; //越界后重新分配
}
}
} catch (IOException e) {
e.printStackTrace();
}
}
}
可以跟以前的Acceptor做個對比，做了如下改動：
①接受到連接后不再直接觸發handler了
②初始化一堆SubReactor（從反應堆），每個交給一個線程處理，注冊讀事件后順序分配給不同的SubReactor去處理自己的selector監聽。
以上，就可以把讀寫處理+業務處理與接受連接的Reactor徹底分開了，接受連接的事件不再受任何讀寫、業務相關的影響，只負責接收，目前即便是業務線程池用光線程發生排隊，也不會影響到連接的接收，很大程度上降低了服務端的接收能力遭遇瓶頸的風險。
step3.改寫AsyncHandler的打印
這里就不po代碼了，具體就是把SubReactor的序號傳給handler，標記觸發Handler的Reactor是哪個。
同樣的，啟動下服務端，再開啟兩個客戶端（跟之前一樣，每個客戶端發10條消息終止連接），運行結果如下：
服務端：
1號SubReactor等待注冊中...
3號SubReactor等待注冊中...
0號SubReactor等待注冊中...
2號SubReactor等待注冊中...
收到來自 /127.0.0.1:60407 的連接
0號SubReactor等待注冊中...
收到來自 /127.0.0.1:60410 的連接
1號SubReactor等待注冊中...
1號SubReactor觸發：收到來自客戶端/127.0.0.1:60410的消息: 客戶端發送的第1條消息
0號SubReactor觸發：收到來自客戶端/127.0.0.1:60407的消息: 客戶端發送的第1條消息
0號SubReactor觸發：收到來自客戶端/127.0.0.1:60407的消息: 客戶端發送的第2條消息
1號SubReactor觸發：收到來自客戶端/127.0.0.1:60410的消息: 客戶端發送的第2條消息
0號SubReactor觸發：收到來自客戶端/127.0.0.1:60407的消息: 客戶端發送的第3條消息
1號SubReactor觸發：收到來自客戶端/127.0.0.1:60410的消息: 客戶端發送的第3條消息
1號SubReactor觸發：收到來自客戶端/127.0.0.1:60410的消息: 客戶端發送的第4條消息
0號SubReactor觸發：收到來自客戶端/127.0.0.1:60407的消息: 客戶端發送的第4條消息
1號SubReactor觸發：收到來自客戶端/127.0.0.1:60410的消息: 客戶端發送的第5條消息
0號SubReactor觸發：收到來自客戶端/127.0.0.1:60407的消息: 客戶端發送的第5條消息
1號SubReactor觸發：收到來自客戶端/127.0.0.1:60410的消息: 客戶端發送的第6條消息
0號SubReactor觸發：收到來自客戶端/127.0.0.1:60407的消息: 客戶端發送的第6條消息
1號SubReactor觸發：收到來自客戶端/127.0.0.1:60410的消息: 客戶端發送的第7條消息
0號SubReactor觸發：收到來自客戶端/127.0.0.1:60407的消息: 客戶端發送的第7條消息
0號SubReactor觸發：收到來自客戶端/127.0.0.1:60407的消息: 客戶端發送的第8條消息
1號SubReactor觸發：收到來自客戶端/127.0.0.1:60410的消息: 客戶端發送的第8條消息
0號SubReactor觸發：收到來自客戶端/127.0.0.1:60407的消息: 客戶端發送的第9條消息
1號SubReactor觸發：收到來自客戶端/127.0.0.1:60410的消息: 客戶端發送的第9條消息
0號SubReactor觸發：收到來自客戶端/127.0.0.1:60407的消息: 客戶端發送的第10條消息
1號SubReactor觸發：收到來自客戶端/127.0.0.1:60410的消息: 客戶端發送的第10條消息
0號SubReactor觸發：read時-------連接關閉
1號SubReactor觸發：read時-------連接關閉
客戶端：
已完成 /127.0.0.1:2333 的連接
已完成 /127.0.0.1:2333 的連接
收到來自服務端的消息: 0號SubReactor觸發：我收到來自/127.0.0.1:60407的信息辣：客戶端發送的第1條消息, 200ok;
收到來自服務端的消息: 1號SubReactor觸發：我收到來自/127.0.0.1:60410的信息辣：客戶端發送的第1條消息, 200ok;
收到來自服務端的消息: 0號SubReactor觸發：我收到來自/127.0.0.1:60407的信息辣：客戶端發送的第2條消息, 200ok;
收到來自服務端的消息: 1號SubReactor觸發：我收到來自/127.0.0.1:60410的信息辣：客戶端發送的第2條消息, 200ok;
收到來自服務端的消息: 1號SubReactor觸發：我收到來自/127.0.0.1:60410的信息辣：客戶端發送的第3條消息, 200ok;
收到來自服務端的消息: 0號SubReactor觸發：我收到來自/127.0.0.1:60407的信息辣：客戶端發送的第3條消息, 200ok;
收到來自服務端的消息: 1號SubReactor觸發：我收到來自/127.0.0.1:60410的信息辣：客戶端發送的第4條消息, 200ok;
收到來自服務端的消息: 0號SubReactor觸發：我收到來自/127.0.0.1:60407的信息辣：客戶端發送的第4條消息, 200ok;
收到來自服務端的消息: 1號SubReactor觸發：我收到來自/127.0.0.1:60410的信息辣：客戶端發送的第5條消息, 200ok;
收到來自服務端的消息: 0號SubReactor觸發：我收到來自/127.0.0.1:60407的信息辣：客戶端發送的第5條消息, 200ok;
收到來自服務端的消息: 1號SubReactor觸發：我收到來自/127.0.0.1:60410的信息辣：客戶端發送的第6條消息, 200ok;
收到來自服務端的消息: 0號SubReactor觸發：我收到來自/127.0.0.1:60407的信息辣：客戶端發送的第6條消息, 200ok;
收到來自服務端的消息: 1號SubReactor觸發：我收到來自/127.0.0.1:60410的信息辣：客戶端發送的第7條消息, 200ok;
收到來自服務端的消息: 0號SubReactor觸發：我收到來自/127.0.0.1:60407的信息辣：客戶端發送的第7條消息, 200ok;
收到來自服務端的消息: 1號SubReactor觸發：我收到來自/127.0.0.1:60410的信息辣：客戶端發送的第8條消息, 200ok;
收到來自服務端的消息: 0號SubReactor觸發：我收到來自/127.0.0.1:60407的信息辣：客戶端發送的第8條消息, 200ok;
收到來自服務端的消息: 0號SubReactor觸發：我收到來自/127.0.0.1:60407的信息辣：客戶端發送的第9條消息, 200ok;
收到來自服務端的消息: 1號SubReactor觸發：我收到來自/127.0.0.1:60410的信息辣：客戶端發送的第9條消息, 200ok;
收到來自服務端的消息: 0號SubReactor觸發：我收到來自/127.0.0.1:60407的信息辣：客戶端發送的第10條消息, 200ok;
收到來自服務端的消息: 1號SubReactor觸發：我收到來自/127.0.0.1:60410的信息辣：客戶端發送的第10條消息, 200ok;
到這里，主從Reactor模型就被改寫完成了，上面的例子只是簡單演示了下這個模型，所有的例子都是從單線程Reactor模型一點點改寫來的，客戶端沒變過，為的是更好的測試服務端在不同模型下的表現。主從Reactor模型應用的比較多，比如著名NIO框架Netty底層模型也是基於主從Reactor模型來實現的。
到這里java nio的東西已經差不多記錄完了，后續會開始netty的學習記錄，當然上述例子弱化了buffer的使用，而且例子中不存在粘包拆包的問題（因為都是請求+應答的方式進行），如果把上面的例子改成客戶端在未收到響應時就連續發送幾條信息，服務端這時再次由寫模式切換到讀模式，就會從Channel里連續拿到這幾條消息，這就導致了粘包問題，那么如何解決類似的問題呢？通常是定義一種協議，來區分消息頭和尾，中間的消息體是我們真正需要的數據，這種協議也就是我們常說的應用層協議，比如HTTP、FTP等，這里不做贅述，之后會通過一個例子來完成這部分的補充說明。
代碼地址
單線程Reactor模型：https://github.com/exceting/DemoAll/tree/master/jdk/src/main/java/demo/jdk/reactor/simple
多線程Reactor模型：同上，Acceptor里的Handler改成AsyncHandler即可
主從多線程Reactor模型：https://github.com/exceting/DemoAll/tree/master/jdk/src/main/java/demo/jdk/reactor/mainsub