前言：單線程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