Reactor模式和NIO

板橋里人 jdon.com 2002/11/08

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

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

經(jīng)典的網(wǎng)絡服務的設計如下圖，在每個線程中完成對數(shù)據(jù)的處理：

但這種模式在用戶負載增加時，性能將下降非常的快。我們需要重新尋找一個新的方案，保持數(shù)據(jù)處理的流暢，很顯然，事件觸發(fā)機制是最好的解決辦法，當有事件發(fā)生時，會觸動handler,然后開始數(shù)據(jù)的處理。

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

Reactor模式參與者

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

如圖：

Java的NIO為reactor模式提供了實現(xiàn)的基礎機制，它的Selector當發(fā)現(xiàn)某個channel有數(shù)據(jù)時，會通過SlectorKey來告知我們，在此我們實現(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注冊該channelSelectionKey sk =serverSocket.register(selector,SelectionKey.OP_ACCEPT);logger.debug("-->Start serverSocket.register!");//利用sk的attache功能綁定Acceptor 如果有事情，觸發(fā)Acceptorsk.attach(new Acceptor());logger.debug("-->attach(new Acceptor()!");}public void run() { // normally in a new Threadtry {while (!Thread.interrupted()){selector.select();Set selected = selector.selectedKeys();Iterator it = selected.iterator();//Selector如果發(fā)現(xiàn)channel有OP_ACCEPT或READ事件發(fā)生，下列遍歷就會進行。while (it.hasNext())//來一個事件 第一次觸發(fā)一個accepter線程//以后觸發(fā)SocketReadHandlerdispatch((SelectionKey)(it.next()));selected.clear();}}catch (IOException ex) {logger.debug("reactor stop!"+ex);}}//運行Acceptor或SocketReadHandlervoid dispatch(SelectionKey k) {Runnable r = (Runnable)(k.attachment());if (r != null){// r.run();}}class Acceptor implements Runnable { // innerpublic void run() {try {logger.debug("-->ready for accept!");SocketChannel c = serverSocket.accept();if (c != null)//調(diào)用Handler來處理channelnew SocketReadHandler(selector, c);}catch(IOException ex) {logger.debug("accept stop!"+ex);}}} }

以上代碼中巧妙使用了SocketChannel的attach功能，將Hanlder和可能會發(fā)生事件的channel鏈接在一起，當發(fā)生事件時，可以立即觸發(fā)相應鏈接的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ā)時，將調(diào)用本類的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);......//激活線程池 處理這些requestrequestHandle(new Request(socket,btt));.....}catch(Exception e) {}}

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

將數(shù)據(jù)讀出后，可以將這些數(shù)據(jù)處理線程做成一個線程池，這樣，數(shù)據(jù)讀出后，立即扔到線程池中，這樣加速處理速度：

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

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

總結

以上是生活随笔為你收集整理的java NIO和Reactor模式的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

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