Reactor模式和NIO

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

當(dāng)前分布式計(jì)算　Web Services盛行天下，這些網(wǎng)絡(luò)服務(wù)的底層都離不開對socket的操作。他們都有一個(gè)共同的結(jié)構(gòu)：
1. Read request
2. Decode request
3. Process service
4. Encode reply
5. Send reply

經(jīng)典的網(wǎng)絡(luò)服務(wù)的設(shè)計(jì)例如以下圖。在每一個(gè)線程中完畢對數(shù)據(jù)的處理：

但這樣的模式在用戶負(fù)載添加時(shí)，性能將下降非常的快。我們須要又一次尋找一個(gè)新的方案，保持?jǐn)?shù)據(jù)處理的流暢。非常顯然。事件觸發(fā)機(jī)制是最好的解決的方法，當(dāng)有事件發(fā)生時(shí)，會(huì)觸動(dòng)handler,然后開始數(shù)據(jù)的處理。

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

Reactor模式參與者

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

如圖：

Java的NIO為reactor模式提供了實(shí)現(xiàn)的基礎(chǔ)機(jī)制，它的Selector當(dāng)發(fā)現(xiàn)某個(gè)channel有數(shù)據(jù)時(shí)，會(huì)通過SlectorKey來告知我們。在此我們實(shí)現(xiàn)事件和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 假設(shè)有事情。觸發(fā)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假設(shè)發(fā)現(xiàn)channel有OP_ACCEPT或READ事件發(fā)生，下列遍歷就會(huì)進(jìn)行。 　　　　　　while (it.hasNext()) 　　　　　　　　//來一個(gè)事件 第一次觸發(fā)一個(gè)accepter線程 　　　　　　　　//以后觸發(fā)SocketReadHandler 　　　　　　　　dispatch((SelectionKey)(it.next())); 　　　　　　　　selected.clear(); 　　　　　　} 　　　　}catch (IOException ex) { 　　　　　　　　logger.debug("reactor stop!"+ex); 　　　　} 　　} 　　//執(zhí)行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) 　　　　　　　　//調(diào)用Handler來處理channel 　　　　　　　　new SocketReadHandler(selector, c); 　　　　　　} 　　　　catch(IOException ex) { 　　　　　　logger.debug("accept stop!"+ex); 　　　　} 　　　　} 　　} }

以上代碼中巧妙使用了SocketChannel的attach功能，將Hanlder和可能會(huì)發(fā)生事件的channel鏈接在一起。當(dāng)發(fā)生事件時(shí)，能夠馬上觸發(fā)對應(yīng)鏈接的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 下一步有事件觸發(fā)時(shí)，將調(diào)用本類的run方法。 　　　　//參看dispatch(SelectionKey k) 　　　　sk.attach(this); 　　　　//同一時(shí)候?qū)electionKey標(biāo)記為可讀，以便讀取。 　　　　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里面又運(yùn)行了一次attach，這樣，覆蓋前面的Acceptor，下次該Handler又有READ事件發(fā)生時(shí)，將直接觸發(fā)Handler.從而開始了數(shù)據(jù)的讀　處理　寫　發(fā)出等流程處理。

將數(shù)據(jù)讀出后，能夠?qū)⑦@些數(shù)據(jù)處理線程做成一個(gè)線程池，這樣，數(shù)據(jù)讀出后，馬上扔到線程池中。這樣加速處理速度：

更進(jìn)一步，我們能夠使用多個(gè)Selector分別處理連接和讀事件。

一個(gè)高性能的Java網(wǎng)絡(luò)服務(wù)機(jī)制就要形成，激動(dòng)人心的集群并行計(jì)算即將實(shí)現(xiàn)。

兩種I/O多路復(fù)用模式：Reactor和Proactor

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

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

傳遞給操作系統(tǒng)的參數(shù)須要包含用戶定義的數(shù)據(jù)緩沖區(qū)地址和數(shù)據(jù)大小。操作系統(tǒng)才干從中得到寫出操作所需數(shù)據(jù)，或?qū)懭霃膕ocket讀到的數(shù)據(jù)。事件分離器捕獲IO操作完畢事件，然后將事件傳遞給相應(yīng)處理器。比方，在windows上，處理器發(fā)起一個(gè)異步IO操作。再由事件分離器等待IOCompletion事件。典型的異步模式實(shí)現(xiàn)，都建立在操作系統(tǒng)支持異步API的基礎(chǔ)之上，我們將這樣的實(shí)現(xiàn)稱為“系統(tǒng)級”異步或“真”異步。由于應(yīng)用程序全然依賴操作系統(tǒng)運(yùn)行真正的IO工作。

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

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

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

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

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

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

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

借此，我們就能夠進(jìn)一步構(gòu)建全然可移植的，平臺無關(guān)的。有通用對外接口的解決方式。

Scalable IO in Java原文

NIO原理與應(yīng)用

用NIO開發(fā)一個(gè)高性能聊天系統(tǒng)

Socket打造高性能server

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

server后端性能大比拼

事件驅(qū)動(dòng)編程

并發(fā)模型

Rx (Reactive Extensions)介紹

Reactive編程

EDA

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

版權(quán)聲明：本文博主原創(chuàng)文章，博客，未經(jīng)同意不得轉(zhuǎn)載。

轉(zhuǎn)載于:https://www.cnblogs.com/bhlsheji/p/4849093.html

總結(jié)

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

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯(cuò)，歡迎將生活随笔推薦給好友。