Reactor模式和NIO

本文可以看作是Doug Lea Scalable IO in Java一文的翻譯。

當前分布式計算　Web Services盛行天下，這些網絡服務的底層都離不開對socket的操作。他們都有一個共同的結構：
1. Read request
2. Decode request
3. Process service
4. Encode reply
5. Send reply

經典的網絡服務的設計例如以下圖。在每一個線程中完畢對數據的處理：

但這樣的模式在用戶負載添加時，性能將下降非常的快。我們須要又一次尋找一個新的方案，保持數據處理的流暢。非常顯然。事件觸發機制是最好的解決的方法，當有事件發生時，會觸動handler,然后開始數據的處理。

Reactor模式類似于AWT中的Event處理：

Reactor模式參與者

1.Reactor 負責響應IO事件。一旦發生。廣播發送給對應的Handler去處理,這類似于AWT的thread
2.Handler 是負責非阻塞行為。類似于AWT ActionListeners；同一時候負責將handlers與event事件綁定，類似于AWT addActionListener

如圖：

Java的NIO為reactor模式提供了實現的基礎機制，它的Selector當發現某個channel有數據時，會通過SlectorKey來告知我們。在此我們實現事件和handler的綁定。

我們來看看Reactor模式代碼:

public class Reactor implements Runnable{ 　　final Selector selector; 　　final ServerSocketChannel serverSocket; 　　Reactor(int port) throws IOException { 　　　　selector = Selector.open(); 　　　　serverSocket = ServerSocketChannel.open(); 　　　　InetSocketAddress address = new InetSocketAddress(InetAddress.getLocalHost(),port); 　　　　serverSocket.socket().bind(address); 　　　　serverSocket.configureBlocking(false); 　　　　//向selector注冊該channel 　　　　 SelectionKey sk =serverSocket.register(selector,SelectionKey.OP_ACCEPT); 　　　　logger.debug("-->Start serverSocket.register!"); 　　　　//利用sk的attache功能綁定Acceptor 假設有事情。觸發Acceptor 　　　　sk.attach(new Acceptor()); 　　　　logger.debug("-->attach(new Acceptor()!"); 　　} 　　public void run() { // normally in a new Thread 　　　　try { 　　　　while (!Thread.interrupted()) 　　　　{ 　　　　　　selector.select(); 　　　　　　Set selected = selector.selectedKeys(); 　　　　　　Iterator it = selected.iterator(); 　　　　　　//Selector假設發現channel有OP_ACCEPT或READ事件發生，下列遍歷就會進行。 　　　　　　while (it.hasNext()) 　　　　　　　　//來一個事件 第一次觸發一個accepter線程 　　　　　　　　//以后觸發SocketReadHandler 　　　　　　　　dispatch((SelectionKey)(it.next())); 　　　　　　　　selected.clear(); 　　　　　　} 　　　　}catch (IOException ex) { 　　　　　　　　logger.debug("reactor stop!"+ex); 　　　　} 　　} 　　//執行Acceptor或SocketReadHandler 　　void dispatch(SelectionKey k) { 　　　　Runnable r = (Runnable)(k.attachment()); 　　　　if (r != null){ 　　　　　　// r.run(); 　　　　} 　　} 　　class Acceptor implements Runnable { // inner 　　　　public void run() { 　　　　try { 　　　　　　logger.debug("-->ready for accept!"); 　　　　　　SocketChannel c = serverSocket.accept(); 　　　　　　if (c != null) 　　　　　　　　//調用Handler來處理channel 　　　　　　　　new SocketReadHandler(selector, c); 　　　　　　} 　　　　catch(IOException ex) { 　　　　　　logger.debug("accept stop!"+ex); 　　　　} 　　　　} 　　} }

以上代碼中巧妙使用了SocketChannel的attach功能，將Hanlder和可能會發生事件的channel鏈接在一起。當發生事件時，能夠馬上觸發對應鏈接的Handler。

再看看Handler代碼:

public class SocketReadHandler implements Runnable { 　　public static Logger logger = Logger.getLogger(SocketReadHandler.class); 　　private Test test=new Test(); 　　final SocketChannel socket; 　　final SelectionKey sk; 　　 static final int READING = 0, SENDING = 1; 　　int state = READING; 　　public SocketReadHandler(Selector sel, SocketChannel c) 　　　　throws IOException { 　　　　socket = c; 　　　　socket.configureBlocking(false); 　　　　 sk = socket.register(sel, 0); 　　　　//將SelectionKey綁定為本Handler 下一步有事件觸發時，將調用本類的run方法。 　　　　//參看dispatch(SelectionKey k) 　　　　sk.attach(this); 　　　　//同一時候將SelectionKey標記為可讀，以便讀取。 　　　　sk.interestOps(SelectionKey.OP_READ); 　　　　sel.wakeup(); 　　} 　　public void run() { 　　　　try{ 　　　　// test.read(socket,input); 　　　　　　readRequest() ; 　　　　}catch(Exception ex){ 　　　　logger.debug("readRequest error"+ex); 　　　　} 　　} /** * 處理讀取data * @param key * @throws Exception */ private void readRequest() throws Exception { 　　ByteBuffer input = ByteBuffer.allocate(1024); 　　input.clear(); 　　try{ 　　　　int bytesRead = socket.read(input); 　　　　...... 　　　　//激活線程池 處理這些request 　　　　requestHandle(new Request(socket,btt)); 　　　　..... 　　}catch(Exception e) { 　　} }

注意在Handler里面又運行了一次attach，這樣，覆蓋前面的Acceptor，下次該Handler又有READ事件發生時，將直接觸發Handler.從而開始了數據的讀　處理　寫　發出等流程處理。

將數據讀出后，能夠將這些數據處理線程做成一個線程池，這樣，數據讀出后，馬上扔到線程池中。這樣加速處理速度：

更進一步，我們能夠使用多個Selector分別處理連接和讀事件。

一個高性能的Java網絡服務機制就要形成，激動人心的集群并行計算即將實現。

兩種I/O多路復用模式：Reactor和Proactor

?一般地,I/O多路復用機制都依賴于一個事件多路分離器(Event Demultiplexer)。分離器對象可將來自事件源的I/O事件分離出來，并分發到相應的read/write事件處理器(Event Handler)。開發者預先注冊須要處理的事件及其事件處理器（或回調函數）；事件分離器負責將請求事件傳遞給事件處理器。兩個與事件分離器有關的模式是Reactor和Proactor。

Reactor模式採用同步IO。而Proactor採用異步IO。
?在Reactor中，事件分離器負責等待文件描寫敘述符或socket為讀寫操作準備就緒，然后將就緒事件傳遞給相應的處理器，最后由處理器負責完畢實際的讀寫工作。
?而在Proactor模式中，處理器--或者兼任處理器的事件分離器，僅僅負責發起異步讀寫操作。IO操作本身由操作系統來完畢。

傳遞給操作系統的參數須要包含用戶定義的數據緩沖區地址和數據大小。操作系統才干從中得到寫出操作所需數據，或寫入從socket讀到的數據。事件分離器捕獲IO操作完畢事件，然后將事件傳遞給相應處理器。比方，在windows上，處理器發起一個異步IO操作。再由事件分離器等待IOCompletion事件。典型的異步模式實現，都建立在操作系統支持異步API的基礎之上，我們將這樣的實現稱為“系統級”異步或“真”異步。由于應用程序全然依賴操作系統運行真正的IO工作。

?舉個樣例。將有助于理解Reactor與Proactor二者的差異，以讀操作為例（類操作類似）。

?在Reactor中實現讀：
?- 注冊讀就緒事件和對應的事件處理器
?- 事件分離器等待事件
?- 事件到來，激活分離器，分離器調用事件相應的處理器。
?- 事件處理器完畢實際的讀操作，處理讀到的數據，注冊新的事件，然后返還控制權。
?與例如以下Proactor（真異步）中的讀過程比較：
?- 處理器發起異步讀操作（注意：操作系統必須支持異步IO）。在這樣的情況下，處理器無視IO就緒事件，它關注的是完畢事件。

?- 事件分離器等待操作完畢事件
?- 在分離器等待過程中。操作系統利用并行的內核線程運行實際的讀操作。并將結果數據存入用戶自己定義緩沖區，最后通知事件分離器讀操作完畢。
?- 事件分離器呼喚處理器。
?- 事件處理器處理用戶自己定義緩沖區中的數據，然后啟動一個新的異步操作，并將控制權返回事件分離器。
?
實踐現狀?
?由Douglas Schmidt等人開發的開源C++開發框架ACE，提供了大量與平臺無關，支持并發的底層類（線程，相互排斥量等），且在高抽象層次上，提供了兩組不同的類--ACE Reactor和ACE Proactor的實現。

只是。盡管二者都與平臺無關，提供的接口卻各異。

?ACE Proactor在windows平臺上具有更為優異的性能表現。由于windows在操作系統提供了高效的異步API支持（見http://msdn2.microsoft.com/en-us/library/aa365198.aspx）。
?然而，并不是全部的操作系統都在系統級大力支持異步。像非常多Unix系統就沒做到。因此，在Unix上，選擇ACE Reactor解決方式可能更好。但這樣一來，為了獲得最好的性能，網絡應用的開發者必須為不同的操作系統維護多份代碼：windows上以ACE Proactor為基礎。而Unix系統上則採用ACE Reactor解決方式。
?
改進方案
? 在這部分，我們將嘗試應對為Proactor和Reactor模式建立可移植框架的挑戰。在改進方案中。我們將Reactor原來位于事件處理器內的read/write操作移至分離器(最好還是將這個思路稱為“模擬異步”)，以此尋求將Reactor多路同步IO轉化為模擬異步IO。以讀操作為樣例，改進步驟例如以下：
? - 注冊讀就緒事件及其處理器，并為分離器提供數據緩沖區地址，須要讀取數據量等信息。

? - 分離器等待事件（如在select()上等待）
? - 事件到來。激活分離器。分離器運行一個非堵塞讀操作（它有完畢這個操作所需的所有信息），最后調用相應處理器。
? - 事件處理器處理用戶自己定義緩沖區的數據，注冊新的事件（當然相同要給出數據緩沖區地址，須要讀取的數據量等信息），最后將控制權返還分離器。

? 如我們所見，通過對多路IO模式功能結構的改造，可將Reactor轉化為Proactor模式。改造前后，模型實際完畢的工作量沒有添加，僅僅只是參與者間對工作職責稍加調換。沒有工作量的改變，自然不會造成性能的削弱。對例如以下各步驟的比較，能夠證明工作量的恒定：
? 標準/典型的Reactor：
? - 步驟1：等待事件到來（Reactor負責）
? - 步驟2：將讀就緒事件分發給用戶定義的處理器（Reactor負責）
? - 步驟3：讀數據（用戶處理器負責）
? - 步驟4：處理數據（用戶處理器負責）
? 改進實現的模擬Proactor：
? - 步驟1：等待事件到來（Proactor負責）
? - 步驟2：得到讀就緒事件，運行讀數據（如今由Proactor負責）
? - 步驟3：將讀完畢事件分發給用戶處理器（Proactor負責）
? - 步驟4：處理數據（用戶處理器負責）??
??
? 對于不提供異步IO API的操作系統來說。這樣的辦法能夠隱藏socket API的交互細節，從而對外暴露一個完整的異步接口。

借此，我們就能夠進一步構建全然可移植的，平臺無關的。有通用對外接口的解決方式。

Scalable IO in Java原文

NIO原理與應用

用NIO開發一個高性能聊天系統

Socket打造高性能server

很多其它NIO專題系列討論....

server后端性能大比拼

事件驅動編程

并發模型

Rx (Reactive Extensions)介紹

Reactive編程

EDA

2002年大神住址：http://www.jdon.com/concurrent/reactor.htm

版權聲明：本文博主原創文章，博客，未經同意不得轉載。

轉載于:https://www.cnblogs.com/bhlsheji/p/4849093.html

總結

以上是生活随笔為你收集整理的Nio得知3——该示范基地：多路复用器模式的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。